feat: utilize split in existing rewards calculation

pkg/postgres/migrations/202411221331_operatorAVSSplitSnapshots/up.go

@@ -0,0 +1,31 @@
+package _202411221331_operatorAVSSplitSnapshots
+
+import (
+	"database/sql"
+
+	"gorm.io/gorm"
+)
+
+type Migration struct {
+}
+
+func (m *Migration) Up(db *sql.DB, grm *gorm.DB) error {
+	queries := []string{
+		`CREATE TABLE IF NOT EXISTS operator_avs_split_snapshots (
+			operator varchar not null,
+			avs varchar not null,
+			split integer not null,
+			snapshot date not null
+		)`,
+	}
+	for _, query := range queries {
+		if _, err := db.Exec(query); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (m *Migration) GetName() string {
+	return "202411221331_operatorAVSSplitSnapshots"
+}

pkg/postgres/migrations/202411221331_operatorPISplitSnapshots/up.go

@@ -0,0 +1,30 @@
+package _202411221331_operatorPISplitSnapshots
+
+import (
+	"database/sql"
+
+	"gorm.io/gorm"
+)
+
+type Migration struct {
+}
+
+func (m *Migration) Up(db *sql.DB, grm *gorm.DB) error {
+	queries := []string{
+		`CREATE TABLE IF NOT EXISTS operator_pi_split_snapshots (
+			operator varchar not null,
+			split integer not null,
+			snapshot date not null
+		)`,
+	}
+	for _, query := range queries {
+		if _, err := db.Exec(query); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (m *Migration) GetName() string {
+	return "202411221331_operatorPISplitSnapshots"
+}

pkg/postgres/migrations/migrator.go

@@ -38,6 +38,8 @@ import (
 	_202411191550_operatorAVSSplits "github.com/Layr-Labs/sidecar/pkg/postgres/migrations/202411191550_operatorAVSSplits"
 	_202411191708_operatorPISplits "github.com/Layr-Labs/sidecar/pkg/postgres/migrations/202411191708_operatorPISplits"
 	_202411191947_cleanupUnusedTables "github.com/Layr-Labs/sidecar/pkg/postgres/migrations/202411191947_cleanupUnusedTables"
+	_202411221331_operatorAVSSplitSnapshots "github.com/Layr-Labs/sidecar/pkg/postgres/migrations/202411221331_operatorAVSSplitSnapshots"
+	_202411221331_operatorPISplitSnapshots "github.com/Layr-Labs/sidecar/pkg/postgres/migrations/202411221331_operatorPISplitSnapshots"
 
 	"go.uber.org/zap"
 	"gorm.io/gorm"
@@ -112,6 +114,8 @@ func (m *Migrator) MigrateAll() error {
 		&_202411191550_operatorAVSSplits.Migration{},
 		&_202411191708_operatorPISplits.Migration{},
 		&_202411191947_cleanupUnusedTables.Migration{},
+		&_202411221331_operatorAVSSplitSnapshots.Migration{},
+		&_202411221331_operatorPISplitSnapshots.Migration{},
 	}
 
 	for _, migration := range migrations {

pkg/rewards/2_goldStakerRewardAmounts.go

@@ -2,6 +2,7 @@ package rewards
 
 import (
 	"database/sql"
+
 	"github.com/Layr-Labs/sidecar/internal/config"
 	"go.uber.org/zap"
 )
@@ -107,26 +108,38 @@ staker_operator_total_tokens AS (
     END as total_staker_operator_payout
   FROM staker_proportion
 ),
--- Calculate the token breakdown for each (staker, operator) pair
+-- Include the operator_avs_split_snapshots table
+operator_avs_splits_cte AS (
+  SELECT
+    operator,
+    avs,
+    snapshot,
+    split
+  FROM operator_avs_split_snapshots
+),
+-- Calculate the token breakdown for each (staker, operator) pair with dynamic split logic
+-- If no split is found, default to 1000 (10%)
 token_breakdowns AS (
-  SELECT *,
+  SELECT sot.*,
     CASE
       WHEN snapshot < @amazonHardforkDate AND reward_submission_date < @amazonHardforkDate THEN
-        cast(total_staker_operator_payout * 0.10 AS DECIMAL(38,0))
+        cast(total_staker_operator_payout * COALESCE(oas.split, 1000) / 10000.0 AS DECIMAL(38,0))
       WHEN snapshot < @nileHardforkDate AND reward_submission_date < @nileHardforkDate THEN
-        (total_staker_operator_payout * 0.10)::text::decimal(38,0)
+        (total_staker_operator_payout * COALESCE(oas.split, 1000) / 10000.0)::text::decimal(38,0)
       ELSE
-        floor(total_staker_operator_payout * 0.10)
+        floor(total_staker_operator_payout * COALESCE(oas.split, 1000) / 10000.0)
     END as operator_tokens,
     CASE
       WHEN snapshot < @amazonHardforkDate AND reward_submission_date < @amazonHardforkDate THEN
-        total_staker_operator_payout - cast(total_staker_operator_payout * 0.10 as DECIMAL(38,0))
+        total_staker_operator_payout - cast(total_staker_operator_payout * COALESCE(oas.split, 1000) / 10000.0 AS DECIMAL(38, 0))
       WHEN snapshot < @nileHardforkDate AND reward_submission_date < @nileHardforkDate THEN
-        total_staker_operator_payout - ((total_staker_operator_payout * 0.10)::text::decimal(38,0))
+        total_staker_operator_payout - ((total_staker_operator_payout * COALESCE(oas.split, 1000) / 10000.0)::text::decimal(38, 0))
       ELSE
-        total_staker_operator_payout - floor(total_staker_operator_payout * 0.10)
+        total_staker_operator_payout - floor(total_staker_operator_payout * COALESCE(oas.split, 1000) / 10000.0)
     END as staker_tokens
-  FROM staker_operator_total_tokens
+  FROM staker_operator_total_tokens sot
+  LEFT JOIN operator_avs_splits_cte oas
+  ON sot.operator = oas.operator AND sot.avs = oas.avs AND sot.snapshot = oas.snapshot
 )
 SELECT * from token_breakdowns
 ORDER BY reward_hash, snapshot, staker, operator

pkg/rewards/5_goldRfaeStakers.go

@@ -2,6 +2,7 @@ package rewards
 
 import (
 	"database/sql"
+
 	"github.com/Layr-Labs/sidecar/internal/config"
 	"go.uber.org/zap"
 )
@@ -106,12 +107,23 @@ staker_operator_total_tokens AS (
     FLOOR(staker_proportion * tokens_per_day_decimal) as total_staker_operator_payout
   FROM staker_proportion
 ),
--- Calculate the token breakdown for each (staker, operator) pair
+-- Include the operator_pi_split_snapshots table
+operator_pi_splits_cte AS (
+  SELECT
+    operator,
+    snapshot,
+    split
+  FROM operator_pi_split_snapshots
+),
+-- Calculate the token breakdown for each (staker, operator) pair with dynamic split logic
+-- If no split is found, default to 1000 (10%)
 token_breakdowns AS (
-  SELECT *,
-    floor(total_staker_operator_payout * 0.10) as operator_tokens,
-    total_staker_operator_payout - floor(total_staker_operator_payout * 0.10) as staker_tokens
-  FROM staker_operator_total_tokens
+  SELECT sot.*,
+    floor(total_staker_operator_payout * COALESCE(ops.split, 1000) / 10000.0) as operator_tokens,
+    total_staker_operator_payout - floor(total_staker_operator_payout * COALESCE(ops.split, 1000) / 10000.0) as staker_tokens
+  FROM staker_operator_total_tokens sot 
+  LEFT JOIN operator_pi_splits_cte ops
+  ON sot.operator = ops.operator AND sot.snapshot = ops.snapshot
 )
 SELECT * from token_breakdowns
 ORDER BY reward_hash, snapshot, staker, operator

pkg/rewards/operatorAvsSplitSnapshots.go

@@ -0,0 +1,92 @@
+package rewards
+
+const operatorAvsSplitSnapshotQuery = `
+WITH operator_avs_splits_with_block_info as (
+	select
+		oas.operator,
+		oas.avs,
+		oas.activated_at::timestamp(6) as activated_at,
+		oas.new_operator_avs_split_bips as split,
+		oas.block_number,
+		oas.log_index,
+		b.block_time::timestamp(6) as block_time
+	from operator_avs_splits as oas
+	left join blocks as b on (b.number = oas.block_number)
+	where activated_at < TIMESTAMP '{{.cutoffDate}}'
+),
+-- Rank the records for each combination of (operator, avs, activation date) by activation time, block time and log index
+ranked_operator_avs_split_records as (
+	SELECT *,
+		   ROW_NUMBER() OVER (PARTITION BY operator, avs, cast(activated_at AS DATE) ORDER BY activated_at DESC, block_time DESC, log_index DESC) AS rn
+	FROM operator_avs_splits_with_block_info
+),
+-- Get the latest record for each day & round up to the snapshot day
+snapshotted_records as (
+ SELECT
+	 operator,
+	 avs,
+	 split,
+	 block_time,
+	 date_trunc('day', activated_at) + INTERVAL '1' day AS snapshot_time
+ from ranked_operator_avs_split_records
+ where rn = 1
+),
+-- Get the range for each operator, avs pairing
+operator_avs_split_windows as (
+ SELECT
+	 operator, avs, split, snapshot_time as start_time,
+	 CASE
+		 -- If the range does not have the end, use the current timestamp truncated to 0 UTC
+		 WHEN LEAD(snapshot_time) OVER (PARTITION BY operator, avs ORDER BY snapshot_time) is null THEN date_trunc('day', TIMESTAMP '{{.cutoffDate}}')
+		 ELSE LEAD(snapshot_time) OVER (PARTITION BY operator, avs ORDER BY snapshot_time)
+		 END AS end_time
+ FROM snapshotted_records
+),
+-- Clean up any records where start_time >= end_time
+cleaned_records as (
+  SELECT * FROM operator_avs_split_windows
+  WHERE start_time < end_time
+),
+-- Generate a snapshot for each day in the range
+final_results as (
+	SELECT
+		operator,
+		avs,
+		split,
+		d AS snapshot
+	FROM
+		cleaned_records
+			CROSS JOIN
+		generate_series(DATE(start_time), DATE(end_time) - interval '1' day, interval '1' day) AS d
+)
+select * from final_results
+`
+
+func (r *RewardsCalculator) GenerateAndInsertOperatorAvsSplitSnapshots(snapshotDate string) error {
+	tableName := "operator_avs_split_snapshots"
+
+	query, err := renderQueryTemplate(operatorAvsSplitSnapshotQuery, map[string]string{
+		"cutoffDate": snapshotDate,
+	})
+	if err != nil {
+		r.logger.Sugar().Errorw("Failed to render query template", "error", err)
+		return err
+	}
+
+	err = r.generateAndInsertFromQuery(tableName, query, nil)
+	if err != nil {
+		r.logger.Sugar().Errorw("Failed to generate operator_avs_split_snapshots", "error", err)
+		return err
+	}
+	return nil
+}
+
+func (r *RewardsCalculator) ListOperatorAvsSplitSnapshots() ([]*OperatorAVSSplitSnapshots, error) {
+	var snapshots []*OperatorAVSSplitSnapshots
+	res := r.grm.Model(&OperatorAVSSplitSnapshots{}).Find(&snapshots)
+	if res.Error != nil {
+		r.logger.Sugar().Errorw("Failed to list operator avs split snapshots", "error", res.Error)
+		return nil, res.Error
+	}
+	return snapshots, nil
+}

pkg/rewards/operatorPISplitSnapshots.go

@@ -0,0 +1,89 @@
+package rewards
+
+const operatorPISplitSnapshotQuery = `
+WITH operator_pi_splits_with_block_info as (
+	select
+		ops.operator,
+		ops.activated_at::timestamp(6) as activated_at,
+		ops.new_operator_avs_split_bips as split,
+		ops.block_number,
+		ops.log_index,
+		b.block_time::timestamp(6) as block_time
+	from operator_pi_splits as ops
+	left join blocks as b on (b.number = ops.block_number)
+	where activated_at < TIMESTAMP '{{.cutoffDate}}'
+),
+-- Rank the records for each combination of (operator, activation date) by activation time, block time and log index
+ranked_operator_pi_split_records as (
+	SELECT *,
+		   ROW_NUMBER() OVER (PARTITION BY operator, cast(activated_at AS DATE) ORDER BY activated_at DESC, block_time DESC, log_index DESC) AS rn
+	FROM operator_pi_splits_with_block_info
+),
+-- Get the latest record for each day & round up to the snapshot day
+snapshotted_records as (
+ SELECT
+	 operator,
+	 split,
+	 block_time,
+	 date_trunc('day', activated_at) + INTERVAL '1' day AS snapshot_time
+ from ranked_operator_pi_split_records
+ where rn = 1
+),
+-- Get the range for each operator
+operator_pi_split_windows as (
+ SELECT
+	 operator, split, snapshot_time as start_time,
+	 CASE
+		 -- If the range does not have the end, use the current timestamp truncated to 0 UTC
+		 WHEN LEAD(snapshot_time) OVER (PARTITION BY operator ORDER BY snapshot_time) is null THEN date_trunc('day', TIMESTAMP '{{.cutoffDate}}')
+		 ELSE LEAD(snapshot_time) OVER (PARTITION BY operator ORDER BY snapshot_time)
+		 END AS end_time
+ FROM snapshotted_records
+),
+-- Clean up any records where start_time >= end_time
+cleaned_records as (
+  SELECT * FROM operator_pi_split_windows
+  WHERE start_time < end_time
+),
+-- Generate a snapshot for each day in the range
+final_results as (
+	SELECT
+		operator,
+		split,
+		d AS snapshot
+	FROM
+		cleaned_records
+			CROSS JOIN
+		generate_series(DATE(start_time), DATE(end_time) - interval '1' day, interval '1' day) AS d
+)
+select * from final_results
+`
+
+func (r *RewardsCalculator) GenerateAndInsertOperatorPISplitSnapshots(snapshotDate string) error {
+	tableName := "operator_pi_split_snapshots"
+
+	query, err := renderQueryTemplate(operatorPISplitSnapshotQuery, map[string]string{
+		"cutoffDate": snapshotDate,
+	})
+	if err != nil {
+		r.logger.Sugar().Errorw("Failed to render query template", "error", err)
+		return err
+	}
+
+	err = r.generateAndInsertFromQuery(tableName, query, nil)
+	if err != nil {
+		r.logger.Sugar().Errorw("Failed to generate operator_pi_split_snapshots", "error", err)
+		return err
+	}
+	return nil
+}
+
+func (r *RewardsCalculator) ListOperatorPISplitSnapshots() ([]*OperatorPISplitSnapshots, error) {
+	var snapshots []*OperatorPISplitSnapshots
+	res := r.grm.Model(&OperatorPISplitSnapshots{}).Find(&snapshots)
+	if res.Error != nil {
+		r.logger.Sugar().Errorw("Failed to list operator pi split snapshots", "error", res.Error)
+		return nil, res.Error
+	}
+	return snapshots, nil
+}

pkg/rewards/rewards.go

@@ -4,19 +4,21 @@ import (
 	"bytes"
 	"errors"
 	"fmt"
+	"time"
+
 	"github.com/Layr-Labs/eigenlayer-rewards-proofs/pkg/distribution"
 	"github.com/Layr-Labs/sidecar/pkg/postgres/helpers"
 	"github.com/Layr-Labs/sidecar/pkg/storage"
 	"github.com/Layr-Labs/sidecar/pkg/utils"
 	gethcommon "github.com/ethereum/go-ethereum/common"
 	"github.com/wealdtech/go-merkletree/v2"
 	"gorm.io/gorm/clause"
-	"time"
+
+	"text/template"
 
 	"github.com/Layr-Labs/sidecar/internal/config"
 	"go.uber.org/zap"
 	"gorm.io/gorm"
-	"text/template"
 )
 
 type RewardsCalculator struct {
@@ -291,6 +293,18 @@ func (rc *RewardsCalculator) generateSnapshotData(snapshotDate string) error {
 	}
 	rc.logger.Sugar().Debugw("Generated staker delegation snapshots")
 
+	if err = rc.GenerateAndInsertOperatorAvsSplitSnapshots(snapshotDate); err != nil {
+		rc.logger.Sugar().Errorw("Failed to generate operator avs split snapshots", "error", err)
+		return err
+	}
+	rc.logger.Sugar().Debugw("Generated operator avs split snapshots")
+
+	if err = rc.GenerateAndInsertOperatorPISplitSnapshots(snapshotDate); err != nil {
+		rc.logger.Sugar().Errorw("Failed to generate operator pi snapshots", "error", err)
+		return err
+	}
+	rc.logger.Sugar().Debugw("Generated operator pi snapshots")
+
 	return nil
 }