Add simplified node tx fetcher

fetch/tx/node.go

@@ -0,0 +1,129 @@
+package tx
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/gnolang/gno/tm2/pkg/bft/types"
+)
+
+type NodeFetcher struct {
+	storage Storage
+	client  Client
+
+	queryInterval time.Duration // block query interval
+}
+
+// NewNodeFetcher creates a new transaction result fetcher instance
+// that gets data from a remote chain
+// TODO add logger
+func NewNodeFetcher(
+	storage Storage,
+	client Client,
+) *NodeFetcher {
+	return &NodeFetcher{
+		storage:       storage,
+		client:        client,
+		queryInterval: 1 * time.Second,
+	}
+}
+
+// FetchTransactions runs the transaction fetcher [BLOCKING]
+func (f *NodeFetcher) FetchTransactions(ctx context.Context) error {
+	// catchupWithChain syncs any transactions that have occurred
+	// between the local last block (in storage) and the chain state (latest head)
+	catchupWithChain := func(lastBlock int64) (int64, error) {
+		// Fetch the latest block from the chain
+		latest, latestErr := f.client.GetLatestBlockNumber()
+		if latestErr != nil {
+			return 0, fmt.Errorf("unable to fetch latest block number, %w", latestErr)
+		}
+
+		// Check if there is a block gap
+		if lastBlock == latest {
+			// No gap, nothing to sync
+			return latest, nil
+		}
+
+		// Catch up to the latest block
+		for block := lastBlock + 1; block <= latest; block++ {
+			if fetchErr := f.saveTxsFromBlock(ctx, block); fetchErr != nil {
+				return 0, fetchErr
+			}
+		}
+
+		// Return the latest available block
+		return latest, nil
+	}
+
+	// Fetch the latest tx from storage
+	lastTx, err := f.storage.GetLatestTx(ctx)
+	if err != nil {
+		return fmt.Errorf("unable to fetch latest transaction, %w", err)
+	}
+
+	// The height present in storage
+	currentHeight := lastTx.Height
+
+	// "Catch up" initially with the chain
+	if currentHeight, err = catchupWithChain(currentHeight); err != nil {
+		return err
+	}
+
+	// Start a listener for monitoring new blocks
+	ticker := time.NewTicker(f.queryInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			// TODO log
+			return nil
+		case <-ticker.C:
+			if currentHeight, err = catchupWithChain(currentHeight); err != nil {
+				return err
+			}
+		}
+	}
+}
+
+// saveTxsFromBlock commits the block transactions to storage
+func (f *NodeFetcher) saveTxsFromBlock(
+	ctx context.Context,
+	blockNum int64,
+) error {
+	// TODO log
+	// Get block info from the chain
+	block, err := f.client.GetBlock(blockNum)
+	if err != nil {
+		return fmt.Errorf("unable to fetch block, %w", err)
+	}
+
+	// Skip empty blocks
+	if block.Block.NumTxs == 0 {
+		return nil
+	}
+
+	// Get the transaction execution results
+	txResults, err := f.client.GetBlockResults(blockNum)
+	if err != nil {
+		return fmt.Errorf("unable to fetch block results, %w", err)
+	}
+
+	// Save the transaction result to the storage
+	for index, tx := range block.Block.Txs {
+		result := &types.TxResult{
+			Height:   block.Block.Height,
+			Index:    uint32(index),
+			Tx:       tx,
+			Response: txResults.Results.DeliverTxs[index],
+		}
+
+		if err := f.storage.SaveTx(ctx, result); err != nil {
+			return fmt.Errorf("unable to save tx, %w", err)
+		}
+	}
+
+	return nil
+}

fetch/tx/types.go

@@ -0,0 +1,30 @@
+package tx
+
+import (
+	"context"
+
+	core_types "github.com/gnolang/gno/tm2/pkg/bft/rpc/core/types"
+	"github.com/gnolang/gno/tm2/pkg/bft/types"
+)
+
+// Storage defines the transaction storage interface
+type Storage interface {
+	// GetLatestTx returns the latest transaction from the storage
+	GetLatestTx(ctx context.Context) (*types.TxResult, error)
+
+	// SaveTx saves the transaction to the permanent storage
+	SaveTx(ctx context.Context, tx *types.TxResult) error
+}
+
+// Client defines the interface for the node (client) communication
+type Client interface {
+	// GetLatestBlockNumber returns the latest block height from the chain
+	GetLatestBlockNumber() (int64, error)
+
+	// GetBlock returns specified block
+	GetBlock(int64) (*core_types.ResultBlock, error)
+
+	// GetBlockResults returns the results of executing the transactions
+	// for the specified block
+	GetBlockResults(int64) (*core_types.ResultBlockResults, error)
+}

go.mod

@@ -3,3 +3,39 @@ module github.com/gnolang/tx-indexer
 go 1.20
 
 require github.com/peterbourgon/ff/v3 v3.4.0
+
+require (
+	github.com/cespare/xxhash v1.1.0 // indirect
+	github.com/cespare/xxhash/v2 v2.1.1 // indirect
+	github.com/dgraph-io/badger/v3 v3.2103.4 // indirect
+	github.com/dgraph-io/ristretto v0.1.1 // indirect
+	github.com/dustin/go-humanize v1.0.0 // indirect
+	github.com/gnolang/goleveldb v0.0.9 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b // indirect
+	github.com/golang/groupcache v0.0.0-20190702054246-869f871628b6 // indirect
+	github.com/golang/protobuf v1.5.3 // indirect
+	github.com/golang/snappy v0.0.4 // indirect
+	github.com/google/flatbuffers v1.12.1 // indirect
+	github.com/jmhodges/levigo v1.0.0 // indirect
+	github.com/klauspost/compress v1.12.3 // indirect
+	github.com/libp2p/go-buffer-pool v0.1.0 // indirect
+	github.com/linxGnu/grocksdb v1.8.5 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
+	github.com/tecbot/gorocksdb v0.0.0-20191217155057-f0fad39f321c // indirect
+	go.etcd.io/bbolt v1.3.8 // indirect
+	go.opencensus.io v0.22.5 // indirect
+	golang.org/x/mod v0.14.0 // indirect
+	golang.org/x/net v0.17.0 // indirect
+	golang.org/x/sys v0.14.0 // indirect
+)
+
+require (
+	github.com/btcsuite/btcd/btcec/v2 v2.3.2 // indirect
+	github.com/btcsuite/btcd/btcutil v1.1.3 // indirect
+	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.2.0 // indirect
+	github.com/gnolang/gno v0.0.0-20231215125729-9262c1a8f949
+	github.com/gnolang/overflow v0.0.0-20170615021017-4d914c927216 // indirect
+	golang.org/x/crypto v0.15.0 // indirect
+	google.golang.org/protobuf v1.31.0 // indirect
+)