feat: emit withdrawn and withdrawable events

consumer/event_consumer.go

@@ -1,12 +1,16 @@
 package consumer
 
 import (
+	"context"
+
 	"github.com/babylonlabs-io/staking-queue-client/client"
 )
 
 type EventConsumer interface {
 	Start() error
-	PushActiveStakingEvent(ev *client.StakingEvent) error
-	PushUnbondingStakingEvent(ev *client.StakingEvent) error
+	PushActiveStakingEvent(ctx context.Context, ev *client.StakingEvent) error
+	PushUnbondingStakingEvent(ctx context.Context, ev *client.StakingEvent) error
+	PushWithdrawableStakingEvent(ctx context.Context, ev *client.StakingEvent) error
+	PushWithdrawnStakingEvent(ctx context.Context, ev *client.StakingEvent) error
 	Stop() error
 }

e2etest/e2e_test.go

@@ -46,8 +46,9 @@ func TestQueueConsumer(t *testing.T) {
 			hex.EncodeToString(bbndatagen.GenRandomByteArray(r, 10)),
 			[]string{hex.EncodeToString(bbndatagen.GenRandomByteArray(r, 10))},
 			1000,
+			[]string{hex.EncodeToString(bbndatagen.GenRandomByteArray(r, 10))},
 		)
-		err = queueConsumer.PushActiveStakingEvent(&stakingEvent)
+		err = queueConsumer.PushActiveStakingEvent(context.TODO(), &stakingEvent)
 		require.NoError(t, err)
 		stakingEventList = append(stakingEventList, &stakingEvent)
 	}

go.mod

