feat: remove queue from the service

cmd/staking-expiry-checker/main.go

@@ -13,7 +13,6 @@ import (
 	"github.com/babylonlabs-io/staking-expiry-checker/internal/db"
 	"github.com/babylonlabs-io/staking-expiry-checker/internal/observability/metrics"
 	"github.com/babylonlabs-io/staking-expiry-checker/internal/poller"
-	"github.com/babylonlabs-io/staking-expiry-checker/internal/queue"
 	"github.com/babylonlabs-io/staking-expiry-checker/internal/services"
 )
 
@@ -53,12 +52,7 @@ func main() {
 		log.Fatal().Err(err).Msg("error while creating btc client")
 	}
 
-	qm, err := queue.NewQueueManager(&cfg.Queue)
-	if err != nil {
-		log.Fatal().Err(err).Msg("error while creating queue manager")
-	}
-
-	delegationService := services.NewService(dbClient, btcClient, qm)
+	delegationService := services.NewService(dbClient, btcClient)
 	if err != nil {
 		log.Fatal().Err(err).Msg("error while creating delegation service")
 	}

config/config-docker.yml

@@ -12,14 +12,6 @@ btc:
   net-params: testnet
   rpc-user: rpcuser
   rpc-pass: rpcpass
-queue:
-  queue_user: user # can be replaced by values in .env file
-  queue_password: password
-  url: "localhost:5672"
-  processing_timeout: 5 # 5 second
-  msg_max_retry_attempts: 10
-  requeue_delay_time: 300
-  queue_type: quorum
 metrics:
   host: 0.0.0.0
   port: 2112

config/config-local.yml

@@ -12,14 +12,6 @@ btc:
   net-params: testnet
   rpc-user: rpcuser
   rpc-pass: rpcpass
-queue:
-  queue_user: user # can be replaced by values in .env file
-  queue_password: password
-  url: "localhost:5672"
-  processing_timeout: 5 # 5 second
-  msg_max_retry_attempts: 3
-  requeue_delay_time: 60
-  queue_type: quorum
 metrics:
   host: 0.0.0.0
   port: 2112

contrib/images/staking-expiry-checker/Dockerfile

@@ -1,4 +1,4 @@
-FROM golang:1.21-alpine AS builder
+FROM golang:1.23.1-alpine AS builder
 
 ARG VERSION="HEAD"
 

go.mod

@@ -1,11 +1,9 @@
 module github.com/babylonlabs-io/staking-expiry-checker
 
-go 1.21.6
+go 1.23.1
 
 require (
-	github.com/babylonlabs-io/staking-queue-client v0.4.1
 	github.com/btcsuite/btcd v0.24.0
-	github.com/rabbitmq/amqp091-go v1.9.0
 	github.com/spf13/viper v1.18.2
 )
 

internal/config/config.go

@@ -5,16 +5,14 @@ import (
 	"os"
 	"strings"
 
-	queue "github.com/babylonlabs-io/staking-queue-client/config"
 	"github.com/spf13/viper"
 )
 
 type Config struct {
-	Poller  PollerConfig      `mapstructure:"poller"`
-	Db      DbConfig          `mapstructure:"db"`
-	Btc     BtcConfig         `mapstructure:"btc"`
-	Queue   queue.QueueConfig `mapstructure:"queue"`
-	Metrics MetricsConfig     `mapstructure:"metrics"`
+	Poller  PollerConfig  `mapstructure:"poller"`
+	Db      DbConfig      `mapstructure:"db"`
+	Btc     BtcConfig     `mapstructure:"btc"`
+	Metrics MetricsConfig `mapstructure:"metrics"`
 }
 
 func (cfg *Config) Validate() error {
@@ -34,10 +32,6 @@ func (cfg *Config) Validate() error {
 		return err
 	}
 
-	if err := cfg.Queue.Validate(); err != nil {
-		return err
-	}
-
 	return nil
 }
 

internal/observability/metrics/metrics.go

@@ -32,7 +32,6 @@ var (
 	metricsRouter              *chi.Mux
 	pollDurationHistogram      *prometheus.HistogramVec
 	btcClientDurationHistogram *prometheus.HistogramVec
-	queueSendErrorCounter      prometheus.Counter
 )
 
 // Init initializes the metrics package.
@@ -89,18 +88,9 @@ func registerMetrics() {
 		[]string{"function", "status"},
 	)
 
-	// add a counter for the number of errors from the fail to push message into queue
-	queueSendErrorCounter = prometheus.NewCounter(
-		prometheus.CounterOpts{
-			Name: "queue_send_error_count",
-			Help: "The total number of errors when sending messages to the queue",
-		},
-	)
-
 	prometheus.MustRegister(
 		pollDurationHistogram,
 		btcClientDurationHistogram,
-		queueSendErrorCounter,
 	)
 }
 
@@ -126,7 +116,3 @@ func RecordBtcClientMetrics[T any](clientRequest func() (T, error)) (T, error) {
 
 	return result, err
 }
-
-func RecordQueueSendError() {
-	queueSendErrorCounter.Inc()
-}

internal/services/service.go

@@ -5,22 +5,18 @@ import (
 
 	"github.com/babylonlabs-io/staking-expiry-checker/internal/btcclient"
 	"github.com/babylonlabs-io/staking-expiry-checker/internal/db"
-	"github.com/babylonlabs-io/staking-expiry-checker/internal/queue"
-	queueclient "github.com/babylonlabs-io/staking-queue-client/client"
 	"github.com/rs/zerolog/log"
 )
 
 type Service struct {
-	db           db.DbInterface
-	btc          btcclient.BtcInterface
-	queueManager *queue.QueueManager
+	db  db.DbInterface
+	btc btcclient.BtcInterface
 }
 
-func NewService(db db.DbInterface, btc btcclient.BtcInterface, qm *queue.QueueManager) *Service {
+func NewService(db db.DbInterface, btc btcclient.BtcInterface) *Service {
 	return &Service{
-		db:           db,
-		btc:          btc,
-		queueManager: qm,
+		db:  db,
+		btc: btc,
 	}
 }
 
@@ -43,11 +39,8 @@ func (s *Service) ProcessExpiredDelegations(ctx context.Context) error {
 		}
 
 		for _, delegation := range expiredDelegations {
-			ev := queueclient.NewExpiredStakingEvent(delegation.StakingTxHashHex, delegation.TxType)
-			if err := s.queueManager.SendExpiredStakingEvent(ctx, ev); err != nil {
-				log.Error().Err(err).Msg("Error sending expired staking event")
-				return err
-			}
+			// TODO: Process the expired delegation.
+			log.Info().Msgf("Found a expired delegation, do nothing now: %v", delegation.ID)
 			// After successfully sending the event, delete the entry from the database.
 			if err := s.db.DeleteExpiredDelegation(ctx, delegation.ID); err != nil {
 				log.Error().Err(err).Msg("Error deleting expired delegation")

tests/config-test.yml

@@ -12,14 +12,6 @@ btc:
   net-params: testnet
   rpc-user: rpcuser
   rpc-pass: rpcpass
-queue:
-  queue_user: user # can be replaced by values in .env file
-  queue_password: password
-  url: "localhost:5672"
-  processing_timeout: 60 # 5 second
-  msg_max_retry_attempts: 2
-  requeue_delay_time: 5
-  queue_type: quorum
 metrics:
   host: 0.0.0.0
   port: 2113