fix(payments): fix payment storage constraint (#1763)

Author: paul-nicolas

components/payments/cmd/api/internal/storage/payments.go

@@ -146,7 +146,7 @@ func (s *Storage) UpsertPayments(ctx context.Context, payments []*models.Payment
 
 	_, err := s.db.NewInsert().
 		Model(&payments).
-		On("CONFLICT (reference) DO UPDATE").
+		On("CONFLICT (reference, connector_id) DO UPDATE").
 		Set("amount = EXCLUDED.amount").
 		Set("type = EXCLUDED.type").
 		Set("status = EXCLUDED.status").

components/payments/cmd/api/internal/storage/payments_test.go

@@ -77,6 +77,45 @@ func insertPayments(t *testing.T, store *Storage, connectorID models.ConnectorID
 	return []models.Payment{p1, p2}
 }
 
+func TestUpsertPayments(t *testing.T) {
+	t.Parallel()
+
+	store := newStore(t)
+
+	connectorID := installConnector(t, store)
+	insertAccounts(t, store, connectorID)
+	insertPayments(t, store, connectorID)
+
+	p1 := models.Payment{
+		ID: models.PaymentID{
+			PaymentReference: models.PaymentReference{
+				Reference: "test_1",
+				Type:      models.PaymentTypePayIn,
+			},
+			ConnectorID: connectorID,
+		},
+		ConnectorID:   connectorID,
+		CreatedAt:     time.Date(2023, 11, 14, 8, 0, 0, 0, time.UTC),
+		Reference:     "test_1",
+		Amount:        big.NewInt(100),
+		InitialAmount: big.NewInt(100),
+		Type:          models.PaymentTypePayIn,
+		Status:        models.PaymentStatusPending,
+		Scheme:        models.PaymentSchemeA2A,
+		Asset:         models.Asset("USD/2"),
+		SourceAccountID: &models.AccountID{
+			Reference:   "test_account",
+			ConnectorID: connectorID,
+		},
+		DestinationAccountID: &models.AccountID{
+			Reference:   "test_account2",
+			ConnectorID: connectorID,
+		},
+	}
+	err := store.UpsertPayments(context.Background(), []*models.Payment{&p1})
+	require.NoError(t, err)
+}
+
 func TestListPayments(t *testing.T) {
 	t.Parallel()
 

components/payments/cmd/connectors/internal/storage/connectors_test.go

@@ -16,6 +16,11 @@ var (
 		Reference: uuid.New(),
 		Provider:  models.ConnectorProviderDummyPay,
 	}
+
+	connectorID2 = models.ConnectorID{
+		Reference: uuid.New(),
+		Provider:  models.ConnectorProviderDummyPay,
+	}
 )
 
 func TestConnectors(t *testing.T) {
@@ -41,6 +46,18 @@ func testInstallConnectors(t *testing.T, store *storage.Storage) {
 	)
 	require.NoError(t, err)
 
+	connector2 := models.Connector{
+		ID:       connectorID2,
+		Name:     "test2",
+		Provider: models.ConnectorProviderDummyPay,
+	}
+	err = store.Install(
+		context.Background(),
+		&connector2,
+		json.RawMessage([]byte(`{"foo": "bar"}`)),
+	)
+	require.NoError(t, err)
+
 	err = store.Install(
 		context.Background(),
 		&connector1,

components/payments/cmd/connectors/internal/storage/payments_test.go

@@ -26,11 +26,60 @@ func TestPayments(t *testing.T) {
 
 	testInstallConnectors(t, store)
 	testCreatePayments(t, store)
+	testCreatePaymentsSameReferenceButDifferentConnectorID(t, store)
 	testUpdatePayment(t, store)
 	testUninstallConnectors(t, store)
 	testPaymentsDeletedAfterConnectorUninstall(t, store)
 }
 
+func testCreatePaymentsSameReferenceButDifferentConnectorID(t *testing.T, store *storage.Storage) {
+	id1 := &models.PaymentID{
+		PaymentReference: models.PaymentReference{
+			Reference: "test-same-reference",
+			Type:      models.PaymentTypePayOut,
+		},
+		ConnectorID: connectorID,
+	}
+	p1 := &models.Payment{
+		ID:          *id1,
+		CreatedAt:   p1T,
+		Reference:   "test-same-reference",
+		Amount:      big.NewInt(100),
+		ConnectorID: connectorID,
+		Type:        models.PaymentTypePayOut,
+		Status:      models.PaymentStatusSucceeded,
+		Scheme:      models.PaymentSchemeCardVisa,
+		Asset:       models.Asset("USD/2"),
+	}
+
+	ids, err := store.UpsertPayments(context.Background(), []*models.Payment{p1})
+	require.NoError(t, err)
+	require.Len(t, ids, 1)
+
+	id2 := &models.PaymentID{
+		PaymentReference: models.PaymentReference{
+			Reference: "test-same-reference2",
+			Type:      models.PaymentTypePayOut,
+		},
+		ConnectorID: connectorID2,
+	}
+	p2 := &models.Payment{
+		ID:          *id2,
+		CreatedAt:   p1T,
+		Reference:   "test-same-reference2",
+		Amount:      big.NewInt(100),
+		ConnectorID: connectorID2,
+		Type:        models.PaymentTypePayOut,
+		Status:      models.PaymentStatusSucceeded,
+		Scheme:      models.PaymentSchemeCardVisa,
+		Asset:       models.Asset("USD/2"),
+	}
+
+	ids, err = store.UpsertPayments(context.Background(), []*models.Payment{p2})
+	require.NoError(t, err)
+	require.Len(t, ids, 1)
+}
+
 func testCreatePayments(t *testing.T, store *storage.Storage) {
 	p1ID = &models.PaymentID{
 		PaymentReference: models.PaymentReference{

components/payments/internal/storage/migrations_v1.x.go

@@ -396,6 +396,21 @@ func registerMigrationsV1(ctx context.Context, migrator *migrations.Migrator) {
 				return fixMissingReferenceTransferInitiation(ctx, tx)
 			},
 		},
+		migrations.Migration{
+			Up: func(tx bun.Tx) error {
+				// Rererence should be unique within a connector, but not accross
+				// connectors. This migration fixes it.
+				_, err := tx.Exec(`
+					ALTER TABLE payments.payment ADD CONSTRAINT payment_reference_connector_id_unique_key UNIQUE (reference, connector_id);
+					ALTER TABLE payments.payment DROP CONSTRAINT IF EXISTS payment_reference_key;
+				`)
+				if err != nil {
+					return err
+				}
+
+				return nil
+			},
+		},
 	)
 }