Move Options handling to the config package

cmd/masa-node/main.go

@@ -16,7 +16,6 @@ import (
 	"github.com/masa-finance/masa-oracle/pkg/api"
 	"github.com/masa-finance/masa-oracle/pkg/config"
 	"github.com/masa-finance/masa-oracle/pkg/db"
-	"github.com/masa-finance/masa-oracle/pkg/masacrypto"
 	"github.com/masa-finance/masa-oracle/pkg/staking"
 )
 
@@ -34,19 +33,15 @@ func main() {
 	if err != nil {
 		logrus.Fatalf("[-] %v", err)
 	}
-	cfg.LogConfig()
-	cfg.SetupLogging()
 
-	keyManager, err := masacrypto.NewKeyManager(cfg.PrivateKey, cfg.PrivateKeyFile)
-	if err != nil {
-		logrus.Fatal("[-] Failed to initialize keys:", err)
-	}
+	cfg.SetupLogging()
+	cfg.LogConfig()
 
 	// Create a cancellable context
 	ctx, cancel := context.WithCancel(context.Background())
 
 	if cfg.Faucet {
-		err := handleFaucet(cfg.RpcUrl, keyManager.EcdsaPrivKey)
+		err := handleFaucet(cfg.RpcUrl, cfg.KeyManager.EcdsaPrivKey)
 		if err != nil {
 			logrus.Errorf("[-] %v", err)
 			os.Exit(1)
@@ -57,7 +52,7 @@ func main() {
 	}
 
 	if cfg.StakeAmount != "" {
-		err := handleStaking(cfg.RpcUrl, keyManager.EcdsaPrivKey, cfg.StakeAmount)
+		err := handleStaking(cfg.RpcUrl, cfg.KeyManager.EcdsaPrivKey, cfg.StakeAmount)
 		if err != nil {
 			logrus.Warningf("%v", err)
 		} else {
@@ -67,7 +62,7 @@ func main() {
 	}
 
 	// Verify the staking event
-	isStaked, err := staking.VerifyStakingEvent(cfg.RpcUrl, keyManager.EthAddress)
+	isStaked, err := staking.VerifyStakingEvent(cfg.RpcUrl, cfg.KeyManager.EthAddress)
 	if err != nil {
 		logrus.Error(err)
 	}
@@ -76,7 +71,7 @@ func main() {
 		logrus.Warn("No staking event found for this address")
 	}
 
-	masaNodeOptions, workHandlerManager, pubKeySub := initOptions(cfg, keyManager)
+	masaNodeOptions, workHandlerManager, pubKeySub := config.InitOptions(cfg)
 	// Create a new OracleNode
 	masaNode, err := node.NewOracleNode(ctx, masaNodeOptions...)
 
@@ -94,14 +89,14 @@ func main() {
 
 	if cfg.AllowedPeer {
 		cfg.AllowedPeerId = masaNode.Host.ID().String()
-		cfg.AllowedPeerPublicKey = keyManager.HexPubKey
+		cfg.AllowedPeerPublicKey = cfg.KeyManager.HexPubKey
 		logrus.Infof("[+] Allowed peer with ID: %s and PubKey: %s", cfg.AllowedPeerId, cfg.AllowedPeerPublicKey)
 	} else {
 		logrus.Warn("[-] This node is not set as the allowed peer")
 	}
 
 	// Init cache resolver
-	db.InitResolverCache(masaNode, keyManager, cfg.AllowedPeerId, cfg.AllowedPeerPublicKey, cfg.Validator)
+	db.InitResolverCache(masaNode, cfg.KeyManager, cfg.AllowedPeerId, cfg.AllowedPeerPublicKey, cfg.Validator)
 
 	// Cancel the context when SIGINT is received
 	go handleSignals(cancel, masaNode, cfg)
@@ -125,7 +120,7 @@ func main() {
 		logrus.Errorf("[-] Error while getting node IP address from %v: %v", multiAddr, err)
 	}
 	// Display the welcome message with the multiaddress and IP address
-	config.DisplayWelcomeMessage(multiAddr.String(), ipAddr, keyManager.EthAddress, isStaked, cfg.Validator, cfg.TwitterScraper, cfg.TelegramScraper, cfg.DiscordScraper, cfg.WebScraper, versioning.ApplicationVersion, versioning.ProtocolVersion)
+	config.DisplayWelcomeMessage(multiAddr.String(), ipAddr, cfg.KeyManager.EthAddress, isStaked, cfg.Validator, cfg.TwitterScraper, cfg.TelegramScraper, cfg.DiscordScraper, cfg.WebScraper, versioning.ApplicationVersion, versioning.ProtocolVersion)
 
 	<-ctx.Done()
 }

node/options.go

@@ -32,7 +32,15 @@ type NodeOption struct {
 	Version              string
 	MasaDir              string
 	CachePath            string
-	KeyManager           *masacrypto.KeyManager
+
+	OracleProtocol       string
+	NodeDataSyncProtocol string
+	NodeGossipTopic      string
+	Rendezvous           string
+	WorkerProtocol       string
+	PageSize             int
+
+	KeyManager *masacrypto.KeyManager
 }
 
 type PubSubHandlers struct {
@@ -167,3 +175,39 @@ func WithKeyManager(km *masacrypto.KeyManager) Option {
 		o.KeyManager = km
 	}
 }
+
+func WithOracleProtocol(s string) Option {
+	return func(o *NodeOption) {
+		o.OracleProtocol = s
+	}
+}
+
+func WithNodeDataSyncProtocol(s string) Option {
+	return func(o *NodeOption) {
+		o.NodeDataSyncProtocol = s
+	}
+}
+
+func WithNodeGossipTopic(s string) Option {
+	return func(o *NodeOption) {
+		o.NodeGossipTopic = s
+	}
+}
+
+func WithRendezvous(s string) Option {
+	return func(o *NodeOption) {
+		o.Rendezvous = s
+	}
+}
+
+func WithWorkerProtocol(s string) Option {
+	return func(o *NodeOption) {
+		o.WorkerProtocol = s
+	}
+}
+
+func WithPageSize(size int) Option {
+	return func(o *NodeOption) {
+		o.PageSize = size
+	}
+}

node/oracle_node.go

@@ -26,7 +26,6 @@ import (
 
 	"github.com/masa-finance/masa-oracle/internal/versioning"
 	"github.com/masa-finance/masa-oracle/pkg/chain"
-	"github.com/masa-finance/masa-oracle/pkg/config"
 	myNetwork "github.com/masa-finance/masa-oracle/pkg/network"
 	"github.com/masa-finance/masa-oracle/pkg/pubsub"
 )
@@ -47,7 +46,6 @@ type OracleNode struct {
 	Blockchain    *chain.Chain
 	Options       NodeOption
 	Context       context.Context
-	Config        *config.AppConfig
 }
 
 // GetMultiAddrs returns the priority multiaddr for this node.
@@ -150,7 +148,7 @@ func NewOracleNode(ctx context.Context, opts ...Option) (*OracleNode, error) {
 		Options:       *o,
 	}
 
-	n.Protocol = n.protocolWithVersion(config.OracleProtocol)
+	n.Protocol = n.protocolWithVersion(n.Options.OracleProtocol)
 	return n, nil
 }
 
@@ -204,7 +202,7 @@ func (node *OracleNode) Start() (err error) {
 	logrus.Infof("[+] Starting node with ID: %s", node.GetMultiAddrs().String())
 
 	node.Host.SetStreamHandler(node.Protocol, node.handleStream)
-	node.Host.SetStreamHandler(node.protocolWithVersion(config.NodeDataSyncProtocol), node.ReceiveNodeData)
+	node.Host.SetStreamHandler(node.protocolWithVersion(node.Options.NodeDataSyncProtocol), node.ReceiveNodeData)
 
 	for pid, n := range node.Options.ProtocolHandlers {
 		node.Host.SetStreamHandler(pid, n)
@@ -215,7 +213,7 @@ func (node *OracleNode) Start() (err error) {
 	}
 
 	if node.Options.IsStaked {
-		node.Host.SetStreamHandler(node.protocolWithVersion(config.NodeGossipTopic), node.GossipNodeData)
+		node.Host.SetStreamHandler(node.protocolWithVersion(node.Options.NodeGossipTopic), node.GossipNodeData)
 	}
 
 	node.Host.Network().Notify(node.NodeTracker)
@@ -236,7 +234,7 @@ func (node *OracleNode) Start() (err error) {
 		return err
 	}
 
-	err = myNetwork.EnableMDNS(node.Host, config.Rendezvous, node.PeerChan)
+	err = myNetwork.EnableMDNS(node.Host, node.Options.Rendezvous, node.PeerChan)
 	if err != nil {
 		return err
 	}

node/oracle_node_listener.go

@@ -14,7 +14,6 @@ import (
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/sirupsen/logrus"
 
-	"github.com/masa-finance/masa-oracle/pkg/config"
 	pubsub2 "github.com/masa-finance/masa-oracle/pkg/pubsub"
 )
 
@@ -42,7 +41,7 @@ func (node *OracleNode) ListenToNodeTracker() {
 				continue
 			}
 			// Publish the JSON data on the node.topic
-			err = node.PublishTopic(config.NodeGossipTopic, jsonData)
+			err = node.PublishTopic(node.Options.NodeGossipTopic, jsonData)
 			if err != nil {
 				logrus.Errorf("[-] Error publishing node data: %v", err)
 			}
@@ -88,10 +87,10 @@ type NodeDataPage struct {
 func (node *OracleNode) SendNodeDataPage(allNodeData []pubsub2.NodeData, stream network.Stream, pageNumber int) {
 	logrus.Debugf("[+] SendNodeDataPage --> %s: Page: %d", stream.Conn().RemotePeer(), pageNumber)
 	totalRecords := len(allNodeData)
-	totalPages := int(math.Ceil(float64(totalRecords) / config.PageSize))
+	totalPages := int(math.Ceil(float64(totalRecords) / float64(node.Options.PageSize)))
 
-	startIndex := pageNumber * config.PageSize
-	endIndex := startIndex + config.PageSize
+	startIndex := pageNumber * node.Options.PageSize
+	endIndex := startIndex + node.Options.PageSize
 	if endIndex > totalRecords {
 		endIndex = totalRecords
 	}
@@ -133,9 +132,9 @@ func (node *OracleNode) SendNodeData(peerID peer.ID) {
 		nodeData = node.NodeTracker.GetUpdatedNodes(sinceTime)
 	}
 	totalRecords := len(nodeData)
-	totalPages := int(math.Ceil(float64(totalRecords) / float64(config.PageSize)))
+	totalPages := int(math.Ceil(float64(totalRecords) / float64(node.Options.PageSize)))
 
-	stream, err := node.Host.NewStream(node.Context, peerID, node.protocolWithVersion(config.NodeDataSyncProtocol))
+	stream, err := node.Host.NewStream(node.Context, peerID, node.protocolWithVersion(node.Options.NodeDataSyncProtocol))
 	if err != nil {
 		// node.NodeTracker.RemoveNodeData(peerID.String())
 		return

node/protocol.go

@@ -9,7 +9,6 @@ import (
 	"github.com/libp2p/go-libp2p/core/network"
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/libp2p/go-libp2p/core/protocol"
-	"github.com/masa-finance/masa-oracle/pkg/config"
 )
 
 const (
@@ -50,7 +49,7 @@ func (node *OracleNode) subscribeToTopics() error {
 	}
 
 	// Subscribe to NodeGossipTopic to participate in the network's gossip protocol.
-	if err := node.SubscribeTopic(config.NodeGossipTopic, node.NodeTracker, false); err != nil {
+	if err := node.SubscribeTopic(node.Options.NodeGossipTopic, node.NodeTracker, false); err != nil {
 		return err
 	}
 

pkg/config/app.go

@@ -9,6 +9,7 @@ import (
 	"strings"
 
 	"github.com/masa-finance/masa-oracle/internal/versioning"
+	"github.com/masa-finance/masa-oracle/pkg/masacrypto"
 
 	"github.com/gotd/contrib/bg"
 	"github.com/joho/godotenv"
@@ -58,6 +59,7 @@ type AppConfig struct {
 	WebScraper         bool   `mapstructure:"webScraper"`
 	APIEnabled         bool   `mapstructure:"api_enabled"`
 
+	KeyManager   *masacrypto.KeyManager
 	TelegramStop bg.StopFunc
 }
 
@@ -72,22 +74,27 @@ type AppConfig struct {
 // In case of any errors it returns nill with an error object
 func GetConfig() (*AppConfig, error) {
 	instance := &AppConfig{}
-
 	instance.setDefaultConfig()
 	// TODO Shouldn't the env vars override the file config, instead of the other way around?
 	instance.setEnvVariableConfig()
 	instance.setFileConfig(viper.GetString("FILE_PATH"))
 
 	if err := instance.setCommandLineConfig(); err != nil {
-		return nil, fmt.Errorf("Error while setting command line options: %v", err)
+		return nil, err
 	}
 
 	if err := viper.Unmarshal(instance); err != nil {
-		return nil, fmt.Errorf("Unable to unmarshal config into struct: %v", err)
+		return nil, fmt.Errorf("Unable to unmarshal config into struct, %v", err)
 	}
 
 	instance.APIEnabled = viper.GetBool("api_enabled")
 
+	keyManager, err := masacrypto.NewKeyManager(instance.PrivateKey, instance.PrivateKeyFile)
+	if err != nil {
+		return nil, fmt.Errorf("Failed to initialize keys: %v", err)
+	}
+	instance.KeyManager = keyManager
+
 	return instance, nil
 }
 

cmd/masa-node/config.go → pkg/config/options.go

@@ -1,16 +1,27 @@
-package main
+package config
 
 import (
 	"github.com/masa-finance/masa-oracle/node"
-	"github.com/masa-finance/masa-oracle/pkg/config"
-	"github.com/masa-finance/masa-oracle/pkg/masacrypto"
-	pubsub "github.com/masa-finance/masa-oracle/pkg/pubsub"
+	"github.com/masa-finance/masa-oracle/pkg/pubsub"
 	"github.com/masa-finance/masa-oracle/pkg/workers"
 )
 
-func initOptions(cfg *config.AppConfig, keyManager *masacrypto.KeyManager) ([]node.Option, *workers.WorkHandlerManager, *pubsub.PublicKeySubscriptionHandler) {
+var constantOptions = []node.Option{
+	node.WithOracleProtocol(OracleProtocol),
+	node.WithNodeDataSyncProtocol(NodeDataSyncProtocol),
+	node.WithNodeGossipTopic(NodeGossipTopic),
+	node.WithRendezvous(Rendezvous),
+	node.WithPageSize(PageSize),
+}
+
+// WithConstantOptions adds options that are set to constant values. We need to add them to
+// the node to avoid a dependency loop.
+func WithConstantOptions(nodes ...node.Option) []node.Option {
+	return append(nodes, constantOptions...)
+}
+
+func InitOptions(cfg *AppConfig) ([]node.Option, *workers.WorkHandlerManager, *pubsub.PublicKeySubscriptionHandler) {
 	// WorkerManager configuration
-	// TODO: this needs to be moved under config, but now it's here as there are import cycles given singletons
 	workerManagerOptions := []workers.WorkerOptionFunc{
 		workers.WithMasaDir(cfg.MasaDir),
 	}
@@ -20,25 +31,26 @@ func initOptions(cfg *config.AppConfig, keyManager *masacrypto.KeyManager) ([]no
 		cachePath = cfg.MasaDir + "/cache"
 	}
 
-	masaNodeOptions := []node.Option{
+	masaNodeOptions := WithConstantOptions(
 		node.EnableStaked,
-		//	config.WithService(),
+		//	WithService(),
 		node.WithEnvironment(cfg.Environment),
 		node.WithVersion(cfg.Version),
 		node.WithPort(cfg.PortNbr),
 		node.WithBootNodes(cfg.Bootnodes...),
 		node.WithMasaDir(cfg.MasaDir),
 		node.WithCachePath(cachePath),
-		node.WithKeyManager(keyManager),
-	}
+		node.WithKeyManager(cfg.KeyManager),
+		node.WithWorkerProtocol(WorkerProtocol),
+	)
 
 	if cfg.TwitterScraper {
 		workerManagerOptions = append(workerManagerOptions, workers.EnableTwitterWorker)
 		masaNodeOptions = append(masaNodeOptions, node.IsTwitterScraper)
 	}
 
 	if cfg.TelegramScraper {
-		// XXX: Telegram scraper is not implemented yet in the worker (?)
+		// TODO: Telegram scraper is not implemented yet in the worker (?)
 		masaNodeOptions = append(masaNodeOptions, node.IsTelegramScraper)
 	}
 
@@ -56,15 +68,14 @@ func initOptions(cfg *config.AppConfig, keyManager *masacrypto.KeyManager) ([]no
 	blockChainEventTracker := node.NewBlockChain()
 	pubKeySub := &pubsub.PublicKeySubscriptionHandler{}
 
-	// TODO: Where the config is involved, move to the config the generation of Node options
 	masaNodeOptions = append(masaNodeOptions, []node.Option{
 		// Register the worker manager
 		node.WithMasaProtocolHandler(
-			config.WorkerProtocol,
+			WorkerProtocol,
 			workHandlerManager.HandleWorkerStream,
 		),
-		node.WithPubSubHandler(config.PublicKeyTopic, pubKeySub, false),
-		node.WithPubSubHandler(config.BlockTopic, blockChainEventTracker, true),
+		node.WithPubSubHandler(PublicKeyTopic, pubKeySub, false),
+		node.WithPubSubHandler(BlockTopic, blockChainEventTracker, true),
 	}...)
 
 	if cfg.Validator {

pkg/network/kdht.go

@@ -37,6 +37,7 @@ func EnableDHT(ctx context.Context, host host.Host, bootstrapNodes []multiaddr.M
 	options = append(options, dht.RoutingTableRefreshPeriod(time.Minute*5)) // Set refresh interval
 	options = append(options, dht.Mode(dht.ModeAutoServer))
 	options = append(options, dht.ProtocolPrefix(prefix))
+	// WTF: Why?
 	options = append(options, dht.NamespacedValidator("db", dbValidator{}))
 
 	kademliaDHT, err := dht.New(ctx, host, options...)