From 7b5649f73de7817418ce71ab1b5e881bae2380aa Mon Sep 17 00:00:00 2001 From: enochakinbode Date: Tue, 28 Jul 2026 15:38:51 +0100 Subject: [PATCH 01/13] Route payouts through BullMQ --- .../src/modules/payouts/payouts.constants.ts | 27 ++++ .../api/src/modules/payouts/payouts.module.ts | 6 +- .../src/modules/payouts/payouts.processor.ts | 39 ++++++ .../src/modules/payouts/payouts.service.ts | 126 ++++++++++++++++-- 4 files changed, 189 insertions(+), 9 deletions(-) create mode 100644 apps/api/src/modules/payouts/payouts.constants.ts create mode 100644 apps/api/src/modules/payouts/payouts.processor.ts diff --git a/apps/api/src/modules/payouts/payouts.constants.ts b/apps/api/src/modules/payouts/payouts.constants.ts new file mode 100644 index 0000000..6804a5b --- /dev/null +++ b/apps/api/src/modules/payouts/payouts.constants.ts @@ -0,0 +1,27 @@ +export const PAYOUTS_QUEUE = 'payouts'; + +export const EXECUTE_PAYOUT_JOB = 'execute-payout'; +export const EXECUTE_PAYOUT_BATCH_JOB = 'execute-payout-batch'; +export const GENERATE_RECURRING_PAYOUT_JOB = 'generate-recurring-payout'; + +export interface ExecutePayoutJobData { + payoutId: string; +} + +export interface ExecutePayoutBatchJobData { + batchId: string; +} + +export interface GenerateRecurringPayoutJobData { + recurringPayoutId: string; +} + +export const PAYOUT_JOB_CLEANUP = { + removeOnComplete: { + age: 7 * 24 * 60 * 60, + count: 1000, + }, + removeOnFail: { + age: 30 * 24 * 60 * 60, + }, +}; diff --git a/apps/api/src/modules/payouts/payouts.module.ts b/apps/api/src/modules/payouts/payouts.module.ts index d1ab203..14714b2 100644 --- a/apps/api/src/modules/payouts/payouts.module.ts +++ b/apps/api/src/modules/payouts/payouts.module.ts @@ -1,6 +1,9 @@ import { Module } from '@nestjs/common'; +import { BullModule } from '@nestjs/bullmq'; import { PayoutsController } from './payouts.controller'; import { PayoutsService } from './payouts.service'; +import { PayoutsProcessor } from './payouts.processor'; +import { PAYOUTS_QUEUE } from './payouts.constants'; import { PrismaModule } from '../prisma/prisma.module'; import { WebhooksModule } from '../webhooks/webhooks.module'; import { StellarModule } from '../stellar/stellar.module'; @@ -16,8 +19,9 @@ import { NotificationsModule } from '../notifications/notifications.module'; AuthModule, EventsModule, NotificationsModule, + BullModule.registerQueue({ name: PAYOUTS_QUEUE }), ], - providers: [PayoutsService], + providers: [PayoutsService, PayoutsProcessor], controllers: [PayoutsController], exports: [PayoutsService], }) diff --git a/apps/api/src/modules/payouts/payouts.processor.ts b/apps/api/src/modules/payouts/payouts.processor.ts new file mode 100644 index 0000000..3bb248c --- /dev/null +++ b/apps/api/src/modules/payouts/payouts.processor.ts @@ -0,0 +1,39 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { Processor, WorkerHost } from '@nestjs/bullmq'; +import { Job } from 'bullmq'; +import { PayoutsService } from './payouts.service'; +import { + EXECUTE_PAYOUT_BATCH_JOB, + EXECUTE_PAYOUT_JOB, + type ExecutePayoutBatchJobData, + type ExecutePayoutJobData, + PAYOUTS_QUEUE, +} from './payouts.constants'; + +type PayoutJobData = ExecutePayoutJobData | ExecutePayoutBatchJobData; + +@Injectable() +@Processor(PAYOUTS_QUEUE) +export class PayoutsProcessor extends WorkerHost { + private readonly logger = new Logger(PayoutsProcessor.name); + + constructor(private readonly payoutsService: PayoutsService) { + super(); + } + + async process(job: Job): Promise { + if (job.name === EXECUTE_PAYOUT_JOB) { + const data = job.data as ExecutePayoutJobData; + await this.payoutsService.processQueuedPayout(data.payoutId); + return; + } + + if (job.name === EXECUTE_PAYOUT_BATCH_JOB) { + const data = job.data as ExecutePayoutBatchJobData; + await this.payoutsService.processQueuedBatch(data.batchId); + return; + } + + this.logger.warn(`Unknown payout job ignored: ${job.name}`); + } +} diff --git a/apps/api/src/modules/payouts/payouts.service.ts b/apps/api/src/modules/payouts/payouts.service.ts index b2653f4..c6c1043 100644 --- a/apps/api/src/modules/payouts/payouts.service.ts +++ b/apps/api/src/modules/payouts/payouts.service.ts @@ -4,7 +4,10 @@ import { Injectable, Logger, NotFoundException, + OnApplicationBootstrap, } from '@nestjs/common'; +import { InjectQueue } from '@nestjs/bullmq'; +import { Queue } from 'bullmq'; import { DestType, Payout, PayoutStatus, Prisma } from '@prisma/client'; import { PrismaService } from '../prisma/prisma.service'; import { WebhooksService } from '../webhooks/webhooks.service'; @@ -14,6 +17,14 @@ import { NotificationsService } from '../notifications/notifications.service'; import { CreatePayoutDto, BulkPayoutDto } from './dto/create-payout.dto'; import { PayoutFiltersDto } from './dto/payout-filters.dto'; import { randomUUID } from 'crypto'; +import { + EXECUTE_PAYOUT_BATCH_JOB, + EXECUTE_PAYOUT_JOB, + PAYOUTS_QUEUE, + PAYOUT_JOB_CLEANUP, + type ExecutePayoutBatchJobData, + type ExecutePayoutJobData, +} from './payouts.constants'; // ── Types ───────────────────────────────────────────────────────────────────── @@ -37,7 +48,7 @@ function assetToString(asset: AssetObject): string { // ── Service ─────────────────────────────────────────────────────────────────── @Injectable() -export class PayoutsService { +export class PayoutsService implements OnApplicationBootstrap { private readonly logger = new Logger(PayoutsService.name); constructor( @@ -46,8 +57,16 @@ export class PayoutsService { private readonly stellar: StellarService, private readonly eventsService: EventsService, private readonly notifications: NotificationsService, + @InjectQueue(PAYOUTS_QUEUE) + private readonly payoutQueue: Queue< + ExecutePayoutJobData | ExecutePayoutBatchJobData + >, ) {} + async onApplicationBootstrap(): Promise { + await this.reconcileScheduledPayoutJobs(); + } + // ── Create single payout ────────────────────────────────────────────────── async create( @@ -112,10 +131,7 @@ export class PayoutsService { ) .catch(() => undefined); - // Process immediately unless scheduled for the future - if (!payout.scheduledAt || payout.scheduledAt <= new Date()) { - this.processPayout(payout).catch(() => undefined); - } + await this.enqueuePayout(payout); return payout; } @@ -130,6 +146,7 @@ export class PayoutsService { const results: BulkPayoutResult['payouts'] = []; let accepted = 0; let rejected = 0; + let dueNow = 0; for (let i = 0; i < dto.payouts.length; i++) { const item = dto.payouts[i]; @@ -172,8 +189,10 @@ export class PayoutsService { results.push({ index: i, payoutId: payout.id }); accepted++; - if (!payout.scheduledAt || payout.scheduledAt <= new Date()) { - this.processPayout(payout).catch(() => undefined); + if (payout.scheduledAt && payout.scheduledAt > new Date()) { + await this.enqueuePayout(payout); + } else { + dueNow++; } } catch (err) { rejected++; @@ -193,6 +212,10 @@ export class PayoutsService { } as Prisma.InputJsonValue) .catch(() => undefined); + if (dueNow > 0) { + await this.enqueueBatch(batchId); + } + return { batchId, total: dto.payouts.length, @@ -287,13 +310,52 @@ export class PayoutsService { ) .catch(() => undefined); - this.processPayout(reset).catch(() => undefined); + await this.enqueuePayout(reset); return reset; } // ── Internal processing ─────────────────────────────────────────────────── + async processQueuedPayout(payoutId: string): Promise { + const payout = await this.prisma.payout.findUnique({ + where: { id: payoutId }, + }); + if (!payout) { + this.logger.warn(`Queued payout ${payoutId} skipped: not found`); + return; + } + + if (payout.status !== PayoutStatus.PENDING) { + this.logger.log( + `Queued payout ${payout.id} skipped: status is ${payout.status}`, + ); + return; + } + + if (payout.scheduledAt && payout.scheduledAt > new Date()) { + await this.enqueuePayout(payout); + return; + } + + await this.processPayout(payout); + } + + async processQueuedBatch(batchId: string): Promise { + const payouts = await this.prisma.payout.findMany({ + where: { + batchId, + status: PayoutStatus.PENDING, + OR: [{ scheduledAt: null }, { scheduledAt: { lte: new Date() } }], + }, + orderBy: { createdAt: 'asc' }, + }); + + for (const payout of payouts) { + await this.processQueuedPayout(payout.id); + } + } + private async processPayout(payout: Payout): Promise { await this.prisma.payout.update({ where: { id: payout.id }, @@ -464,4 +526,52 @@ export class PayoutsService { createdAt: payout.createdAt.toISOString(), }; } + + private async enqueuePayout(payout: Payout): Promise { + const delay = payout.scheduledAt + ? Math.max(0, payout.scheduledAt.getTime() - Date.now()) + : 0; + + await this.payoutQueue.add( + EXECUTE_PAYOUT_JOB, + { payoutId: payout.id }, + { + jobId: `payout:${payout.id}`, + delay, + attempts: 1, + ...PAYOUT_JOB_CLEANUP, + }, + ); + } + + private async enqueueBatch(batchId: string): Promise { + await this.payoutQueue.add( + EXECUTE_PAYOUT_BATCH_JOB, + { batchId }, + { + jobId: `payout-batch:${batchId}`, + attempts: 1, + ...PAYOUT_JOB_CLEANUP, + }, + ); + } + + private async reconcileScheduledPayoutJobs(): Promise { + const scheduled = await this.prisma.payout.findMany({ + where: { + status: PayoutStatus.PENDING, + scheduledAt: { not: null }, + }, + }); + + for (const payout of scheduled) { + await this.enqueuePayout(payout); + } + + if (scheduled.length > 0) { + this.logger.log( + `Reconciled ${scheduled.length} scheduled payout job(s)`, + ); + } + } } From fd95cafbf7e8254f9bfeb9c5b6ca6954fe04c717 Mon Sep 17 00:00:00 2001 From: enochakinbode Date: Tue, 28 Jul 2026 15:40:18 +0100 Subject: [PATCH 02/13] Add payout scheduling state --- .../migration.sql | 99 +++++++++++++++++++ apps/api/prisma/schema.prisma | 74 ++++++++++++++ 2 files changed, 173 insertions(+) create mode 100644 apps/api/prisma/migrations/20260728120000_payout_scheduling_state/migration.sql diff --git a/apps/api/prisma/migrations/20260728120000_payout_scheduling_state/migration.sql b/apps/api/prisma/migrations/20260728120000_payout_scheduling_state/migration.sql new file mode 100644 index 0000000..fc62222 --- /dev/null +++ b/apps/api/prisma/migrations/20260728120000_payout_scheduling_state/migration.sql @@ -0,0 +1,99 @@ +-- AlterEnum +ALTER TYPE "PayoutStatus" ADD VALUE IF NOT EXISTS 'REQUIRES_CONFIRMATION'; + +-- CreateEnum +CREATE TYPE "RecurringPayoutFrequency" AS ENUM ('DAILY', 'WEEKLY', 'BI_WEEKLY', 'MONTHLY'); + +-- CreateEnum +CREATE TYPE "MerchantLedgerEntryType" AS ENUM ('CREDIT', 'PAYOUT_DEBIT', 'PAYOUT_RELEASE'); + +-- AlterTable +ALTER TABLE "Payout" + ADD COLUMN "confirmedAt" TIMESTAMP(3), + ADD COLUMN "recurringPayoutId" TEXT; + +-- CreateTable +CREATE TABLE "RecurringPayout" ( + "id" TEXT NOT NULL, + "merchantId" TEXT NOT NULL, + "recipientId" TEXT, + "recipientName" TEXT NOT NULL, + "destinationType" "DestType" NOT NULL, + "destination" JSONB NOT NULL, + "amount" DECIMAL(36,18) NOT NULL, + "currency" TEXT NOT NULL, + "frequency" "RecurringPayoutFrequency" NOT NULL, + "nextRunAt" TIMESTAMP(3) NOT NULL, + "lastRunAt" TIMESTAMP(3), + "active" BOOLEAN NOT NULL DEFAULT true, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "RecurringPayout_pkey" PRIMARY KEY ("id") +); + +-- CreateTable +CREATE TABLE "MerchantBalance" ( + "merchantId" TEXT NOT NULL, + "availableAmount" DECIMAL(36,18) NOT NULL DEFAULT 0, + "reservedAmount" DECIMAL(36,18) NOT NULL DEFAULT 0, + "currency" TEXT NOT NULL DEFAULT 'USD', + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "MerchantBalance_pkey" PRIMARY KEY ("merchantId") +); + +-- CreateTable +CREATE TABLE "MerchantLedgerEntry" ( + "id" TEXT NOT NULL, + "merchantId" TEXT NOT NULL, + "payoutId" TEXT, + "paymentId" TEXT, + "type" "MerchantLedgerEntryType" NOT NULL, + "amount" DECIMAL(36,18) NOT NULL, + "currency" TEXT NOT NULL, + "description" TEXT, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "MerchantLedgerEntry_pkey" PRIMARY KEY ("id") +); + +-- CreateIndex +CREATE INDEX "Payout_scheduledAt_idx" ON "Payout"("scheduledAt"); + +-- CreateIndex +CREATE INDEX "Payout_recurringPayoutId_idx" ON "Payout"("recurringPayoutId"); + +-- CreateIndex +CREATE INDEX "RecurringPayout_merchantId_idx" ON "RecurringPayout"("merchantId"); + +-- CreateIndex +CREATE INDEX "RecurringPayout_active_nextRunAt_idx" ON "RecurringPayout"("active", "nextRunAt"); + +-- CreateIndex +CREATE INDEX "RecurringPayout_recipientId_idx" ON "RecurringPayout"("recipientId"); + +-- CreateIndex +CREATE INDEX "MerchantLedgerEntry_merchantId_createdAt_idx" ON "MerchantLedgerEntry"("merchantId", "createdAt"); + +-- CreateIndex +CREATE INDEX "MerchantLedgerEntry_payoutId_idx" ON "MerchantLedgerEntry"("payoutId"); + +-- CreateIndex +CREATE INDEX "MerchantLedgerEntry_paymentId_idx" ON "MerchantLedgerEntry"("paymentId"); + +-- AddForeignKey +ALTER TABLE "Payout" ADD CONSTRAINT "Payout_recurringPayoutId_fkey" FOREIGN KEY ("recurringPayoutId") REFERENCES "RecurringPayout"("id") ON DELETE SET NULL ON UPDATE CASCADE; + +-- AddForeignKey +ALTER TABLE "RecurringPayout" ADD CONSTRAINT "RecurringPayout_merchantId_fkey" FOREIGN KEY ("merchantId") REFERENCES "Merchant"("id") ON DELETE CASCADE ON UPDATE CASCADE; + +-- AddForeignKey +ALTER TABLE "RecurringPayout" ADD CONSTRAINT "RecurringPayout_recipientId_fkey" FOREIGN KEY ("recipientId") REFERENCES "Recipient"("id") ON DELETE SET NULL ON UPDATE CASCADE; + +-- AddForeignKey +ALTER TABLE "MerchantBalance" ADD CONSTRAINT "MerchantBalance_merchantId_fkey" FOREIGN KEY ("merchantId") REFERENCES "Merchant"("id") ON DELETE CASCADE ON UPDATE CASCADE; + +-- AddForeignKey +ALTER TABLE "MerchantLedgerEntry" ADD CONSTRAINT "MerchantLedgerEntry_merchantId_fkey" FOREIGN KEY ("merchantId") REFERENCES "Merchant"("id") ON DELETE CASCADE ON UPDATE CASCADE; diff --git a/apps/api/prisma/schema.prisma b/apps/api/prisma/schema.prisma index 123c330..82c93aa 100644 --- a/apps/api/prisma/schema.prisma +++ b/apps/api/prisma/schema.prisma @@ -49,6 +49,9 @@ model Merchant { notifications Notification[] settlementKey MerchantSettlementKey? recipients Recipient[] + recurringPayouts RecurringPayout[] + balance MerchantBalance? + ledgerEntries MerchantLedgerEntry[] } model ApiKey { @@ -302,20 +305,76 @@ model Payout { status PayoutStatus @default(PENDING) stellarTxHash String? scheduledAt DateTime? + confirmedAt DateTime? completedAt DateTime? failureReason String? batchId String? idempotencyKey String? @unique recipientId String? + recurringPayoutId String? createdAt DateTime @default(now()) merchant Merchant @relation(fields: [merchantId], references: [id]) recipient Recipient? @relation(fields: [recipientId], references: [id]) + recurringPayout RecurringPayout? @relation(fields: [recurringPayoutId], references: [id], onDelete: SetNull) @@index([merchantId]) @@index([status]) + @@index([scheduledAt]) @@index([batchId]) @@index([createdAt]) @@index([recipientId]) + @@index([recurringPayoutId]) +} + +model RecurringPayout { + id String @id @default(cuid()) + merchantId String + recipientId String? + recipientName String + destinationType DestType + destination Json + amount Decimal @db.Decimal(36, 18) + currency String + frequency RecurringPayoutFrequency + nextRunAt DateTime + lastRunAt DateTime? + active Boolean @default(true) + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + merchant Merchant @relation(fields: [merchantId], references: [id], onDelete: Cascade) + recipient Recipient? @relation(fields: [recipientId], references: [id], onDelete: SetNull) + payouts Payout[] + + @@index([merchantId]) + @@index([active, nextRunAt]) + @@index([recipientId]) +} + +model MerchantBalance { + merchantId String @id + availableAmount Decimal @default(0) @db.Decimal(36, 18) + reservedAmount Decimal @default(0) @db.Decimal(36, 18) + currency String @default("USD") + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + merchant Merchant @relation(fields: [merchantId], references: [id], onDelete: Cascade) +} + +model MerchantLedgerEntry { + id String @id @default(cuid()) + merchantId String + payoutId String? + paymentId String? + type MerchantLedgerEntryType + amount Decimal @db.Decimal(36, 18) + currency String + description String? + createdAt DateTime @default(now()) + merchant Merchant @relation(fields: [merchantId], references: [id], onDelete: Cascade) + + @@index([merchantId, createdAt]) + @@index([payoutId]) + @@index([paymentId]) } // Saved payout destination. Reconstructed to match migration @@ -331,6 +390,7 @@ model Recipient { updatedAt DateTime @updatedAt merchant Merchant @relation(fields: [merchantId], references: [id], onDelete: Cascade) payouts Payout[] + recurringPayouts RecurringPayout[] @@unique([merchantId, name]) @@index([merchantId]) @@ -346,12 +406,26 @@ enum DestType { enum PayoutStatus { PENDING + REQUIRES_CONFIRMATION PROCESSING COMPLETED FAILED CANCELLED } +enum RecurringPayoutFrequency { + DAILY + WEEKLY + BI_WEEKLY + MONTHLY +} + +enum MerchantLedgerEntryType { + CREDIT + PAYOUT_DEBIT + PAYOUT_RELEASE +} + // ── Webhooks ────────────────────────────────────────────────── model WebhookEvent { From 4d39a88f7a75013b8705669ae98978b5ee280afd Mon Sep 17 00:00:00 2001 From: enochakinbode Date: Tue, 28 Jul 2026 15:43:03 +0100 Subject: [PATCH 03/13] Enforce payout balance checks --- .../src/modules/payments/payments.service.ts | 77 ++++++++ .../src/modules/payouts/payouts.service.ts | 183 ++++++++++++++---- 2 files changed, 221 insertions(+), 39 deletions(-) diff --git a/apps/api/src/modules/payments/payments.service.ts b/apps/api/src/modules/payments/payments.service.ts index 98c0b7b..f96ae67 100644 --- a/apps/api/src/modules/payments/payments.service.ts +++ b/apps/api/src/modules/payments/payments.service.ts @@ -12,6 +12,7 @@ import { import { ConfigService } from '@nestjs/config'; import Stripe from 'stripe'; import { + MerchantLedgerEntryType, PaymentStatus, Payment, Prisma, @@ -177,6 +178,11 @@ export class PaymentsService implements OnModuleInit { ): Promise { this.logger.log(`Updating payment ${id} status to ${status}`); + const existingPayment = await this.prisma.payment.findUnique({ + where: { id }, + select: { status: true }, + }); + const updatedPayment = await this.prisma.payment.update({ where: { id }, data: { @@ -224,6 +230,10 @@ export class PaymentsService implements OnModuleInit { if (status === PaymentStatus.COMPLETED) { const receivedAmount = updatedPayment.destAmount?.toString() ?? '0'; + if (existingPayment?.status !== PaymentStatus.COMPLETED) { + await this.creditMerchantBalance(updatedPayment); + } + void this.notificationsService .notifyPaymentReceived( updatedPayment.merchantId, @@ -244,6 +254,45 @@ export class PaymentsService implements OnModuleInit { return updatedPayment; } + private async creditMerchantBalance(payment: Payment): Promise { + const amount = payment.destAmount ?? payment.sourceAmount; + const currency = payment.destAsset || payment.sourceAsset; + if (!amount || !currency) return; + + const existingCredit = await this.prisma.merchantLedgerEntry.findFirst({ + where: { + merchantId: payment.merchantId, + paymentId: payment.id, + type: MerchantLedgerEntryType.CREDIT, + }, + }); + if (existingCredit) return; + + await this.prisma.$transaction([ + this.prisma.merchantBalance.upsert({ + where: { merchantId: payment.merchantId }, + create: { + merchantId: payment.merchantId, + availableAmount: amount, + currency, + }, + update: { + availableAmount: { increment: amount }, + }, + }), + this.prisma.merchantLedgerEntry.create({ + data: { + merchantId: payment.merchantId, + paymentId: payment.id, + type: MerchantLedgerEntryType.CREDIT, + amount, + currency, + description: 'Payment completion credit', + }, + }), + ]); + } + async findExpiredLocked(): Promise { const now = new Date(); return await this.prisma.payment.findMany({ @@ -1115,6 +1164,9 @@ export class PaymentsService implements OnModuleInit { succeededAt: new Date().toISOString(), }, }); + const creditAmount = payment.destAmount ?? payment.sourceAmount ?? 0; + const creditCurrency = payment.destAsset || payment.sourceAsset || 'USD'; + const shouldCredit = payment.status !== PaymentStatus.COMPLETED; await this.prisma.$transaction([ this.prisma.payment.update({ @@ -1125,6 +1177,31 @@ export class PaymentsService implements OnModuleInit { metadata: updatedMetadata, }, }), + ...(shouldCredit + ? [ + this.prisma.merchantBalance.upsert({ + where: { merchantId: payment.merchantId }, + create: { + merchantId: payment.merchantId, + availableAmount: creditAmount, + currency: creditCurrency, + }, + update: { + availableAmount: { increment: creditAmount }, + }, + }), + this.prisma.merchantLedgerEntry.create({ + data: { + merchantId: payment.merchantId, + paymentId: payment.id, + type: MerchantLedgerEntryType.CREDIT, + amount: creditAmount, + currency: creditCurrency, + description: 'Payment completion credit', + }, + }), + ] + : []), this.prisma.webhookEvent.create({ data: { merchantId: payment.merchantId, diff --git a/apps/api/src/modules/payouts/payouts.service.ts b/apps/api/src/modules/payouts/payouts.service.ts index c6c1043..77bea07 100644 --- a/apps/api/src/modules/payouts/payouts.service.ts +++ b/apps/api/src/modules/payouts/payouts.service.ts @@ -8,7 +8,13 @@ import { } from '@nestjs/common'; import { InjectQueue } from '@nestjs/bullmq'; import { Queue } from 'bullmq'; -import { DestType, Payout, PayoutStatus, Prisma } from '@prisma/client'; +import { + DestType, + MerchantLedgerEntryType, + Payout, + PayoutStatus, + Prisma, +} from '@prisma/client'; import { PrismaService } from '../prisma/prisma.service'; import { WebhooksService } from '../webhooks/webhooks.service'; import { StellarService } from '../stellar/stellar.service'; @@ -357,58 +363,28 @@ export class PayoutsService implements OnApplicationBootstrap { } private async processPayout(payout: Payout): Promise { - await this.prisma.payout.update({ - where: { id: payout.id }, - data: { status: PayoutStatus.PROCESSING }, - }); + const started = await this.startPayoutProcessing(payout.id); + if (!started) return; try { - const destination = payout.destination as Record; + const destination = started.destination as Record; const isStellarPayout = - payout.destinationType === DestType.STELLAR || - (payout.destinationType === DestType.CRYPTO_WALLET && + started.destinationType === DestType.STELLAR || + (started.destinationType === DestType.CRYPTO_WALLET && typeof destination.network === 'string' && destination.network.toLowerCase() === 'stellar'); if (isStellarPayout) { - await this.processStellarPayout(payout, destination); + await this.processStellarPayout(started, destination); } // BANK_ACCOUNT and MOBILE_MONEY remain PROCESSING until external system // delivers and calls back to update the status. } catch (err) { const failureReason = err instanceof Error ? err.message : 'Processing failed'; - const failed = await this.prisma.payout.update({ - where: { id: payout.id }, - data: { status: PayoutStatus.FAILED, failureReason }, - }); - this.eventsService.emitPayoutStatus( - payout.merchantId, - payout.id, - PayoutStatus.FAILED, - { - amount: failed.amount.toString(), - currency: failed.currency, - failureReason, - updatedAt: new Date(), - }, - ); - void this.notifications - .notifyPayoutFailed( - payout.merchantId, - payout.id, - payout.recipientName, - failureReason, - ) - .catch(() => undefined); - this.webhooks - .dispatch(payout.merchantId, 'payout.failed', { - ...this.webhookPayload(failed), - failureReason, - } as Prisma.InputJsonValue) - .catch(() => undefined); - this.logger.error(`Payout ${payout.id} failed: ${failureReason}`); + await this.releaseDebitedFunds(started); + await this.failPayout(started, failureReason); } } @@ -491,6 +467,135 @@ export class PayoutsService implements OnApplicationBootstrap { .catch(() => undefined); } + private async startPayoutProcessing(payoutId: string): Promise { + const result = await this.prisma.$transaction(async (tx) => { + await tx.$executeRaw` + INSERT INTO "MerchantBalance" ("merchantId", "availableAmount", "reservedAmount", "currency", "updatedAt") + SELECT "merchantId", 0, 0, "currency", NOW() + FROM "Payout" + WHERE "id" = ${payoutId} + ON CONFLICT ("merchantId") DO NOTHING + `; + + const payout = await tx.payout.findUnique({ where: { id: payoutId } }); + if (!payout || payout.status !== PayoutStatus.PENDING) { + return { payout: null, failureReason: null }; + } + + const rows = await tx.$queryRaw< + Array<{ + availableAmount: Prisma.Decimal; + currency: string; + }> + >` + SELECT "availableAmount", "currency" + FROM "MerchantBalance" + WHERE "merchantId" = ${payout.merchantId} + FOR UPDATE + `; + const balance = rows[0]; + + if (!balance || balance.currency !== payout.currency) { + return { + payout, + failureReason: `Insufficient balance for ${payout.currency}`, + }; + } + + if (balance.availableAmount.lt(payout.amount)) { + return { + payout, + failureReason: `Insufficient balance for ${payout.currency}`, + }; + } + + await tx.merchantBalance.update({ + where: { merchantId: payout.merchantId }, + data: { + availableAmount: { decrement: payout.amount }, + }, + }); + await tx.merchantLedgerEntry.create({ + data: { + merchantId: payout.merchantId, + payoutId: payout.id, + type: MerchantLedgerEntryType.PAYOUT_DEBIT, + amount: payout.amount, + currency: payout.currency, + description: 'Payout execution debit', + }, + }); + + const started = await tx.payout.update({ + where: { id: payout.id }, + data: { status: PayoutStatus.PROCESSING }, + }); + return { payout: started, failureReason: null }; + }); + + if (result.failureReason && result.payout) { + await this.failPayout(result.payout, result.failureReason); + return null; + } + + return result.payout; + } + + private async releaseDebitedFunds(payout: Payout): Promise { + await this.prisma.$transaction(async (tx) => { + await tx.merchantBalance.update({ + where: { merchantId: payout.merchantId }, + data: { availableAmount: { increment: payout.amount } }, + }); + await tx.merchantLedgerEntry.create({ + data: { + merchantId: payout.merchantId, + payoutId: payout.id, + type: MerchantLedgerEntryType.PAYOUT_RELEASE, + amount: payout.amount, + currency: payout.currency, + description: 'Payout execution funds released after failure', + }, + }); + }); + } + + private async failPayout( + payout: Payout, + failureReason: string, + ): Promise { + const failed = await this.prisma.payout.update({ + where: { id: payout.id }, + data: { status: PayoutStatus.FAILED, failureReason }, + }); + this.eventsService.emitPayoutStatus( + payout.merchantId, + payout.id, + PayoutStatus.FAILED, + { + amount: failed.amount.toString(), + currency: failed.currency, + failureReason, + updatedAt: new Date(), + }, + ); + void this.notifications + .notifyPayoutFailed( + payout.merchantId, + payout.id, + payout.recipientName, + failureReason, + ) + .catch(() => undefined); + this.webhooks + .dispatch(payout.merchantId, 'payout.failed', { + ...this.webhookPayload(failed), + failureReason, + } as Prisma.InputJsonValue) + .catch(() => undefined); + this.logger.error(`Payout ${payout.id} failed: ${failureReason}`); + } + // ── Helpers ─────────────────────────────────────────────────────────────── private formatResponse(payout: Payout) { From 478936f630809e980e0037cb641078087d2da916 Mon Sep 17 00:00:00 2001 From: enochakinbode Date: Tue, 28 Jul 2026 15:45:31 +0100 Subject: [PATCH 04/13] Add recurring payout scheduling --- .../modules/payouts/dto/create-payout.dto.ts | 26 ++ .../src/modules/payouts/payouts.controller.ts | 52 ++++ .../src/modules/payouts/payouts.processor.ts | 15 +- .../src/modules/payouts/payouts.service.ts | 249 +++++++++++++++++- 4 files changed, 339 insertions(+), 3 deletions(-) diff --git a/apps/api/src/modules/payouts/dto/create-payout.dto.ts b/apps/api/src/modules/payouts/dto/create-payout.dto.ts index 8909d5a..76b24bd 100644 --- a/apps/api/src/modules/payouts/dto/create-payout.dto.ts +++ b/apps/api/src/modules/payouts/dto/create-payout.dto.ts @@ -65,6 +65,32 @@ export const CreatePayoutSchema = z.object({ export type CreatePayoutDto = z.infer; +export const RecurringPayoutFrequencySchema = z.enum([ + 'DAILY', + 'WEEKLY', + 'BI_WEEKLY', + 'MONTHLY', +]); + +export const CreateRecurringPayoutSchema = CreatePayoutSchema.omit({ + scheduledAt: true, +}).extend({ + frequency: RecurringPayoutFrequencySchema, + startsAt: z.coerce.date().optional(), +}); + +export const UpdateRecurringPayoutSchema = + CreateRecurringPayoutSchema.partial().extend({ + active: z.boolean().optional(), + }); + +export type CreateRecurringPayoutDto = z.infer< + typeof CreateRecurringPayoutSchema +>; +export type UpdateRecurringPayoutDto = z.infer< + typeof UpdateRecurringPayoutSchema +>; + // ── Bulk payout ──────────────────────────────────────────────────────────────── export const BulkPayoutSchema = z.object({ diff --git a/apps/api/src/modules/payouts/payouts.controller.ts b/apps/api/src/modules/payouts/payouts.controller.ts index 5399c1b..34885dc 100644 --- a/apps/api/src/modules/payouts/payouts.controller.ts +++ b/apps/api/src/modules/payouts/payouts.controller.ts @@ -6,9 +6,11 @@ import { HttpCode, HttpStatus, Param, + Patch, Post, Query, UseGuards, + Delete, } from '@nestjs/common'; import { PayoutsService } from './payouts.service'; import { @@ -16,6 +18,10 @@ import { CreatePayoutDto, BulkPayoutSchema, BulkPayoutDto, + CreateRecurringPayoutSchema, + CreateRecurringPayoutDto, + UpdateRecurringPayoutSchema, + UpdateRecurringPayoutDto, } from './dto/create-payout.dto'; import { PayoutFiltersSchema, @@ -64,6 +70,52 @@ export class PayoutsController { return this.payoutsService.list(merchantId, filters); } + @Post('recurring') + @UseGuards(JwtAuthGuard) + @HttpCode(HttpStatus.CREATED) + async createRecurring( + @CurrentMerchant('id') merchantId: string, + @Body(new ZodValidationPipe(CreateRecurringPayoutSchema)) + dto: CreateRecurringPayoutDto, + ) { + return this.payoutsService.createRecurring(merchantId, dto); + } + + @Get('recurring') + @UseGuards(JwtAuthGuard) + async listRecurring(@CurrentMerchant('id') merchantId: string) { + return this.payoutsService.listRecurring(merchantId); + } + + @Get('recurring/:id') + @UseGuards(JwtAuthGuard) + async getRecurring( + @CurrentMerchant('id') merchantId: string, + @Param('id') id: string, + ) { + return this.payoutsService.getRecurring(id, merchantId); + } + + @Patch('recurring/:id') + @UseGuards(JwtAuthGuard) + async updateRecurring( + @CurrentMerchant('id') merchantId: string, + @Param('id') id: string, + @Body(new ZodValidationPipe(UpdateRecurringPayoutSchema)) + dto: UpdateRecurringPayoutDto, + ) { + return this.payoutsService.updateRecurring(id, merchantId, dto); + } + + @Delete('recurring/:id') + @UseGuards(JwtAuthGuard) + async deleteRecurring( + @CurrentMerchant('id') merchantId: string, + @Param('id') id: string, + ) { + return this.payoutsService.deleteRecurring(id, merchantId); + } + // ── GET /v1/payouts/:id ─────────────────────────────────────────────────── @Get(':id') async getOne( diff --git a/apps/api/src/modules/payouts/payouts.processor.ts b/apps/api/src/modules/payouts/payouts.processor.ts index 3bb248c..d8ede0c 100644 --- a/apps/api/src/modules/payouts/payouts.processor.ts +++ b/apps/api/src/modules/payouts/payouts.processor.ts @@ -5,12 +5,17 @@ import { PayoutsService } from './payouts.service'; import { EXECUTE_PAYOUT_BATCH_JOB, EXECUTE_PAYOUT_JOB, + GENERATE_RECURRING_PAYOUT_JOB, + type GenerateRecurringPayoutJobData, type ExecutePayoutBatchJobData, type ExecutePayoutJobData, PAYOUTS_QUEUE, } from './payouts.constants'; -type PayoutJobData = ExecutePayoutJobData | ExecutePayoutBatchJobData; +type PayoutJobData = + | ExecutePayoutJobData + | ExecutePayoutBatchJobData + | GenerateRecurringPayoutJobData; @Injectable() @Processor(PAYOUTS_QUEUE) @@ -34,6 +39,14 @@ export class PayoutsProcessor extends WorkerHost { return; } + if (job.name === GENERATE_RECURRING_PAYOUT_JOB) { + const data = job.data as GenerateRecurringPayoutJobData; + await this.payoutsService.processRecurringPayout( + data.recurringPayoutId, + ); + return; + } + this.logger.warn(`Unknown payout job ignored: ${job.name}`); } } diff --git a/apps/api/src/modules/payouts/payouts.service.ts b/apps/api/src/modules/payouts/payouts.service.ts index 77bea07..ebc63c9 100644 --- a/apps/api/src/modules/payouts/payouts.service.ts +++ b/apps/api/src/modules/payouts/payouts.service.ts @@ -14,22 +14,31 @@ import { Payout, PayoutStatus, Prisma, + RecurringPayout, + RecurringPayoutFrequency, } from '@prisma/client'; import { PrismaService } from '../prisma/prisma.service'; import { WebhooksService } from '../webhooks/webhooks.service'; import { StellarService } from '../stellar/stellar.service'; import { EventsService } from '../events/events/events.service'; import { NotificationsService } from '../notifications/notifications.service'; -import { CreatePayoutDto, BulkPayoutDto } from './dto/create-payout.dto'; +import { + CreatePayoutDto, + BulkPayoutDto, + CreateRecurringPayoutDto, + UpdateRecurringPayoutDto, +} from './dto/create-payout.dto'; import { PayoutFiltersDto } from './dto/payout-filters.dto'; import { randomUUID } from 'crypto'; import { EXECUTE_PAYOUT_BATCH_JOB, EXECUTE_PAYOUT_JOB, + GENERATE_RECURRING_PAYOUT_JOB, PAYOUTS_QUEUE, PAYOUT_JOB_CLEANUP, type ExecutePayoutBatchJobData, type ExecutePayoutJobData, + type GenerateRecurringPayoutJobData, } from './payouts.constants'; // ── Types ───────────────────────────────────────────────────────────────────── @@ -65,12 +74,15 @@ export class PayoutsService implements OnApplicationBootstrap { private readonly notifications: NotificationsService, @InjectQueue(PAYOUTS_QUEUE) private readonly payoutQueue: Queue< - ExecutePayoutJobData | ExecutePayoutBatchJobData + | ExecutePayoutJobData + | ExecutePayoutBatchJobData + | GenerateRecurringPayoutJobData >, ) {} async onApplicationBootstrap(): Promise { await this.reconcileScheduledPayoutJobs(); + await this.reconcileRecurringPayoutJobs(); } // ── Create single payout ────────────────────────────────────────────────── @@ -275,6 +287,116 @@ export class PayoutsService implements OnApplicationBootstrap { return payout; } + // ── Recurring payouts ──────────────────────────────────────────────────── + + async createRecurring( + merchantId: string, + dto: CreateRecurringPayoutDto, + ): Promise { + const finalDto = await this.resolveRecipientDetails(merchantId, dto); + const startsAt = dto.startsAt ?? this.nextRunAfter(new Date(), dto.frequency); + + const recurring = await this.prisma.recurringPayout.create({ + data: { + merchantId, + recipientId: dto.recipientId ?? null, + recipientName: finalDto.recipientName, + destinationType: finalDto.destinationType as DestType, + destination: finalDto.destination as Prisma.InputJsonValue, + amount: dto.amount, + currency: dto.currency, + frequency: dto.frequency as RecurringPayoutFrequency, + nextRunAt: startsAt, + }, + }); + + await this.enqueueRecurringPayout(recurring); + return recurring; + } + + async listRecurring(merchantId: string): Promise { + return this.prisma.recurringPayout.findMany({ + where: { merchantId }, + orderBy: { createdAt: 'desc' }, + }); + } + + async getRecurring( + id: string, + merchantId: string, + ): Promise { + const recurring = await this.prisma.recurringPayout.findUnique({ + where: { id }, + }); + if (!recurring || recurring.merchantId !== merchantId) { + throw new NotFoundException('Recurring payout not found'); + } + return recurring; + } + + async updateRecurring( + id: string, + merchantId: string, + dto: UpdateRecurringPayoutDto, + ): Promise { + const existing = await this.getRecurring(id, merchantId); + const finalDto = + dto.recipientId || dto.destination + ? await this.resolveRecipientDetails(merchantId, { + recipientId: dto.recipientId ?? existing.recipientId ?? undefined, + recipientName: dto.recipientName ?? existing.recipientName, + destinationType: + dto.destinationType ?? existing.destinationType, + destination: + dto.destination ?? + (existing.destination as unknown as CreatePayoutDto['destination']), + amount: dto.amount ?? existing.amount.toString(), + currency: dto.currency ?? existing.currency, + }) + : null; + + const updated = await this.prisma.recurringPayout.update({ + where: { id }, + data: { + ...(dto.recipientId !== undefined && { + recipientId: dto.recipientId, + }), + ...(finalDto && { recipientName: finalDto.recipientName }), + ...(finalDto && { + destinationType: finalDto.destinationType as DestType, + destination: finalDto.destination as Prisma.InputJsonValue, + }), + ...(dto.amount !== undefined && { amount: dto.amount }), + ...(dto.currency !== undefined && { currency: dto.currency }), + ...(dto.frequency !== undefined && { + frequency: dto.frequency as RecurringPayoutFrequency, + }), + ...(dto.startsAt !== undefined && { nextRunAt: dto.startsAt }), + ...(dto.active !== undefined && { active: dto.active }), + }, + }); + + await this.removeRecurringPayoutJob(updated.id); + if (updated.active) { + await this.enqueueRecurringPayout(updated); + } + + return updated; + } + + async deleteRecurring( + id: string, + merchantId: string, + ): Promise { + const existing = await this.getRecurring(id, merchantId); + const updated = await this.prisma.recurringPayout.update({ + where: { id: existing.id }, + data: { active: false }, + }); + await this.removeRecurringPayoutJob(existing.id); + return updated; + } + // ── Cancel ──────────────────────────────────────────────────────────────── async cancel(id: string, merchantId: string): Promise { @@ -362,6 +484,39 @@ export class PayoutsService implements OnApplicationBootstrap { } } + async processRecurringPayout(recurringPayoutId: string): Promise { + const recurring = await this.prisma.recurringPayout.findUnique({ + where: { id: recurringPayoutId }, + }); + if (!recurring || !recurring.active) return; + + const now = new Date(); + const payout = await this.prisma.payout.create({ + data: { + merchantId: recurring.merchantId, + recipientId: recurring.recipientId, + recipientName: recurring.recipientName, + destinationType: recurring.destinationType, + destination: recurring.destination as Prisma.InputJsonValue, + amount: recurring.amount, + currency: recurring.currency, + status: PayoutStatus.PENDING, + scheduledAt: now, + recurringPayoutId: recurring.id, + }, + }); + + await this.prisma.recurringPayout.update({ + where: { id: recurring.id }, + data: { + lastRunAt: now, + nextRunAt: this.nextRunAfter(now, recurring.frequency), + }, + }); + + await this.enqueuePayout(payout); + } + private async processPayout(payout: Payout): Promise { const started = await this.startPayoutProcessing(payout.id); if (!started) return; @@ -661,6 +816,31 @@ export class PayoutsService implements OnApplicationBootstrap { ); } + private async enqueueRecurringPayout( + recurring: RecurringPayout, + ): Promise { + await this.payoutQueue.add( + GENERATE_RECURRING_PAYOUT_JOB, + { recurringPayoutId: recurring.id }, + { + jobId: `recurring-payout:${recurring.id}`, + delay: Math.max(0, recurring.nextRunAt.getTime() - Date.now()), + repeat: { + every: this.frequencyToMilliseconds(recurring.frequency), + key: this.recurringJobKey(recurring.id), + }, + attempts: 1, + ...PAYOUT_JOB_CLEANUP, + }, + ); + } + + private async removeRecurringPayoutJob(id: string): Promise { + await this.payoutQueue + .removeRepeatableByKey(this.recurringJobKey(id)) + .catch(() => undefined); + } + private async reconcileScheduledPayoutJobs(): Promise { const scheduled = await this.prisma.payout.findMany({ where: { @@ -679,4 +859,69 @@ export class PayoutsService implements OnApplicationBootstrap { ); } } + + private async reconcileRecurringPayoutJobs(): Promise { + const recurringPayouts = await this.prisma.recurringPayout.findMany({ + where: { active: true }, + }); + + for (const recurring of recurringPayouts) { + await this.enqueueRecurringPayout(recurring); + } + + if (recurringPayouts.length > 0) { + this.logger.log( + `Reconciled ${recurringPayouts.length} recurring payout job(s)`, + ); + } + } + + private async resolveRecipientDetails( + merchantId: string, + dto: T, + ): Promise { + if (!dto.recipientId) return dto; + + const recipient = await this.prisma.recipient.findFirst({ + where: { id: dto.recipientId, merchantId }, + }); + if (!recipient) { + throw new NotFoundException('Recipient not found'); + } + + return { + ...dto, + recipientName: recipient.name, + destinationType: recipient.type, + destination: recipient.details as unknown as T['destination'], + }; + } + + private frequencyToMilliseconds( + frequency: RecurringPayoutFrequency, + ): number { + switch (frequency) { + case RecurringPayoutFrequency.DAILY: + return 24 * 60 * 60 * 1000; + case RecurringPayoutFrequency.WEEKLY: + return 7 * 24 * 60 * 60 * 1000; + case RecurringPayoutFrequency.BI_WEEKLY: + return 14 * 24 * 60 * 60 * 1000; + case RecurringPayoutFrequency.MONTHLY: + return 30 * 24 * 60 * 60 * 1000; + } + } + + private nextRunAfter( + date: Date, + frequency: RecurringPayoutFrequency | `${RecurringPayoutFrequency}`, + ): Date { + return new Date(date.getTime() + this.frequencyToMilliseconds( + frequency as RecurringPayoutFrequency, + )); + } + + private recurringJobKey(id: string): string { + return `recurring-payout:${id}`; + } } From 2db5c06e48a0d523c23938fe8c19133158bf46f8 Mon Sep 17 00:00:00 2001 From: enochakinbode Date: Tue, 28 Jul 2026 15:49:43 +0100 Subject: [PATCH 05/13] Require confirmation for high value payouts --- .../migration.sql | 4 + apps/api/prisma/schema.prisma | 2 + .../notifications/notifications.service.ts | 18 +++ .../modules/notifications/templates/index.ts | 25 ++- .../modules/payouts/dto/create-payout.dto.ts | 6 + .../modules/payouts/dto/payout-filters.dto.ts | 9 +- .../src/modules/payouts/payouts.controller.ts | 12 ++ .../src/modules/payouts/payouts.service.ts | 143 ++++++++++++++++-- 8 files changed, 206 insertions(+), 13 deletions(-) create mode 100644 apps/api/prisma/migrations/20260728123000_payout_confirmation_codes/migration.sql diff --git a/apps/api/prisma/migrations/20260728123000_payout_confirmation_codes/migration.sql b/apps/api/prisma/migrations/20260728123000_payout_confirmation_codes/migration.sql new file mode 100644 index 0000000..8b3b1e0 --- /dev/null +++ b/apps/api/prisma/migrations/20260728123000_payout_confirmation_codes/migration.sql @@ -0,0 +1,4 @@ +-- AlterTable +ALTER TABLE "Payout" + ADD COLUMN "confirmationCodeHash" TEXT, + ADD COLUMN "confirmationCodeExpiresAt" TIMESTAMP(3); diff --git a/apps/api/prisma/schema.prisma b/apps/api/prisma/schema.prisma index 82c93aa..b33fd29 100644 --- a/apps/api/prisma/schema.prisma +++ b/apps/api/prisma/schema.prisma @@ -306,6 +306,8 @@ model Payout { stellarTxHash String? scheduledAt DateTime? confirmedAt DateTime? + confirmationCodeHash String? + confirmationCodeExpiresAt DateTime? completedAt DateTime? failureReason String? batchId String? diff --git a/apps/api/src/modules/notifications/notifications.service.ts b/apps/api/src/modules/notifications/notifications.service.ts index 7284815..863d646 100644 --- a/apps/api/src/modules/notifications/notifications.service.ts +++ b/apps/api/src/modules/notifications/notifications.service.ts @@ -431,4 +431,22 @@ export class NotificationsService { html: templates.payoutConfirmationTemplate(payout), }); } + + async sendPayoutApprovalCode( + merchantEmail: string, + payout: { + code: string; + payoutId: string; + recipientName: string; + amount: string; + currency: string; + expiresInMinutes: number; + }, + ): Promise { + await this.dispatch({ + to: merchantEmail, + subject: `Confirm payout ${payout.payoutId}`, + html: templates.payoutApprovalCodeTemplate(payout), + }); + } } diff --git a/apps/api/src/modules/notifications/templates/index.ts b/apps/api/src/modules/notifications/templates/index.ts index 1230178..1c9ca1b 100644 --- a/apps/api/src/modules/notifications/templates/index.ts +++ b/apps/api/src/modules/notifications/templates/index.ts @@ -292,10 +292,33 @@ export const payoutConfirmationTemplate = (payout: Payout) =>

