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..41d71a5 --- /dev/null +++ b/apps/api/prisma/migrations/20260728120000_payout_scheduling_state/migration.sql @@ -0,0 +1,102 @@ +-- 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, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "MerchantBalance_pkey" PRIMARY KEY ("merchantId", "currency") +); + +-- 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 UNIQUE INDEX "MerchantLedgerEntry_paymentId_type_key" ON "MerchantLedgerEntry"("paymentId", "type"); + +-- 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/migrations/20260728123000_payout_confirmation_codes/migration.sql b/apps/api/prisma/migrations/20260728123000_payout_confirmation_codes/migration.sql new file mode 100644 index 0000000..1e29187 --- /dev/null +++ b/apps/api/prisma/migrations/20260728123000_payout_confirmation_codes/migration.sql @@ -0,0 +1,5 @@ +-- AlterTable +ALTER TABLE "Payout" + ADD COLUMN "confirmationCodeHash" TEXT, + 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 123c330..31d8a4d 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[] + balances MerchantBalance[] + ledgerEntries MerchantLedgerEntry[] } model ApiKey { @@ -302,20 +305,85 @@ model Payout { status PayoutStatus @default(PENDING) stellarTxHash String? scheduledAt DateTime? + confirmedAt DateTime? + confirmationCodeHash String? + confirmationCodeExpiresAt DateTime? + confirmationAttempts Int @default(0) 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]) +} + +// One row per merchant per currency — balances in different currencies must +// never be summed into a single amount. +model MerchantBalance { + merchantId String + availableAmount Decimal @default(0) @db.Decimal(36, 18) + reservedAmount Decimal @default(0) @db.Decimal(36, 18) + currency String + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + merchant Merchant @relation(fields: [merchantId], references: [id], onDelete: Cascade) + + @@id([merchantId, currency]) +} + +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) + + // A payment may be credited at most once, even under concurrent webhooks. + @@unique([paymentId, type]) + @@index([merchantId, createdAt]) + @@index([payoutId]) + @@index([paymentId]) } // Saved payout destination. Reconstructed to match migration @@ -331,6 +399,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 +415,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 { 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/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/payments/payments.service.spec.ts b/apps/api/src/modules/payments/payments.service.spec.ts index bf09d11..d7356ec 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,17 @@ 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') { + const transaction = arg as (client: typeof prisma) => Promise; + return transaction(prisma); + } + return arg; + }); const module: TestingModule = await Test.createTestingModule({ providers: [ diff --git a/apps/api/src/modules/payments/payments.service.ts b/apps/api/src/modules/payments/payments.service.ts index 98c0b7b..ba2c4d7 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,58 @@ 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; + + 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 { const now = new Date(); return await this.prisma.payment.findMany({ @@ -1115,7 +1177,6 @@ export class PaymentsService implements OnModuleInit { succeededAt: new Date().toISOString(), }, }); - await this.prisma.$transaction([ this.prisma.payment.update({ where: { id: payment.id }, @@ -1146,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/dto/create-payout.dto.ts b/apps/api/src/modules/payouts/dto/create-payout.dto.ts index 8909d5a..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,12 +59,44 @@ 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(), }); 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 +>; + +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..fcff493 100644 --- a/apps/api/src/modules/payouts/dto/payout-filters.dto.ts +++ b/apps/api/src/modules/payouts/dto/payout-filters.dto.ts @@ -2,12 +2,19 @@ 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']) .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.constants.ts b/apps/api/src/modules/payouts/payouts.constants.ts new file mode 100644 index 0000000..443e0b2 --- /dev/null +++ b/apps/api/src/modules/payouts/payouts.constants.ts @@ -0,0 +1,30 @@ +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; +} + +export interface ExecutePayoutBatchJobData { + batchId: string; +} + +export interface GenerateRecurringPayoutJobData { + recurringPayoutId: string; +} + +export type RunRecurringPayoutsJobData = Record; + +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.controller.ts b/apps/api/src/modules/payouts/payouts.controller.ts index 5399c1b..d2083a1 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,12 @@ import { CreatePayoutDto, BulkPayoutSchema, BulkPayoutDto, + CreateRecurringPayoutSchema, + CreateRecurringPayoutDto, + UpdateRecurringPayoutSchema, + UpdateRecurringPayoutDto, + ConfirmPayoutSchema, + ConfirmPayoutDto, } from './dto/create-payout.dto'; import { PayoutFiltersSchema, @@ -64,6 +72,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( @@ -73,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.module.ts b/apps/api/src/modules/payouts/payouts.module.ts index d1ab203..722b87a 100644 --- a/apps/api/src/modules/payouts/payouts.module.ts +++ b/apps/api/src/modules/payouts/payouts.module.ts @@ -1,23 +1,29 @@ 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'; 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, + 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..19e8a5b --- /dev/null +++ b/apps/api/src/modules/payouts/payouts.processor.ts @@ -0,0 +1,56 @@ +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, + GENERATE_RECURRING_PAYOUT_JOB, + RUN_RECURRING_PAYOUTS_JOB, + type GenerateRecurringPayoutJobData, + type ExecutePayoutBatchJobData, + type ExecutePayoutJobData, + PAYOUTS_QUEUE, +} from './payouts.constants'; + +type PayoutJobData = + | ExecutePayoutJobData + | ExecutePayoutBatchJobData + | GenerateRecurringPayoutJobData; + +@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; + } + + if (job.name === GENERATE_RECURRING_PAYOUT_JOB) { + const data = job.data as GenerateRecurringPayoutJobData; + await this.payoutsService.processRecurringPayout(data.recurringPayoutId); + 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.spec.ts b/apps/api/src/modules/payouts/payouts.service.spec.ts new file mode 100644 index 0000000..a944dde --- /dev/null +++ b/apps/api/src/modules/payouts/payouts.service.spec.ts @@ -0,0 +1,581 @@ +import { Prisma, PayoutStatus, DestType } from '@prisma/client'; +import * as bcrypt from 'bcrypt'; +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; + confirmationAttempts: number; + 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(), + remove: 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(), + updateMany: 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, + confirmationAttempts: overrides.confirmationAttempts ?? 0, + 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.remove.mockResolvedValue(1); + 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.updateMany.mockResolvedValue({ count: 1 }); + 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.updateMany).toHaveBeenCalledWith({ + where: { id: 'recurring_1', active: true, nextRunAt: now }, + 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('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 }), + ]); + 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 b2653f4..1444e0b 100644 --- a/apps/api/src/modules/payouts/payouts.service.ts +++ b/apps/api/src/modules/payouts/payouts.service.ts @@ -4,16 +4,48 @@ import { Injectable, Logger, NotFoundException, + OnApplicationBootstrap, } from '@nestjs/common'; -import { DestType, Payout, PayoutStatus, Prisma } from '@prisma/client'; +import { InjectQueue } from '@nestjs/bullmq'; +import { ConfigService } from '@nestjs/config'; +import { Queue } from 'bullmq'; +import { + DestType, + MerchantLedgerEntryType, + 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 { MerchantSettlementService } from '../merchant/merchant-settlement.service'; +import { + CreatePayoutDto, + 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, + 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 ───────────────────────────────────────────────────────────────────── @@ -34,20 +66,54 @@ 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}`; +} + +const MAX_CONFIRMATION_ATTEMPTS = 5; + // ── Service ─────────────────────────────────────────────────────────────────── @Injectable() -export class PayoutsService { +export class PayoutsService implements OnApplicationBootstrap { private readonly logger = new Logger(PayoutsService.name); constructor( 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, + @InjectQueue(PAYOUTS_QUEUE) + private readonly payoutQueue: Queue< + | ExecutePayoutJobData + | ExecutePayoutBatchJobData + | GenerateRecurringPayoutJobData + | RunRecurringPayoutsJobData + >, ) {} + 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(); + } + // ── Create single payout ────────────────────────────────────────────────── async create( @@ -89,6 +155,7 @@ export class PayoutsService { }; } + const confirmation = await this.buildConfirmationFields(dto.amount); const payout = await this.prisma.payout.create({ data: { merchantId, @@ -98,8 +165,12 @@ export class PayoutsService { 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, }, }); @@ -112,9 +183,10 @@ export class PayoutsService { ) .catch(() => undefined); - // Process immediately unless scheduled for the future - if (!payout.scheduledAt || payout.scheduledAt <= new Date()) { - this.processPayout(payout).catch(() => undefined); + if (confirmation) { + await this.sendPayoutConfirmationCode(payout, confirmation.code); + } else { + await this.enqueuePayout(payout); } return payout; @@ -130,6 +202,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]; @@ -155,6 +228,7 @@ export class PayoutsService { }; } + const confirmation = await this.buildConfirmationFields(item.amount); const payout = await this.prisma.payout.create({ data: { merchantId, @@ -164,16 +238,24 @@ export class PayoutsService { 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()) { - this.processPayout(payout).catch(() => undefined); + if (confirmation) { + await this.sendPayoutConfirmationCode(payout, confirmation.code); + } else if (payout.scheduledAt && payout.scheduledAt > new Date()) { + await this.enqueuePayout(payout); + } else { + dueNow++; } } catch (err) { rejected++; @@ -193,6 +275,10 @@ export class PayoutsService { } as Prisma.InputJsonValue) .catch(() => undefined); + if (dueNow > 0) { + await this.enqueueBatch(batchId); + } + return { batchId, total: dto.payouts.length, @@ -246,21 +332,189 @@ export class PayoutsService { 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) { + 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'); + } + + const confirmed = await this.prisma.payout.update({ + where: { id: payout.id }, + data: { + status: PayoutStatus.PENDING, + confirmedAt: new Date(), + confirmationCodeHash: null, + confirmationCodeExpiresAt: null, + confirmationAttempts: 0, + }, + }); + + await this.enqueuePayout(confirmed); + return confirmed; + } + + // ── 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 || dto.destinationType + ? 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, + 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 { 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`, ); } - 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 ───────────────────────────────────────────────────────────────── @@ -274,9 +528,20 @@ export class PayoutsService { ); } + 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, + confirmationAttempts: 0, + }, }); this.webhooks @@ -287,66 +552,209 @@ export class PayoutsService { ) .catch(() => undefined); - this.processPayout(reset).catch(() => undefined); + if (confirmation) { + await this.sendPayoutConfirmationCode(reset, confirmation.code); + } else { + await this.enqueuePayout(reset); + } return reset; } // ── Internal processing ─────────────────────────────────────────────────── - private async processPayout(payout: Payout): Promise { - await this.prisma.payout.update({ - where: { id: payout.id }, - data: { status: PayoutStatus.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); + } + } + + async processRecurringPayout(recurringPayoutId: string): Promise { + const recurring = await this.prisma.recurringPayout.findUnique({ + where: { id: recurringPayoutId }, + }); + 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(), + ); + 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: confirmation + ? PayoutStatus.REQUIRES_CONFIRMATION + : PayoutStatus.PENDING, + scheduledAt: now, + confirmationCodeHash: confirmation?.hash ?? null, + confirmationCodeExpiresAt: confirmation?.expiresAt ?? null, + recurringPayoutId: recurring.id, + }, + }); + + if (confirmation) { + await this.sendPayoutConfirmationCode(payout, confirmation.code); + } else { + await this.enqueuePayout(payout); + } + } + + 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); + } + } + + // ── 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; 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); } } @@ -355,10 +763,14 @@ export class PayoutsService { destination: Record, ): Promise { const destAddress = String(destination.address); - const destAsset = - typeof destination.asset === 'string' ? destination.asset : 'native'; - const sourceAsset = 'native'; + const destAsset = this.normalizeStellarAsset(destination.asset); + // 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, + ); let txHash: string; @@ -369,6 +781,7 @@ export class PayoutsService { asset: sourceAsset, amount: sourceAmount, destinationAccount: destAddress, + sourceSecret, }); } else { const { paths } = await this.stellar.findStrictSendPaths({ @@ -379,13 +792,18 @@ export class PayoutsService { }); 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, @@ -394,6 +812,7 @@ export class PayoutsService { destinationAsset: destAsset, destinationMinAmount: destMin, path: pathStrings, + sourceSecret, }); } @@ -429,6 +848,164 @@ export class PayoutsService { .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'; + + 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` + INSERT INTO "MerchantBalance" ("merchantId", "availableAmount", "reservedAmount", "currency", "updatedAt") + SELECT "merchantId", 0, 0, "currency", NOW() + FROM "Payout" + WHERE "id" = ${payoutId} + ON CONFLICT ("merchantId", "currency") 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; + }> + >` + SELECT "availableAmount" + FROM "MerchantBalance" + WHERE "merchantId" = ${payout.merchantId} AND "currency" = ${payout.currency} + FOR UPDATE + `; + const balance = rows[0]; + + if (!balance || balance.availableAmount.lt(payout.amount)) { + return { + payout, + failureReason: `Insufficient balance for ${payout.currency}`, + }; + } + + await tx.merchantBalance.update({ + where: { + merchantId_currency: { + merchantId: payout.merchantId, + currency: payout.currency, + }, + }, + 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_currency: { + merchantId: payout.merchantId, + currency: payout.currency, + }, + }, + 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) { @@ -464,4 +1041,207 @@ 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; + + // 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 }, + { + 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 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}`, + repeat: { + every, + offset: recurring.nextRunAt.getTime() % every, + 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 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: { + 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)`); + } + } + + 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): Date { + return new Date(date.getTime() + this.frequencyToMilliseconds(frequency)); + } + + 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, + }); + } + + private async getMerchantPayoutSourceSecret( + merchantId: string, + ): Promise { + const settlementKey = await this.prisma.merchantSettlementKey.findUnique({ + where: { merchantId }, + }); + 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); + } } 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'); }