fix(payments): atomic processing with idempotency key and recovery

src/__tests__/payment-recovery-cron.spec.ts

@@ -0,0 +1,90 @@
+import { Test } from '@nestjs/testing';
+import { ScheduleModule } from '@nestjs/schedule';
+import { PaymentRecoveryService } from '../payments/payment-recovery.service';
+import { WebhookService } from '../webhook/webhook.service';
+import { StellarService } from '../stellar/stellar.service';
+import { db } from '../db/index';
+
+jest.mock('../db/index', () => ({
+  db: {
+    query: {
+      payments: { findFirst: jest.fn() },
+    },
+    execute: jest.fn(),
+    update: jest.fn(),
+    transaction: jest.fn(),
+  },
+}));
+
+const mockDb = db as jest.Mocked<typeof db>;
+
+/**
+ * Verifies the @Cron(EVERY_5_MINUTES) wiring itself — that recoverStaleSessions
+ * actually runs automatically on the documented schedule once registered with
+ * Nest's SchedulerRegistry — as opposed to the other payment-recovery specs,
+ * which call the recovery methods directly and never exercise the schedule.
+ *
+ * Spying on a @Cron-decorated method does not work with @nestjs/schedule:
+ * registering before app.init() strips the metadata schedule discovery relies
+ * on (no job gets registered at all), and registering after init has no effect
+ * because the CronJob already captured the real method. So instead of spying,
+ * this lets the real handler run on the fake-timer schedule and observes it
+ * through its (mocked) db.execute call.
+ */
+describe('PaymentRecoveryService cron schedule', () => {
+  beforeEach(() => {
+    jest.clearAllMocks();
+    (mockDb.execute as jest.Mock).mockResolvedValue([]);
+    // Anchor 1s past a 5-minute boundary so EVERY_5_MINUTES has a deterministic
+    // ~5-minute gap to the next fire, instead of depending on which second the
+    // real wall clock happens to be at when the test runs.
+    jest.useFakeTimers({ now: new Date('2026-01-01T00:00:01.000Z') });
+  });
+
+  afterEach(() => {
+    jest.useRealTimers();
+  });
+
+  async function bootstrap() {
+    const moduleRef = await Test.createTestingModule({
+      imports: [ScheduleModule.forRoot()],
+      providers: [
+        PaymentRecoveryService,
+        { provide: WebhookService, useValue: { dispatchWebhook: jest.fn() } },
+        {
+          provide: StellarService,
+          useValue: {
+            getPaymentsForAccount: jest.fn().mockResolvedValue([]),
+            verifyTransaction: jest.fn(),
+          },
+        },
+      ],
+    }).compile();
+
+    const app = moduleRef.createNestApplication();
+    await app.init();
+    return app;
+  }
+
+  it('does not run before 5 minutes have elapsed', async () => {
+    const app = await bootstrap();
+
+    await jest.advanceTimersByTimeAsync(4 * 60 * 1000);
+    expect(mockDb.execute).not.toHaveBeenCalled();
+
+    await app.close();
+  });
+
+  it('runs the recovery scan automatically every 5 minutes', async () => {
+    const app = await bootstrap();
+
+    await jest.advanceTimersByTimeAsync(5 * 60 * 1000 + 1000);
+    expect(mockDb.execute).toHaveBeenCalled();
+
+    const callsAfterFirstRun = (mockDb.execute as jest.Mock).mock.calls.length;
+    await jest.advanceTimersByTimeAsync(5 * 60 * 1000);
+    expect((mockDb.execute as jest.Mock).mock.calls.length).toBeGreaterThan(callsAfterFirstRun);
+
+    await app.close();
+  });
+});

src/__tests__/payment-recovery.spec.ts

@@ -1,5 +1,6 @@
 import { PaymentRecoveryService } from '../payments/payment-recovery.service';
 import { WebhookService } from '../webhook/webhook.service';