- +
Amount${formatAmount(payout.amount, payout.currency)}
${formatAmount(Number(payout.amount), payout.currency)}
Destination ${esc(payout.destination)}
Status ${esc(payout.status)}
`); + +export const payoutApprovalCodeTemplate = (payout: { + code: string; + payoutId: string; + recipientName: string; + amount: string; + currency: string; + expiresInMinutes: number; +}) => + layout(` +

Confirm Payout

+

+ Use this code to confirm payout ${esc(payout.payoutId)} to ${esc(payout.recipientName)}. + It expires in ${payout.expiresInMinutes} minutes. +

+
+ ${esc(payout.code)} +
+ + + +
Amount${formatAmount(Number(payout.amount), payout.currency)}
+ `); diff --git a/apps/api/src/modules/payouts/dto/create-payout.dto.ts b/apps/api/src/modules/payouts/dto/create-payout.dto.ts index 76b24bd..2ae3a36 100644 --- a/apps/api/src/modules/payouts/dto/create-payout.dto.ts +++ b/apps/api/src/modules/payouts/dto/create-payout.dto.ts @@ -91,6 +91,12 @@ export type UpdateRecurringPayoutDto = z.infer< typeof UpdateRecurringPayoutSchema >; +export const ConfirmPayoutSchema = z.object({ + code: z.string().regex(/^\d{6}$/, 'code must be a 6-digit string'), +}); + +export type ConfirmPayoutDto = z.infer; + // ── Bulk payout ──────────────────────────────────────────────────────────────── export const BulkPayoutSchema = z.object({ diff --git a/apps/api/src/modules/payouts/dto/payout-filters.dto.ts b/apps/api/src/modules/payouts/dto/payout-filters.dto.ts index a55c2ca..3b6a7f2 100644 --- a/apps/api/src/modules/payouts/dto/payout-filters.dto.ts +++ b/apps/api/src/modules/payouts/dto/payout-filters.dto.ts @@ -2,7 +2,14 @@ import { z } from 'zod'; export const PayoutFiltersSchema = z.object({ status: z - .enum(['PENDING', 'PROCESSING', 'COMPLETED', 'FAILED', 'CANCELLED']) + .enum([ + 'PENDING', + 'REQUIRES_CONFIRMATION', + 'PROCESSING', + 'COMPLETED', + 'FAILED', + 'CANCELLED', + ]) .optional(), destinationType: z .enum(['BANK_ACCOUNT', 'MOBILE_MONEY', 'CRYPTO_WALLET', 'STELLAR']) diff --git a/apps/api/src/modules/payouts/payouts.controller.ts b/apps/api/src/modules/payouts/payouts.controller.ts index 34885dc..d2083a1 100644 --- a/apps/api/src/modules/payouts/payouts.controller.ts +++ b/apps/api/src/modules/payouts/payouts.controller.ts @@ -22,6 +22,8 @@ import { CreateRecurringPayoutDto, UpdateRecurringPayoutSchema, UpdateRecurringPayoutDto, + ConfirmPayoutSchema, + ConfirmPayoutDto, } from './dto/create-payout.dto'; import { PayoutFiltersSchema, @@ -125,6 +127,16 @@ export class PayoutsController { return this.payoutsService.getById(id, merchantId); } + @Post(':id/confirm') + @UseGuards(JwtAuthGuard) + async confirm( + @CurrentMerchant('id') merchantId: string, + @Param('id') id: string, + @Body(new ZodValidationPipe(ConfirmPayoutSchema)) dto: ConfirmPayoutDto, + ) { + return this.payoutsService.confirm(id, merchantId, dto); + } + // ── POST /v1/payouts/:id/cancel ─────────────────────────────────────────── @Post(':id/cancel') @UseGuards(JwtAuthGuard) diff --git a/apps/api/src/modules/payouts/payouts.service.ts b/apps/api/src/modules/payouts/payouts.service.ts index ebc63c9..bd38eb2 100644 --- a/apps/api/src/modules/payouts/payouts.service.ts +++ b/apps/api/src/modules/payouts/payouts.service.ts @@ -7,6 +7,7 @@ import { OnApplicationBootstrap, } from '@nestjs/common'; import { InjectQueue } from '@nestjs/bullmq'; +import { ConfigService } from '@nestjs/config'; import { Queue } from 'bullmq'; import { DestType, @@ -27,9 +28,11 @@ import { BulkPayoutDto, CreateRecurringPayoutDto, UpdateRecurringPayoutDto, + ConfirmPayoutDto, } from './dto/create-payout.dto'; import { PayoutFiltersDto } from './dto/payout-filters.dto'; -import { randomUUID } from 'crypto'; +import { randomInt, randomUUID } from 'crypto'; +import * as bcrypt from 'bcrypt'; import { EXECUTE_PAYOUT_BATCH_JOB, EXECUTE_PAYOUT_JOB, @@ -72,6 +75,7 @@ export class PayoutsService implements OnApplicationBootstrap { private readonly stellar: StellarService, private readonly eventsService: EventsService, private readonly notifications: NotificationsService, + private readonly config: ConfigService, @InjectQueue(PAYOUTS_QUEUE) private readonly payoutQueue: Queue< | ExecutePayoutJobData @@ -126,6 +130,7 @@ export class PayoutsService implements OnApplicationBootstrap { }; } + const confirmation = await this.buildConfirmationFields(dto.amount); const payout = await this.prisma.payout.create({ data: { merchantId, @@ -135,8 +140,12 @@ export class PayoutsService implements OnApplicationBootstrap { destination: finalDto.destination as Prisma.InputJsonValue, amount: dto.amount, currency: dto.currency, - status: PayoutStatus.PENDING, + status: confirmation + ? PayoutStatus.REQUIRES_CONFIRMATION + : PayoutStatus.PENDING, scheduledAt: dto.scheduledAt ?? null, + confirmationCodeHash: confirmation?.hash ?? null, + confirmationCodeExpiresAt: confirmation?.expiresAt ?? null, idempotencyKey: idempotencyKey ?? null, }, }); @@ -149,7 +158,11 @@ export class PayoutsService implements OnApplicationBootstrap { ) .catch(() => undefined); - await this.enqueuePayout(payout); + if (confirmation) { + await this.sendPayoutConfirmationCode(payout, confirmation.code); + } else { + await this.enqueuePayout(payout); + } return payout; } @@ -190,6 +203,7 @@ export class PayoutsService implements OnApplicationBootstrap { }; } + const confirmation = await this.buildConfirmationFields(item.amount); const payout = await this.prisma.payout.create({ data: { merchantId, @@ -199,15 +213,21 @@ export class PayoutsService implements OnApplicationBootstrap { destination: finalItem.destination as Prisma.InputJsonValue, amount: item.amount, currency: item.currency, - status: PayoutStatus.PENDING, + status: confirmation + ? PayoutStatus.REQUIRES_CONFIRMATION + : PayoutStatus.PENDING, scheduledAt: item.scheduledAt ?? null, + confirmationCodeHash: confirmation?.hash ?? null, + confirmationCodeExpiresAt: confirmation?.expiresAt ?? null, batchId, }, }); results.push({ index: i, payoutId: payout.id }); accepted++; - if (payout.scheduledAt && payout.scheduledAt > new Date()) { + if (confirmation) { + await this.sendPayoutConfirmationCode(payout, confirmation.code); + } else if (payout.scheduledAt && payout.scheduledAt > new Date()) { await this.enqueuePayout(payout); } else { dueNow++; @@ -287,6 +307,44 @@ export class PayoutsService implements OnApplicationBootstrap { return payout; } + async confirm( + id: string, + merchantId: string, + dto: ConfirmPayoutDto, + ): Promise { + const payout = await this.getById(id, merchantId); + if (payout.status !== PayoutStatus.REQUIRES_CONFIRMATION) { + throw new BadRequestException( + `Cannot confirm a payout in ${payout.status} status`, + ); + } + if ( + !payout.confirmationCodeHash || + !payout.confirmationCodeExpiresAt || + payout.confirmationCodeExpiresAt.getTime() < Date.now() + ) { + throw new BadRequestException('Payout confirmation code is expired'); + } + + const ok = await bcrypt.compare(dto.code, payout.confirmationCodeHash); + if (!ok) { + throw new BadRequestException('Invalid payout confirmation code'); + } + + const confirmed = await this.prisma.payout.update({ + where: { id: payout.id }, + data: { + status: PayoutStatus.PENDING, + confirmedAt: new Date(), + confirmationCodeHash: null, + confirmationCodeExpiresAt: null, + }, + }); + + await this.enqueuePayout(confirmed); + return confirmed; + } + // ── Recurring payouts ──────────────────────────────────────────────────── async createRecurring( @@ -402,9 +460,12 @@ export class PayoutsService implements OnApplicationBootstrap { async cancel(id: string, merchantId: string): Promise { const payout = await this.getById(id, merchantId); - if (payout.status !== PayoutStatus.PENDING) { + if ( + payout.status !== PayoutStatus.PENDING && + payout.status !== PayoutStatus.REQUIRES_CONFIRMATION + ) { throw new BadRequestException( - `Cannot cancel a payout in ${payout.status} status — only PENDING payouts can be cancelled`, + `Cannot cancel a payout in ${payout.status} status — only pending payouts can be cancelled`, ); } @@ -425,9 +486,19 @@ export class PayoutsService implements OnApplicationBootstrap { ); } + const confirmation = await this.buildConfirmationFields( + payout.amount.toString(), + ); const reset = await this.prisma.payout.update({ where: { id }, - data: { status: PayoutStatus.PENDING, failureReason: null }, + data: { + status: confirmation + ? PayoutStatus.REQUIRES_CONFIRMATION + : PayoutStatus.PENDING, + failureReason: null, + confirmationCodeHash: confirmation?.hash ?? null, + confirmationCodeExpiresAt: confirmation?.expiresAt ?? null, + }, }); this.webhooks @@ -438,7 +509,11 @@ export class PayoutsService implements OnApplicationBootstrap { ) .catch(() => undefined); - await this.enqueuePayout(reset); + if (confirmation) { + await this.sendPayoutConfirmationCode(reset, confirmation.code); + } else { + await this.enqueuePayout(reset); + } return reset; } @@ -491,6 +566,9 @@ export class PayoutsService implements OnApplicationBootstrap { if (!recurring || !recurring.active) return; const now = new Date(); + const confirmation = await this.buildConfirmationFields( + recurring.amount.toString(), + ); const payout = await this.prisma.payout.create({ data: { merchantId: recurring.merchantId, @@ -500,8 +578,12 @@ export class PayoutsService implements OnApplicationBootstrap { destination: recurring.destination as Prisma.InputJsonValue, amount: recurring.amount, currency: recurring.currency, - status: PayoutStatus.PENDING, + status: confirmation + ? PayoutStatus.REQUIRES_CONFIRMATION + : PayoutStatus.PENDING, scheduledAt: now, + confirmationCodeHash: confirmation?.hash ?? null, + confirmationCodeExpiresAt: confirmation?.expiresAt ?? null, recurringPayoutId: recurring.id, }, }); @@ -514,7 +596,11 @@ export class PayoutsService implements OnApplicationBootstrap { }, }); - await this.enqueuePayout(payout); + if (confirmation) { + await this.sendPayoutConfirmationCode(payout, confirmation.code); + } else { + await this.enqueuePayout(payout); + } } private async processPayout(payout: Payout): Promise { @@ -924,4 +1010,39 @@ export class PayoutsService implements OnApplicationBootstrap { private recurringJobKey(id: string): string { return `recurring-payout:${id}`; } + + private async buildConfirmationFields( + amount: string, + ): Promise<{ code: string; hash: string; expiresAt: Date } | null> { + const threshold = Number(this.config.get('PAYOUT_2FA_THRESHOLD')); + if (!Number.isFinite(threshold) || threshold <= 0) return null; + if (Number(amount) <= threshold) return null; + + const code = randomInt(0, 1_000_000).toString().padStart(6, '0'); + return { + code, + hash: await bcrypt.hash(code, 10), + expiresAt: new Date(Date.now() + 10 * 60 * 1000), + }; + } + + private async sendPayoutConfirmationCode( + payout: Payout, + code: string, + ): Promise { + const merchant = await this.prisma.merchant.findUnique({ + where: { id: payout.merchantId }, + select: { email: true }, + }); + if (!merchant) return; + + await this.notifications.sendPayoutApprovalCode(merchant.email, { + code, + payoutId: payout.id, + recipientName: payout.recipientName, + amount: payout.amount.toString(), + currency: payout.currency, + expiresInMinutes: 10, + }); + } } From f35c22d2187db132a9428b07bc779b2ad863c5ee Mon Sep 17 00:00:00 2001 From: enochakinbode Date: Tue, 28 Jul 2026 15:51:14 +0100 Subject: [PATCH 06/13] Send Stellar payouts from merchant wallets --- apps/api/src/modules/payouts/payouts.module.ts | 2 ++ .../api/src/modules/payouts/payouts.service.ts | 18 ++++++++++++++++++ .../api/src/modules/stellar/stellar.service.ts | 11 ++++++++--- 3 files changed, 28 insertions(+), 3 deletions(-) diff --git a/apps/api/src/modules/payouts/payouts.module.ts b/apps/api/src/modules/payouts/payouts.module.ts index 14714b2..722b87a 100644 --- a/apps/api/src/modules/payouts/payouts.module.ts +++ b/apps/api/src/modules/payouts/payouts.module.ts @@ -10,12 +10,14 @@ import { StellarModule } from '../stellar/stellar.module'; import { AuthModule } from '../auth/auth.module'; import { EventsModule } from '../events/events.module'; import { NotificationsModule } from '../notifications/notifications.module'; +import { MerchantModule } from '../merchant/merchant.module'; @Module({ imports: [ PrismaModule, WebhooksModule, StellarModule, + MerchantModule, AuthModule, EventsModule, NotificationsModule, diff --git a/apps/api/src/modules/payouts/payouts.service.ts b/apps/api/src/modules/payouts/payouts.service.ts index bd38eb2..dfee4a2 100644 --- a/apps/api/src/modules/payouts/payouts.service.ts +++ b/apps/api/src/modules/payouts/payouts.service.ts @@ -23,6 +23,7 @@ import { WebhooksService } from '../webhooks/webhooks.service'; import { StellarService } from '../stellar/stellar.service'; import { EventsService } from '../events/events/events.service'; import { NotificationsService } from '../notifications/notifications.service'; +import { MerchantSettlementService } from '../merchant/merchant-settlement.service'; import { CreatePayoutDto, BulkPayoutDto, @@ -73,6 +74,7 @@ export class PayoutsService implements OnApplicationBootstrap { private readonly prisma: PrismaService, private readonly webhooks: WebhooksService, private readonly stellar: StellarService, + private readonly settlement: MerchantSettlementService, private readonly eventsService: EventsService, private readonly notifications: NotificationsService, private readonly config: ConfigService, @@ -638,6 +640,9 @@ export class PayoutsService implements OnApplicationBootstrap { typeof destination.asset === 'string' ? destination.asset : 'native'; const sourceAsset = 'native'; const sourceAmount = payout.amount.toString(); + const sourceSecret = await this.getMerchantPayoutSourceSecret( + payout.merchantId, + ); let txHash: string; @@ -648,6 +653,7 @@ export class PayoutsService implements OnApplicationBootstrap { asset: sourceAsset, amount: sourceAmount, destinationAccount: destAddress, + sourceSecret, }); } else { const { paths } = await this.stellar.findStrictSendPaths({ @@ -673,6 +679,7 @@ export class PayoutsService implements OnApplicationBootstrap { destinationAsset: destAsset, destinationMinAmount: destMin, path: pathStrings, + sourceSecret, }); } @@ -1045,4 +1052,15 @@ export class PayoutsService implements OnApplicationBootstrap { expiresInMinutes: 10, }); } + + private async getMerchantPayoutSourceSecret( + merchantId: string, + ): Promise { + const settlementKey = await this.prisma.merchantSettlementKey.findUnique({ + where: { merchantId }, + }); + if (!settlementKey?.managed) return undefined; + + return this.settlement.decryptSeed(settlementKey); + } } diff --git a/apps/api/src/modules/stellar/stellar.service.ts b/apps/api/src/modules/stellar/stellar.service.ts index f0e49eb..07b533f 100644 --- a/apps/api/src/modules/stellar/stellar.service.ts +++ b/apps/api/src/modules/stellar/stellar.service.ts @@ -182,10 +182,11 @@ export class StellarService { destinationMinAmount: string; destinationAccount: string; path: string[]; + sourceSecret?: string; }): Promise { this.logger.log('Executing Stellar path payment'); - const keypair = this.requireKeypair(); + const keypair = this.requireKeypair(params.sourceSecret); const account = await this.horizonServer.loadAccount(keypair.publicKey()); const tx = new StellarSdk.TransactionBuilder(account, { @@ -220,10 +221,11 @@ export class StellarService { asset: string; amount: string; destinationAccount: string; + sourceSecret?: string; }): Promise { this.logger.log('Executing Stellar direct payment'); - const keypair = this.requireKeypair(); + const keypair = this.requireKeypair(params.sourceSecret); const account = await this.horizonServer.loadAccount(keypair.publicKey()); const tx = new StellarSdk.TransactionBuilder(account, { @@ -285,7 +287,10 @@ export class StellarService { // ── Private helpers ──────────────────────────────────────────────────────── - private requireKeypair(): StellarSdk.Keypair { + private requireKeypair(sourceSecret?: string): StellarSdk.Keypair { + if (sourceSecret) { + return StellarSdk.Keypair.fromSecret(sourceSecret); + } if (!this.relayKeypair) { throw new Error('STELLAR_RELAY_KEYPAIR_SECRET is not configured'); } From 95897d15df62501d91f5aa581469f3616ee9376c Mon Sep 17 00:00:00 2001 From: enochakinbode Date: Tue, 28 Jul 2026 15:52:36 +0100 Subject: [PATCH 07/13] Update payment balance test mocks --- .../modules/payments/payments.service.spec.ts | 16 +++++++++++++++- 1 file changed, 15 insertions(+), 1 deletion(-) diff --git a/apps/api/src/modules/payments/payments.service.spec.ts b/apps/api/src/modules/payments/payments.service.spec.ts index bf09d11..e37b2f6 100644 --- a/apps/api/src/modules/payments/payments.service.spec.ts +++ b/apps/api/src/modules/payments/payments.service.spec.ts @@ -57,6 +57,13 @@ describe('PaymentsService', () => { webhookEvent: { create: jest.fn(), }, + merchantBalance: { + upsert: jest.fn(), + }, + merchantLedgerEntry: { + create: jest.fn(), + findFirst: jest.fn(), + }, $transaction: jest.fn(), }; @@ -112,7 +119,14 @@ describe('PaymentsService', () => { prisma.payment.findUnique.mockResolvedValue(paymentRecord); prisma.payment.update.mockResolvedValue(paymentRecord); prisma.webhookEvent.create.mockResolvedValue({}); - prisma.$transaction.mockResolvedValue(undefined); + prisma.merchantBalance.upsert.mockResolvedValue({}); + prisma.merchantLedgerEntry.create.mockResolvedValue({}); + prisma.merchantLedgerEntry.findFirst.mockResolvedValue(null); + prisma.$transaction.mockImplementation(async (arg: unknown) => { + if (Array.isArray(arg)) return Promise.all(arg); + if (typeof arg === 'function') return arg(prisma); + return arg; + }); const module: TestingModule = await Test.createTestingModule({ providers: [ From 6870d6bf455b8d01eb0b469f9c47e4b00bf04ced Mon Sep 17 00:00:00 2001 From: enochakinbode Date: Tue, 28 Jul 2026 16:33:34 +0100 Subject: [PATCH 08/13] Use BullMQ safe payout job IDs --- apps/api/src/modules/payouts/payouts.service.ts | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/apps/api/src/modules/payouts/payouts.service.ts b/apps/api/src/modules/payouts/payouts.service.ts index dfee4a2..2f1c169 100644 --- a/apps/api/src/modules/payouts/payouts.service.ts +++ b/apps/api/src/modules/payouts/payouts.service.ts @@ -889,7 +889,7 @@ export class PayoutsService implements OnApplicationBootstrap { EXECUTE_PAYOUT_JOB, { payoutId: payout.id }, { - jobId: `payout:${payout.id}`, + jobId: `payout-${payout.id}`, delay, attempts: 1, ...PAYOUT_JOB_CLEANUP, @@ -902,7 +902,7 @@ export class PayoutsService implements OnApplicationBootstrap { EXECUTE_PAYOUT_BATCH_JOB, { batchId }, { - jobId: `payout-batch:${batchId}`, + jobId: `payout-batch-${batchId}`, attempts: 1, ...PAYOUT_JOB_CLEANUP, }, @@ -916,7 +916,7 @@ export class PayoutsService implements OnApplicationBootstrap { GENERATE_RECURRING_PAYOUT_JOB, { recurringPayoutId: recurring.id }, { - jobId: `recurring-payout:${recurring.id}`, + jobId: `recurring-payout-${recurring.id}`, delay: Math.max(0, recurring.nextRunAt.getTime() - Date.now()), repeat: { every: this.frequencyToMilliseconds(recurring.frequency), @@ -1015,7 +1015,7 @@ export class PayoutsService implements OnApplicationBootstrap { } private recurringJobKey(id: string): string { - return `recurring-payout:${id}`; + return `recurring-payout-${id}`; } private async buildConfirmationFields( From e06025fd836eebea8d848a618e2e694cd329993e Mon Sep 17 00:00:00 2001 From: enochakinbode Date: Tue, 28 Jul 2026 16:39:45 +0100 Subject: [PATCH 09/13] Align recurring payout repeat times --- apps/api/src/modules/payouts/payouts.service.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/apps/api/src/modules/payouts/payouts.service.ts b/apps/api/src/modules/payouts/payouts.service.ts index 2f1c169..260bac5 100644 --- a/apps/api/src/modules/payouts/payouts.service.ts +++ b/apps/api/src/modules/payouts/payouts.service.ts @@ -912,14 +912,15 @@ export class PayoutsService implements OnApplicationBootstrap { private async enqueueRecurringPayout( recurring: RecurringPayout, ): Promise { + const every = this.frequencyToMilliseconds(recurring.frequency); await this.payoutQueue.add( GENERATE_RECURRING_PAYOUT_JOB, { recurringPayoutId: recurring.id }, { jobId: `recurring-payout-${recurring.id}`, - delay: Math.max(0, recurring.nextRunAt.getTime() - Date.now()), repeat: { - every: this.frequencyToMilliseconds(recurring.frequency), + every, + offset: recurring.nextRunAt.getTime() % every, key: this.recurringJobKey(recurring.id), }, attempts: 1, From 292bce13e90c8c736de50a7c2adf60c4ab72edfb Mon Sep 17 00:00:00 2001 From: enochakinbode Date: Tue, 28 Jul 2026 16:43:10 +0100 Subject: [PATCH 10/13] Run due recurring payouts from BullMQ --- .../src/modules/payouts/payouts.constants.ts | 3 ++ .../src/modules/payouts/payouts.processor.ts | 6 ++++ .../src/modules/payouts/payouts.service.ts | 35 +++++++++++++++++++ 3 files changed, 44 insertions(+) diff --git a/apps/api/src/modules/payouts/payouts.constants.ts b/apps/api/src/modules/payouts/payouts.constants.ts index 6804a5b..443e0b2 100644 --- a/apps/api/src/modules/payouts/payouts.constants.ts +++ b/apps/api/src/modules/payouts/payouts.constants.ts @@ -3,6 +3,7 @@ export const PAYOUTS_QUEUE = 'payouts'; export const EXECUTE_PAYOUT_JOB = 'execute-payout'; export const EXECUTE_PAYOUT_BATCH_JOB = 'execute-payout-batch'; export const GENERATE_RECURRING_PAYOUT_JOB = 'generate-recurring-payout'; +export const RUN_RECURRING_PAYOUTS_JOB = 'run-recurring-payouts'; export interface ExecutePayoutJobData { payoutId: string; @@ -16,6 +17,8 @@ export interface GenerateRecurringPayoutJobData { recurringPayoutId: string; } +export type RunRecurringPayoutsJobData = Record; + export const PAYOUT_JOB_CLEANUP = { removeOnComplete: { age: 7 * 24 * 60 * 60, diff --git a/apps/api/src/modules/payouts/payouts.processor.ts b/apps/api/src/modules/payouts/payouts.processor.ts index d8ede0c..78eac91 100644 --- a/apps/api/src/modules/payouts/payouts.processor.ts +++ b/apps/api/src/modules/payouts/payouts.processor.ts @@ -6,6 +6,7 @@ import { EXECUTE_PAYOUT_BATCH_JOB, EXECUTE_PAYOUT_JOB, GENERATE_RECURRING_PAYOUT_JOB, + RUN_RECURRING_PAYOUTS_JOB, type GenerateRecurringPayoutJobData, type ExecutePayoutBatchJobData, type ExecutePayoutJobData, @@ -47,6 +48,11 @@ export class PayoutsProcessor extends WorkerHost { return; } + if (job.name === RUN_RECURRING_PAYOUTS_JOB) { + await this.payoutsService.processDueRecurringPayouts(); + return; + } + this.logger.warn(`Unknown payout job ignored: ${job.name}`); } } diff --git a/apps/api/src/modules/payouts/payouts.service.ts b/apps/api/src/modules/payouts/payouts.service.ts index 260bac5..7ba5d16 100644 --- a/apps/api/src/modules/payouts/payouts.service.ts +++ b/apps/api/src/modules/payouts/payouts.service.ts @@ -38,11 +38,13 @@ import { EXECUTE_PAYOUT_BATCH_JOB, EXECUTE_PAYOUT_JOB, GENERATE_RECURRING_PAYOUT_JOB, + RUN_RECURRING_PAYOUTS_JOB, PAYOUTS_QUEUE, PAYOUT_JOB_CLEANUP, type ExecutePayoutBatchJobData, type ExecutePayoutJobData, type GenerateRecurringPayoutJobData, + type RunRecurringPayoutsJobData, } from './payouts.constants'; // ── Types ───────────────────────────────────────────────────────────────────── @@ -83,12 +85,14 @@ export class PayoutsService implements OnApplicationBootstrap { | ExecutePayoutJobData | ExecutePayoutBatchJobData | GenerateRecurringPayoutJobData + | RunRecurringPayoutsJobData >, ) {} async onApplicationBootstrap(): Promise { await this.reconcileScheduledPayoutJobs(); await this.reconcileRecurringPayoutJobs(); + await this.enqueueRecurringDispatcher(); } // ── Create single payout ────────────────────────────────────────────────── @@ -605,6 +609,21 @@ export class PayoutsService implements OnApplicationBootstrap { } } + async processDueRecurringPayouts(): Promise { + const due = await this.prisma.recurringPayout.findMany({ + where: { + active: true, + nextRunAt: { lte: new Date() }, + }, + orderBy: { nextRunAt: 'asc' }, + take: 100, + }); + + for (const recurring of due) { + await this.processRecurringPayout(recurring.id); + } + } + private async processPayout(payout: Payout): Promise { const started = await this.startPayoutProcessing(payout.id); if (!started) return; @@ -935,6 +954,22 @@ export class PayoutsService implements OnApplicationBootstrap { .catch(() => undefined); } + private async enqueueRecurringDispatcher(): Promise { + await this.payoutQueue.add( + RUN_RECURRING_PAYOUTS_JOB, + {}, + { + jobId: 'run-recurring-payouts', + repeat: { + every: 30_000, + key: 'run-recurring-payouts', + }, + attempts: 1, + ...PAYOUT_JOB_CLEANUP, + }, + ); + } + private async reconcileScheduledPayoutJobs(): Promise { const scheduled = await this.prisma.payout.findMany({ where: { From ac0e139cf62a54a55c18b1a739e6b62bebc41b70 Mon Sep 17 00:00:00 2001 From: enochakinbode Date: Tue, 28 Jul 2026 20:42:13 +0100 Subject: [PATCH 11/13] Support seeded payout scheduling tests --- apps/api/prisma/seed.ts | 234 +++++++++++++++++- .../modules/payouts/dto/create-payout.dto.ts | 2 +- .../modules/payouts/dto/payout-filters.dto.ts | 2 +- .../src/modules/payouts/payouts.service.ts | 25 +- 4 files changed, 255 insertions(+), 8 deletions(-) diff --git a/apps/api/prisma/seed.ts b/apps/api/prisma/seed.ts index 8eae162..e7cf619 100644 --- a/apps/api/prisma/seed.ts +++ b/apps/api/prisma/seed.ts @@ -10,11 +10,18 @@ async function main() { // Clean existing data (order matters due to FK constraints) await prisma.webhookEvent.deleteMany(); + await prisma.notification.deleteMany(); + await prisma.merchantLedgerEntry.deleteMany(); await prisma.payment.deleteMany(); await prisma.quote.deleteMany(); await prisma.paymentLink.deleteMany(); await prisma.invoice.deleteMany(); await prisma.payout.deleteMany(); + await prisma.recurringPayout.deleteMany(); + await prisma.recipient.deleteMany(); + await prisma.merchantBalance.deleteMany(); + await prisma.merchantSettlementKey.deleteMany(); + await prisma.apiKey.deleteMany(); await prisma.teamMember.deleteMany(); await prisma.merchant.deleteMany(); @@ -91,6 +98,102 @@ async function main() { console.log('Created team members'); + // ── Balances / Recipients ─────────────────────────────────── + + await prisma.merchantBalance.createMany({ + data: [ + { + merchantId: acme.id, + availableAmount: '25000.00', + reservedAmount: '0', + currency: 'USDC', + }, + { + merchantId: globex.id, + availableAmount: '7500.00', + reservedAmount: '0', + currency: 'USDC', + }, + { + merchantId: initech.id, + availableAmount: '50.00', + reservedAmount: '0', + currency: 'USDC', + }, + ], + }); + + await prisma.merchantLedgerEntry.createMany({ + data: [ + { + merchantId: acme.id, + type: 'CREDIT', + amount: '25000.00', + currency: 'USDC', + description: 'Seeded settlement balance for payout testing', + }, + { + merchantId: globex.id, + type: 'CREDIT', + amount: '7500.00', + currency: 'USDC', + description: 'Seeded settlement balance for payout testing', + }, + { + merchantId: initech.id, + type: 'CREDIT', + amount: '50.00', + currency: 'USDC', + description: 'Seeded low balance for insufficient-funds testing', + }, + ], + }); + + const [acmeStellarRecipient, acmeTreasuryRecipient, globexPartnerRecipient] = + await Promise.all([ + prisma.recipient.create({ + data: { + merchantId: acme.id, + name: 'Acme Stellar Ops Wallet', + type: 'STELLAR', + details: { + type: 'STELLAR', + address: 'GDQP2KPQGKIHYJGXNUIYOMHARUARCA7DJT5FO2FFOOBD3XCDGYOO5SS', + asset: 'USDC', + }, + isDefault: true, + }, + }), + prisma.recipient.create({ + data: { + merchantId: acme.id, + name: 'Acme Treasury Wallet', + type: 'CRYPTO_WALLET', + details: { + type: 'CRYPTO_WALLET', + address: '0x742d35Cc6634C0532925a3b844Bc9e7595f2bD18', + network: 'ethereum', + asset: 'USDC', + }, + }, + }), + prisma.recipient.create({ + data: { + merchantId: globex.id, + name: 'Globex Stellar Partner', + type: 'STELLAR', + details: { + type: 'STELLAR', + address: 'GBZXN7PIRZGNMHGA7MUUUF4GWPY5AHDKSTEBKHL5USQV7IRG3OVRRM', + asset: 'USDC', + }, + isDefault: true, + }, + }), + ]); + + console.log('Created merchant balances, ledger credits, and recipients'); + // ── Quotes ─────────────────────────────────────────────────── const now = new Date(); @@ -429,6 +532,11 @@ async function main() { // ── Payouts ────────────────────────────────────────────────── + const oneMinuteAgo = new Date(now.getTime() - 60_000); + const twoMinutesLater = new Date(now.getTime() + 120_000); + const fiveMinutesLater = new Date(now.getTime() + 300_000); + const confirmationCodeHash = await hash('123456', 10); + await prisma.payout.createMany({ data: [ // Completed bank payout @@ -451,12 +559,14 @@ async function main() { // Completed crypto payout { merchantId: acme.id, + recipientId: acmeTreasuryRecipient.id, recipientName: 'Acme Treasury Wallet', destinationType: 'CRYPTO_WALLET', destination: { - type: 'crypto', - chain: 'ethereum', + type: 'CRYPTO_WALLET', + network: 'ethereum', address: '0x742d35Cc6634C0532925a3b844Bc9e7595f2bD18', + asset: 'USDC', }, amount: '2500.00', currency: 'USDC', @@ -482,17 +592,91 @@ async function main() { // Pending Stellar payout { merchantId: globex.id, + recipientId: globexPartnerRecipient.id, recipientName: 'Globex Partner', destinationType: 'STELLAR', destination: { - type: 'stellar', - address: 'GDQP2KPQGKIHYJGXNUIYOMHARUARCA7DJT5FO2FFOOBD3XCDGYOO5SS', + type: 'STELLAR', + address: 'GBZXN7PIRZGNMHGA7MUUUF4GWPY5AHDKSTEBKHL5USQV7IRG3OVRRM', + asset: 'USDC', }, amount: '800.00', currency: 'USDC', status: 'PENDING', scheduledAt: new Date(now.getTime() + 86_400_000), }, + // Due scheduled payout: API restart reconciliation should enqueue it + // and BullMQ should execute it shortly after startup. + { + merchantId: acme.id, + recipientId: acmeStellarRecipient.id, + recipientName: 'Acme Due Scheduled Payout', + destinationType: 'STELLAR', + destination: { + type: 'STELLAR', + address: 'GDQP2KPQGKIHYJGXNUIYOMHARUARCA7DJT5FO2FFOOBD3XCDGYOO5SS', + asset: 'USDC', + }, + amount: '25.00', + currency: 'USDC', + status: 'PENDING', + scheduledAt: oneMinuteAgo, + idempotencyKey: 'seed-due-scheduled-payout', + }, + // Future scheduled payout: should stay pending until the delayed + // BullMQ job reaches its scheduled time. + { + merchantId: acme.id, + recipientId: acmeStellarRecipient.id, + recipientName: 'Acme Future Scheduled Payout', + destinationType: 'STELLAR', + destination: { + type: 'STELLAR', + address: 'GDQP2KPQGKIHYJGXNUIYOMHARUARCA7DJT5FO2FFOOBD3XCDGYOO5SS', + asset: 'USDC', + }, + amount: '30.00', + currency: 'USDC', + status: 'PENDING', + scheduledAt: twoMinutesLater, + idempotencyKey: 'seed-future-scheduled-payout', + }, + // Confirmation-required payout. Use code 123456 against + // POST /v1/payouts/:id/confirm, then BullMQ should execute it. + { + merchantId: acme.id, + recipientId: acmeStellarRecipient.id, + recipientName: 'Acme High Value Approval Test', + destinationType: 'STELLAR', + destination: { + type: 'STELLAR', + address: 'GDQP2KPQGKIHYJGXNUIYOMHARUARCA7DJT5FO2FFOOBD3XCDGYOO5SS', + asset: 'USDC', + }, + amount: '7500.00', + currency: 'USDC', + status: 'REQUIRES_CONFIRMATION', + confirmationCodeHash, + confirmationCodeExpiresAt: fiveMinutesLater, + idempotencyKey: 'seed-confirmation-required-payout', + }, + // Insufficient-balance payout: Initech has only 50 USDC available, so + // processing this should fail clearly without debiting funds. + { + merchantId: initech.id, + recipientName: 'Initech Insufficient Balance Test', + destinationType: 'STELLAR', + destination: { + type: 'STELLAR', + address: 'GCKFBEIYV2U22IO2BJ4KVJOIP7XPWQGQFKKWXR6DOSJBV7STMAQSMTGG', + asset: 'USDC', + }, + amount: '500.00', + currency: 'USDC', + status: 'PENDING', + scheduledAt: oneMinuteAgo, + idempotencyKey: 'seed-insufficient-balance-payout', + }, // Failed payout { merchantId: initech.id, @@ -528,6 +712,45 @@ async function main() { console.log('Created payouts'); + await prisma.recurringPayout.createMany({ + data: [ + { + merchantId: acme.id, + recipientId: acmeStellarRecipient.id, + recipientName: 'Acme Daily Recurring Payout', + destinationType: 'STELLAR', + destination: { + type: 'STELLAR', + address: 'GDQP2KPQGKIHYJGXNUIYOMHARUARCA7DJT5FO2FFOOBD3XCDGYOO5SS', + asset: 'USDC', + }, + amount: '15.00', + currency: 'USDC', + frequency: 'DAILY', + nextRunAt: oneMinuteAgo, + active: true, + }, + { + merchantId: globex.id, + recipientId: globexPartnerRecipient.id, + recipientName: 'Globex Weekly Partner Payout', + destinationType: 'STELLAR', + destination: { + type: 'STELLAR', + address: 'GBZXN7PIRZGNMHGA7MUUUF4GWPY5AHDKSTEBKHL5USQV7IRG3OVRRM', + asset: 'USDC', + }, + amount: '125.00', + currency: 'USDC', + frequency: 'WEEKLY', + nextRunAt: twoMinutesLater, + active: true, + }, + ], + }); + + console.log('Created recurring payout configs'); + // ── Webhook Events ─────────────────────────────────────────── const completedPayments = payments.filter((p) => p.status === 'COMPLETED'); @@ -590,6 +813,9 @@ async function main() { console.log(`Created ${webhookData.length} webhook events`); console.log('\nSeed completed successfully!'); + console.log( + 'Restart the API after seeding so BullMQ reconciles seeded scheduled and recurring payouts into Redis.', + ); } main() diff --git a/apps/api/src/modules/payouts/dto/create-payout.dto.ts b/apps/api/src/modules/payouts/dto/create-payout.dto.ts index 2ae3a36..e1b83f3 100644 --- a/apps/api/src/modules/payouts/dto/create-payout.dto.ts +++ b/apps/api/src/modules/payouts/dto/create-payout.dto.ts @@ -59,7 +59,7 @@ export const CreatePayoutSchema = z.object({ .refine((v) => parseFloat(v) > 0, { message: 'amount must be greater than 0', }), - currency: z.string().length(3).toUpperCase(), + currency: z.string().min(3).max(12).toUpperCase(), scheduledAt: z.coerce.date().optional(), }); diff --git a/apps/api/src/modules/payouts/dto/payout-filters.dto.ts b/apps/api/src/modules/payouts/dto/payout-filters.dto.ts index 3b6a7f2..fcff493 100644 --- a/apps/api/src/modules/payouts/dto/payout-filters.dto.ts +++ b/apps/api/src/modules/payouts/dto/payout-filters.dto.ts @@ -14,7 +14,7 @@ export const PayoutFiltersSchema = z.object({ destinationType: z .enum(['BANK_ACCOUNT', 'MOBILE_MONEY', 'CRYPTO_WALLET', 'STELLAR']) .optional(), - currency: z.string().length(3).toUpperCase().optional(), + currency: z.string().min(3).max(12).toUpperCase().optional(), dateFrom: z.coerce.date().optional(), dateTo: z.coerce.date().optional(), batchId: z.string().optional(), diff --git a/apps/api/src/modules/payouts/payouts.service.ts b/apps/api/src/modules/payouts/payouts.service.ts index 7ba5d16..5e45904 100644 --- a/apps/api/src/modules/payouts/payouts.service.ts +++ b/apps/api/src/modules/payouts/payouts.service.ts @@ -66,6 +66,14 @@ function assetToString(asset: AssetObject): string { return `${asset.code}:${asset.issuer}`; } +function stellarUsdcAsset(network?: string): string { + const isMainnet = network?.toLowerCase() === 'mainnet'; + const issuer = isMainnet + ? 'GA5ZSEJYB37JRC5AVCIA5MOP4RHTM335X2KGX3IHOJAPP5RE34K4KZVN' + : 'GBBD47IF6LWK7P7MDEVSCWR7DPUWV3NY3DTQEVFL4NAT4AQH3ZLLFLA5'; + return `USDC:${issuer}`; +} + // ── Service ─────────────────────────────────────────────────────────────────── @Injectable() @@ -655,8 +663,7 @@ export class PayoutsService implements OnApplicationBootstrap { destination: Record, ): Promise { const destAddress = String(destination.address); - const destAsset = - typeof destination.asset === 'string' ? destination.asset : 'native'; + const destAsset = this.normalizeStellarAsset(destination.asset); const sourceAsset = 'native'; const sourceAmount = payout.amount.toString(); const sourceSecret = await this.getMerchantPayoutSourceSecret( @@ -734,6 +741,20 @@ export class PayoutsService implements OnApplicationBootstrap { .catch(() => undefined); } + private normalizeStellarAsset(asset: unknown): string { + if (typeof asset !== 'string' || asset.trim() === '') return 'native'; + + const normalized = asset.trim(); + const upper = normalized.toUpperCase(); + if (upper === 'XLM' || normalized.toLowerCase() === 'native') { + return 'native'; + } + if (upper === 'USDC') { + return stellarUsdcAsset(this.config.get('STELLAR_NETWORK')); + } + return normalized; + } + private async startPayoutProcessing(payoutId: string): Promise { const result = await this.prisma.$transaction(async (tx) => { await tx.$executeRaw` From cb6fffa8ac3cd886bd7a93b85a620fb1619a2932 Mon Sep 17 00:00:00 2001 From: enochakinbode Date: Tue, 28 Jul 2026 20:48:34 +0100 Subject: [PATCH 12/13] Test payout scheduling jobs --- .../modules/payments/payments.service.spec.ts | 5 +- .../src/modules/payouts/payouts.processor.ts | 4 +- .../modules/payouts/payouts.service.spec.ts | 496 ++++++++++++++++++ .../src/modules/payouts/payouts.service.ts | 34 +- 4 files changed, 513 insertions(+), 26 deletions(-) create mode 100644 apps/api/src/modules/payouts/payouts.service.spec.ts diff --git a/apps/api/src/modules/payments/payments.service.spec.ts b/apps/api/src/modules/payments/payments.service.spec.ts index e37b2f6..d7356ec 100644 --- a/apps/api/src/modules/payments/payments.service.spec.ts +++ b/apps/api/src/modules/payments/payments.service.spec.ts @@ -124,7 +124,10 @@ describe('PaymentsService', () => { prisma.merchantLedgerEntry.findFirst.mockResolvedValue(null); prisma.$transaction.mockImplementation(async (arg: unknown) => { if (Array.isArray(arg)) return Promise.all(arg); - if (typeof arg === 'function') return arg(prisma); + if (typeof arg === 'function') { + const transaction = arg as (client: typeof prisma) => Promise; + return transaction(prisma); + } return arg; }); diff --git a/apps/api/src/modules/payouts/payouts.processor.ts b/apps/api/src/modules/payouts/payouts.processor.ts index 78eac91..19e8a5b 100644 --- a/apps/api/src/modules/payouts/payouts.processor.ts +++ b/apps/api/src/modules/payouts/payouts.processor.ts @@ -42,9 +42,7 @@ export class PayoutsProcessor extends WorkerHost { if (job.name === GENERATE_RECURRING_PAYOUT_JOB) { const data = job.data as GenerateRecurringPayoutJobData; - await this.payoutsService.processRecurringPayout( - data.recurringPayoutId, - ); + await this.payoutsService.processRecurringPayout(data.recurringPayoutId); return; } diff --git a/apps/api/src/modules/payouts/payouts.service.spec.ts b/apps/api/src/modules/payouts/payouts.service.spec.ts new file mode 100644 index 0000000..59bfc37 --- /dev/null +++ b/apps/api/src/modules/payouts/payouts.service.spec.ts @@ -0,0 +1,496 @@ +import { Prisma, PayoutStatus, DestType } from '@prisma/client'; +import { PayoutsService } from './payouts.service'; +import { + EXECUTE_PAYOUT_BATCH_JOB, + EXECUTE_PAYOUT_JOB, + GENERATE_RECURRING_PAYOUT_JOB, + RUN_RECURRING_PAYOUTS_JOB, +} from './payouts.constants'; + +interface MockPayout { + id: string; + merchantId: string; + recipientId: string | null; + recipientName: string; + destinationType: DestType; + destination: Record; + amount: Prisma.Decimal; + currency: string; + status: PayoutStatus; + stellarTxHash: string | null; + scheduledAt: Date | null; + confirmedAt: Date | null; + confirmationCodeHash: string | null; + confirmationCodeExpiresAt: Date | null; + completedAt: Date | null; + failureReason: string | null; + batchId: string | null; + idempotencyKey: string | null; + recurringPayoutId: string | null; + createdAt: Date; +} + +interface MockRecurringPayout { + id: string; + merchantId: string; + recipientId: string | null; + recipientName: string; + destinationType: DestType; + destination: Record; + amount: Prisma.Decimal; + currency: string; + frequency: 'DAILY' | 'WEEKLY' | 'BI_WEEKLY' | 'MONTHLY'; + nextRunAt: Date; + lastRunAt: Date | null; + active: boolean; + createdAt: Date; + updatedAt: Date; +} + +describe('PayoutsService', () => { + const now = new Date('2026-07-28T12:00:00.000Z'); + const merchantId = 'merchant_123'; + + const queue = { + add: jest.fn(), + removeRepeatableByKey: jest.fn(), + }; + const prisma = { + payout: { + findUnique: jest.fn(), + findMany: jest.fn(), + create: jest.fn(), + update: jest.fn(), + count: jest.fn(), + }, + recurringPayout: { + findUnique: jest.fn(), + findMany: jest.fn(), + create: jest.fn(), + update: jest.fn(), + }, + recipient: { + findFirst: jest.fn(), + }, + merchant: { + findUnique: jest.fn(), + }, + merchantSettlementKey: { + findUnique: jest.fn(), + }, + $transaction: jest.fn(), + }; + const webhooks = { + dispatch: jest.fn(), + }; + const stellar = { + sendPayment: jest.fn(), + findStrictSendPaths: jest.fn(), + executePathPayment: jest.fn(), + }; + const settlement = { + decryptSeed: jest.fn(), + }; + const eventsService = { + emitPayoutStatus: jest.fn(), + }; + const notifications = { + notifyPayoutCompleted: jest.fn(), + notifyPayoutFailed: jest.fn(), + sendPayoutApprovalCode: jest.fn(), + }; + const config = { + get: jest.fn(), + }; + + let service: PayoutsService; + let payoutCounter = 0; + + function makePayout(overrides: Partial = {}): MockPayout { + payoutCounter += 1; + return { + id: overrides.id ?? `payout_${payoutCounter}`, + merchantId: overrides.merchantId ?? merchantId, + recipientId: overrides.recipientId ?? null, + recipientName: overrides.recipientName ?? 'Recipient', + destinationType: overrides.destinationType ?? DestType.BANK_ACCOUNT, + destination: overrides.destination ?? { + type: 'BANK_ACCOUNT', + accountNumber: '1234567890', + country: 'US', + }, + amount: overrides.amount ?? new Prisma.Decimal('10'), + currency: overrides.currency ?? 'USDC', + status: overrides.status ?? PayoutStatus.PENDING, + stellarTxHash: overrides.stellarTxHash ?? null, + scheduledAt: overrides.scheduledAt ?? null, + confirmedAt: overrides.confirmedAt ?? null, + confirmationCodeHash: overrides.confirmationCodeHash ?? null, + confirmationCodeExpiresAt: overrides.confirmationCodeExpiresAt ?? null, + completedAt: overrides.completedAt ?? null, + failureReason: overrides.failureReason ?? null, + batchId: overrides.batchId ?? null, + idempotencyKey: overrides.idempotencyKey ?? null, + recurringPayoutId: overrides.recurringPayoutId ?? null, + createdAt: overrides.createdAt ?? now, + }; + } + + function makeRecurring( + overrides: Partial = {}, + ): MockRecurringPayout { + return { + id: overrides.id ?? 'recurring_123', + merchantId: overrides.merchantId ?? merchantId, + recipientId: overrides.recipientId ?? null, + recipientName: overrides.recipientName ?? 'Recurring Recipient', + destinationType: overrides.destinationType ?? DestType.BANK_ACCOUNT, + destination: overrides.destination ?? { + type: 'BANK_ACCOUNT', + accountNumber: '1234567890', + country: 'US', + }, + amount: overrides.amount ?? new Prisma.Decimal('15'), + currency: overrides.currency ?? 'USDC', + frequency: overrides.frequency ?? 'DAILY', + nextRunAt: overrides.nextRunAt ?? now, + lastRunAt: overrides.lastRunAt ?? null, + active: overrides.active ?? true, + createdAt: overrides.createdAt ?? now, + updatedAt: overrides.updatedAt ?? now, + }; + } + + beforeEach(() => { + jest.useFakeTimers().setSystemTime(now); + jest.clearAllMocks(); + payoutCounter = 0; + + queue.add.mockResolvedValue({}); + queue.removeRepeatableByKey.mockResolvedValue(undefined); + webhooks.dispatch.mockResolvedValue(undefined); + stellar.sendPayment.mockResolvedValue('stellar_tx_hash'); + stellar.findStrictSendPaths.mockResolvedValue({ + paths: [{ path: [], destinationAmount: '10' }], + }); + stellar.executePathPayment.mockResolvedValue('stellar_path_tx_hash'); + settlement.decryptSeed.mockReturnValue(undefined); + eventsService.emitPayoutStatus.mockReturnValue(undefined); + notifications.notifyPayoutCompleted.mockResolvedValue(undefined); + notifications.notifyPayoutFailed.mockResolvedValue(undefined); + notifications.sendPayoutApprovalCode.mockResolvedValue(undefined); + config.get.mockReturnValue(undefined); + prisma.payout.findUnique.mockResolvedValue(null); + prisma.payout.findMany.mockResolvedValue([]); + prisma.payout.count.mockResolvedValue(0); + prisma.payout.create.mockImplementation( + async ({ data }: { data: Partial }) => makePayout(data), + ); + prisma.payout.update.mockImplementation( + async ({ + where, + data, + }: { + where: { id: string }; + data: Partial; + }) => makePayout({ id: where.id, ...data }), + ); + prisma.recurringPayout.findUnique.mockResolvedValue(null); + prisma.recurringPayout.findMany.mockResolvedValue([]); + prisma.recurringPayout.create.mockImplementation( + async ({ data }: { data: Partial }) => + makeRecurring(data), + ); + prisma.recurringPayout.update.mockImplementation( + async ({ + where, + data, + }: { + where: { id: string }; + data: Partial; + }) => makeRecurring({ id: where.id, ...data }), + ); + prisma.recipient.findFirst.mockResolvedValue(null); + prisma.merchant.findUnique.mockResolvedValue({ + email: 'merchant@test.dev', + }); + prisma.merchantSettlementKey.findUnique.mockResolvedValue(null); + prisma.$transaction.mockImplementation(async (callback: unknown) => { + if (typeof callback !== 'function') return callback; + const tx = { + $executeRaw: jest.fn().mockResolvedValue(undefined), + $queryRaw: jest.fn().mockResolvedValue([ + { + availableAmount: new Prisma.Decimal('1000'), + currency: 'USDC', + }, + ]), + payout: { + findUnique: jest.fn().mockResolvedValue(makePayout()), + update: jest + .fn() + .mockImplementation( + async ({ + where, + data, + }: { + where: { id: string }; + data: Partial; + }) => makePayout({ id: where.id, ...data }), + ), + }, + merchantBalance: { + update: jest.fn().mockResolvedValue({}), + }, + merchantLedgerEntry: { + create: jest.fn().mockResolvedValue({}), + }, + }; + const transaction = callback as (client: typeof tx) => Promise; + return transaction(tx); + }); + + service = new PayoutsService( + prisma as never, + webhooks as never, + stellar as never, + settlement as never, + eventsService as never, + notifications as never, + config as never, + queue as never, + ); + }); + + afterEach(() => { + jest.useRealTimers(); + }); + + it('queues scheduled payouts as delayed BullMQ jobs', async () => { + const scheduledAt = new Date(now.getTime() + 60_000); + const payout = makePayout({ id: 'scheduled_1', scheduledAt }); + prisma.payout.create.mockResolvedValueOnce(payout); + + await service.create(merchantId, { + recipientName: 'Scheduled Recipient', + destinationType: 'BANK_ACCOUNT', + destination: { + type: 'BANK_ACCOUNT', + accountNumber: '1234567890', + country: 'US', + }, + amount: '10', + currency: 'USDC', + scheduledAt, + }); + + expect(queue.add).toHaveBeenCalledWith( + EXECUTE_PAYOUT_JOB, + { payoutId: 'scheduled_1' }, + expect.objectContaining({ + jobId: 'payout-scheduled_1', + delay: 60_000, + attempts: 1, + }), + ); + }); + + it('groups due bulk payouts into a single batch job', async () => { + prisma.payout.create + .mockResolvedValueOnce(makePayout({ id: 'bulk_1' })) + .mockResolvedValueOnce(makePayout({ id: 'bulk_2' })); + + const result = await service.createBulk(merchantId, { + payouts: [ + { + recipientName: 'Bulk One', + destinationType: 'BANK_ACCOUNT', + destination: { + type: 'BANK_ACCOUNT', + accountNumber: '1234567890', + country: 'US', + }, + amount: '10', + currency: 'USDC', + }, + { + recipientName: 'Bulk Two', + destinationType: 'BANK_ACCOUNT', + destination: { + type: 'BANK_ACCOUNT', + accountNumber: '9876543210', + country: 'US', + }, + amount: '20', + currency: 'USDC', + }, + ], + }); + + expect(result.accepted).toBe(2); + expect(queue.add).toHaveBeenCalledTimes(1); + expect(queue.add).toHaveBeenCalledWith( + EXECUTE_PAYOUT_BATCH_JOB, + { batchId: result.batchId }, + expect.objectContaining({ + jobId: `payout-batch-${result.batchId}`, + attempts: 1, + }), + ); + }); + + it('requires confirmation for payouts above the configured threshold', async () => { + config.get.mockImplementation((key: string) => + key === 'PAYOUT_2FA_THRESHOLD' ? '100' : undefined, + ); + const highValuePayout = makePayout({ + id: 'approval_1', + amount: new Prisma.Decimal('500'), + status: PayoutStatus.REQUIRES_CONFIRMATION, + confirmationCodeHash: 'hashed-code', + confirmationCodeExpiresAt: new Date(now.getTime() + 600_000), + }); + prisma.payout.create.mockResolvedValueOnce(highValuePayout); + + const payout = await service.create(merchantId, { + recipientName: 'High Value Recipient', + destinationType: 'BANK_ACCOUNT', + destination: { + type: 'BANK_ACCOUNT', + accountNumber: '1234567890', + country: 'US', + }, + amount: '500', + currency: 'USDC', + }); + + expect(payout.status).toBe(PayoutStatus.REQUIRES_CONFIRMATION); + expect(queue.add).not.toHaveBeenCalled(); + expect(notifications.sendPayoutApprovalCode).toHaveBeenCalledWith( + 'merchant@test.dev', + expect.objectContaining({ + payoutId: 'approval_1', + amount: '500', + currency: 'USDC', + }), + ); + }); + + it('fails insufficient-balance payouts without debiting funds', async () => { + const pending = makePayout({ + id: 'poor_1', + amount: new Prisma.Decimal('50'), + }); + const txBalanceUpdate = jest.fn(); + const txLedgerCreate = jest.fn(); + prisma.payout.findUnique.mockResolvedValueOnce(pending); + prisma.$transaction.mockImplementationOnce(async (callback: unknown) => { + const tx = { + $executeRaw: jest.fn().mockResolvedValue(undefined), + $queryRaw: jest.fn().mockResolvedValue([ + { + availableAmount: new Prisma.Decimal('10'), + currency: 'USDC', + }, + ]), + payout: { + findUnique: jest.fn().mockResolvedValue(pending), + update: jest.fn(), + }, + merchantBalance: { + update: txBalanceUpdate, + }, + merchantLedgerEntry: { + create: txLedgerCreate, + }, + }; + const transaction = callback as (client: typeof tx) => Promise; + return transaction(tx); + }); + + await service.processQueuedPayout('poor_1'); + + expect(txBalanceUpdate).not.toHaveBeenCalled(); + expect(txLedgerCreate).not.toHaveBeenCalled(); + expect(prisma.payout.update).toHaveBeenCalledWith({ + where: { id: 'poor_1' }, + data: { + status: PayoutStatus.FAILED, + failureReason: 'Insufficient balance for USDC', + }, + }); + }); + + it('generates a payout from due recurring configuration and schedules the next run', async () => { + const recurring = makeRecurring({ + id: 'recurring_1', + frequency: 'DAILY', + nextRunAt: now, + }); + const generated = makePayout({ + id: 'generated_1', + recurringPayoutId: 'recurring_1', + scheduledAt: now, + }); + prisma.recurringPayout.findUnique.mockResolvedValueOnce(recurring); + prisma.payout.create.mockResolvedValueOnce(generated); + + await service.processRecurringPayout('recurring_1'); + + expect(prisma.payout.create).toHaveBeenCalledWith({ + data: expect.objectContaining({ + merchantId, + recurringPayoutId: 'recurring_1', + status: PayoutStatus.PENDING, + scheduledAt: now, + }), + }); + expect(prisma.recurringPayout.update).toHaveBeenCalledWith({ + where: { id: 'recurring_1' }, + data: { + lastRunAt: now, + nextRunAt: new Date('2026-07-29T12:00:00.000Z'), + }, + }); + expect(queue.add).toHaveBeenCalledWith( + EXECUTE_PAYOUT_JOB, + { payoutId: 'generated_1' }, + expect.objectContaining({ jobId: 'payout-generated_1' }), + ); + }); + + it('reconciles persisted scheduled and recurring jobs on bootstrap', async () => { + prisma.payout.findMany.mockResolvedValueOnce([ + makePayout({ id: 'scheduled_restart', scheduledAt: now }), + ]); + prisma.recurringPayout.findMany.mockResolvedValueOnce([ + makeRecurring({ id: 'recurring_restart', nextRunAt: now }), + ]); + + await service.onApplicationBootstrap(); + + expect(queue.add).toHaveBeenCalledWith( + EXECUTE_PAYOUT_JOB, + { payoutId: 'scheduled_restart' }, + expect.objectContaining({ jobId: 'payout-scheduled_restart' }), + ); + expect(queue.add).toHaveBeenCalledWith( + GENERATE_RECURRING_PAYOUT_JOB, + { recurringPayoutId: 'recurring_restart' }, + expect.objectContaining({ + jobId: 'recurring-payout-recurring_restart', + repeat: expect.objectContaining({ + key: 'recurring-payout-recurring_restart', + }), + }), + ); + expect(queue.add).toHaveBeenCalledWith( + RUN_RECURRING_PAYOUTS_JOB, + {}, + expect.objectContaining({ + jobId: 'run-recurring-payouts', + repeat: expect.objectContaining({ every: 30_000 }), + }), + ); + }); +}); diff --git a/apps/api/src/modules/payouts/payouts.service.ts b/apps/api/src/modules/payouts/payouts.service.ts index 5e45904..1ab521b 100644 --- a/apps/api/src/modules/payouts/payouts.service.ts +++ b/apps/api/src/modules/payouts/payouts.service.ts @@ -366,7 +366,8 @@ export class PayoutsService implements OnApplicationBootstrap { dto: CreateRecurringPayoutDto, ): Promise { const finalDto = await this.resolveRecipientDetails(merchantId, dto); - const startsAt = dto.startsAt ?? this.nextRunAfter(new Date(), dto.frequency); + const startsAt = + dto.startsAt ?? this.nextRunAfter(new Date(), dto.frequency); const recurring = await this.prisma.recurringPayout.create({ data: { @@ -393,10 +394,7 @@ export class PayoutsService implements OnApplicationBootstrap { }); } - async getRecurring( - id: string, - merchantId: string, - ): Promise { + async getRecurring(id: string, merchantId: string): Promise { const recurring = await this.prisma.recurringPayout.findUnique({ where: { id }, }); @@ -417,8 +415,7 @@ export class PayoutsService implements OnApplicationBootstrap { ? await this.resolveRecipientDetails(merchantId, { recipientId: dto.recipientId ?? existing.recipientId ?? undefined, recipientName: dto.recipientName ?? existing.recipientName, - destinationType: - dto.destinationType ?? existing.destinationType, + destinationType: dto.destinationType ?? existing.destinationType, destination: dto.destination ?? (existing.destination as unknown as CreatePayoutDto['destination']), @@ -435,7 +432,7 @@ export class PayoutsService implements OnApplicationBootstrap { }), ...(finalDto && { recipientName: finalDto.recipientName }), ...(finalDto && { - destinationType: finalDto.destinationType as DestType, + destinationType: finalDto.destinationType, destination: finalDto.destination as Prisma.InputJsonValue, }), ...(dto.amount !== undefined && { amount: dto.amount }), @@ -755,7 +752,9 @@ export class PayoutsService implements OnApplicationBootstrap { return normalized; } - private async startPayoutProcessing(payoutId: string): Promise { + private async startPayoutProcessing( + payoutId: string, + ): Promise { const result = await this.prisma.$transaction(async (tx) => { await tx.$executeRaw` INSERT INTO "MerchantBalance" ("merchantId", "availableAmount", "reservedAmount", "currency", "updatedAt") @@ -1004,9 +1003,7 @@ export class PayoutsService implements OnApplicationBootstrap { } if (scheduled.length > 0) { - this.logger.log( - `Reconciled ${scheduled.length} scheduled payout job(s)`, - ); + this.logger.log(`Reconciled ${scheduled.length} scheduled payout job(s)`); } } @@ -1047,9 +1044,7 @@ export class PayoutsService implements OnApplicationBootstrap { }; } - private frequencyToMilliseconds( - frequency: RecurringPayoutFrequency, - ): number { + private frequencyToMilliseconds(frequency: RecurringPayoutFrequency): number { switch (frequency) { case RecurringPayoutFrequency.DAILY: return 24 * 60 * 60 * 1000; @@ -1062,13 +1057,8 @@ export class PayoutsService implements OnApplicationBootstrap { } } - private nextRunAfter( - date: Date, - frequency: RecurringPayoutFrequency | `${RecurringPayoutFrequency}`, - ): Date { - return new Date(date.getTime() + this.frequencyToMilliseconds( - frequency as RecurringPayoutFrequency, - )); + private nextRunAfter(date: Date, frequency: RecurringPayoutFrequency): Date { + return new Date(date.getTime() + this.frequencyToMilliseconds(frequency)); } private recurringJobKey(id: string): string { From cd2de919cad055fd9cd321c3c830235c8e81c409 Mon Sep 17 00:00:00 2001 From: Collins Ikechukwu Date: Fri, 31 Jul 2026 09:36:34 +0100 Subject: [PATCH 13/13] fix: close money-correctness gaps in payout scheduling - Guard recurring runs with an atomic nextRunAt claim so the repeatable job and the due-recurring dispatcher cannot double-generate a payout - Remove stale BullMQ jobs before re-enqueuing so retried payouts are not silently deduped against completed jobs from earlier attempts - Key MerchantBalance by (merchantId, currency) so credits in different currencies are never summed into one amount - Enforce at-most-once payment credits with a unique (paymentId, type) ledger constraint and a single transactional credit path shared by both Stripe webhook handlers - Add completeExternalPayout/failExternalPayout so external settlement outcomes release debited funds instead of leaking them - Derive the Stellar source asset from the payout currency instead of always sending XLM; fail unsupported currencies and missing paths - Cap payout confirmation attempts at 5, then fail the payout - Remove cancelled payouts' queued jobs; warn on relay-key fallback and invalid PAYOUT_2FA_THRESHOLD; handle destinationType-only updates --- .../migration.sql | 7 +- .../migration.sql | 3 +- apps/api/prisma/schema.prisma | 13 +- .../src/modules/payments/payments.service.ts | 108 +++++----- .../modules/payouts/payouts.service.spec.ts | 89 +++++++- .../src/modules/payouts/payouts.service.ts | 194 +++++++++++++++--- 6 files changed, 315 insertions(+), 99 deletions(-) diff --git a/apps/api/prisma/migrations/20260728120000_payout_scheduling_state/migration.sql b/apps/api/prisma/migrations/20260728120000_payout_scheduling_state/migration.sql index fc62222..41d71a5 100644 --- a/apps/api/prisma/migrations/20260728120000_payout_scheduling_state/migration.sql +++ b/apps/api/prisma/migrations/20260728120000_payout_scheduling_state/migration.sql @@ -37,11 +37,11 @@ CREATE TABLE "MerchantBalance" ( "merchantId" TEXT NOT NULL, "availableAmount" DECIMAL(36,18) NOT NULL DEFAULT 0, "reservedAmount" DECIMAL(36,18) NOT NULL DEFAULT 0, - "currency" TEXT NOT NULL DEFAULT 'USD', + "currency" TEXT NOT NULL, "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, "updatedAt" TIMESTAMP(3) NOT NULL, - CONSTRAINT "MerchantBalance_pkey" PRIMARY KEY ("merchantId") + CONSTRAINT "MerchantBalance_pkey" PRIMARY KEY ("merchantId", "currency") ); -- CreateTable @@ -74,6 +74,9 @@ CREATE INDEX "RecurringPayout_active_nextRunAt_idx" ON "RecurringPayout"("active -- CreateIndex CREATE INDEX "RecurringPayout_recipientId_idx" ON "RecurringPayout"("recipientId"); +-- CreateIndex +CREATE UNIQUE INDEX "MerchantLedgerEntry_paymentId_type_key" ON "MerchantLedgerEntry"("paymentId", "type"); + -- CreateIndex CREATE INDEX "MerchantLedgerEntry_merchantId_createdAt_idx" ON "MerchantLedgerEntry"("merchantId", "createdAt"); diff --git a/apps/api/prisma/migrations/20260728123000_payout_confirmation_codes/migration.sql b/apps/api/prisma/migrations/20260728123000_payout_confirmation_codes/migration.sql index 8b3b1e0..1e29187 100644 --- a/apps/api/prisma/migrations/20260728123000_payout_confirmation_codes/migration.sql +++ b/apps/api/prisma/migrations/20260728123000_payout_confirmation_codes/migration.sql @@ -1,4 +1,5 @@ -- AlterTable ALTER TABLE "Payout" ADD COLUMN "confirmationCodeHash" TEXT, - ADD COLUMN "confirmationCodeExpiresAt" TIMESTAMP(3); + ADD COLUMN "confirmationCodeExpiresAt" TIMESTAMP(3), + ADD COLUMN "confirmationAttempts" INTEGER NOT NULL DEFAULT 0; diff --git a/apps/api/prisma/schema.prisma b/apps/api/prisma/schema.prisma index b33fd29..31d8a4d 100644 --- a/apps/api/prisma/schema.prisma +++ b/apps/api/prisma/schema.prisma @@ -50,7 +50,7 @@ model Merchant { settlementKey MerchantSettlementKey? recipients Recipient[] recurringPayouts RecurringPayout[] - balance MerchantBalance? + balances MerchantBalance[] ledgerEntries MerchantLedgerEntry[] } @@ -308,6 +308,7 @@ model Payout { confirmedAt DateTime? confirmationCodeHash String? confirmationCodeExpiresAt DateTime? + confirmationAttempts Int @default(0) completedAt DateTime? failureReason String? batchId String? @@ -352,14 +353,18 @@ model RecurringPayout { @@index([recipientId]) } +// One row per merchant per currency — balances in different currencies must +// never be summed into a single amount. model MerchantBalance { - merchantId String @id + merchantId String availableAmount Decimal @default(0) @db.Decimal(36, 18) reservedAmount Decimal @default(0) @db.Decimal(36, 18) - currency String @default("USD") + currency String createdAt DateTime @default(now()) updatedAt DateTime @updatedAt merchant Merchant @relation(fields: [merchantId], references: [id], onDelete: Cascade) + + @@id([merchantId, currency]) } model MerchantLedgerEntry { @@ -374,6 +379,8 @@ model MerchantLedgerEntry { createdAt DateTime @default(now()) merchant Merchant @relation(fields: [merchantId], references: [id], onDelete: Cascade) + // A payment may be credited at most once, even under concurrent webhooks. + @@unique([paymentId, type]) @@index([merchantId, createdAt]) @@index([payoutId]) @@index([paymentId]) diff --git a/apps/api/src/modules/payments/payments.service.ts b/apps/api/src/modules/payments/payments.service.ts index f96ae67..ba2c4d7 100644 --- a/apps/api/src/modules/payments/payments.service.ts +++ b/apps/api/src/modules/payments/payments.service.ts @@ -254,43 +254,56 @@ export class PaymentsService implements OnModuleInit { return updatedPayment; } + // Idempotent: the unique (paymentId, type) ledger constraint guarantees a + // payment is credited at most once, even under concurrent webhook delivery. + // The ledger entry is created first so a duplicate aborts the transaction + // before the balance increment. private async creditMerchantBalance(payment: Payment): Promise { const amount = payment.destAmount ?? payment.sourceAmount; const currency = payment.destAsset || payment.sourceAsset; if (!amount || !currency) return; - const existingCredit = await this.prisma.merchantLedgerEntry.findFirst({ - where: { - merchantId: payment.merchantId, - paymentId: payment.id, - type: MerchantLedgerEntryType.CREDIT, - }, - }); - if (existingCredit) return; - - await this.prisma.$transaction([ - this.prisma.merchantBalance.upsert({ - where: { merchantId: payment.merchantId }, - create: { - merchantId: payment.merchantId, - availableAmount: amount, - currency, - }, - update: { - availableAmount: { increment: amount }, - }, - }), - this.prisma.merchantLedgerEntry.create({ - data: { - merchantId: payment.merchantId, - paymentId: payment.id, - type: MerchantLedgerEntryType.CREDIT, - amount, - currency, - description: 'Payment completion credit', - }, - }), - ]); + try { + await this.prisma.$transaction([ + this.prisma.merchantLedgerEntry.create({ + data: { + merchantId: payment.merchantId, + paymentId: payment.id, + type: MerchantLedgerEntryType.CREDIT, + amount, + currency, + description: 'Payment completion credit', + }, + }), + this.prisma.merchantBalance.upsert({ + where: { + merchantId_currency: { + merchantId: payment.merchantId, + currency, + }, + }, + create: { + merchantId: payment.merchantId, + availableAmount: amount, + currency, + }, + update: { + availableAmount: { increment: amount }, + }, + }), + ]); + } catch (err) { + if ( + err instanceof Prisma.PrismaClientKnownRequestError && + err.code === 'P2002' + ) { + this.logger.log( + `Payment ${payment.id} already credited — skipping duplicate credit`, + ); + return; + } + throw err; + } } async findExpiredLocked(): Promise { @@ -1164,10 +1177,6 @@ export class PaymentsService implements OnModuleInit { succeededAt: new Date().toISOString(), }, }); - const creditAmount = payment.destAmount ?? payment.sourceAmount ?? 0; - const creditCurrency = payment.destAsset || payment.sourceAsset || 'USD'; - const shouldCredit = payment.status !== PaymentStatus.COMPLETED; - await this.prisma.$transaction([ this.prisma.payment.update({ where: { id: payment.id }, @@ -1177,31 +1186,6 @@ export class PaymentsService implements OnModuleInit { metadata: updatedMetadata, }, }), - ...(shouldCredit - ? [ - this.prisma.merchantBalance.upsert({ - where: { merchantId: payment.merchantId }, - create: { - merchantId: payment.merchantId, - availableAmount: creditAmount, - currency: creditCurrency, - }, - update: { - availableAmount: { increment: creditAmount }, - }, - }), - this.prisma.merchantLedgerEntry.create({ - data: { - merchantId: payment.merchantId, - paymentId: payment.id, - type: MerchantLedgerEntryType.CREDIT, - amount: creditAmount, - currency: creditCurrency, - description: 'Payment completion credit', - }, - }), - ] - : []), this.prisma.webhookEvent.create({ data: { merchantId: payment.merchantId, @@ -1223,6 +1207,8 @@ export class PaymentsService implements OnModuleInit { }, }), ]); + + await this.creditMerchantBalance(payment); } private async handlePaymentIntentFailed(event: Stripe.Event) { diff --git a/apps/api/src/modules/payouts/payouts.service.spec.ts b/apps/api/src/modules/payouts/payouts.service.spec.ts index 59bfc37..a944dde 100644 --- a/apps/api/src/modules/payouts/payouts.service.spec.ts +++ b/apps/api/src/modules/payouts/payouts.service.spec.ts @@ -1,4 +1,5 @@ import { Prisma, PayoutStatus, DestType } from '@prisma/client'; +import * as bcrypt from 'bcrypt'; import { PayoutsService } from './payouts.service'; import { EXECUTE_PAYOUT_BATCH_JOB, @@ -22,6 +23,7 @@ interface MockPayout { confirmedAt: Date | null; confirmationCodeHash: string | null; confirmationCodeExpiresAt: Date | null; + confirmationAttempts: number; completedAt: Date | null; failureReason: string | null; batchId: string | null; @@ -53,6 +55,7 @@ describe('PayoutsService', () => { const queue = { add: jest.fn(), + remove: jest.fn(), removeRepeatableByKey: jest.fn(), }; const prisma = { @@ -68,6 +71,7 @@ describe('PayoutsService', () => { findMany: jest.fn(), create: jest.fn(), update: jest.fn(), + updateMany: jest.fn(), }, recipient: { findFirst: jest.fn(), @@ -127,6 +131,7 @@ describe('PayoutsService', () => { confirmedAt: overrides.confirmedAt ?? null, confirmationCodeHash: overrides.confirmationCodeHash ?? null, confirmationCodeExpiresAt: overrides.confirmationCodeExpiresAt ?? null, + confirmationAttempts: overrides.confirmationAttempts ?? 0, completedAt: overrides.completedAt ?? null, failureReason: overrides.failureReason ?? null, batchId: overrides.batchId ?? null, @@ -167,6 +172,7 @@ describe('PayoutsService', () => { payoutCounter = 0; queue.add.mockResolvedValue({}); + queue.remove.mockResolvedValue(1); queue.removeRepeatableByKey.mockResolvedValue(undefined); webhooks.dispatch.mockResolvedValue(undefined); stellar.sendPayment.mockResolvedValue('stellar_tx_hash'); @@ -196,6 +202,7 @@ describe('PayoutsService', () => { }) => makePayout({ id: where.id, ...data }), ); prisma.recurringPayout.findUnique.mockResolvedValue(null); + prisma.recurringPayout.updateMany.mockResolvedValue({ count: 1 }); prisma.recurringPayout.findMany.mockResolvedValue([]); prisma.recurringPayout.create.mockImplementation( async ({ data }: { data: Partial }) => @@ -445,8 +452,8 @@ describe('PayoutsService', () => { scheduledAt: now, }), }); - expect(prisma.recurringPayout.update).toHaveBeenCalledWith({ - where: { id: 'recurring_1' }, + expect(prisma.recurringPayout.updateMany).toHaveBeenCalledWith({ + where: { id: 'recurring_1', active: true, nextRunAt: now }, data: { lastRunAt: now, nextRunAt: new Date('2026-07-29T12:00:00.000Z'), @@ -459,6 +466,84 @@ describe('PayoutsService', () => { ); }); + it('does not generate a payout when the recurring run is not yet due', async () => { + const recurring = makeRecurring({ + id: 'recurring_future', + nextRunAt: new Date(now.getTime() + 60_000), + }); + prisma.recurringPayout.findUnique.mockResolvedValueOnce(recurring); + + await service.processRecurringPayout('recurring_future'); + + expect(prisma.recurringPayout.updateMany).not.toHaveBeenCalled(); + expect(prisma.payout.create).not.toHaveBeenCalled(); + }); + + it('does not generate a payout when another worker already claimed the run', async () => { + const recurring = makeRecurring({ id: 'recurring_raced', nextRunAt: now }); + prisma.recurringPayout.findUnique.mockResolvedValueOnce(recurring); + prisma.recurringPayout.updateMany.mockResolvedValueOnce({ count: 0 }); + + await service.processRecurringPayout('recurring_raced'); + + expect(prisma.payout.create).not.toHaveBeenCalled(); + expect(queue.add).not.toHaveBeenCalled(); + }); + + it('removes any stale job before re-enqueuing a payout so retries execute', async () => { + const failed = makePayout({ id: 'retry_1', status: PayoutStatus.FAILED }); + prisma.payout.findUnique.mockResolvedValueOnce(failed); + prisma.payout.update.mockResolvedValueOnce( + makePayout({ id: 'retry_1', status: PayoutStatus.PENDING }), + ); + + await service.retry('retry_1', merchantId); + + expect(queue.remove).toHaveBeenCalledWith('payout-retry_1'); + expect(queue.add).toHaveBeenCalledWith( + EXECUTE_PAYOUT_JOB, + { payoutId: 'retry_1' }, + expect.objectContaining({ jobId: 'payout-retry_1' }), + ); + const removeOrder = queue.remove.mock.invocationCallOrder[0]; + const addOrder = queue.add.mock.invocationCallOrder[0]; + expect(removeOrder).toBeLessThan(addOrder); + }); + + it('fails the payout after too many invalid confirmation attempts', async () => { + const pending = makePayout({ + id: 'confirm_1', + status: PayoutStatus.REQUIRES_CONFIRMATION, + confirmationCodeHash: bcrypt.hashSync('654321', 4), + confirmationCodeExpiresAt: new Date(now.getTime() + 600_000), + }); + prisma.payout.findUnique.mockResolvedValueOnce(pending); + prisma.payout.update.mockResolvedValueOnce( + makePayout({ + id: 'confirm_1', + status: PayoutStatus.REQUIRES_CONFIRMATION, + confirmationAttempts: 5, + }), + ); + + await expect( + service.confirm('confirm_1', merchantId, { code: '000000' }), + ).rejects.toThrow('Too many invalid confirmation attempts'); + + expect(prisma.payout.update).toHaveBeenCalledWith({ + where: { id: 'confirm_1' }, + data: { confirmationAttempts: { increment: 1 } }, + }); + expect(prisma.payout.update).toHaveBeenCalledWith({ + where: { id: 'confirm_1' }, + data: { + status: PayoutStatus.FAILED, + failureReason: 'Too many invalid confirmation attempts', + }, + }); + expect(queue.add).not.toHaveBeenCalled(); + }); + it('reconciles persisted scheduled and recurring jobs on bootstrap', async () => { prisma.payout.findMany.mockResolvedValueOnce([ makePayout({ id: 'scheduled_restart', scheduledAt: now }), diff --git a/apps/api/src/modules/payouts/payouts.service.ts b/apps/api/src/modules/payouts/payouts.service.ts index 1ab521b..1444e0b 100644 --- a/apps/api/src/modules/payouts/payouts.service.ts +++ b/apps/api/src/modules/payouts/payouts.service.ts @@ -74,6 +74,8 @@ function stellarUsdcAsset(network?: string): string { return `USDC:${issuer}`; } +const MAX_CONFIRMATION_ATTEMPTS = 5; + // ── Service ─────────────────────────────────────────────────────────────────── @Injectable() @@ -98,6 +100,15 @@ export class PayoutsService implements OnApplicationBootstrap { ) {} async onApplicationBootstrap(): Promise { + const rawThreshold = this.config.get('PAYOUT_2FA_THRESHOLD'); + if (rawThreshold !== undefined) { + const threshold = Number(rawThreshold); + if (!Number.isFinite(threshold) || threshold <= 0) { + this.logger.warn( + `PAYOUT_2FA_THRESHOLD is set to "${rawThreshold}" which is not a positive number — high-value payout confirmation is disabled`, + ); + } + } await this.reconcileScheduledPayoutJobs(); await this.reconcileRecurringPayoutJobs(); await this.enqueueRecurringDispatcher(); @@ -342,6 +353,19 @@ export class PayoutsService implements OnApplicationBootstrap { const ok = await bcrypt.compare(dto.code, payout.confirmationCodeHash); if (!ok) { + const attempted = await this.prisma.payout.update({ + where: { id: payout.id }, + data: { confirmationAttempts: { increment: 1 } }, + }); + if (attempted.confirmationAttempts >= MAX_CONFIRMATION_ATTEMPTS) { + await this.failPayout( + attempted, + 'Too many invalid confirmation attempts', + ); + throw new BadRequestException( + 'Too many invalid confirmation attempts — payout failed, use retry to request a new code', + ); + } throw new BadRequestException('Invalid payout confirmation code'); } @@ -352,6 +376,7 @@ export class PayoutsService implements OnApplicationBootstrap { confirmedAt: new Date(), confirmationCodeHash: null, confirmationCodeExpiresAt: null, + confirmationAttempts: 0, }, }); @@ -411,7 +436,7 @@ export class PayoutsService implements OnApplicationBootstrap { ): Promise { const existing = await this.getRecurring(id, merchantId); const finalDto = - dto.recipientId || dto.destination + dto.recipientId || dto.destination || dto.destinationType ? await this.resolveRecipientDetails(merchantId, { recipientId: dto.recipientId ?? existing.recipientId ?? undefined, recipientName: dto.recipientName ?? existing.recipientName, @@ -480,10 +505,16 @@ export class PayoutsService implements OnApplicationBootstrap { ); } - return this.prisma.payout.update({ + const cancelled = await this.prisma.payout.update({ where: { id }, data: { status: PayoutStatus.CANCELLED }, }); + + // Best-effort queue cleanup; the status guard in processQueuedPayout + // protects correctness if the job cannot be removed. + await this.payoutQueue.remove(`payout-${id}`).catch(() => undefined); + + return cancelled; } // ── Retry ───────────────────────────────────────────────────────────────── @@ -509,6 +540,7 @@ export class PayoutsService implements OnApplicationBootstrap { failureReason: null, confirmationCodeHash: confirmation?.hash ?? null, confirmationCodeExpiresAt: confirmation?.expiresAt ?? null, + confirmationAttempts: 0, }, }); @@ -577,6 +609,24 @@ export class PayoutsService implements OnApplicationBootstrap { if (!recurring || !recurring.active) return; const now = new Date(); + if (recurring.nextRunAt > now) return; + + // Atomically claim this run: both the per-record repeatable job and the + // due-recurring dispatcher can race here, and only one may generate a + // payout. The nextRunAt equality check makes the claim first-wins. + const claimed = await this.prisma.recurringPayout.updateMany({ + where: { + id: recurring.id, + active: true, + nextRunAt: recurring.nextRunAt, + }, + data: { + lastRunAt: now, + nextRunAt: this.nextRunAfter(now, recurring.frequency), + }, + }); + if (claimed.count === 0) return; + const confirmation = await this.buildConfirmationFields( recurring.amount.toString(), ); @@ -599,14 +649,6 @@ export class PayoutsService implements OnApplicationBootstrap { }, }); - await this.prisma.recurringPayout.update({ - where: { id: recurring.id }, - data: { - lastRunAt: now, - nextRunAt: this.nextRunAfter(now, recurring.frequency), - }, - }); - if (confirmation) { await this.sendPayoutConfirmationCode(payout, confirmation.code); } else { @@ -629,6 +671,67 @@ export class PayoutsService implements OnApplicationBootstrap { } } + // ── External settlement callbacks ───────────────────────────────────────── + // BANK_ACCOUNT and MOBILE_MONEY payouts are debited and left PROCESSING + // until the external settlement system reports the outcome. These are the + // only valid terminal transitions for such payouts: a failure MUST release + // the debited funds, never a bare status update. + + async completeExternalPayout(id: string): Promise { + const payout = await this.prisma.payout.findUnique({ where: { id } }); + if (!payout) throw new NotFoundException('Payout not found'); + if (payout.status !== PayoutStatus.PROCESSING) { + throw new BadRequestException( + `Cannot complete a payout in ${payout.status} status`, + ); + } + + const completed = await this.prisma.payout.update({ + where: { id }, + data: { + status: PayoutStatus.COMPLETED, + completedAt: new Date(), + }, + }); + + this.eventsService.emitPayoutStatus( + payout.merchantId, + payout.id, + PayoutStatus.COMPLETED, + { + amount: completed.amount.toString(), + currency: completed.currency, + updatedAt: new Date(), + }, + ); + void this.notifications + .notifyPayoutCompleted(payout.merchantId, payout.id, payout.recipientName) + .catch(() => undefined); + this.webhooks + .dispatch( + payout.merchantId, + 'payout.completed', + this.webhookPayload(completed) as Prisma.InputJsonValue, + ) + .catch(() => undefined); + + return completed; + } + + async failExternalPayout(id: string, reason: string): Promise { + const payout = await this.prisma.payout.findUnique({ where: { id } }); + if (!payout) throw new NotFoundException('Payout not found'); + if (payout.status !== PayoutStatus.PROCESSING) { + throw new BadRequestException( + `Cannot fail a payout in ${payout.status} status`, + ); + } + + await this.releaseDebitedFunds(payout); + await this.failPayout(payout, reason); + return this.prisma.payout.findUniqueOrThrow({ where: { id } }); + } + private async processPayout(payout: Payout): Promise { const started = await this.startPayoutProcessing(payout.id); if (!started) return; @@ -661,7 +764,9 @@ export class PayoutsService implements OnApplicationBootstrap { ): Promise { const destAddress = String(destination.address); const destAsset = this.normalizeStellarAsset(destination.asset); - const sourceAsset = 'native'; + // The on-chain transfer must move the same money that was debited from the + // merchant ledger, so the source asset derives from the payout currency. + const sourceAsset = this.stellarSourceAssetForCurrency(payout.currency); const sourceAmount = payout.amount.toString(); const sourceSecret = await this.getMerchantPayoutSourceSecret( payout.merchantId, @@ -687,13 +792,18 @@ export class PayoutsService implements OnApplicationBootstrap { }); const bestPath = paths[0]; + if (!bestPath) { + throw new Error( + `No conversion path from ${sourceAsset} to ${destAsset}`, + ); + } // Convert asset objects to the 'native' | 'CODE:issuer' strings parseAsset expects - const pathStrings: string[] = (bestPath?.path ?? []).map((a) => + const pathStrings: string[] = (bestPath.path ?? []).map((a) => assetToString(a as AssetObject), ); - const destMin = ( - parseFloat(bestPath?.destinationAmount ?? sourceAmount) * 0.99 - ).toFixed(7); + const destMin = (parseFloat(bestPath.destinationAmount) * 0.99).toFixed( + 7, + ); txHash = await this.stellar.executePathPayment({ sourceAsset, @@ -738,6 +848,17 @@ export class PayoutsService implements OnApplicationBootstrap { .catch(() => undefined); } + private stellarSourceAssetForCurrency(currency: string): string { + const upper = currency.toUpperCase(); + if (upper === 'XLM' || upper === 'NATIVE') return 'native'; + if (upper === 'USDC') { + return stellarUsdcAsset(this.config.get('STELLAR_NETWORK')); + } + throw new Error( + `Payout currency ${currency} cannot be sent on Stellar — no on-chain asset mapping`, + ); + } + private normalizeStellarAsset(asset: unknown): string { if (typeof asset !== 'string' || asset.trim() === '') return 'native'; @@ -761,7 +882,7 @@ export class PayoutsService implements OnApplicationBootstrap { SELECT "merchantId", 0, 0, "currency", NOW() FROM "Payout" WHERE "id" = ${payoutId} - ON CONFLICT ("merchantId") DO NOTHING + ON CONFLICT ("merchantId", "currency") DO NOTHING `; const payout = await tx.payout.findUnique({ where: { id: payoutId } }); @@ -772,24 +893,16 @@ export class PayoutsService implements OnApplicationBootstrap { const rows = await tx.$queryRaw< Array<{ availableAmount: Prisma.Decimal; - currency: string; }> >` - SELECT "availableAmount", "currency" + SELECT "availableAmount" FROM "MerchantBalance" - WHERE "merchantId" = ${payout.merchantId} + WHERE "merchantId" = ${payout.merchantId} AND "currency" = ${payout.currency} FOR UPDATE `; const balance = rows[0]; - if (!balance || balance.currency !== payout.currency) { - return { - payout, - failureReason: `Insufficient balance for ${payout.currency}`, - }; - } - - if (balance.availableAmount.lt(payout.amount)) { + if (!balance || balance.availableAmount.lt(payout.amount)) { return { payout, failureReason: `Insufficient balance for ${payout.currency}`, @@ -797,7 +910,12 @@ export class PayoutsService implements OnApplicationBootstrap { } await tx.merchantBalance.update({ - where: { merchantId: payout.merchantId }, + where: { + merchantId_currency: { + merchantId: payout.merchantId, + currency: payout.currency, + }, + }, data: { availableAmount: { decrement: payout.amount }, }, @@ -831,7 +949,12 @@ export class PayoutsService implements OnApplicationBootstrap { private async releaseDebitedFunds(payout: Payout): Promise { await this.prisma.$transaction(async (tx) => { await tx.merchantBalance.update({ - where: { merchantId: payout.merchantId }, + where: { + merchantId_currency: { + merchantId: payout.merchantId, + currency: payout.currency, + }, + }, data: { availableAmount: { increment: payout.amount } }, }); await tx.merchantLedgerEntry.create({ @@ -924,6 +1047,12 @@ export class PayoutsService implements OnApplicationBootstrap { ? Math.max(0, payout.scheduledAt.getTime() - Date.now()) : 0; + // BullMQ silently ignores add() when a job with this id still exists in + // any state — including completed jobs kept by removeOnComplete. Remove + // any leftover (e.g. from a previous attempt of a retried payout) first; + // an active job cannot be removed, and the jobId dedupe then applies. + await this.payoutQueue.remove(`payout-${payout.id}`).catch(() => undefined); + await this.payoutQueue.add( EXECUTE_PAYOUT_JOB, { payoutId: payout.id }, @@ -1106,7 +1235,12 @@ export class PayoutsService implements OnApplicationBootstrap { const settlementKey = await this.prisma.merchantSettlementKey.findUnique({ where: { merchantId }, }); - if (!settlementKey?.managed) return undefined; + if (!settlementKey?.managed) { + this.logger.warn( + `Merchant ${merchantId} has no managed settlement key — payout will be funded from the platform relay account`, + ); + return undefined; + } return this.settlement.decryptSeed(settlementKey); }