@@ -6,7 +6,7 @@ require (
 	cosmossdk.io/math v1.4.0
 	github.com/avast/retry-go/v4 v4.5.1
 	github.com/babylonlabs-io/babylon v1.0.0-rc.2
-	github.com/babylonlabs-io/staking-queue-client v0.4.7-0.20241212112557-9ac7de686075
+	github.com/babylonlabs-io/staking-queue-client v0.4.7-0.20250116064256-c4b08ada1f40
 	github.com/btcsuite/btcd v0.24.3-0.20241011125836-24eb815168f4
 	github.com/btcsuite/btcd/btcec/v2 v2.3.4
 	github.com/btcsuite/btcd/btcutil v1.1.6

go.sum

@@ -1429,8 +1429,8 @@ github.com/aws/aws-sdk-go v1.44.312/go.mod h1:aVsgQcEevwlmQ7qHE9I3h+dtQgpqhFB+i8
 github.com/aws/aws-sdk-go-v2 v0.18.0/go.mod h1:JWVYvqSMppoMJC0x5wdwiImzgXTI9FuZwxzkQq9wy+g=
 github.com/babylonlabs-io/babylon v1.0.0-rc.2 h1:H7OpEDNNOXyC+9TUo4vVYLlHNhOQ8m9KqWP1qzjEt0c=
 github.com/babylonlabs-io/babylon v1.0.0-rc.2/go.mod h1:B8ma8IjGUEKhmoRfwv60Qa7DtUXssCgtmD89huQ4+5I=
-github.com/babylonlabs-io/staking-queue-client v0.4.7-0.20241212112557-9ac7de686075 h1:gB+jslBkK5/ror4sn9NHldKjLu4nE88jgD43d2L3osc=
-github.com/babylonlabs-io/staking-queue-client v0.4.7-0.20241212112557-9ac7de686075/go.mod h1:n3fr3c+9LNiJlyETmcrVk94Zn76rAADhGZKxX+rVf+Q=
+github.com/babylonlabs-io/staking-queue-client v0.4.7-0.20250116064256-c4b08ada1f40 h1:k6lZ1HOcQ4xcUMxIKvcWwSddvV3Sigi8dP/bWTJlT2I=
+github.com/babylonlabs-io/staking-queue-client v0.4.7-0.20250116064256-c4b08ada1f40/go.mod h1:n3fr3c+9LNiJlyETmcrVk94Zn76rAADhGZKxX+rVf+Q=
 github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
 github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
 github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8=

internal/db/model/delegation.go

@@ -151,3 +151,11 @@ func (d *BTCDelegationDetails) HasInclusionProof() bool {
 	// Ref: https://github.com/babylonlabs-io/babylon/blob/b1a4b483f60458fcf506adf1d80aaa6c8c10f8a4/x/btcstaking/types/btc_delegation.go#L47
 	return d.StartHeight > 0 && d.EndHeight > 0
 }
+
+func ToStateStrings(stateHistory []StateRecord) []string {
+	states := make([]string, len(stateHistory))
+	for i, record := range stateHistory {
+		states[i] = record.State.String()
+	}
+	return states
+}

internal/services/consumer_events.go

@@ -11,33 +11,73 @@ import (
 
 func (s *Service) emitActiveDelegationEvent(
 	ctx context.Context,
-	stakingTxHashHex string,
-	stakerBtcPkHex string,
-	finalityProviderBtcPksHex []string,
-	stakingAmount uint64,
+	delegation *model.BTCDelegationDetails,
 ) *types.Error {
+	stateHistoryStrs := model.ToStateStrings(delegation.StateHistory)
 	stakingEvent := queuecli.NewActiveStakingEvent(
-		stakingTxHashHex,
-		stakerBtcPkHex,
-		finalityProviderBtcPksHex,
-		stakingAmount,
+		delegation.StakingTxHashHex,
+		delegation.StakerBtcPkHex,
+		delegation.FinalityProviderBtcPksHex,
+		delegation.StakingAmount,
+		stateHistoryStrs,
 	)
 
-	if err := s.queueManager.PushActiveStakingEvent(&stakingEvent); err != nil {
+	if err := s.queueManager.PushActiveStakingEvent(ctx, &stakingEvent); err != nil {
 		return types.NewInternalServiceError(fmt.Errorf("failed to push the staking event to the queue: %w", err))
 	}
 	return nil
 }
 
-func (s *Service) emitUnbondingDelegationEvent(ctx context.Context, delegation *model.BTCDelegationDetails) *types.Error {
+func (s *Service) emitUnbondingDelegationEvent(
+	ctx context.Context,
+	delegation *model.BTCDelegationDetails,
+) *types.Error {
+	stateHistoryStrs := model.ToStateStrings(delegation.StateHistory)
 	ev := queuecli.NewUnbondingStakingEvent(
 		delegation.StakingTxHashHex,
 		delegation.StakerBtcPkHex,
 		delegation.FinalityProviderBtcPksHex,
 		delegation.StakingAmount,
+		stateHistoryStrs,
 	)
-	if err := s.queueManager.PushUnbondingStakingEvent(&ev); err != nil {
+	if err := s.queueManager.PushUnbondingStakingEvent(ctx, &ev); err != nil {
 		return types.NewInternalServiceError(fmt.Errorf("failed to push the unbonding event to the queue: %w", err))
 	}
 	return nil
 }
+
+func (s *Service) emitWithdrawableDelegationEvent(
+	ctx context.Context,
+	delegation *model.BTCDelegationDetails,
+) *types.Error {
+	stateHistoryStrs := model.ToStateStrings(delegation.StateHistory)
+	ev := queuecli.NewWithdrawableStakingEvent(
+		delegation.StakingTxHashHex,
+		delegation.StakerBtcPkHex,
+		delegation.FinalityProviderBtcPksHex,
+		delegation.StakingAmount,
+		stateHistoryStrs,
+	)
+	if err := s.queueManager.PushWithdrawableStakingEvent(ctx, &ev); err != nil {
+		return types.NewInternalServiceError(fmt.Errorf("failed to push the withdrawable event to the queue: %w", err))
+	}
+	return nil
+}
+
+func (s *Service) emitWithdrawnDelegationEvent(
+	ctx context.Context,
+	delegation *model.BTCDelegationDetails,
+) *types.Error {
+	stateHistoryStrs := model.ToStateStrings(delegation.StateHistory)
+	ev := queuecli.NewWithdrawnStakingEvent(
+		delegation.StakingTxHashHex,
+		delegation.StakerBtcPkHex,
+		delegation.FinalityProviderBtcPksHex,
+		delegation.StakingAmount,
+		stateHistoryStrs,
+	)
+	if err := s.queueManager.PushWithdrawnStakingEvent(ctx, &ev); err != nil {
+		return types.NewInternalServiceError(fmt.Errorf("failed to push the withdrawn event to the queue: %w", err))
+	}
+	return nil
+}

internal/services/delegation.go

@@ -160,10 +160,7 @@ func (s *Service) processCovenantQuorumReachedEvent(
 
 		err = s.emitActiveDelegationEvent(
 			ctx,
-			delegation.StakingTxHashHex,
-			delegation.StakerBtcPkHex,
-			delegation.FinalityProviderBtcPksHex,
-			delegation.StakingAmount,
+			delegation,
 		)
 		if err != nil {
 			return err
@@ -238,10 +235,7 @@ func (s *Service) processBTCDelegationInclusionProofReceivedEvent(
 
 		err = s.emitActiveDelegationEvent(
 			ctx,
-			inclusionProofEvent.StakingTxHash,
-			delegation.StakerBtcPkHex,
-			delegation.FinalityProviderBtcPksHex,
-			delegation.StakingAmount,
+			delegation,
 		)
 		if err != nil {
 			return err

internal/services/expiry_checker.go

@@ -71,6 +71,11 @@ func (s *Service) checkExpiry(ctx context.Context) *types.Error {
 					fmt.Errorf("failed to update BTC delegation state to withdrawable: %w", err),
 				)
 			}
+		} else {
+			// This means the state transitioned to withdrawable so we need to emit the event
+			if err := s.emitWithdrawableDelegationEvent(ctx, delegation); err != nil {
+				return err
+			}
 		}
 
 		if err := s.db.DeleteExpiredDelegation(ctx, delegation.StakingTxHashHex); err != nil {

internal/services/watch_btc_events.go

@@ -128,6 +128,14 @@ func (s *Service) watchForSpendSlashingChange(
 			return
 		}
 
+		if err := s.emitWithdrawnDelegationEvent(quitCtx, delegation); err != nil {
+			log.Error().
+				Err(err).
+				Str("staking_tx", delegation.StakingTxHashHex).
+				Msg("failed to emit withdrawn delegation event")
+			return
+		}
+
 		// Update to withdrawn state
 		delegationSubState := subState
 		if err := s.db.UpdateBTCDelegationState(
@@ -379,6 +387,10 @@ func (s *Service) handleWithdrawal(
 		return fmt.Errorf("current state %s is not qualified for withdrawal", *delegationState)
 	}
 
+	if err := s.emitWithdrawnDelegationEvent(ctx, delegation); err != nil {
+		return fmt.Errorf("failed to emit withdrawn delegation event: %w", err)
+	}
+
 	// Update to withdrawn state
 	log.Debug().
 		Str("staking_tx", delegation.StakingTxHashHex).

internal/types/state.go

@@ -62,8 +62,10 @@ func QualifiedStatesForWithdrawn() []DelegationState {
 // QualifiedStatesForWithdrawable returns the qualified current states for Withdrawable event
 // The "StateWithdrawable" is included b/c sub state can be changed to if
 // user did not withdraw ontime. e.g TIMELOCK change to TIMELOCK_SLASHING
+// The "StateActive" is included b/c the slashing tx can be detected before babylon events
+// at point where last state is active
 func QualifiedStatesForWithdrawable() []DelegationState {
-	return []DelegationState{StateUnbonding, StateSlashed, StateWithdrawable}
+	return []DelegationState{StateActive, StateUnbonding, StateSlashed, StateWithdrawable}
 }
 
 type DelegationSubState string