+import { StellarService } from '../stellar/stellar.service';
 import { db } from '../db/index';
 
 jest.mock('../db/index', () => ({
@@ -9,6 +10,7 @@ jest.mock('../db/index', () => ({
     },
     execute: jest.fn(),
     update: jest.fn(),
+    transaction: jest.fn(),
   },
 }));
 
@@ -20,9 +22,23 @@ const createMockUpdateChain = () => {
   return { set, where };
 };
 
+const createMockTx = () => ({
+  insert: jest.fn().mockReturnValue({
+    values: jest.fn().mockReturnValue({
+      onConflictDoNothing: jest.fn().mockResolvedValue(undefined),
+    }),
+  }),
+  update: jest.fn().mockReturnValue({
+    set: jest.fn().mockReturnValue({
+      where: jest.fn().mockResolvedValue(undefined),
+    }),
+  }),
+});
+
 describe('PaymentRecoveryService', () => {
   let service: PaymentRecoveryService;
   let mockWebhooks: jest.Mocked<WebhookService>;
+  let mockStellar: jest.Mocked<StellarService>;
 
   beforeEach(() => {
     jest.clearAllMocks();
@@ -31,12 +47,137 @@ describe('PaymentRecoveryService', () => {
       dispatchWebhook: jest.fn(),
     } as any;
 
-    service = new PaymentRecoveryService(mockWebhooks);
+    mockStellar = {
+      getPaymentsForAccount: jest.fn().mockResolvedValue([]),
+      verifyTransaction: jest.fn().mockResolvedValue(false),
+    } as any;
+
+    service = new PaymentRecoveryService(mockWebhooks, mockStellar);
+  });
+
+  describe('recoverStuckProcessing', () => {
+    const stuckRow = {
+      id: 'session-stuck',
+      merchant_id: 'merchant-1',
+      memo: 'memo-stuck',
+      amount: '10.0000000',
+      asset_code: 'USDC',
+      asset_issuer: null,
+      receiving_account: 'GPLATFORM',
+    };
+
+    it('does nothing when no sessions are stuck in processing', async () => {
+      (mockDb.execute as jest.Mock).mockResolvedValueOnce([]);
+
+      await (service as any).recoverStuckProcessing();
+
+      expect(mockStellar.getPaymentsForAccount).not.toHaveBeenCalled();
+      expect(mockDb.update).not.toHaveBeenCalled();
+      expect(mockDb.transaction).not.toHaveBeenCalled();
+    });
+
+    it('completes the payment when Horizon confirms a matching transaction', async () => {
+      (mockDb.execute as jest.Mock).mockResolvedValueOnce([stuckRow]);
+      (mockStellar.getPaymentsForAccount as jest.Mock).mockResolvedValue([
+        {
+          type: 'payment',
+          transaction_memo: 'memo-stuck',
+          transaction_hash: 'tx-recovered',
+          amount: '10.0000000',
+          asset_code: 'USDC',
+          from: 'GSENDER',
+        },
+      ]);
+      (mockStellar.verifyTransaction as jest.Mock).mockResolvedValue(true);
+
+      const mockTx = createMockTx();
+      (mockDb.transaction as jest.Mock).mockImplementation(async (fn: (tx: any) => Promise<void>) =>
+        fn(mockTx),
+      );
+
+      await (service as any).recoverStuckProcessing();
+
+      expect(mockStellar.verifyTransaction).toHaveBeenCalledWith('tx-recovered');
+      expect(mockTx.insert).toHaveBeenCalledTimes(1);
+      expect(mockTx.update).toHaveBeenCalledTimes(1);
+      expect(mockWebhooks.dispatchWebhook).toHaveBeenCalledWith(
+        'merchant-1',
+        'payment.confirmed',
+        expect.objectContaining({ sessionId: 'session-stuck', txHash: 'tx-recovered' }),
+      );
+    });
+
+    it('reverts to pending when no matching payment is found on Horizon', async () => {
+      (mockDb.execute as jest.Mock).mockResolvedValueOnce([stuckRow]);
+      (mockStellar.getPaymentsForAccount as jest.Mock).mockResolvedValue([]);
+
+      const chain = createMockUpdateChain();
+      (mockDb.update as jest.Mock).mockReturnValue({ set: chain.set });
+
+      await (service as any).recoverStuckProcessing();
+
+      expect(mockDb.transaction).not.toHaveBeenCalled();
+      expect(chain.set).toHaveBeenCalledWith(expect.objectContaining({ status: 'pending' }));
+    });
+
+    it('reverts to pending when the matching transaction did not actually confirm', async () => {
+      (mockDb.execute as jest.Mock).mockResolvedValueOnce([stuckRow]);
+      (mockStellar.getPaymentsForAccount as jest.Mock).mockResolvedValue([
+        {
+          type: 'payment',
+          transaction_memo: 'memo-stuck',
+          transaction_hash: 'tx-failed',
+          amount: '10.0000000',
+          asset_code: 'USDC',
+          from: 'GSENDER',
+        },
+      ]);
+      (mockStellar.verifyTransaction as jest.Mock).mockResolvedValue(false);
+
+      const chain = createMockUpdateChain();
+      (mockDb.update as jest.Mock).mockReturnValue({ set: chain.set });
+
+      await (service as any).recoverStuckProcessing();
+
+      expect(mockDb.transaction).not.toHaveBeenCalled();
+      expect(chain.set).toHaveBeenCalledWith(expect.objectContaining({ status: 'pending' }));
+    });
+
+    it('reverts to pending without calling Horizon when the session has no memo', async () => {
+      (mockDb.execute as jest.Mock).mockResolvedValueOnce([{ ...stuckRow, memo: null }]);
+
+      const chain = createMockUpdateChain();
+      (mockDb.update as jest.Mock).mockReturnValue({ set: chain.set });
+
+      await (service as any).recoverStuckProcessing();
+
+      expect(mockStellar.getPaymentsForAccount).not.toHaveBeenCalled();
+      expect(chain.set).toHaveBeenCalledWith(expect.objectContaining({ status: 'pending' }));
+    });
+
+    it('continues reconciling remaining sessions if one fails', async () => {
+      (mockDb.execute as jest.Mock).mockResolvedValueOnce([
+        { ...stuckRow, id: 'session-a' },
+        { ...stuckRow, id: 'session-b' },
+      ]);
+      (mockStellar.getPaymentsForAccount as jest.Mock)
+        .mockRejectedValueOnce(new Error('Horizon down'))
+        .mockResolvedValueOnce([]);
+
+      const chain = createMockUpdateChain();
+      (mockDb.update as jest.Mock).mockReturnValue({ set: chain.set });
+
+      await (service as any).recoverStuckProcessing();
+
+      expect(chain.set).toHaveBeenCalledTimes(1);
+      expect(chain.where).toHaveBeenCalled();
+    });
   });
 
   describe('recoverPendingWithPayments', () => {
     it('marks pending sessions with existing payments as paid', async () => {
       (mockDb.execute as jest.Mock)
+        .mockResolvedValueOnce([])
         .mockResolvedValueOnce([
           {
             id: 'session-1',
@@ -65,6 +206,7 @@ describe('PaymentRecoveryService', () => {
 
     it('handles multiple pending sessions', async () => {
       (mockDb.execute as jest.Mock)
+        .mockResolvedValueOnce([])
         .mockResolvedValueOnce([
           {
             id: 'session-1',
@@ -97,6 +239,7 @@ describe('PaymentRecoveryService', () => {
 
     it('continues if one session recovery fails', async () => {
       (mockDb.execute as jest.Mock)
+        .mockResolvedValueOnce([])
         .mockResolvedValueOnce([
           {
             id: 'session-1',
@@ -138,6 +281,7 @@ describe('PaymentRecoveryService', () => {
   describe('recoverStuckPending', () => {
     it('marks stale pending sessions without payments as expired', async () => {
       (mockDb.execute as jest.Mock)
+        .mockResolvedValueOnce([])
         .mockResolvedValueOnce([])
         .mockResolvedValueOnce([{ id: 'stale-session', memo: 'memo-1' }])
         .mockResolvedValueOnce([]);
@@ -153,6 +297,7 @@ describe('PaymentRecoveryService', () => {
 
     it('marks stale pending sessions with payments as paid', async () => {
       (mockDb.execute as jest.Mock)
+        .mockResolvedValueOnce([])
         .mockResolvedValueOnce([])
         .mockResolvedValueOnce([{ id: 'stale-session', memo: 'memo-1' }])
         .mockResolvedValueOnce([]);
@@ -167,9 +312,24 @@ describe('PaymentRecoveryService', () => {
     });
   });
 
+  describe('orchestration', () => {
+    it('runs all four recovery scenarios in order', async () => {
+      (mockDb.execute as jest.Mock)
+        .mockResolvedValueOnce([])
+        .mockResolvedValueOnce([])
+        .mockResolvedValueOnce([])
+        .mockResolvedValueOnce([]);
+
+      await service.recoverStaleSessions();
+
+      expect(mockDb.execute).toHaveBeenCalledTimes(4);
+    });
+  });
+
   describe('empty results', () => {
     it('does nothing when no sessions need recovery', async () => {
       (mockDb.execute as jest.Mock)
+        .mockResolvedValueOnce([])
         .mockResolvedValueOnce([])
         .mockResolvedValueOnce([])
         .mockResolvedValueOnce([]);