refactor rpc to a wrapper for easier reuse

Author: iuwqyir

cmd/orchestrator.go

@@ -6,8 +6,8 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/rs/zerolog/log"
 	"github.com/spf13/cobra"
-	"github.com/thirdweb-dev/indexer/internal/common"
 	"github.com/thirdweb-dev/indexer/internal/orchestrator"
+	"github.com/thirdweb-dev/indexer/internal/rpc"
 )
 
 var (
@@ -23,7 +23,7 @@ var (
 
 func RunOrchestrator(cmd *cobra.Command, args []string) {
 	log.Info().Msg("Starting indexer")
-	rpc, err := common.InitializeRPC()
+	rpc, err := rpc.Initialize()
 	if err != nil {
 		log.Fatal().Err(err).Msg("Failed to initialize RPC")
 	}

internal/common/block.go

@@ -35,3 +35,5 @@ type BlockData struct {
 	Logs         []Log
 	Traces       []Trace
 }
+
+type RawBlock = map[string]interface{}

internal/common/log.go

@@ -16,3 +16,5 @@ type Log struct {
 	Data             string   `json:"data"`
 	Topics           []string `json:"topics"`
 }
+
+type RawLogs = []map[string]interface{}

internal/common/rpc.go

@@ -27,3 +27,5 @@ type Trace struct {
 	RewardType       string   `json:"reward_type"`
 	RefundAddress    string   `json:"refund_address"`
 }
+
+type RawTraces = []map[string]interface{}

internal/common/trace.go

@@ -0,0 +1,18 @@
+package common
+
+import "math/big"
+
+func BigIntSliceToChunks(values []*big.Int, chunkSize int) [][]*big.Int {
+	if chunkSize >= len(values) || chunkSize <= 0 {
+		return [][]*big.Int{values}
+	}
+	var chunks [][]*big.Int
+	for i := 0; i < len(values); i += chunkSize {
+		end := i + chunkSize
+		if end > len(values) {
+			end = len(values)
+		}
+		chunks = append(chunks, values[i:end])
+	}
+	return chunks
+}

internal/common/utils.go

@@ -5,19 +5,18 @@ import (
 	"time"
 
 	"github.com/rs/zerolog/log"
-	"github.com/thirdweb-dev/indexer/internal/common"
 	"github.com/thirdweb-dev/indexer/internal/metrics"
+	"github.com/thirdweb-dev/indexer/internal/rpc"
 )
 
 const DEFAULT_CHAIN_TRACKER_POLL_INTERVAL = 300000 // 5 minutes
 
 type ChainTracker struct {
-	rpc               common.RPC
+	rpc               rpc.Client
 	triggerIntervalMs int
 }
 
-func NewChainTracker(rpc common.RPC) *ChainTracker {
-
+func NewChainTracker(rpc rpc.Client) *ChainTracker {
 	return &ChainTracker{
 		rpc:               rpc,
 		triggerIntervalMs: DEFAULT_CHAIN_TRACKER_POLL_INTERVAL,

internal/orchestrator/chain_tracker.go

@@ -11,6 +11,7 @@ import (
 	config "github.com/thirdweb-dev/indexer/configs"
 	"github.com/thirdweb-dev/indexer/internal/common"
 	"github.com/thirdweb-dev/indexer/internal/metrics"
+	"github.com/thirdweb-dev/indexer/internal/rpc"
 	"github.com/thirdweb-dev/indexer/internal/storage"
 )
 
@@ -22,10 +23,10 @@ type Committer struct {
 	blocksPerCommit   int
 	storage           storage.IStorage
 	pollFromBlock     *big.Int
-	rpc               common.RPC
+	rpc               rpc.Client
 }
 
-func NewCommitter(rpc common.RPC, storage storage.IStorage) *Committer {
+func NewCommitter(rpc rpc.Client, storage storage.IStorage) *Committer {
 	triggerInterval := config.Cfg.Committer.Interval
 	if triggerInterval == 0 {
 		triggerInterval = DEFAULT_COMMITTER_TRIGGER_INTERVAL

internal/orchestrator/committer.go

@@ -9,6 +9,7 @@ import (
 	config "github.com/thirdweb-dev/indexer/configs"
 	"github.com/thirdweb-dev/indexer/internal/common"
 	"github.com/thirdweb-dev/indexer/internal/metrics"
+	"github.com/thirdweb-dev/indexer/internal/rpc"
 	"github.com/thirdweb-dev/indexer/internal/storage"
 	"github.com/thirdweb-dev/indexer/internal/worker"
 )
@@ -20,10 +21,10 @@ type FailureRecoverer struct {
 	failuresPerPoll   int
 	triggerIntervalMs int
 	storage           storage.IStorage
-	rpc               common.RPC
+	rpc               rpc.Client
 }
 
-func NewFailureRecoverer(rpc common.RPC, storage storage.IStorage) *FailureRecoverer {
+func NewFailureRecoverer(rpc rpc.Client, storage storage.IStorage) *FailureRecoverer {
 	failuresPerPoll := config.Cfg.FailureRecoverer.BlocksPerRun
 	if failuresPerPoll == 0 {
 		failuresPerPoll = DEFAULT_FAILURES_PER_POLL
@@ -80,7 +81,7 @@ func (fr *FailureRecoverer) Start() {
 	select {}
 }
 
-func (fr *FailureRecoverer) handleWorkerResults(blockFailures []common.BlockFailure, results []worker.WorkerResult) {
+func (fr *FailureRecoverer) handleWorkerResults(blockFailures []common.BlockFailure, results []rpc.GetFullBlockResult) {
 	log.Debug().Msgf("Failure Recoverer recovered %d blocks", len(results))
 	blockFailureMap := make(map[*big.Int]common.BlockFailure)
 	for _, failure := range blockFailures {
@@ -105,10 +106,10 @@ func (fr *FailureRecoverer) handleWorkerResults(blockFailures []common.BlockFail
 			})
 		} else {
 			successfulResults = append(successfulResults, common.BlockData{
-				Block:        result.Block,
-				Logs:         result.Logs,
-				Transactions: result.Transactions,
-				Traces:       result.Traces,
+				Block:        result.Data.Block,
+				Logs:         result.Data.Logs,
+				Transactions: result.Data.Transactions,
+				Traces:       result.Data.Traces,
 			})
 			failuresToDelete = append(failuresToDelete, blockFailureForBlock)
 		}

internal/orchestrator/failure_recoverer.go

@@ -4,19 +4,19 @@ import (
 	"sync"
 
 	config "github.com/thirdweb-dev/indexer/configs"
-	"github.com/thirdweb-dev/indexer/internal/common"
+	"github.com/thirdweb-dev/indexer/internal/rpc"
 	"github.com/thirdweb-dev/indexer/internal/storage"
 )
 
 type Orchestrator struct {
-	rpc                     common.RPC
+	rpc                     rpc.Client
 	storage                 storage.IStorage
 	pollerEnabled           bool
 	failureRecovererEnabled bool
 	committerEnabled        bool
 }
 
-func NewOrchestrator(rpc common.RPC) (*Orchestrator, error) {
+func NewOrchestrator(rpc rpc.Client) (*Orchestrator, error) {
 	storage, err := storage.NewStorageConnector(&config.Cfg.Storage)
 	if err != nil {
 		return nil, err