fix(staking): stake amount history (#46)

* fix(staking): stake amount history (#45)

* fix(staking): stake amount history api

* fix(staking): stake amount history api

* fix(staking): stake amount history api

* fix(staking): stake amount history api

Author: gezerui0124

db/cl_staking_event.go

@@ -1,6 +1,11 @@
 package db
 
-import "gorm.io/gorm"
+import (
+	"errors"
+	"time"
+
+	"gorm.io/gorm"
+)
 
 type CLStakingEvent struct {
 	ID          uint64 `gorm:"primarykey"`
@@ -12,6 +17,11 @@ type CLStakingEvent struct {
 	Amount      string `gorm:"not null;column:amount"`
 }
 
+type CLSuccessfulStakingEvent struct {
+	CLStakingEvent
+	BlockTime time.Time `gorm:"not null;column:block_time"`
+}
+
 func (CLStakingEvent) TableName() string {
 	return "cl_staking_events"
 }
@@ -25,3 +35,22 @@ func BatchCreateCLStakingEvents(db *gorm.DB, indexer string, events []*CLStaking
 		return UpdateIndexPoint(tx, indexer, height)
 	})
 }
+
+func GetSuccessfulCLStakingEventsAfter(db *gorm.DB, eventTypes []string, blockHeight int64) ([]*CLSuccessfulStakingEvent, error) {
+	var events []*CLSuccessfulStakingEvent
+
+	if err := db.
+		Table("cl_staking_events AS e").
+		Joins("JOIN cl_blocks AS b ON e.block_height = b.height").
+		Select("e.*, b.time AS block_time").
+		Where("e.event_type IN (?)", eventTypes).
+		Where("e.status_ok = ?", true).
+		Where("e.block_height > ?", blockHeight).
+		Scan(&events).Error; errors.Is(err, gorm.ErrRecordNotFound) {
+		return events, nil
+	} else if err != nil {
+		return nil, err
+	}
+
+	return events, nil
+}

db/cl_total_stake_hist.go

@@ -0,0 +1,86 @@
+package db
+
+import (
+	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
+)
+
+type CLTotalStakeHist struct {
+	ID uint64 `gorm:"primarykey"`
+
+	TotalStakeAmount int64 `gorm:"not null;column:total_stake_amount"`
+	UpdatedAtBlock   int64 `gorm:"not null;column:updated_at_block;index:idx_cl_total_stake_hist_updated_at_block,unique"`
+	UpdatedAtTime    int64 `gorm:"not null;column:updated_at_time;index:idx_cl_total_stake_hist_updated_at_time,unique"`
+}
+
+func (CLTotalStakeHist) TableName() string {
+	return "cl_total_stake_hists"
+}
+
+func GetLatestCLTotalStakeHist(db *gorm.DB) (*CLTotalStakeHist, error) {
+	var stake CLTotalStakeHist
+
+	if err := db.Order("updated_at_time DESC").First(&stake).Error; err != nil {
+		return nil, err
+	}
+
+	return &stake, nil
+}
+
+func GetCLTotalStakeHists(db *gorm.DB) ([]*CLTotalStakeHist, error) {
+	var stakes []*CLTotalStakeHist
+
+	if err := db.Order("updated_at_time ASC").Find(&stakes).Error; err != nil {
+		return nil, err
+	}
+
+	return stakes, nil
+}
+
+func GetLatestCLTotalStakeHistBefore(db *gorm.DB, timestamp int64) (*CLTotalStakeHist, error) {
+	var stake CLTotalStakeHist
+
+	if err := db.Where("updated_at_time <= ?", timestamp).Order("updated_at_time DESC").First(&stake).Error; err != nil {
+		return nil, err
+	}
+
+	return &stake, nil
+}
+
+func GetCLTotalStakeHistsAfter(db *gorm.DB, timestamp int64) ([]*CLTotalStakeHist, error) {
+	var stakes []*CLTotalStakeHist
+
+	if err := db.Where("updated_at_time > ?", timestamp).Order("updated_at_time ASC").Find(&stakes).Error; err != nil {
+		return nil, err
+	}
+
+	return stakes, nil
+}
+
+func UpsertCLGenesisTotalStakeHist(db *gorm.DB, indexer string, stake *CLTotalStakeHist) error {
+	return db.Transaction(func(tx *gorm.DB) error {
+		if err := tx.Clauses(clause.OnConflict{
+			Columns:   []clause.Column{{Name: "updated_at_time"}},
+			DoNothing: true,
+		}).Create(stake).Error; err != nil {
+			return err
+		}
+
+		return nil
+	})
+}
+
+func BatchUpsertCLTotalStakeHists(db *gorm.DB, indexer string, stakes []*CLTotalStakeHist) error {
+	return db.Transaction(func(tx *gorm.DB) error {
+		if err := tx.Clauses(clause.OnConflict{
+			Columns: []clause.Column{{Name: "updated_at_time"}},
+			DoUpdates: clause.Assignments(map[string]interface{}{
+				"total_stake_amount": gorm.Expr("excluded.total_stake_amount"),
+			}),
+		}).CreateInBatches(stakes, 100).Error; err != nil {
+			return err
+		}
+
+		return nil
+	})
+}

pkg/indexer/cl_staking_event.go

@@ -203,7 +203,7 @@ func (c *CLStakingEventIndexer) getStakingEvents(from, to int64) ([]*db.CLStakin
 					return nil, fmt.Errorf("event %s: sender address not found", eventType)
 				}
 
-				if strings.ToLower(delAddr) != strings.ToLower(senderAddr) {
+				if !strings.EqualFold(delAddr, senderAddr) {
 					eventType = EventType2Behalf[eventType]
 				}
 			case TypeUnjail:
@@ -225,7 +225,7 @@ func (c *CLStakingEventIndexer) getStakingEvents(from, to int64) ([]*db.CLStakin
 					return nil, fmt.Errorf("convert validator compressed key to address from event %s failed: %w", eventType, err)
 				}
 
-				if strings.ToLower(valAddr.String()) != strings.ToLower(senderAddr) {
+				if !strings.EqualFold(valAddr.String(), senderAddr) {
 					eventType = EventType2Behalf[eventType]
 				}
 			}

pkg/indexer/cl_total_stake.go

@@ -21,8 +21,6 @@ const (
 	genesisTime = "2025-01-19T15:00:00.00000000Z"
 	timeLayout  = "2006-01-02T15:04:05.999999999Z"
 
-	clUnbondingPeriod = 1209600 * time.Second // 14 days
-
 	stakeEventType   = "delegate_success"
 	unstakeEventType = "undelegate_success"
 )

pkg/indexer/cl_total_stake_hist.go

@@ -0,0 +1,149 @@
+package indexer
+
+import (
+	"context"
+	"fmt"
+	"sort"
+	"strconv"
+	"time"
+
+	"github.com/rs/zerolog/log"
+	"gorm.io/gorm"
+
+	"github.com/piplabs/story-staking-api/db"
+)
+
+const (
+	genesisBlockHeight = 1
+)
+
+var _ Indexer = (*CLTotalStakeHistIndexer)(nil)
+
+type CLTotalStakeHistIndexer struct {
+	ctx context.Context
+
+	dbOperator *gorm.DB
+}
+
+func NewCLTotalStakeHistIndexer(ctx context.Context, dbOperator *gorm.DB) (*CLTotalStakeHistIndexer, error) {
+	c := &CLTotalStakeHistIndexer{
+		ctx: ctx,
+
+		dbOperator: dbOperator,
+	}
+
+	if err := c.init(); err != nil {
+		return nil, fmt.Errorf("init indexer %s failed: %w", c.Name(), err)
+	}
+
+	return c, nil
+}
+
+func (c *CLTotalStakeHistIndexer) Name() string {
+	return "cl_total_stake_hist"
+}
+
+func (c *CLTotalStakeHistIndexer) Run() {
+	log.Info().Str("indexer", c.Name()).Msg("Start indexing")
+
+	ticker := time.NewTicker(10 * time.Second)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-c.ctx.Done():
+			return
+		case <-ticker.C:
+			if err := c.index(); err != nil {
+				log.Error().Err(err).Str("indexer", c.Name()).Msg("index cl total stake hist failed")
+			}
+		}
+	}
+}
+
+func (c *CLTotalStakeHistIndexer) init() error {
+	gt, err := time.Parse(timeLayout, genesisTime)
+	if err != nil {
+		return fmt.Errorf("parse genesis time failed: %w", err)
+	}
+
+	if err := db.UpsertCLGenesisTotalStakeHist(c.dbOperator, c.Name(), &db.CLTotalStakeHist{
+		TotalStakeAmount: genesisStakeAmount,
+		UpdatedAtBlock:   genesisBlockHeight,
+		UpdatedAtTime:    gt.Unix(),
+	}); err != nil {
+		return fmt.Errorf("upsert genesis cl total stake failed: %w", err)
+	}
+
+	return nil
+}
+
+func (c *CLTotalStakeHistIndexer) index() error {
+	latestHist, err := db.GetLatestCLTotalStakeHist(c.dbOperator)
+	if err != nil {
+		return fmt.Errorf("get latest cl total stake hist failed: %w", err)
+	}
+
+	events, err := db.GetSuccessfulCLStakingEventsAfter(c.dbOperator, []string{TypeStake, TypeUnstake}, latestHist.UpdatedAtBlock)
+	if err != nil {
+		return fmt.Errorf("get successful cl staking events failed: %w", err)
+	} else if len(events) == 0 {
+		return nil
+	}
+
+	blk2StakeChange, blk2BlockTime := make(map[int64]int64), make(map[int64]int64)
+	for _, event := range events {
+		var amount int64
+		if event.Amount == "" {
+			// special bypass for aeneid
+			amount = 0
+		} else {
+			amt, err := strconv.ParseInt(event.Amount, 10, 64)
+			if err != nil {
+				return fmt.Errorf("parse amount %s failed: %w", event.Amount, err)
+			}
+			amount = amt
+		}
+
+		switch event.EventType {
+		case TypeStake:
+			blk2StakeChange[event.BlockHeight] += amount
+		case TypeUnstake:
+			blk2StakeChange[event.BlockHeight] -= amount
+		}
+
+		blk2BlockTime[event.BlockHeight] = event.BlockTime.Unix()
+	}
+
+	type stakeChange struct {
+		BlockHeight       int64
+		BlockTime         int64
+		StakeChangeAmount int64
+	}
+
+	scs := make([]*stakeChange, 0)
+	for blkno, amt := range blk2StakeChange {
+		scs = append(scs, &stakeChange{
+			BlockHeight:       blkno,
+			BlockTime:         blk2BlockTime[blkno],
+			StakeChangeAmount: amt,
+		})
+	}
+
+	sort.Slice(scs, func(i, j int) bool {
+		return scs[i].BlockHeight < scs[j].BlockHeight
+	})
+
+	totalStakeAmount := latestHist.TotalStakeAmount
+	clTotalStakeHists := make([]*db.CLTotalStakeHist, 0)
+	for _, sc := range scs {
+		totalStakeAmount += sc.StakeChangeAmount
+		clTotalStakeHists = append(clTotalStakeHists, &db.CLTotalStakeHist{
+			TotalStakeAmount: totalStakeAmount,
+			UpdatedAtBlock:   sc.BlockHeight,
+			UpdatedAtTime:    sc.BlockTime,
+		})
+	}
+
+	return db.BatchUpsertCLTotalStakeHists(c.dbOperator, c.Name(), clTotalStakeHists)
+}