refactor access node's transactions and collections storage module

Author: zhangchiqing

cmd/access/node_builder/access_node_builder.go

@@ -346,6 +346,8 @@ type FlowAccessNodeBuilder struct {
 	ExecutionDataTracker         tracker.Storage
 	VersionControl               *version.VersionControl
 	StopControl                  *stop.StopControl
+	Transactions                 storage.Transactions
+	Collections                  storage.Collections
 
 	// The sync engine participants provider is the libp2p peer store for the access node
 	// which is not available until after the network has started.
@@ -574,6 +576,14 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 		AdminCommand("read-execution-data", func(config *cmd.NodeConfig) commands.AdminCommand {
 			return stateSyncCommands.NewReadExecutionDataCommand(builder.ExecutionDataStore)
 		}).
+		Module("transactions and collections storage", func(node *cmd.NodeConfig) error {
+			// TODO: replace with node.ProtocolDB
+			db := badgerimpl.ToDB(node.DB)
+			transactions := store.NewTransactions(node.Metrics.Cache, db)
+			builder.Collections = store.NewCollections(db, transactions)
+			builder.Transactions = transactions
+			return nil
+		}).
 		Module("execution data datastore and blobstore", func(node *cmd.NodeConfig) error {
 			datastoreDir := filepath.Join(builder.executionDataDir, "blobstore")
 			err := os.MkdirAll(datastoreDir, 0700)
@@ -966,8 +976,8 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 					builder.Storage.RegisterIndex,
 					builder.Storage.Headers,
 					builder.Storage.Events,
-					builder.Storage.Collections,
-					builder.Storage.Transactions,
+					notNil(builder.Collections),
+					notNil(builder.Transactions),
 					builder.Storage.LightTransactionResults,
 					builder.RootChainID.Chain(),
 					indexerDerivedChainData,
@@ -1613,7 +1623,7 @@ func (builder *FlowAccessNodeBuilder) Initialize() error {
 	builder.EnqueueNetworkInit()
 
 	builder.AdminCommand("get-transactions", func(conf *cmd.NodeConfig) commands.AdminCommand {
-		return storageCommands.NewGetTransactionsCommand(conf.State, conf.Storage.Payloads, conf.Storage.Collections)
+		return storageCommands.NewGetTransactionsCommand(conf.State, conf.Storage.Payloads, notNil(builder.Collections))
 	})
 
 	// if this is an access node that supports public followers, enqueue the public network
@@ -2034,8 +2044,8 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 				HistoricalAccessNodes: builder.HistoricalAccessRPCs,
 				Blocks:                node.Storage.Blocks,
 				Headers:               node.Storage.Headers,
-				Collections:           node.Storage.Collections,
-				Transactions:          node.Storage.Transactions,
+				Collections:           notNil(builder.Collections),
+				Transactions:          notNil(builder.Transactions),
 				ExecutionReceipts:     node.Storage.Receipts,
 				ExecutionResults:      node.Storage.Results,
 				TxResultErrorMessages: node.Storage.TransactionResultErrorMessages,
@@ -2137,8 +2147,8 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 				builder.RequestEng,
 				node.Storage.Blocks,
 				node.Storage.Headers,
-				node.Storage.Collections,
-				node.Storage.Transactions,
+				notNil(builder.Collections),
+				notNil(builder.Transactions),
 				node.Storage.Results,
 				node.Storage.Receipts,
 				builder.collectionExecutedMetric,
@@ -2384,3 +2394,15 @@ func (builder *FlowAccessNodeBuilder) initPublicLibp2pNode(networkKey crypto.Pri
 
 	return libp2pNode, nil
 }
+
+// notNil ensures that the input is not nil and returns it
+// the usage is to ensure the dependencies are initialized before initializing a module.
+// for instance, the IngestionEngine depends on storage.Collections, which is initialized in a
+// different function, so we need to ensure that the storage.Collections is initialized before
+// creating the IngestionEngine.
+func notNil[T any](dep T) T {
+	if any(dep) == nil {
+		panic("dependency is nil")
+	}
+	return dep
+}