feat: remove queue from the service

bin/local-startup.sh

@@ -8,16 +8,4 @@ else
     echo "Starting MongoDB"
     # Start MongoDB
     docker compose up -d mongodb
-fi
-
-# Check if the RabbitMQ container is already running
-RABBITMQ_CONTAINER_NAME="rabbitmq"
-if [ $(docker ps -q -f name=^/${RABBITMQ_CONTAINER_NAME}$) ]; then
-    echo "RabbitMQ container already running. Skipping RabbitMQ startup."
-else
-    echo "Starting RabbitMQ"
-    # Start RabbitMQ
-    docker compose up -d rabbitmq
-    # Wait for RabbitMQ to start
-    sleep 10
 fi

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"
 
@@ -31,13 +31,13 @@
     make build
 
 # Final minimal image with binary only
 FROM alpine:3.16 as run
 
 RUN addgroup --gid 1138 -S staking-expiry-checker && adduser --uid 1138 -S staking-expiry-checker -G staking-expiry-checker
 RUN apk add bash curl jq
 
 # Label should match your github repo
 LABEL org.opencontainers.image.source="https://github.com/babylonlabs-io/staking-expiry-checker:${VERSION}"
 
 
 # Copy over binaries from the build-env

docker-compose.yml

@@ -7,7 +7,6 @@ services:
       - CONFIG=/home/staking-expiry-checker/config.yml
     depends_on:
       - mongodb
-      - rabbitmq
     volumes:
       - ./config/config-docker.yml:/home/staking-expiry-checker/config.yml:Z
   mongodb:
@@ -22,14 +21,3 @@ services:
     volumes:
       - ./bin/init-mongo.sh:/init-mongo.sh
     entrypoint: [ "/init-mongo.sh" ]
-  rabbitmq:
-    image: rabbitmq:3-management
-    container_name: rabbitmq
-    ports:
-      - "5672:5672" # AMQP protocol port
-      - "15672:15672" # Management UI port
-    environment:
-      RABBITMQ_DEFAULT_USER: user
-      RABBITMQ_DEFAULT_PASS: password
-    volumes:
-      - "./rabbitmq_data:/var/lib/rabbitmq"

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
 )
 

go.sum

@@ -1,6 +1,4 @@
 github.com/aead/siphash v1.0.1/go.mod h1:Nywa3cDsYNNK3gaciGTWPwHt0wlpNV15vwmswBAUSII=
-github.com/babylonlabs-io/staking-queue-client v0.4.1 h1:AW+jtrNxZYN/isRx+njqjHbUU9CzhF42Ke6roK+0N3I=
-github.com/babylonlabs-io/staking-queue-client v0.4.1/go.mod h1:n3fr3c+9LNiJlyETmcrVk94Zn76rAADhGZKxX+rVf+Q=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/btcsuite/btcd v0.20.1-beta/go.mod h1:wVuoA8VJLEcwgqHBwHmzLRazpKxTv13Px/pDuV7OomQ=
@@ -83,11 +81,8 @@ github.com/jrick/logrotate v1.0.0/go.mod h1:LNinyqDIJnpAur+b8yyulnQw/wDuN1+BYKlT
 github.com/kkdai/bstream v0.0.0-20161212061736-f391b8402d23/go.mod h1:J+Gs4SYgM6CZQHDETBtE9HaSEkGmuNXF86RwHhHUvq4=
 github.com/klauspost/compress v1.17.0 h1:Rnbp4K9EjcDuVuHtd0dgA4qNuv9yKDYKK1ulpJwgrqM=
 github.com/klauspost/compress v1.17.0/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
-github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
-github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
-github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/magiconair/properties v1.8.7 h1:IeQXZAiQcpL9mgcAe1Nu6cX9LLw6ExEHKjN0VQdvPDY=
@@ -124,10 +119,8 @@ github.com/prometheus/common v0.48.0 h1:QO8U2CdOzSn1BBsmXJXduaaW+dY/5QLjfB8svtSz
 github.com/prometheus/common v0.48.0/go.mod h1:0/KsvlIEfPQCQ5I2iNSAWKPZziNCvRs5EC6ILDTlAPc=
 github.com/prometheus/procfs v0.12.0 h1:jluTpSng7V9hY0O2R9DzzJHYb2xULk9VTR1V1R/k6Bo=
 github.com/prometheus/procfs v0.12.0/go.mod h1:pcuDEFsWDnvcgNzo4EEweacyhjeA9Zk3cnaOZAZEfOo=
-github.com/rabbitmq/amqp091-go v1.9.0 h1:qrQtyzB4H8BQgEuJwhmVQqVHB9O4+MNDJCCAcpc3Aoo=
-github.com/rabbitmq/amqp091-go v1.9.0/go.mod h1:+jPrT9iY2eLjRaMSRHUhc3z14E/l85kv/f+6luSD3pc=
-github.com/rogpeppe/go-internal v1.11.0 h1:cWPaGQEPrBb5/AsnsZesgZZ9yb1OQ+GOISoDNXVBh4M=
-github.com/rogpeppe/go-internal v1.11.0/go.mod h1:ddIwULY96R17DhadqLgMfk9H9tvdUzkipdSkR5nkCZA=
+github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
+github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog=
 github.com/rs/xid v1.5.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg=
 github.com/rs/zerolog v1.32.0 h1:keLypqrlIjaFsbmJOBdB/qvyF8KEtCWHwobLp5l/mQ0=
 github.com/rs/zerolog v1.32.0/go.mod h1:/7mN4D5sKwJLZQ2b/znpjC3/GQWY/xaDXUM0kKWRHss=
@@ -173,8 +166,6 @@ github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d/go.mod h1:rHwXgn7Jul
 github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
 go.mongodb.org/mongo-driver v1.14.0 h1:P98w8egYRjYe3XDjxhYJagTokP/H6HzlsnojRgZRd80=
 go.mongodb.org/mongo-driver v1.14.0/go.mod h1:Vzb0Mk/pa7e6cWw85R4F/endUC3u0U9jGcNU603k65c=
-go.uber.org/goleak v1.2.1 h1:NBol2c7O1ZokfZ0LEU9K6Whx/KnwvepVetCUhtKja4A=
-go.uber.org/goleak v1.2.1/go.mod h1:qlT2yGI9QafXHhZZLxlSuNsMw3FFLxBr+tBRlmO1xH4=
 go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
 go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
 golang.org/x/crypto v0.0.0-20170930174604-9419663f5a44/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
@@ -241,7 +232,6 @@ google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2
 google.golang.org/protobuf v1.32.0 h1:pPC6BG5ex8PDFnkbrGU3EixyhKcQ2aDuBS36lqK/C7I=
 google.golang.org/protobuf v1.32.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
 gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=

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