Change wallet to submitter

cmd/gravitybeam/collector.go

@@ -25,12 +25,17 @@ type Collector struct {
 	logger            *supportlog.Entry
 	horizonClient     horizonclient.ClientInterface
 	pubSub            *pubsub.PubSub
-	topic             *pubsub.Topic
+	listenTopic       *pubsub.Topic
+	publishTopic      *pubsub.Topic
 	store             *Store
 }
 
 func NewCollector(config CollectorConfig) (*Collector, error) {
-	topic, err := config.PubSub.Join("starbridge-stellar-transactions-signed")
+	listenTopic, err := config.PubSub.Join("starbridge-stellar-transactions-signed")
+	if err != nil {
+		return nil, err
+	}
+	publishTopic, err := config.PubSub.Join("starbridge-stellar-transactions-signed-aggregated")
 	if err != nil {
 		return nil, err
 	}
@@ -40,17 +45,18 @@ func NewCollector(config CollectorConfig) (*Collector, error) {
 		horizonClient:     config.HorizonClient,
 		store:             config.Store,
 		pubSub:            config.PubSub,
-		topic:             topic,
+		listenTopic:       listenTopic,
+		publishTopic:      publishTopic,
 	}
 	return c, nil
 }
 
 func (c *Collector) Collect() error {
-	sub, err := c.topic.Subscribe()
+	sub, err := c.listenTopic.Subscribe()
 	if err != nil {
 		return err
 	}
-	c.logger.Infof("Subscribed to topic %s", c.topic.String())
+	c.logger.Infof("Subscribed to topic %s", c.listenTopic.String())
 	ctx := context.Background()
 	for {
 		logger := c.logger
@@ -90,13 +96,10 @@ func (c *Collector) Collect() error {
 			return fmt.Errorf("marshaling tx %s: %w", hash, err)
 		}
 
-		for _, a := range Accounts(tx) {
-			logger.Infof("publishing to topic for %s", a)
-			//nolint:staticcheck // SA1019 ignore this!
-			err = c.pubSub.Publish("starbridge-stellar-transactions-signed-"+a, txBytes)
-			if err != nil {
-				return fmt.Errorf("publishing tx %s for account %s: %w", hash, a, err)
-			}
+		logger.Infof("publishing aggregated")
+		err = c.publishTopic.Publish(ctx, txBytes)
+		if err != nil {
+			return fmt.Errorf("publishing tx %s: %w", hash, err)
 		}
 	}
 }

cmd/gravitybeam/main.go

@@ -9,6 +9,7 @@ import (
 
 	libp2p "github.com/libp2p/go-libp2p"
 	"github.com/libp2p/go-libp2p-core/host"
+	libp2pnetwork "github.com/libp2p/go-libp2p-core/network"
 	"github.com/libp2p/go-libp2p-core/peer"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"github.com/libp2p/go-libp2p/p2p/discovery/mdns"
@@ -57,6 +58,11 @@ func run(args []string, logger *supportlog.Entry) error {
 	if err != nil {
 		return err
 	}
+	host.Network().Notify(&libp2pnetwork.NotifyBundle{
+		ConnectedF: func(n libp2pnetwork.Network, c libp2pnetwork.Conn) {
+			logger.Infof("Connected to: %s", c.RemotePeer().Pretty())
+		},
+	})
 	hostAddrInfo := peer.AddrInfo{
 		ID:    host.ID(),
 		Addrs: host.Addrs(),

cmd/starbridge/main.go

@@ -10,6 +10,7 @@ import (
 
 	libp2p "github.com/libp2p/go-libp2p"
 	"github.com/libp2p/go-libp2p-core/host"
+	libp2pnetwork "github.com/libp2p/go-libp2p-core/network"
 	"github.com/libp2p/go-libp2p-core/peer"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"github.com/libp2p/go-libp2p/p2p/discovery/mdns"
@@ -77,6 +78,11 @@ func run(args []string, logger *supportlog.Entry) error {
 	if err != nil {
 		return err
 	}
+	host.Network().Notify(&libp2pnetwork.NotifyBundle{
+		ConnectedF: func(n libp2pnetwork.Network, c libp2pnetwork.Conn) {
+			logger.Infof("Connected to: %s", c.RemotePeer().Pretty())
+		},
+	})
 	hostAddrInfo := peer.AddrInfo{
 		ID:    host.ID(),
 		Addrs: host.Addrs(),

cmd/stellarwallet/collector.go → cmd/submitter/collector.go

@@ -9,34 +9,30 @@ import (
 
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"github.com/stellar/go/clients/horizonclient"
-	"github.com/stellar/go/keypair"
 	supportlog "github.com/stellar/go/support/log"
 	"github.com/stellar/go/txnbuild"
 )
 
 type CollectorConfig struct {
-	Address           *keypair.FromAddress
 	NetworkPassphrase string
 	Logger            *supportlog.Entry
 	HorizonClient     horizonclient.ClientInterface
 	PubSub            *pubsub.PubSub
 }
 
 type Collector struct {
-	address           *keypair.FromAddress
 	networkPassphrase string
 	logger            *supportlog.Entry
 	horizonClient     horizonclient.ClientInterface
 	topic             *pubsub.Topic
 }
 
 func NewCollector(config CollectorConfig) (*Collector, error) {
-	topic, err := config.PubSub.Join("starbridge-stellar-transactions-signed-" + config.Address.Address())
+	topic, err := config.PubSub.Join("starbridge-stellar-transactions-signed-aggregated")
 	if err != nil {
 		return nil, err
 	}
 	c := &Collector{
-		address:           config.Address,
 		networkPassphrase: config.NetworkPassphrase,
 		logger:            config.Logger,
 		horizonClient:     config.HorizonClient,
@@ -78,12 +74,6 @@ func (c *Collector) Collect() error {
 		}
 		logger = logger.WithField("tx", hex.EncodeToString(hash[:]))
 
-		if !IsRecipient(tx, c.address) {
-			logger.Infof("tx does not have address as recipient, ignoring")
-			continue
-		}
-		logger.Infof("tx has address as recipient")
-
 		tx, err = AuthorizedTransaction(c.horizonClient, hash, tx)
 		if errors.Is(err, ErrNotAuthorized) {
 			logger.Infof("tx not yet authorized")

cmd/stellarwallet/main.go → cmd/submitter/main.go

@@ -9,13 +9,13 @@ import (
 
 	libp2p "github.com/libp2p/go-libp2p"
 	"github.com/libp2p/go-libp2p-core/host"
+	libp2pnetwork "github.com/libp2p/go-libp2p-core/network"
 	"github.com/libp2p/go-libp2p-core/peer"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"github.com/libp2p/go-libp2p/p2p/discovery/mdns"
 	ff "github.com/peterbourgon/ff/v3"
 	"github.com/sirupsen/logrus"
 	"github.com/stellar/go/clients/horizonclient"
-	"github.com/stellar/go/keypair"
 	supportlog "github.com/stellar/go/support/log"
 )
 
@@ -30,14 +30,12 @@ func main() {
 }
 
 func run(args []string, logger *supportlog.Entry) error {
-	fs := flag.NewFlagSet("stellarwallet", flag.ExitOnError)
+	fs := flag.NewFlagSet("submitter", flag.ExitOnError)
 
-	addressStr := ""
 	portP2P := "0"
 	peers := ""
 	horizonURL := "https://horizon-testnet.stellar.org"
 
-	fs.StringVar(&addressStr, "address", addressStr, "Stellar address this wallet is monitoring (also via ADDRESS)")
 	fs.StringVar(&portP2P, "port-p2p", portP2P, "Port to accept P2P requests on (also via PORT_P2P)")
 	fs.StringVar(&peers, "peers", peers, "Comma-separated list of addresses of peers to connect to on start (also via PEERS)")
 	fs.StringVar(&horizonURL, "horizon", horizonURL, "Horizon URL (also via HORIZON_URL)")
@@ -49,11 +47,6 @@ func run(args []string, logger *supportlog.Entry) error {
 
 	logger.Info("Starting...")
 
-	address, err := keypair.ParseAddress(addressStr)
-	if err != nil {
-		return err
-	}
-
 	horizonClient := &horizonclient.Client{HorizonURL: horizonURL}
 
 	networkDetails, err := horizonClient.Root()
@@ -65,6 +58,11 @@ func run(args []string, logger *supportlog.Entry) error {
 	if err != nil {
 		return err
 	}
+	host.Network().Notify(&libp2pnetwork.NotifyBundle{
+		ConnectedF: func(n libp2pnetwork.Network, c libp2pnetwork.Conn) {
+			logger.Infof("Connected to: %s", c.RemotePeer().Pretty())
+		},
+	})
 	hostAddrInfo := peer.AddrInfo{
 		ID:    host.ID(),
 		Addrs: host.Addrs(),
@@ -113,7 +111,6 @@ func run(args []string, logger *supportlog.Entry) error {
 
 	logger.Info("Subscribing to transactions...")
 	collector, err := NewCollector(CollectorConfig{
-		Address:           address,
 		NetworkPassphrase: networkDetails.NetworkPassphrase,
 		Logger:            logger,
 		HorizonClient:     horizonClient,

go.mod

@@ -117,7 +117,7 @@ require (
 	go.uber.org/atomic v1.9.0 // indirect
 	go.uber.org/multierr v1.7.0 // indirect
 	go.uber.org/zap v1.19.0 // indirect
-	golang.org/x/crypto v0.0.0-20211215153901-e495a2d5b3d3
+	golang.org/x/crypto v0.0.0-20211215153901-e495a2d5b3d3 // indirect
 	golang.org/x/mod v0.4.2 // indirect
 	golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2 // indirect
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect

go.sum

@@ -1094,8 +1094,6 @@ github.com/spf13/viper v1.3.2/go.mod h1:ZiWeW+zYFKm7srdB9IoDzzZXaJaI5eL9QjNiN/DM
 github.com/src-d/envconfig v1.0.0/go.mod h1:Q9YQZ7BKITldTBnoxsE5gOeB5y66RyPXeue/R4aaNBc=
 github.com/status-im/keycard-go v0.0.0-20190316090335-8537d3370df4 h1:Gb2Tyox57NRNuZ2d3rmvB3pcmbu7O1RS3m8WRx7ilrg=
 github.com/status-im/keycard-go v0.0.0-20190316090335-8537d3370df4/go.mod h1:RZLeN1LMWmRsyYjvAu+I6Dm9QmlDaIIt+Y+4Kd7Tp+Q=
-github.com/stellar/go v0.0.0-20220121002512-4920b27104df h1:A5V6kIyKZsQ7B44NevNoZBw1mVsD/gn8SOlY6uNyNF8=
-github.com/stellar/go v0.0.0-20220121002512-4920b27104df/go.mod h1:ZjEIAKldwfIdxI1N/vNX1TqUGwwcLWAa5XO4JdiAUIc=
 github.com/stellar/go v0.0.0-20220121195101-cd93d3e009f0 h1:fQv9g3Ybc9M4rM0ufRoVp1A/06wsZSvdvKj71UP6KKA=
 github.com/stellar/go v0.0.0-20220121195101-cd93d3e009f0/go.mod h1:ZjEIAKldwfIdxI1N/vNX1TqUGwwcLWAa5XO4JdiAUIc=
 github.com/stellar/go-xdr v0.0.0-20211103144802-8017fc4bdfee h1:fbVs0xmXpBvVS4GBeiRmAE3Le70ofAqFMch1GTiq/e8=