feat: add RewardsClaimed metaState model to capture state that isnt EigenState

cmd/database.go

@@ -15,6 +15,7 @@ import (
 	"github.com/Layr-Labs/sidecar/pkg/eventBus"
 	"github.com/Layr-Labs/sidecar/pkg/fetcher"
 	"github.com/Layr-Labs/sidecar/pkg/indexer"
+	"github.com/Layr-Labs/sidecar/pkg/metaState/metaStateManager"
 	"github.com/Layr-Labs/sidecar/pkg/pipeline"
 	"github.com/Layr-Labs/sidecar/pkg/postgres"
 	"github.com/Layr-Labs/sidecar/pkg/postgres/migrations"
@@ -87,6 +88,7 @@ var runDatabaseCmd = &cobra.Command{
 		}
 
 		sm := stateManager.NewEigenStateManager(l, grm)
+		msm := metaStateManager.NewMetaStateManager(grm, l, cfg)
 
 		if err := eigenState.LoadEigenStateModels(sm, grm, l, cfg); err != nil {
 			l.Sugar().Fatalw("Failed to load eigen state models", zap.Error(err))
@@ -107,7 +109,7 @@ var runDatabaseCmd = &cobra.Command{
 
 		rcq := rewardsCalculatorQueue.NewRewardsCalculatorQueue(rc, l)
 
-		_ = pipeline.NewPipeline(fetchr, idxr, mds, sm, rc, rcq, cfg, sdc, eb, l)
+		_ = pipeline.NewPipeline(fetchr, idxr, mds, sm, msm, rc, rcq, cfg, sdc, eb, l)
 
 		l.Sugar().Infow("Done")
 	},

cmd/debugger/main.go

@@ -11,6 +11,7 @@ import (
 	"github.com/Layr-Labs/sidecar/pkg/eventBus"
 	"github.com/Layr-Labs/sidecar/pkg/fetcher"
 	"github.com/Layr-Labs/sidecar/pkg/indexer"
+	"github.com/Layr-Labs/sidecar/pkg/metaState/metaStateManager"
 	"github.com/Layr-Labs/sidecar/pkg/pipeline"
 	"github.com/Layr-Labs/sidecar/pkg/postgres"
 	"github.com/Layr-Labs/sidecar/pkg/proofs"
@@ -80,6 +81,7 @@ func main() {
 	}
 
 	sm := stateManager.NewEigenStateManager(l, grm)
+	msm := metaStateManager.NewMetaStateManager(grm, l, cfg)
 
 	if err := eigenState.LoadEigenStateModels(sm, grm, l, cfg); err != nil {
 		l.Sugar().Fatalw("Failed to load eigen state models", zap.Error(err))
@@ -100,14 +102,14 @@ func main() {
 
 	rcq := rewardsCalculatorQueue.NewRewardsCalculatorQueue(rc, l)
 
-	p := pipeline.NewPipeline(fetchr, idxr, mds, sm, rc, rcq, cfg, sdc, eb, l)
+	p := pipeline.NewPipeline(fetchr, idxr, mds, sm, msm, rc, rcq, cfg, sdc, eb, l)
 	rps := proofs.NewRewardsProofsStore(rc, l)
 
 	// Create new sidecar instance
 	// Create new sidecar instance
 	_ = sidecar.NewSidecar(&sidecar.SidecarConfig{
 		GenesisBlockNumber: cfg.GetGenesisBlockNumber(),
-	}, cfg, mds, p, sm, rc, rcq, rps, l, client)
+	}, cfg, mds, p, sm, msm, rc, rcq, rps, l, client)
 
 	rpc := rpcServer.NewRpcServer(&rpcServer.RpcServerConfig{
 		GrpcPort: cfg.RpcConfig.GrpcPort,

cmd/run.go

@@ -13,6 +13,7 @@ import (
 	"github.com/Layr-Labs/sidecar/pkg/eventBus"
 	"github.com/Layr-Labs/sidecar/pkg/fetcher"
 	"github.com/Layr-Labs/sidecar/pkg/indexer"
+	"github.com/Layr-Labs/sidecar/pkg/metaState/metaStateManager"
 	"github.com/Layr-Labs/sidecar/pkg/pipeline"
 	"github.com/Layr-Labs/sidecar/pkg/postgres"
 	"github.com/Layr-Labs/sidecar/pkg/proofs"
@@ -97,6 +98,7 @@ var runCmd = &cobra.Command{
 		}
 
 		sm := stateManager.NewEigenStateManager(l, grm)
+		msm := metaStateManager.NewMetaStateManager(grm, l, cfg)
 
 		if err := eigenState.LoadEigenStateModels(sm, grm, l, cfg); err != nil {
 			l.Sugar().Fatalw("Failed to load eigen state models", zap.Error(err))
@@ -121,12 +123,12 @@ var runCmd = &cobra.Command{
 
 		go rcq.Process()
 
-		p := pipeline.NewPipeline(fetchr, idxr, mds, sm, rc, rcq, cfg, sdc, eb, l)
+		p := pipeline.NewPipeline(fetchr, idxr, mds, sm, msm, rc, rcq, cfg, sdc, eb, l)
 
 		// Create new sidecar instance
 		sidecar := sidecar.NewSidecar(&sidecar.SidecarConfig{
 			GenesisBlockNumber: cfg.GetGenesisBlockNumber(),
-		}, cfg, mds, p, sm, rc, rcq, rps, l, client)
+		}, cfg, mds, p, sm, msm, rc, rcq, rps, l, client)
 
 		rpc := rpcServer.NewRpcServer(&rpcServer.RpcServerConfig{
 			GrpcPort: cfg.RpcConfig.GrpcPort,

pkg/metaState/baseModel/baseModel.go

@@ -0,0 +1,88 @@
+package baseModel
+
+import (
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"github.com/Layr-Labs/sidecar/pkg/parser"
+	"github.com/Layr-Labs/sidecar/pkg/storage"
+	"go.uber.org/zap"
+	"gorm.io/gorm"
+	"slices"
+	"strings"
+)
+
+func IsInterestingLog(contractsEvents map[string][]string, log *storage.TransactionLog) bool {
+	logAddress := strings.ToLower(log.Address)
+	if eventNames, ok := contractsEvents[logAddress]; ok {
+		if slices.Contains(eventNames, log.EventName) {
+			return true
+		}
+	}
+	return false
+}
+
+func ParseLogArguments(log *storage.TransactionLog, l *zap.Logger) ([]parser.Argument, error) {
+	arguments := make([]parser.Argument, 0)
+	err := json.Unmarshal([]byte(log.Arguments), &arguments)
+	if err != nil {
+		l.Sugar().Errorw("Failed to unmarshal arguments",
+			zap.Error(err),
+			zap.String("transactionHash", log.TransactionHash),
+			zap.Uint64("transactionIndex", log.TransactionIndex),
+		)
+		return nil, err
+	}
+	return arguments, nil
+}
+
+func ParseLogOutput[T any](log *storage.TransactionLog, l *zap.Logger) (*T, error) {
+	var outputData *T
+	err := json.Unmarshal([]byte(log.OutputData), &outputData)
+	if err != nil {
+		l.Sugar().Errorw("Failed to unmarshal outputData",
+			zap.Error(err),
+			zap.String("transactionHash", log.TransactionHash),
+			zap.Uint64("transactionIndex", log.TransactionIndex),
+		)
+		return nil, err
+	}
+	return outputData, nil
+}
+
+func DeleteState(tableName string, startBlockNumber uint64, endBlockNumber uint64, db *gorm.DB, l *zap.Logger) error {
+	if endBlockNumber != 0 && endBlockNumber < startBlockNumber {
+		l.Sugar().Errorw("Invalid block range",
+			zap.Uint64("startBlockNumber", startBlockNumber),
+			zap.Uint64("endBlockNumber", endBlockNumber),
+		)
+		return errors.New("Invalid block range; endBlockNumber must be greater than or equal to startBlockNumber")
+	}
+
+	// tokenizing the table name apparently doesnt work, so we need to use Sprintf to include it.
+	query := fmt.Sprintf(`
+		delete from %s
+		where block_number >= @startBlockNumber
+	`, tableName)
+	if endBlockNumber > 0 {
+		query += " and block_number <= @endBlockNumber"
+	}
+	res := db.Exec(query,
+		sql.Named("tableName", tableName),
+		sql.Named("startBlockNumber", startBlockNumber),
+		sql.Named("endBlockNumber", endBlockNumber))
+	if res.Error != nil {
+		l.Sugar().Errorw("Failed to delete state", zap.Error(res.Error))
+		return res.Error
+	}
+	return nil
+}
+
+func CastCommittedStateToInterface[T any](committedState []*T) []interface{} {
+	state := make([]interface{}, len(committedState))
+	for i, v := range committedState {
+		state[i] = *v
+	}
+	return state
+}

pkg/metaState/metaState.go

@@ -0,0 +1,23 @@
+package metaState
+
+import (
+	"github.com/Layr-Labs/sidecar/internal/config"
+	"github.com/Layr-Labs/sidecar/pkg/metaState/metaStateManager"
+	"github.com/Layr-Labs/sidecar/pkg/metaState/rewardsClaimed"
+	"go.uber.org/zap"
+	"gorm.io/gorm"
+)
+
+func LoadMetaStateModels(
+	msm *metaStateManager.MetaStateManager,
+	db *gorm.DB,
+	l *zap.Logger,
+	cfg *config.Config,
+) error {
+	if _, err := rewardsClaimed.NewRewardsClaimedModel(db, l, cfg, msm); err != nil {
+		l.Sugar().Errorw("Failed to create RewardsClaimedModel", zap.Error(err))
+		return err
+	}
+
+	return nil
+}

pkg/metaState/metaStateManager/metaStateManager.go

@@ -0,0 +1,90 @@
+package metaStateManager
+
+import (
+	"github.com/Layr-Labs/sidecar/internal/config"
+	"github.com/Layr-Labs/sidecar/pkg/metaState/types"
+	"github.com/Layr-Labs/sidecar/pkg/storage"
+	"go.uber.org/zap"
+	"gorm.io/gorm"
+)
+
+type MetaStateManager struct {
+	db              *gorm.DB
+	logger          *zap.Logger
+	globalConfig    *config.Config
+	metaStateModels []types.IMetaStateModel
+}
+
+func NewMetaStateManager(db *gorm.DB, l *zap.Logger, gc *config.Config) *MetaStateManager {
+	return &MetaStateManager{
+		db:              db,
+		logger:          l,
+		globalConfig:    gc,
+		metaStateModels: make([]types.IMetaStateModel, 0),
+	}
+}
+
+func (msm *MetaStateManager) RegisterMetaStateModel(model types.IMetaStateModel) {
+	msm.metaStateModels = append(msm.metaStateModels, model)
+}
+
+func (msm *MetaStateManager) InitProcessingForBlock(blockNumber uint64) error {
+	for _, model := range msm.metaStateModels {
+		if err := model.SetupStateForBlock(blockNumber); err != nil {
+			msm.logger.Sugar().Errorw("Failed to setup state for block",
+				"blockNumber", blockNumber,
+				"model", model,
+				"error", err,
+			)
+			return err
+		}
+	}
+	return nil
+}
+
+func (msm *MetaStateManager) CleanupProcessedStateForBlock(blockNumber uint64) error {
+	for _, model := range msm.metaStateModels {
+		if err := model.CleanupProcessedStateForBlock(blockNumber); err != nil {
+			msm.logger.Sugar().Errorw("Failed to cleanup state for block",
+				"blockNumber", blockNumber,
+				"model", model,
+				"error", err,
+			)
+			return err
+		}
+	}
+	return nil
+}
+
+func (msm *MetaStateManager) HandleTransactionLog(log *storage.TransactionLog) error {
+	for _, model := range msm.metaStateModels {
+		if model.IsInterestingLog(log) {
+			if _, err := model.HandleTransactionLog(log); err != nil {
+				msm.logger.Sugar().Errorw("Failed to handle log",
+					"log", log,
+					"model", model,
+					"error", err,
+				)
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func (msm *MetaStateManager) CommitFinalState(blockNumber uint64) (map[string][]interface{}, error) {
+	committedState := make(map[string][]interface{})
+	for _, model := range msm.metaStateModels {
+		state, err := model.CommitFinalState(blockNumber)
+		if err != nil {
+			msm.logger.Sugar().Errorw("Failed to commit final state",
+				"blockNumber", blockNumber,
+				"model", model,
+				"error", err,
+			)
+			return nil, err
+		}
+		committedState[model.ModelName()] = state
+	}
+	return committedState, nil
+}