feat: sync global params

cmd/babylon-staking-indexer/main.go

@@ -55,7 +55,7 @@ func main() {
 		log.Fatal().Err(err).Msg("error while creating queue manager")
 	}
 
-	service := services.NewService(dbClient, btcClient, bbnClient, qm)
+	service := services.NewService(cfg, dbClient, btcClient, bbnClient, qm)
 	if err != nil {
 		log.Fatal().Err(err).Msg("error while creating delegation service")
 	}

config/config-local.yml

@@ -1,6 +1,3 @@
-poller:
-  interval: 5s
-  log-level: debug
 db:
   username: root
   password: example
@@ -15,6 +12,8 @@ btc:
 bbn:
   rpc-addr: https://rpc.devnet.babylonchain.io:443
   timeout: 30s
+poller:
+  param-polling-interval: 10s
 queue:
   queue_user: user # can be replaced by values in .env file
   queue_password: password

internal/clients/bbnclient/bbnclient.go

@@ -2,14 +2,16 @@ package bbnclient
 
 import (
 	"context"
+	"fmt"
 	"net/http"
-
-	ctypes "github.com/cometbft/cometbft/rpc/core/types"
-	"github.com/rs/zerolog/log"
+	"strings"
 
 	"github.com/babylonlabs-io/babylon-staking-indexer/internal/types"
 	"github.com/babylonlabs-io/babylon/client/config"
 	"github.com/babylonlabs-io/babylon/client/query"
+	bbntypes "github.com/babylonlabs-io/babylon/x/btcstaking/types"
+	ctypes "github.com/cometbft/cometbft/rpc/core/types"
+	"github.com/rs/zerolog/log"
 )
 
 type BbnClient struct {
@@ -28,7 +30,9 @@ func (c *BbnClient) GetLatestBlockNumber(ctx context.Context) (int64, *types.Err
 	status, err := c.queryClient.RPCClient.Status(ctx)
 	if err != nil {
 		return 0, types.NewErrorWithMsg(
-			http.StatusInternalServerError, types.InternalServiceError, err.Error(),
+			http.StatusInternalServerError,
+			types.InternalServiceError,
+			fmt.Sprintf("failed to get latest block number by fetching status: %s", err.Error()),
 		)
 	}
 	return status.SyncInfo.LatestBlockHeight, nil
@@ -38,13 +42,57 @@ func (c *BbnClient) GetBlockResults(ctx context.Context, blockHeight int64) (*ct
 	resp, err := c.queryClient.RPCClient.BlockResults(ctx, &blockHeight)
 	if err != nil {
 		return nil, types.NewErrorWithMsg(
-			http.StatusInternalServerError, types.InternalServiceError, err.Error(),
+			http.StatusInternalServerError,
+			types.InternalServiceError,
+			fmt.Sprintf("failed to get block results for block %d: %s", blockHeight, err.Error()),
 		)
 	}
-
 	return resp, nil
 }
 
+func (c *BbnClient) GetCheckpointParams(ctx context.Context) (*CheckpointParams, *types.Error) {
+	params, err := c.queryClient.BTCCheckpointParams()
+	if err != nil {
+		return nil, types.NewErrorWithMsg(
+			http.StatusInternalServerError,
+			types.InternalServiceError,
+			fmt.Sprintf("failed to get checkpoint params: %s", err.Error()),
+		)
+	}
+	return &params.Params, nil
+}
+
+func (c *BbnClient) GetAllStakingParams(ctx context.Context) (map[uint32]StakingParams, *types.Error) {
+	allParams := make(map[uint32]StakingParams) // Map to store versioned staking parameters
+	version := uint32(0)
+
+	for {
+		params, err := c.queryClient.BTCStakingParamsByVersion(version)
+		if err != nil {
+			if strings.Contains(err.Error(), bbntypes.ErrParamsNotFound.Error()) {
+				// Break the loop if an error occurs (indicating no more versions)
+				break
+			}
+			return nil, types.NewErrorWithMsg(
+				http.StatusInternalServerError,
+				types.InternalServiceError,
+				fmt.Sprintf("failed to get staking params for version %d: %s", version, err.Error()),
+			)
+		}
+		allParams[version] = *FromBbnStakingParams(params.Params)
+		version++
+	}
+	if len(allParams) == 0 {
+		return nil, types.NewErrorWithMsg(
+			http.StatusNotFound,
+			types.NotFound,
+			"no staking params found",
+		)
+	}
+
+	return allParams, nil
+}
+
 func (c *BbnClient) getBlockResults(ctx context.Context, blockHeight *int64) (*ctypes.ResultBlockResults, *types.Error) {
 	resp, err := c.queryClient.RPCClient.BlockResults(ctx, blockHeight)
 	if err != nil {

internal/clients/bbnclient/interface.go

@@ -8,6 +8,8 @@ import (
 )
 
 type BbnInterface interface {
+	GetCheckpointParams(ctx context.Context) (*CheckpointParams, *types.Error)
+	GetAllStakingParams(ctx context.Context) (map[uint32]StakingParams, *types.Error)
 	GetLatestBlockNumber(ctx context.Context) (int64, *types.Error)
 	GetBlockResults(
 		ctx context.Context, blockHeight int64,

internal/clients/bbnclient/types.go

@@ -0,0 +1,48 @@
+package bbnclient
+
+import (
+	"encoding/hex"
+
+	checkpointtypes "github.com/babylonlabs-io/babylon/x/btccheckpoint/types"
+	stakingtypes "github.com/babylonlabs-io/babylon/x/btcstaking/types"
+)
+
+// StakingParams represents the staking parameters of the BBN chain
+// Reference: https://github.com/babylonlabs-io/babylon/blob/main/proto/babylon/btcstaking/v1/params.proto
+type StakingParams struct {
+	CovenantPks                  []string
+	CovenantQuorum               uint32
+	MinStakingValueSat           int64
+	MaxStakingValueSat           int64
+	MinStakingTimeBlocks         uint32
+	MaxStakingTimeBlocks         uint32
+	SlashingPkScript             string
+	MinSlashingTxFeeSat          int64
+	SlashingRate                 string
+	MinUnbondingTimeBlocks       uint32
+	UnbondingFeeSat              int64
+	MinCommissionRate            string
+	MaxActiveFinalityProviders   uint32
+	DelegationCreationBaseGasFee uint64
+}
+
+func FromBbnStakingParams(params stakingtypes.Params) *StakingParams {
+	return &StakingParams{
+		CovenantPks:                  params.CovenantPksHex(),
+		CovenantQuorum:               params.CovenantQuorum,
+		MinStakingValueSat:           params.MinStakingValueSat,
+		MaxStakingValueSat:           params.MaxStakingValueSat,
+		MinStakingTimeBlocks:         params.MinStakingTimeBlocks,
+		MaxStakingTimeBlocks:         params.MaxStakingTimeBlocks,
+		SlashingPkScript:             hex.EncodeToString(params.SlashingPkScript),
+		MinSlashingTxFeeSat:          params.MinSlashingTxFeeSat,
+		SlashingRate:                 params.SlashingRate.String(),
+		MinUnbondingTimeBlocks:       params.MinUnbondingTimeBlocks,
+		UnbondingFeeSat:              params.UnbondingFeeSat,
+		MinCommissionRate:            params.MinCommissionRate.String(),
+		MaxActiveFinalityProviders:   params.MaxActiveFinalityProviders,
+		DelegationCreationBaseGasFee: params.DelegationCreationBaseGasFee,
+	}
+}
+
+type CheckpointParams = checkpointtypes.Params

internal/config/config.go

@@ -14,6 +14,7 @@ type Config struct {
 	Db      DbConfig                     `mapstructure:"db"`
 	Btc     BtcConfig                    `mapstructure:"btc"`
 	Bbn     bbnconfig.BabylonQueryConfig `mapstructure:"bbn"`
+	Poller  PollerConfig                 `mapstructure:"poller"`
 	Queue   queue.QueueConfig            `mapstructure:"queue"`
 	Metrics MetricsConfig                `mapstructure:"metrics"`
 }
@@ -42,6 +43,10 @@ func (cfg *Config) Validate() error {
 		return err
 	}
 
+	if err := cfg.Poller.Validate(); err != nil {
+		return err
+	}
+
 	return nil
 }
 

internal/config/poller.go

@@ -2,39 +2,17 @@ package config
 
 import (
 	"errors"
-	"fmt"
 	"time"
-
-	"github.com/rs/zerolog"
 )
 
 type PollerConfig struct {
-	Interval time.Duration `mapstructure:"interval"`
-	LogLevel string        `mapstructure:"log-level"`
+	ParamPollingInterval time.Duration `mapstructure:"param-polling-interval"`
 }
 
 func (cfg *PollerConfig) Validate() error {
-	if cfg.Interval < 0 {
-		return errors.New("poll interval cannot be negative")
-	}
-
-	if err := cfg.ValidateServiceLogLevel(); err != nil {
-		return err
+	if cfg.ParamPollingInterval < 0 {
+		return errors.New("param-polling-interval must be positive")
 	}
 
 	return nil
 }
-
-func (cfg *PollerConfig) ValidateServiceLogLevel() error {
-	// If log level is not set, we don't need to validate it, a default value will be used in service
-	if cfg.LogLevel == "" {
-		return nil
-	}
-
-	if parsedLevel, err := zerolog.ParseLevel(cfg.LogLevel); err != nil {
-		return fmt.Errorf("invalid log level: %w", err)
-	} else if parsedLevel < zerolog.DebugLevel || parsedLevel > zerolog.FatalLevel {
-		return fmt.Errorf("only log levels from debug to fatal are supported")
-	}
-	return nil
-}

internal/db/interface.go

@@ -7,17 +7,60 @@ import (
 )
 
 type DbInterface interface {
+	/**
+	 * Ping checks the database connection.
+	 * @param ctx The context
+	 * @return An error if the operation failed
+	 */
 	Ping(ctx context.Context) error
+	/**
+	 * SaveNewFinalityProvider saves a new finality provider to the database.
+	 * If the finality provider already exists, DuplicateKeyError will be returned.
+	 * @param ctx The context
+	 * @param fpDoc The finality provider details
+	 * @return An error if the operation failed
+	 */
 	SaveNewFinalityProvider(
 		ctx context.Context, fpDoc *model.FinalityProviderDetails,
 	) error
+	/**
+	 * UpdateFinalityProviderState updates the finality provider state.
+	 * @param ctx The context
+	 * @param btcPk The BTC public key
+	 * @param newState The new state
+	 * @return An error if the operation failed
+	 */
 	UpdateFinalityProviderState(
 		ctx context.Context, btcPk string, newState string,
 	) error
+	/**
+	 * UpdateFinalityProviderDetailsFromEvent updates the finality provider details based on the event.
+	 * Only the fields that are not empty in the event will be updated.
+	 * @param ctx The context
+	 * @param detailsToUpdate The finality provider details to update
+	 * @return An error if the operation failed
+	 */
 	UpdateFinalityProviderDetailsFromEvent(
 		ctx context.Context, detailsToUpdate *model.FinalityProviderDetails,
 	) error
+	/**
+	 * GetFinalityProviderByBtcPk retrieves the finality provider details by the BTC public key.
+	 * If the finality provider does not exist, a NotFoundError will be returned.
+	 * @param ctx The context
+	 * @param btcPk The BTC public key
+	 * @return The finality provider details or an error
+	 */
 	GetFinalityProviderByBtcPk(
 		ctx context.Context, btcPk string,
 	) (*model.FinalityProviderDetails, error)
+	/**
+	 * SaveGlobalParams saves the global parameters to the database.
+	 * If the document already exists by type and version, it will be skipped.
+	 * @param ctx The context
+	 * @param param The global parameters document
+	 * @return An error if the operation failed
+	 */
+	SaveGlobalParams(
+		ctx context.Context, param *model.GolablParamDocument,
+	) error
 }

internal/db/model/delegation.go

@@ -2,14 +2,10 @@ package model
 
 import (
 	"github.com/babylonlabs-io/babylon-staking-indexer/internal/types"
-	"go.mongodb.org/mongo-driver/bson/primitive"
 )
 
-const DelegationCollection = "delegation"
-
 type DelegationDocument struct {
-	ID               primitive.ObjectID    `bson:"_id"`
-	StakingTxHashHex string                `bson:"staking_tx_hash_hex"`
+	StakingTxHashHex string                `bson:"_id"` // Primary key
 	State            types.DelegationState `bson:"state"`
 	// TODO: Placeholder for more fields
 }

internal/db/model/params.go

@@ -0,0 +1,7 @@
+package model
+
+type GolablParamDocument struct {
+	Type    string      `bson:"type"`
+	Version uint32      `bson:"version"`
+	Params  interface{} `bson:"params"`
+}

internal/db/model/setup.go

@@ -15,6 +15,8 @@ import (
 
 const (
 	FinalityProviderDetailsCollection = "finality_provider_details"
+	DelegationCollection              = "delegation"
+	GlobalParamsCollection            = "global_params"
 )
 
 type index struct {
@@ -24,6 +26,8 @@ type index struct {
 
 var collections = map[string][]index{
 	FinalityProviderDetailsCollection: {{Indexes: map[string]int{}}},
+	DelegationCollection:              {{Indexes: map[string]int{}}},
+	GlobalParamsCollection:            {{Indexes: map[string]int{}}},
 }
 
 func Setup(ctx context.Context, cfg *config.Config) error {

internal/db/params.go

@@ -0,0 +1,32 @@
+package db
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/babylonlabs-io/babylon-staking-indexer/internal/db/model"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo/options"
+)
+
+func (db *Database) SaveGlobalParams(
+	ctx context.Context, param *model.GolablParamDocument,
+) error {
+	collection := db.client.Database(db.dbName).
+		Collection(model.GlobalParamsCollection)
+
+	filter := bson.M{
+		"type":    param.Type,
+		"version": param.Version,
+	}
+
+	update := bson.M{
+		"$setOnInsert": param, // Only insert if the document doesn't exist
+	}
+
+	_, err := collection.UpdateOne(ctx, filter, update, options.Update().SetUpsert(true))
+	if err != nil {
+		return fmt.Errorf("error while upserting global params document: %w with type %s and version %d", err, param.Type, param.Version)
+	}
+	return nil
+}

internal/services/bootstrap.go

@@ -21,7 +21,7 @@ const (
 // height and processing events. If any errors occur during the process,
 // it will retry with exponential backoff, up to a maximum of maxRetries.
 // The method runs asynchronously to allow non-blocking operation.
-func (s *Service) bootstrapBbn(ctx context.Context) {
+func (s *Service) BootstrapBbn(ctx context.Context) {
 	go func() {
 		bootstrapCtx, cancel := context.WithCancel(ctx)
 		defer cancel()

internal/services/events.go

@@ -36,7 +36,7 @@ func NewBbnEvent(category EventCategory, event abcitypes.Event) BbnEvent {
 
 // startBbnEventProcessor continuously listens for events from the channel and
 // processes them in the main thread
-func (s *Service) startBbnEventProcessor(ctx context.Context) {
+func (s *Service) StartBbnEventProcessor(ctx context.Context) {
 	for event := range s.bbnEventProcessor {
 		if event.Event.Type == "" {
 			log.Warn().Msg("Empty event received, skipping")
@@ -67,7 +67,6 @@ func (s *Service) processBbnTxEvent(ctx context.Context, event abcitypes.Event)
 		s.processNewFinalityProviderEvent(ctx, event)
 	case EventFinalityProviderEditedType:
 		s.processFinalityProviderEditedEvent(ctx, event)
-
 	}
 }