Add state change table for AVS operators to make windowing easier

internal/eigenState/avsOperators/avsOperators.go

@@ -2,6 +2,7 @@ package avsOperators
 
 import (
 	"database/sql"
+	"errors"
 	"fmt"
 	"slices"
 	"sort"
@@ -44,6 +45,14 @@ type RegisteredAvsOperatorDiff struct {
 	Registered  bool
 }
 
+type AvsOperatorStateChange struct {
+	Avs         string
+	Operator    string
+	Registered  bool
+	LogIndex    uint64
+	BlockNumber uint64
+}
+
 func NewSlotID(avs string, operator string) types.SlotID {
 	return types.SlotID(fmt.Sprintf("%s_%s", avs, operator))
 }
@@ -58,6 +67,9 @@ type AvsOperatorsModel struct {
 
 	// Accumulates state changes for SlotIds, grouped by block number
 	stateAccumulator map[uint64]map[types.SlotID]*AccumulatedStateChange
+
+	// Keep track of each distinct change, rather than accumulated change, to add to the delta table
+	deltaAccumulator map[uint64][]*AvsOperatorStateChange
 }
 
 // NewAvsOperators creates a new AvsOperatorsModel.
@@ -76,6 +88,8 @@ func NewAvsOperators(
 		globalConfig: globalConfig,
 
 		stateAccumulator: make(map[uint64]map[types.SlotID]*AccumulatedStateChange),
+
+		deltaAccumulator: make(map[uint64][]*AvsOperatorStateChange),
 	}
 	esm.RegisterState(s, 0)
 	return s, nil
@@ -120,6 +134,15 @@ func (a *AvsOperatorsModel) GetStateTransitions() (types.StateTransitions[Accumu
 			registered = uint64(val.(float64)) == 1
 		}
 
+		// Store the change in the delta accumulator
+		a.deltaAccumulator[log.BlockNumber] = append(a.deltaAccumulator[log.BlockNumber], &AvsOperatorStateChange{
+			Avs:         avs,
+			Operator:    operator,
+			Registered:  registered,
+			LogIndex:    log.LogIndex,
+			BlockNumber: log.BlockNumber,
+		})
+
 		slotID := NewSlotID(avs, operator)
 		record, ok := a.stateAccumulator[log.BlockNumber][slotID]
 		if !ok {
@@ -173,6 +196,7 @@ func (a *AvsOperatorsModel) IsInterestingLog(log *storage.TransactionLog) bool {
 
 func (a *AvsOperatorsModel) InitBlockProcessing(blockNumber uint64) error {
 	a.stateAccumulator[blockNumber] = make(map[types.SlotID]*AccumulatedStateChange)
+	a.deltaAccumulator[blockNumber] = make([]*AvsOperatorStateChange, 0)
 	return nil
 }
 
@@ -249,6 +273,22 @@ func (a *AvsOperatorsModel) prepareState(blockNumber uint64) ([]RegisteredAvsOpe
 	return inserts, deletes, nil
 }
 
+func (a *AvsOperatorsModel) writeDeltaRecordsToDeltaTable(blockNumber uint64) error {
+	records, ok := a.deltaAccumulator[blockNumber]
+	if !ok {
+		msg := "Delta accumulator was not initialized"
+		a.logger.Sugar().Errorw(msg, zap.Uint64("blockNumber", blockNumber))
+		return errors.New(msg)
+	}
+
+	res := a.DB.Model(&AvsOperatorStateChange{}).Clauses(clause.Returning{}).Create(&records)
+	if res.Error != nil {
+		a.logger.Sugar().Errorw("Failed to insert delta records", zap.Error(res.Error))
+		return res.Error
+	}
+	return nil
+}
+
 // CommitFinalState commits the final state for the given block number.
 func (a *AvsOperatorsModel) CommitFinalState(blockNumber uint64) error {
 	err := a.clonePreviousBlocksToNewBlock(blockNumber)
@@ -280,6 +320,11 @@ func (a *AvsOperatorsModel) CommitFinalState(blockNumber uint64) error {
 			return res.Error
 		}
 	}
+
+	if err = a.writeDeltaRecordsToDeltaTable(blockNumber); err != nil {
+		return err
+	}
+
 	return nil
 }
 

internal/eigenState/avsOperators/avsOperators_test.go

@@ -40,6 +40,21 @@ func setup() (
 func teardown(model *AvsOperatorsModel) {
 	model.DB.Exec("delete from avs_operator_changes")
 	model.DB.Exec("delete from registered_avs_operators")
+	model.DB.Exec("delete from avs_operator_state_changes")
+}
+
+func getInsertedDeltaRecordsForBlock(blockNumber uint64, model *AvsOperatorsModel) ([]*AvsOperatorStateChange, error) {
+	results := []*AvsOperatorStateChange{}
+
+	res := model.DB.Model(&AvsOperatorStateChange{}).Where("block_number = ?", blockNumber).Find(&results)
+	return results, res.Error
+}
+
+func getInsertedDeltaRecords(model *AvsOperatorsModel) ([]*AvsOperatorStateChange, error) {
+	results := []*AvsOperatorStateChange{}
+
+	res := model.DB.Model(&AvsOperatorStateChange{}).Order("block_number asc").Find(&results)
+	return results, res.Error
 }
 
 func Test_AvsOperatorState(t *testing.T) {
@@ -84,7 +99,22 @@ func Test_AvsOperatorState(t *testing.T) {
 		assert.Nil(t, err)
 		assert.NotNil(t, res)
 
-		teardown(avsOperatorState)
+		err = avsOperatorState.CommitFinalState(blockNumber)
+		assert.Nil(t, err)
+
+		inserted, err := getInsertedDeltaRecordsForBlock(blockNumber, avsOperatorState)
+		assert.Nil(t, err)
+		assert.Equal(t, 1, len(inserted))
+
+		assert.Equal(t, "0xdf25bdcdcdd9a3dd8c9069306c4dba8d90dd8e8e", inserted[0].Avs)
+		assert.Equal(t, "0x870679e138bcdf293b7ff14dd44b70fc97e12fc0", inserted[0].Operator)
+		assert.Equal(t, true, inserted[0].Registered)
+		assert.Equal(t, blockNumber, inserted[0].BlockNumber)
+		assert.Equal(t, uint64(400), inserted[0].LogIndex)
+
+		t.Cleanup(func() {
+			teardown(avsOperatorState)
+		})
 	})
 	t.Run("Should register AvsOperatorState and generate the table for the block", func(t *testing.T) {
 		esm := stateManager.NewEigenStateManager(l, grm)
@@ -134,7 +164,9 @@ func Test_AvsOperatorState(t *testing.T) {
 		assert.Nil(t, err)
 		assert.True(t, len(stateRoot) > 0)
 
-		teardown(avsOperatorState)
+		t.Cleanup(func() {
+			teardown(avsOperatorState)
+		})
 	})
 	t.Run("Should correctly generate state across multiple blocks", func(t *testing.T) {
 		esm := stateManager.NewEigenStateManager(l, grm)
@@ -217,6 +249,12 @@ func Test_AvsOperatorState(t *testing.T) {
 			assert.True(t, len(stateRoot) > 0)
 		}
 
-		teardown(avsOperatorState)
+		inserted, err := getInsertedDeltaRecords(avsOperatorState)
+		assert.Nil(t, err)
+		assert.Equal(t, len(logs), len(inserted))
+
+		t.Cleanup(func() {
+			teardown(avsOperatorState)
+		})
 	})
 }

internal/eigenState/operatorShares/operatorShares.go

@@ -278,7 +278,6 @@ func (osm *OperatorSharesModel) prepareState(blockNumber uint64) ([]OperatorShar
 	// Map the existing records to a map for easier lookup
 	mappedRecords := make(map[types.SlotID]OperatorShares)
 	for _, record := range existingRecords {
-		fmt.Printf("Existing OperatorShares %+v\n", record)
 		slotID := NewSlotID(record.Operator, record.Strategy)
 		mappedRecords[slotID] = record
 	}

internal/sqlite/migrations/202409161057_avsOperatorDeltas/up.go

@@ -0,0 +1,36 @@
+package _202409161057_avsOperatorDeltas
+
+import (
+	"fmt"
+	"gorm.io/gorm"
+)
+
+type SqliteMigration struct {
+}
+
+func (m *SqliteMigration) Up(grm *gorm.DB) error {
+	queries := []string{
+		`create table if not exists avs_operator_state_changes (
+			operator TEXT NOT NULL,
+			avs TEXT NOT NULL,
+			block_number INTEGER NOT NULL,
+			log_index INTEGER NOT NULL,
+			created_at DATETIME default current_timestamp,
+			registered integer not null,
+			unique(operator, avs, block_number, log_index)
+		);
+		`,
+	}
+
+	for _, query := range queries {
+		if res := grm.Exec(query); res.Error != nil {
+			fmt.Printf("Failed to execute query: %s\n", query)
+			return res.Error
+		}
+	}
+	return nil
+}
+
+func (m *SqliteMigration) GetName() string {
+	return "202409161057_avsOperatorDeltas"
+}

internal/sqlite/migrations/migrator.go

@@ -3,6 +3,7 @@ package migrations
 import (
 	"database/sql"
 	"fmt"
+	_202409161057_avsOperatorDeltas "github.com/Layr-Labs/go-sidecar/internal/sqlite/migrations/202409161057_avsOperatorDeltas"
 	"time"
 
 	_202409061249_bootstrapDb "github.com/Layr-Labs/go-sidecar/internal/sqlite/migrations/202409061249_bootstrapDb"
@@ -52,6 +53,7 @@ func (m *SqliteMigrator) MigrateAll() error {
 		&_202409101144_submittedDistributionRoot.SqliteMigration{},
 		&_202409101540_rewardSubmissions.SqliteMigration{},
 		&_202409111509_removeOperatorRestakedStrategiesBlockConstraint.SqliteMigration{},
+		&_202409161057_avsOperatorDeltas.SqliteMigration{},
 	}
 
 	m.Logger.Sugar().Info("Running migrations")

pkg/rewards/operatorAvsRegistrationWindows.go

@@ -0,0 +1,9 @@
+package rewards
+
+const query = `
+
+`
+
+func (r *RewardsCalculator) GenerateOperatorAvsRegistrationWindows(snapshotDate string) {
+
+}

pkg/rewards/operatorAvsStrategyWindows.go

@@ -0,0 +1,5 @@
+package rewards
+
+func (r *RewardsCalculator) GenerateOperatorAvsStrategyWindows(snapshotDate string) {
+
+}