Handle CircleRecipients for payment retries.

internal/data/circle_recipient.go

@@ -9,6 +9,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/lib/pq"
+
 	"github.com/stellar/stellar-disbursement-platform-backend/db"
 )
 
@@ -123,6 +125,39 @@ func (m CircleRecipientModel) Insert(ctx context.Context, receiverWalletID strin
 	return &circleRecipient, nil
 }
 
+// ResetRecipientsForRetryIfNeeded resets the status of the circle recipients for the given payment IDs to NULL if the status is not active.
+func (m CircleRecipientModel) ResetRecipientsForRetryIfNeeded(ctx context.Context, sqlExec db.SQLExecuter, paymentIDs ...string) ([]*CircleRecipient, error) {
+	if len(paymentIDs) == 0 {
+		return nil, fmt.Errorf("at least one payment ID is required: %w", ErrMissingInput)
+	}
+
+	const query = `
+		UPDATE
+			circle_recipients
+		SET
+			status = NULL,
+			sync_attempts = 0,
+			last_sync_attempt_at = NULL,
+			response_body = NULL
+		WHERE 
+			receiver_wallet_id IN (
+				SELECT DISTINCT receiver_wallet_id
+				FROM payments
+				WHERE id = ANY($1)
+			)
+		 	AND (status != $2 OR status IS NULL)
+		RETURNING
+	` + circleRecipientFields
+
+	var updatedRecipients []*CircleRecipient
+	err := sqlExec.SelectContext(ctx, &updatedRecipients, query, pq.Array(paymentIDs), CircleRecipientStatusActive)
+	if err != nil {
+		return nil, fmt.Errorf("getting context: %w", err)
+	}
+
+	return updatedRecipients, nil
+}
+
 func (m CircleRecipientModel) Update(ctx context.Context, receiverWalletID string, update CircleRecipientUpdate) (*CircleRecipient, error) {
 	if receiverWalletID == "" {
 		return nil, fmt.Errorf("receiverWalletID is required")

internal/data/circle_recipient_test.go

@@ -2,6 +2,7 @@ package data
 
 import (
 	"context"
+	"fmt"
 	"testing"
 	"time"
 
@@ -159,3 +160,177 @@ func Test_CircleRecipientModel_GetByReceiverWalletID(t *testing.T) {
 		assert.Equal(t, insertedCircleRecipient, fetchedCircleRecipient)
 	})
 }
+
+func Test_CircleRecipientModel_ResetRecipientsForRetryIfNeeded(t *testing.T) {
+	dbt := dbtest.Open(t)
+	defer dbt.Close()
+	dbConnectionPool, err := db.OpenDBConnectionPool(dbt.DSN)
+	require.NoError(t, err)
+	defer dbConnectionPool.Close()
+
+	ctx := context.Background()
+
+	m, err := NewModels(dbConnectionPool)
+	require.NoError(t, err)
+
+	asset := CreateAssetFixture(t, ctx, dbConnectionPool, "FOO1", "GA5ZSEJYB37JRC5AVCIA5MOP4RHTM335X2KGX3IHOJAPP5RE34K4KZVV")
+	receiver := CreateReceiverFixture(t, ctx, dbConnectionPool, &Receiver{})
+
+	walletA := CreateWalletFixture(t, ctx, dbConnectionPool, "walletA", "https://www.a.com", "www.a.com", "a://")
+	disbursementA := CreateDisbursementFixture(t, ctx, dbConnectionPool, m.Disbursements, &Disbursement{
+		Wallet:                              walletA,
+		Status:                              ReadyDisbursementStatus,
+		Asset:                               asset,
+		ReceiverRegistrationMessageTemplate: "Disbursement SMS Registration Message Template A1",
+	})
+	rwA := CreateReceiverWalletFixture(t, ctx, dbConnectionPool, receiver.ID, walletA.ID, RegisteredReceiversWalletStatus)
+	paymentA1 := CreatePaymentFixture(t, ctx, dbConnectionPool, m.Payment, &Payment{
+		ReceiverWallet: rwA,
+		Disbursement:   disbursementA,
+		Asset:          *asset,
+		Status:         ReadyPaymentStatus,
+		Amount:         "100",
+	})
+	paymentA2 := CreatePaymentFixture(t, ctx, dbConnectionPool, m.Payment, &Payment{
+		ReceiverWallet: rwA,
+		Disbursement:   disbursementA,
+		Asset:          *asset,
+		Status:         ReadyPaymentStatus,
+		Amount:         "100",
+	})
+
+	walletB := CreateWalletFixture(t, ctx, dbConnectionPool, "walletB", "https://www.b.com", "www.b.com", "b://")
+	disbursementB := CreateDisbursementFixture(t, ctx, dbConnectionPool, m.Disbursements, &Disbursement{
+		Wallet:                              walletB,
+		Status:                              ReadyDisbursementStatus,
+		Asset:                               asset,
+		ReceiverRegistrationMessageTemplate: "Disbursement SMS Registration Message Template A1",
+	})
+	rwB := CreateReceiverWalletFixture(t, ctx, dbConnectionPool, receiver.ID, walletB.ID, RegisteredReceiversWalletStatus)
+	paymentB1 := CreatePaymentFixture(t, ctx, dbConnectionPool, m.Payment, &Payment{
+		ReceiverWallet: rwB,
+		Disbursement:   disbursementB,
+		Asset:          *asset,
+		Status:         ReadyPaymentStatus,
+		Amount:         "100",
+	})
+	paymentB2 := CreatePaymentFixture(t, ctx, dbConnectionPool, m.Payment, &Payment{
+		ReceiverWallet: rwB,
+		Disbursement:   disbursementA,
+		Asset:          *asset,
+		Status:         ReadyPaymentStatus,
+		Amount:         "100",
+	})
+
+	t.Run("🔴 fails if no Payment IDs are provided", func(t *testing.T) {
+		circleRecipients, err := m.CircleRecipient.ResetRecipientsForRetryIfNeeded(ctx, dbConnectionPool)
+		require.ErrorContains(t, err, "at least one payment ID is required")
+		require.Nil(t, circleRecipients)
+	})
+
+	now := time.Now()
+	type TestCases struct {
+		name              string
+		prepareFixturesFn func(t *testing.T, models *Models)
+		paymentIDs        []string
+		wantUpdatedIDs    []string
+	}
+	testCases := []TestCases{
+		{
+			name:           "🟡 nothing happens if no paymentIDs are found",
+			paymentIDs:     []string{"non-existent", "non-existent-2"},
+			wantUpdatedIDs: []string{},
+		},
+		{
+			name:           "🟡 nothing happens if no circle recipients are found for the given payment IDs",
+			paymentIDs:     []string{paymentA1.ID, paymentA2.ID, paymentB1.ID, paymentB2.ID},
+			wantUpdatedIDs: []string{},
+		},
+		{
+			name: "🟢 only 'paymentA*' related recipients were updated",
+			prepareFixturesFn: func(t *testing.T, models *Models) {
+				for _, rwID := range []string{rwA.ID, rwB.ID} {
+					CreateCircleRecipientFixture(t, ctx, dbConnectionPool, CircleRecipient{
+						IdempotencyKey:   uuid.NewString(),
+						ReceiverWalletID: rwID,
+						UpdatedAt:        now,
+						CreatedAt:        now,
+						Status:           CircleRecipientStatusDenied,
+					})
+				}
+			},
+			paymentIDs:     []string{paymentA1.ID, paymentA2.ID},
+			wantUpdatedIDs: []string{rwA.ID},
+		},
+		{
+			name: "🟢 only 'payment*1' related recipients were updated",
+			prepareFixturesFn: func(t *testing.T, models *Models) {
+				for _, rwID := range []string{rwA.ID, rwB.ID} {
+					CreateCircleRecipientFixture(t, ctx, dbConnectionPool, CircleRecipient{
+						IdempotencyKey:   uuid.NewString(),
+						ReceiverWalletID: rwID,
+						UpdatedAt:        now,
+						CreatedAt:        now,
+						Status:           CircleRecipientStatusDenied,
+					})
+				}
+			},
+			paymentIDs:     []string{paymentA1.ID, paymentB1.ID},
+			wantUpdatedIDs: []string{rwA.ID, rwB.ID},
+		},
+	}
+
+	for _, nonActiveStatus := range []CircleRecipientStatus{
+		CircleRecipientStatusDenied,
+		CircleRecipientStatusInactive,
+		CircleRecipientStatusPending,
+		"",
+	} {
+		testCases = append(testCases, TestCases{
+			name: fmt.Sprintf("🟢 all recipients were updated [status=%s]", nonActiveStatus),
+			prepareFixturesFn: func(t *testing.T, models *Models) {
+				for _, rwID := range []string{rwA.ID, rwB.ID} {
+					CreateCircleRecipientFixture(t, ctx, dbConnectionPool, CircleRecipient{
+						IdempotencyKey:   uuid.NewString(),
+						ReceiverWalletID: rwID,
+						UpdatedAt:        now,
+						CreatedAt:        now,
+						Status:           nonActiveStatus,
+					})
+				}
+			},
+			paymentIDs:     []string{paymentA1.ID, paymentA2.ID, paymentB1.ID, paymentB2.ID},
+			wantUpdatedIDs: []string{rwA.ID, rwB.ID},
+		})
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			defer DeleteAllCircleRecipientsFixtures(t, ctx, m.DBConnectionPool)
+			if tc.prepareFixturesFn != nil {
+				tc.prepareFixturesFn(t, m)
+			}
+
+			circleRecipients, err := m.CircleRecipient.ResetRecipientsForRetryIfNeeded(ctx, dbConnectionPool, tc.paymentIDs...)
+			require.NoError(t, err)
+
+			updatedIDs := make([]string, 0, len(circleRecipients))
+			for _, cr := range circleRecipients {
+				updatedIDs = append(updatedIDs, cr.ReceiverWalletID)
+			}
+			require.ElementsMatch(t, tc.wantUpdatedIDs, updatedIDs)
+
+			if len(updatedIDs) > 0 {
+				for _, rwID := range updatedIDs {
+					circleRecipient, err := m.CircleRecipient.GetByReceiverWalletID(ctx, rwID)
+					require.NoError(t, err)
+					require.NotNil(t, circleRecipient)
+					assert.Empty(t, circleRecipient.Status)
+					assert.Empty(t, circleRecipient.SyncAttempts)
+					assert.Empty(t, circleRecipient.LastSyncAttemptAt)
+					assert.Empty(t, circleRecipient.ResponseBody)
+				}
+			}
+		})
+	}
+}

internal/serve/httphandler/payments_handler.go

@@ -178,6 +178,11 @@ func (p PaymentsHandler) RetryPayments(rw http.ResponseWriter, req *http.Request
 				return nil, fmt.Errorf("retrying failed payments: %w", err)
 			}
 
+			_, err = p.Models.CircleRecipient.ResetRecipientsForRetryIfNeeded(ctx, dbTx, reqBody.PaymentIDs...)
+			if err != nil {
+				return nil, fmt.Errorf("resetting circle recipients for retry if needed: %w", err)
+			}
+
 			// Producing event to send ready payments to TSS
 			var payments []*data.Payment
 			payments, err = p.Models.Payment.GetReadyByID(ctx, dbTx, reqBody.PaymentIDs...)