diff --git a/src/blockchain/blockchain.module.ts b/src/blockchain/blockchain.module.ts index 51603bc..68e4700 100644 --- a/src/blockchain/blockchain.module.ts +++ b/src/blockchain/blockchain.module.ts @@ -17,6 +17,12 @@ import { EventPublishProcessor } from './event-publish.processor'; import { MintFinalizerService } from './mint-finalizer.service'; import { MintQueueService, TICKET_MINT_QUEUE } from './mint-queue.service'; import { MintTicketProcessor } from './mint-ticket.processor'; +import { PayoutFinalizerService } from './payout-finalizer.service'; +import { PayoutProcessor } from './payout.processor'; +import { + PayoutQueueService, + TICKET_PAYOUT_QUEUE, +} from './payout-queue.service'; import { RefundFinalizerService } from './refund-finalizer.service'; import { RefundProcessor } from './refund.processor'; import { @@ -58,6 +64,9 @@ import { TicketCheckinProcessor } from './ticket-checkin.processor'; // Consumer side of the refund queue; EventsModule registers the // producer side. registerQueue is idempotent across modules. BullModule.registerQueue({ name: TICKET_REFUND_QUEUE }), + // Payout (on-chain withdraw) queue. Producer + consumer live here; + // callers inject PayoutQueueService (exported below). + BullModule.registerQueue({ name: TICKET_PAYOUT_QUEUE }), // Consumer side of the Circle webhook queue. The WebhooksModule // registers the producer side; registerQueue is idempotent. BullModule.registerQueue({ name: CIRCLE_WEBHOOK_QUEUE }), @@ -73,6 +82,9 @@ import { TicketCheckinProcessor } from './ticket-checkin.processor'; RefundFinalizerService, RefundQueueService, RefundProcessor, + PayoutFinalizerService, + PayoutQueueService, + PayoutProcessor, CheckinQueueService, TicketCheckinProcessor, CircleWebhookProcessor, @@ -84,6 +96,7 @@ import { TicketCheckinProcessor } from './ticket-checkin.processor'; MintFinalizerService, MintQueueService, RefundQueueService, + PayoutQueueService, CheckinQueueService, ], }) diff --git a/src/blockchain/circle-webhook.processor.spec.ts b/src/blockchain/circle-webhook.processor.spec.ts index 569412b..f0e3e6b 100644 --- a/src/blockchain/circle-webhook.processor.spec.ts +++ b/src/blockchain/circle-webhook.processor.spec.ts @@ -44,6 +44,7 @@ function setup(opts: { const reconcile = jest.fn().mockResolvedValue(undefined); const finalize = jest.fn().mockResolvedValue(true); const refundFinalize = jest.fn().mockResolvedValue(true); + const payoutFinalize = jest.fn().mockResolvedValue(undefined); const enqueueMint = jest.fn().mockResolvedValue(undefined); const findDeposit = jest.fn().mockResolvedValue(opts.deposit ?? null); @@ -70,6 +71,7 @@ function setup(opts: { { reconcile } as never, { finalize } as never, { finalize: refundFinalize } as never, + { finalize: payoutFinalize } as never, { enqueueMint } as never, ); @@ -81,6 +83,7 @@ function setup(opts: { reconcile, finalize, refundFinalize, + payoutFinalize, enqueueMint, findDeposit, updateDeposit, diff --git a/src/blockchain/circle-webhook.processor.ts b/src/blockchain/circle-webhook.processor.ts index 46f8fc0..73f91d4 100644 --- a/src/blockchain/circle-webhook.processor.ts +++ b/src/blockchain/circle-webhook.processor.ts @@ -11,6 +11,7 @@ import { PrismaService } from '../prisma/prisma.service'; import { CircleContractService } from './circle-contract.service'; import { MintFinalizerService } from './mint-finalizer.service'; import { MintQueueService } from './mint-queue.service'; +import { PayoutFinalizerService } from './payout-finalizer.service'; import { RefundFinalizerService } from './refund-finalizer.service'; import { CIRCLE_WEBHOOK_JOB, @@ -61,6 +62,7 @@ export class CircleWebhookProcessor extends WorkerHost { private readonly circle: CircleContractService, private readonly finalizer: MintFinalizerService, private readonly refundFinalizer: RefundFinalizerService, + private readonly payoutFinalizer: PayoutFinalizerService, private readonly mintQueue: MintQueueService, ) { super(); @@ -274,6 +276,16 @@ export class CircleWebhookProcessor extends WorkerHost { ); } await this.refundFinalizer.finalize(bt.ticketId, tx.txHash); + } else if (bt.type === BlockchainTxType.WITHDRAW && bt.eventId) { + if (!tx.txHash) { + throw new Error( + `Confirmed withdraw webhook for ${circleTxId} has no txHash`, + ); + } + await this.payoutFinalizer.finalize( + { eventId: bt.eventId, chain: bt.chain ?? '' }, + tx.txHash, + ); } else if (bt.type === BlockchainTxType.CHECKIN) { // The door already flipped the ticket to USED (#24); the // confirmed CHECKIN tx is the on-chain audit record, now diff --git a/src/blockchain/payout-finalizer.service.spec.ts b/src/blockchain/payout-finalizer.service.spec.ts new file mode 100644 index 0000000..204fdb2 --- /dev/null +++ b/src/blockchain/payout-finalizer.service.spec.ts @@ -0,0 +1,74 @@ +import { PayoutStatus } from '@prisma/client'; +import { PayoutFinalizerService } from './payout-finalizer.service'; + +const EVENT = 'evt-1'; +const CHAIN = 'BASE-SEPOLIA'; + +function setup( + opts: { + activePayout?: { id: string } | null; + balances?: bigint[]; + } = {}, +) { + const { activePayout = { id: 'p-1' }, balances = [0n] } = opts; + + const findFirst = jest.fn().mockResolvedValue(activePayout); + const updateMany = jest.fn().mockResolvedValue({ count: 1 }); + const findManyTypes = jest + .fn() + .mockResolvedValue( + balances.map((_, i) => ({ onChainTicketId: BigInt(i + 1) })), + ); + + let call = 0; + const getTicketBalance = jest.fn(() => + Promise.resolve(balances[call++] ?? 0n), + ); + // No TicketBalanceWithdrawn log → extractWithdrawn returns null; the + // completion path still runs. + const getProvider = jest.fn(() => ({ + getTransactionReceipt: jest.fn().mockResolvedValue({ logs: [] }), + })); + + const prisma = { + payout: { findFirst, updateMany }, + ticketType: { findMany: findManyTypes }, + }; + const read = { getProvider, getTicketBalance }; + + const service = new PayoutFinalizerService(prisma as never, read as never); + return { service, updateMany, findFirst }; +} + +describe('PayoutFinalizerService.finalize — auto-complete', () => { + it('completes the payout once escrow is fully drained', async () => { + const m = setup({ balances: [0n, 0n] }); + + await m.service.finalize({ eventId: EVENT, chain: CHAIN }, '0xhash'); + + expect(m.updateMany).toHaveBeenCalledWith( + expect.objectContaining({ + data: expect.objectContaining({ + status: PayoutStatus.COMPLETED, + providerReference: '0xhash', + }), + }), + ); + }); + + it('leaves the payout PROCESSING while escrow remains', async () => { + const m = setup({ balances: [0n, 5n] }); + + await m.service.finalize({ eventId: EVENT, chain: CHAIN }, '0xhash'); + + expect(m.updateMany).not.toHaveBeenCalled(); + }); + + it('no-ops when the event has no active payout', async () => { + const m = setup({ activePayout: null }); + + await m.service.finalize({ eventId: EVENT, chain: CHAIN }, '0xhash'); + + expect(m.updateMany).not.toHaveBeenCalled(); + }); +}); diff --git a/src/blockchain/payout-finalizer.service.ts b/src/blockchain/payout-finalizer.service.ts new file mode 100644 index 0000000..42244c6 --- /dev/null +++ b/src/blockchain/payout-finalizer.service.ts @@ -0,0 +1,161 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { PayoutStatus } from '@prisma/client'; +import { Interface, type LogDescription } from 'ethers'; +import { PrismaService } from '../prisma/prisma.service'; +import { diamondAbi } from './abis'; +import { BlockchainReadService } from './blockchain-read.service'; +import { FEE_TYPE_USDC, USDC_DECIMALS } from './onchain-fees'; + +/** Format a 6-dp USDC base-unit amount as a decimal string, for logs. */ +function formatUsdc(raw: bigint): string { + const s = raw.toString().padStart(USDC_DECIMALS + 1, '0'); + return `${s.slice(0, -USDC_DECIMALS)}.${s.slice(-USDC_DECIMALS)}`; +} + +/** + * Shared post-payout finalization. Given a confirmed + * `withdrawTicketBalance` tx hash, parses the `TicketBalanceWithdrawn` + * event to recover the amount + destination and writes an audit log line + * (tx hash + destination + amount — the #37 acceptance criterion). + * + * Both completion paths drive it identically: + * - the polling fallback in PayoutProcessor, and + * - the Circle webhook handler, authoritative once + * `circle.webhooksEnabled` is on. + * + * A payout request (#46) may fan out several per-ticket-type withdraws. + * Each confirmed withdraw drives this hook; once the event's remaining + * escrow across all ticket types hits zero, the event's active Payout + * row is closed to COMPLETED. The compare-and-set on status makes it + * idempotent under webhook re-delivery / poll races. + */ +@Injectable() +export class PayoutFinalizerService { + private readonly logger = new Logger(PayoutFinalizerService.name); + private readonly iface = new Interface(diamondAbi); + + constructor( + private readonly prisma: PrismaService, + private readonly read: BlockchainReadService, + ) {} + + async finalize( + input: { eventId: string; chain: string }, + txHash: string, + ): Promise { + const withdrawn = await this.extractWithdrawn(input.chain, txHash); + + if (withdrawn) { + this.logger.log( + `Payout settled (event=${input.eventId}, ticketId=${withdrawn.ticketId}, ` + + `amount=${formatUsdc(withdrawn.fee)} USDC, to=${withdrawn.to}, txHash=${txHash})`, + ); + } else { + // Confirmed on-chain but no TicketBalanceWithdrawn in the receipt — + // treat as a zero/no-op withdraw rather than failing the payout. + this.logger.log( + `Payout confirmed with no TicketBalanceWithdrawn event (event=${input.eventId}, txHash=${txHash}) — nothing withdrawn`, + ); + } + + await this.maybeCompletePayout(input.eventId, input.chain, txHash); + } + + // ---------- internals ---------- + + /** + * Close the event's active Payout once its escrow is fully drained. + * No-op when the event has no in-flight payout (e.g. a withdraw + * triggered outside the #46 request flow) or escrow remains. + */ + private async maybeCompletePayout( + eventId: string, + chain: string, + txHash: string, + ): Promise { + const payout = await this.prisma.payout.findFirst({ + where: { + eventId, + status: { in: [PayoutStatus.PENDING, PayoutStatus.PROCESSING] }, + }, + select: { id: true }, + }); + if (!payout) return; + + const ticketTypes = await this.prisma.ticketType.findMany({ + where: { eventId, onChainTicketId: { not: null } }, + select: { onChainTicketId: true }, + }); + + let remaining = 0n; + for (const t of ticketTypes) { + if (t.onChainTicketId === null) continue; + remaining += await this.read.getTicketBalance( + chain, + t.onChainTicketId, + FEE_TYPE_USDC, + ); + } + + if (remaining > 0n) { + this.logger.log( + `Payout ${payout.id} still has ${formatUsdc(remaining)} USDC escrow outstanding — leaving PROCESSING`, + ); + return; + } + + // Compare-and-set: a concurrent finalizer that already closed the + // row updates zero rows and we skip. + const { count } = await this.prisma.payout.updateMany({ + where: { + id: payout.id, + status: { in: [PayoutStatus.PENDING, PayoutStatus.PROCESSING] }, + }, + data: { + status: PayoutStatus.COMPLETED, + processedAt: new Date(), + providerReference: txHash, + }, + }); + + if (count > 0) { + this.logger.log( + `Payout ${payout.id} completed (event=${eventId}, txHash=${txHash})`, + ); + } + } + + private async extractWithdrawn( + chain: string, + txHash: string, + ): Promise<{ ticketId: bigint; fee: bigint; to: string } | null> { + const provider = this.read.getProvider(chain); + const receipt = await provider.getTransactionReceipt(txHash); + if (!receipt) { + throw new Error(`No receipt for tx ${txHash} on ${chain}`); + } + + let parsed: LogDescription | null = null; + for (const log of receipt.logs) { + try { + const p = this.iface.parseLog({ + topics: Array.from(log.topics), + data: log.data, + }); + if (p?.name === 'TicketBalanceWithdrawn') { + parsed = p; + break; + } + } catch { + // not a known facet event — skip + } + } + + if (!parsed) return null; + return { + ticketId: parsed.args.ticketId as bigint, + fee: parsed.args.fee as bigint, + to: parsed.args.to as string, + }; + } +} diff --git a/src/blockchain/payout-queue.service.ts b/src/blockchain/payout-queue.service.ts new file mode 100644 index 0000000..a82a097 --- /dev/null +++ b/src/blockchain/payout-queue.service.ts @@ -0,0 +1,76 @@ +import { InjectQueue } from '@nestjs/bullmq'; +import { Injectable, Logger } from '@nestjs/common'; +import { BlockchainTxStatus, BlockchainTxType } from '@prisma/client'; +import { Queue } from 'bullmq'; +import { PrismaService } from '../prisma/prisma.service'; + +export const TICKET_PAYOUT_QUEUE = 'ticket-payout'; +export const PAYOUT_TICKET_JOB = 'payout-ticket'; + +export interface PayoutTicketJobData { + /** TicketType whose on-chain escrow balance is being withdrawn. */ + ticketTypeId: string; + eventId: string; + blockchainTxId: string; +} + +/** + * Producer for the `ticket-payout` queue. Withdraws an organizer's + * escrowed USDC from the Diamond via `withdrawTicketBalance`, one job + * per TicketType (the on-chain unit — escrow accrues per on-chain + * ticketId). + * + * Only crypto revenue for REFUNDABLE events is escrowed on-chain and + * therefore withdrawable here: non-refundable crypto pays the organizer + * instantly at mint, and fiat is split to the organizer's bank at + * purchase time via Paystack/Monnify subaccounts — neither touches this + * queue. + * + * Trigger-agnostic: the caller (the #46 organizer request endpoint, or a + * future payout cron) decides which ticket types to enqueue. The + * BlockchainTransaction row is written eagerly so there's always an audit + * row; the worker attaches the Circle transaction id when it submits. + */ +@Injectable() +export class PayoutQueueService { + private readonly logger = new Logger(PayoutQueueService.name); + + constructor( + private readonly prisma: PrismaService, + @InjectQueue(TICKET_PAYOUT_QUEUE) private readonly queue: Queue, + ) {} + + async enqueuePayout(ticketTypeId: string, eventId: string): Promise { + const blockchainTx = await this.prisma.blockchainTransaction.create({ + data: { + eventId, + type: BlockchainTxType.WITHDRAW, + status: BlockchainTxStatus.PENDING, + }, + }); + + await this.queue.add( + PAYOUT_TICKET_JOB, + { + ticketTypeId, + eventId, + blockchainTxId: blockchainTx.id, + } satisfies PayoutTicketJobData, + { + // Withdraw moves real on-chain value to the organizer. Five + // attempts with exponential backoff rides out transient Circle + // outages; the processor short-circuits terminal reverts + // (insufficient balance, unauthorized) so those don't burn the + // full budget. + attempts: 5, + backoff: { type: 'exponential', delay: 10_000 }, + removeOnComplete: { age: 60 * 60 * 24, count: 1000 }, + removeOnFail: { age: 60 * 60 * 24 * 7 }, + }, + ); + + this.logger.log( + `Queued ticket-payout for ticketType=${ticketTypeId} event=${eventId} blockchainTxId=${blockchainTx.id}`, + ); + } +} diff --git a/src/blockchain/payout.processor.spec.ts b/src/blockchain/payout.processor.spec.ts new file mode 100644 index 0000000..b16dad4 --- /dev/null +++ b/src/blockchain/payout.processor.spec.ts @@ -0,0 +1,180 @@ +import { + BlockchainTxStatus, + BlockchainTxType, + WalletCreationStatus, +} from '@prisma/client'; +import { UnrecoverableError, type Job } from 'bullmq'; +import { PayoutProcessor } from './payout.processor'; +import { FEE_TYPE_USDC } from './onchain-fees'; +import { PAYOUT_TICKET_JOB, PayoutTicketJobData } from './payout-queue.service'; + +function ticketTypeRow(overrides: Record = {}) { + return { + onChainTicketId: 9n, + event: { id: 'e-1', chain: 'BASE-SEPOLIA', organizerId: 'org-1' }, + ...overrides, + }; +} + +function setup( + row: Record | null, + opts: { + balance?: bigint; + poll?: Record; + webhooksEnabled?: boolean; + wallet?: Record | null; + } = {}, +) { + const findTicketType = jest.fn().mockResolvedValue(row); + const findWallet = jest.fn().mockResolvedValue( + opts.wallet === undefined + ? { + circleWalletId: 'cw-org', + address: '0xORG', + creationStatus: WalletCreationStatus.CREATED, + } + : opts.wallet, + ); + const update = jest.fn().mockResolvedValue({}); + const getTicketBalance = jest + .fn() + .mockResolvedValue(opts.balance ?? 1_000_000n); + const executeContract = jest + .fn() + .mockResolvedValue({ circleTransactionId: 'payout-1' }); + const pollUntilTerminal = jest + .fn() + .mockResolvedValue(opts.poll ?? { state: 'CONFIRMED', txHash: '0xhash' }); + const finalize = jest.fn().mockResolvedValue(undefined); + const get = jest.fn().mockReturnValue(opts.webhooksEnabled ?? false); + + const proc = new PayoutProcessor( + { + ticketType: { findUnique: findTicketType }, + userWallet: { findFirst: findWallet }, + blockchainTransaction: { update }, + } as never, + { executeContract, pollUntilTerminal } as never, + { getTicketBalance } as never, + { finalize } as never, + { get } as never, + ); + return { + proc, + executeContract, + pollUntilTerminal, + finalize, + update, + getTicketBalance, + }; +} + +function job(): Job { + return { + name: PAYOUT_TICKET_JOB, + data: { ticketTypeId: 'tt-1', eventId: 'e-1', blockchainTxId: 'bt-1' }, + attemptsMade: 0, + opts: { attempts: 5 }, + } as Job; +} + +describe('PayoutProcessor', () => { + it('withdraws escrow to the organizer wallet, signed by the organizer', async () => { + const m = setup(ticketTypeRow()); + + await m.proc.process(job()); + + expect(m.getTicketBalance).toHaveBeenCalledWith( + 'BASE-SEPOLIA', + 9n, + FEE_TYPE_USDC, + ); + expect(m.executeContract).toHaveBeenCalledWith( + expect.objectContaining({ + method: 'withdrawTicketBalance', + args: [9n, FEE_TYPE_USDC, '0xORG'], + walletId: 'cw-org', // organizer signs — they own the balance + txType: BlockchainTxType.WITHDRAW, + existingBlockchainTransactionId: 'bt-1', + chain: 'BASE-SEPOLIA', + }), + ); + expect(m.finalize).toHaveBeenCalledWith( + { eventId: 'e-1', chain: 'BASE-SEPOLIA' }, + '0xhash', + ); + }); + + it('zero escrow balance: no-op, closes the audit row as CONFIRMED', async () => { + const m = setup(ticketTypeRow(), { balance: 0n }); + + await m.proc.process(job()); + + expect(m.executeContract).not.toHaveBeenCalled(); + expect(m.update).toHaveBeenCalledWith( + expect.objectContaining({ + where: { id: 'bt-1' }, + data: { status: BlockchainTxStatus.CONFIRMED }, + }), + ); + }); + + it('webhook-authoritative: submits and returns without inline finalize', async () => { + const m = setup(ticketTypeRow(), { webhooksEnabled: true }); + + await m.proc.process(job()); + + expect(m.executeContract).toHaveBeenCalled(); + expect(m.pollUntilTerminal).not.toHaveBeenCalled(); + expect(m.finalize).not.toHaveBeenCalled(); + }); + + it('unpublished ticket type (no onChainTicketId): fails terminally', async () => { + const m = setup(ticketTypeRow({ onChainTicketId: null })); + + await expect(m.proc.process(job())).rejects.toBeInstanceOf( + UnrecoverableError, + ); + expect(m.executeContract).not.toHaveBeenCalled(); + }); + + it('organizer wallet not ready: retryable', async () => { + const m = setup(ticketTypeRow(), { wallet: null }); + + await expect(m.proc.process(job())).rejects.not.toBeInstanceOf( + UnrecoverableError, + ); + expect(m.executeContract).not.toHaveBeenCalled(); + }); + + it('terminal revert (insufficient balance): no retry, marks FAILED', async () => { + const m = setup(ticketTypeRow(), { + poll: { state: 'FAILED', errorReason: 'InsufficientWithdrawBalance' }, + }); + + await expect(m.proc.process(job())).rejects.toBeInstanceOf( + UnrecoverableError, + ); + expect(m.finalize).not.toHaveBeenCalled(); + expect(m.update).toHaveBeenCalledWith( + expect.objectContaining({ + data: expect.objectContaining({ status: BlockchainTxStatus.FAILED }), + }), + ); + }); + + it('withdraw period not reached: retryable (not Unrecoverable)', async () => { + const m = setup(ticketTypeRow(), { + poll: { state: 'FAILED', errorReason: 'WithdrawPeriodNotReached' }, + }); + + await expect(m.proc.process(job())).rejects.not.toBeInstanceOf( + UnrecoverableError, + ); + expect(m.update).toHaveBeenCalledWith( + expect.objectContaining({ + data: expect.objectContaining({ status: BlockchainTxStatus.PENDING }), + }), + ); + }); +}); diff --git a/src/blockchain/payout.processor.ts b/src/blockchain/payout.processor.ts new file mode 100644 index 0000000..1475655 --- /dev/null +++ b/src/blockchain/payout.processor.ts @@ -0,0 +1,257 @@ +import { Processor, WorkerHost } from '@nestjs/bullmq'; +import { Logger } from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; +import { + BlockchainTxStatus, + BlockchainTxType, + WalletCreationStatus, +} from '@prisma/client'; +import { Job, UnrecoverableError } from 'bullmq'; +import { Interface } from 'ethers'; +import { PrismaService } from '../prisma/prisma.service'; +import { diamondAbi } from './abis'; +import { BlockchainReadService } from './blockchain-read.service'; +import { CircleContractService } from './circle-contract.service'; +import { FEE_TYPE_USDC } from './onchain-fees'; +import { PayoutFinalizerService } from './payout-finalizer.service'; +import { + PAYOUT_TICKET_JOB, + PayoutTicketJobData, + TICKET_PAYOUT_QUEUE, +} from './payout-queue.service'; + +/** + * Contract reverts that no amount of retrying will fix — the withdraw is + * simply not allowed. We short-circuit Bull's retry budget for these + * (UnrecoverableError → straight to the failed set) and mark the + * BlockchainTransaction FAILED for auditor review. + * + * `WithdrawPeriodNotReached` is intentionally NOT here: the caller is + * expected to gate on the withdraw period before enqueuing, but if a job + * lands early we let it ride the normal retry/backoff rather than hard- + * fail it. + */ +const TERMINAL_PAYOUT_ERRORS = new Set([ + 'InsufficientWithdrawBalance', + 'FiatBalanceNotWithdrawable', + 'AccessControlUnauthorizedAccount', + 'TicketDoesNotExist', +]); + +/** + * Consumes `ticket-payout` jobs and withdraws an organizer's escrowed + * USDC from the Diamond via `withdrawTicketBalance(uint64 ticketId, + * FeeType feeType, address to)`, signed by the ORGANIZER's Circle wallet + * (they are the ticket admin — only they may withdraw) and paid out to + * that same wallet. + * + * Preflight reads the on-chain balance and no-ops a zero balance (the + * natural idempotency guard: a second run after a successful withdraw + * reads zero and skips). + * + * Completion mirrors the mint/refund workers: + * - `circle.webhooksEnabled` ON: submit and return — the Circle webhook + * handler reconciles and calls PayoutFinalizerService. + * - OFF (fallback): poll until terminal here, then finalize inline. + */ +@Processor(TICKET_PAYOUT_QUEUE) +export class PayoutProcessor extends WorkerHost { + private readonly logger = new Logger(PayoutProcessor.name); + private readonly iface = new Interface(diamondAbi); + + constructor( + private readonly prisma: PrismaService, + private readonly circle: CircleContractService, + private readonly read: BlockchainReadService, + private readonly finalizer: PayoutFinalizerService, + private readonly config: ConfigService, + ) { + super(); + } + + async process(job: Job): Promise { + if (job.name !== PAYOUT_TICKET_JOB) { + this.logger.warn( + `Unexpected job name on ${TICKET_PAYOUT_QUEUE}: ${job.name}`, + ); + return; + } + + const { ticketTypeId, eventId, blockchainTxId } = job.data; + + const ticketType = await this.prisma.ticketType.findUnique({ + where: { id: ticketTypeId }, + select: { + onChainTicketId: true, + event: { select: { id: true, chain: true, organizerId: true } }, + }, + }); + + if (!ticketType) { + throw new Error(`TicketType ${ticketTypeId} not found`); + } + if (ticketType.onChainTicketId === null) { + // Never published on-chain → no escrow to withdraw. Terminal. + await this.markFailed( + blockchainTxId, + 'TicketType has no onChainTicketId', + ); + throw new UnrecoverableError( + `TicketType ${ticketTypeId} has no onChainTicketId; nothing to withdraw`, + ); + } + + const { chain, organizerId } = ticketType.event; + const onChainTicketId = ticketType.onChainTicketId; + + const organizerWallet = await this.prisma.userWallet.findFirst({ + where: { + userId: organizerId, + chain, + creationStatus: WalletCreationStatus.CREATED, + }, + }); + if (!organizerWallet?.circleWalletId || !organizerWallet.address) { + // Transient during provisioning: retry with backoff. + throw new Error( + `Organizer ${organizerId} has no ready wallet on ${chain} to sign the withdraw`, + ); + } + + // Preflight: nothing to withdraw → close the audit row and stop. + const balance = await this.read.getTicketBalance( + chain, + onChainTicketId, + FEE_TYPE_USDC, + ); + if (balance === 0n) { + this.logger.log( + `Payout no-op: ticketType=${ticketTypeId} (onChainTicketId=${onChainTicketId}) has zero escrow balance`, + ); + await this.prisma.blockchainTransaction.update({ + where: { id: blockchainTxId }, + data: { status: BlockchainTxStatus.CONFIRMED }, + }); + return; + } + + try { + const { circleTransactionId } = await this.circle.executeContract({ + method: 'withdrawTicketBalance', + args: [ + onChainTicketId, // uint64 _ticketId + FEE_TYPE_USDC, // enum FeeType _feeType (USDC) + organizerWallet.address, // address _to (payout destination) + ], + chain, + txType: BlockchainTxType.WITHDRAW, + eventId, + existingBlockchainTransactionId: blockchainTxId, + walletId: organizerWallet.circleWalletId, // organizer signs — they own the balance + }); + + this.logger.log( + `withdrawTicketBalance submitted (ticketType=${ticketTypeId}, circleTxId=${circleTransactionId})`, + ); + + // Webhook is authoritative for completion when wired. + if (this.config.get('circle.webhooksEnabled')) { + this.logger.log( + `withdrawTicketBalance awaiting Circle webhook for completion (ticketType=${ticketTypeId})`, + ); + return; + } + + // Fallback: poll until terminal. A withdraw is one block. + const final = await this.circle.pollUntilTerminal(circleTransactionId, { + intervalMs: 4_000, + timeoutMs: 180_000, + }); + + if (final.state !== 'CONFIRMED' && final.state !== 'COMPLETE') { + const reason = final.errorReason ?? '(no reason)'; + if (this.isTerminalPayoutError(reason)) { + throw new UnrecoverableError( + `withdrawTicketBalance not permitted (ticketType=${ticketTypeId}): ${this.describeError(reason)}`, + ); + } + throw new Error( + `withdrawTicketBalance on-chain state ${final.state}: ${reason}`, + ); + } + if (!final.txHash) { + throw new Error('Circle reported terminal success without a txHash'); + } + + await this.finalizer.finalize({ eventId, chain }, final.txHash); + } catch (error) { + const message = error instanceof Error ? error.message : 'unknown error'; + const terminal = + error instanceof UnrecoverableError || + this.isTerminalPayoutError(message); + const isFinal = + terminal || job.attemptsMade + 1 >= (job.opts.attempts ?? 1); + + await this.prisma.blockchainTransaction.update({ + where: { id: blockchainTxId }, + data: { + status: isFinal + ? BlockchainTxStatus.FAILED + : BlockchainTxStatus.PENDING, + error: message.slice(0, 500), + }, + }); + + if (terminal && !(error instanceof UnrecoverableError)) { + this.logger.error( + `Payout not permitted (ticketType=${ticketTypeId}): ${message}`, + ); + throw new UnrecoverableError(message); + } + + if (isFinal) { + this.logger.error( + `Payout failed (ticketType=${ticketTypeId}, attempts=${job.attemptsMade + 1}): ${message}`, + ); + } else { + this.logger.warn( + `Payout attempt ${job.attemptsMade + 1} failed (ticketType=${ticketTypeId}): ${message}`, + ); + } + throw error; + } + } + + // ---------- internals ---------- + + private async markFailed( + blockchainTxId: string, + error: string, + ): Promise { + await this.prisma.blockchainTransaction.update({ + where: { id: blockchainTxId }, + data: { status: BlockchainTxStatus.FAILED, error: error.slice(0, 500) }, + }); + } + + private isTerminalPayoutError(reason: string): boolean { + const name = this.decodeErrorName(reason); + if (name && TERMINAL_PAYOUT_ERRORS.has(name)) return true; + return [...TERMINAL_PAYOUT_ERRORS].some((e) => reason.includes(e)); + } + + private describeError(reason: string): string { + return this.decodeErrorName(reason) ?? reason; + } + + /** Parse a custom-error name out of raw revert data, if present. */ + private decodeErrorName(reason: string): string | null { + const match = reason.match(/0x[0-9a-fA-F]{8,}/); + if (!match) return null; + try { + return this.iface.parseError(match[0])?.name ?? null; + } catch { + return null; + } + } +} diff --git a/src/organizer/dto/query-payouts.dto.ts b/src/organizer/dto/query-payouts.dto.ts new file mode 100644 index 0000000..e8d13f0 --- /dev/null +++ b/src/organizer/dto/query-payouts.dto.ts @@ -0,0 +1,14 @@ +import { IsEnum, IsOptional } from 'class-validator'; +import { PayoutStatus } from '@prisma/client'; +import { PaginationDto } from '../../common/dto/pagination.dto'; + +/** + * Query for `GET /organizer/payouts`. Paginated; `status` optionally + * filters the list. The summary is always computed across ALL of the + * organizer's payouts, independent of the filter. + */ +export class QueryPayoutsDto extends PaginationDto { + @IsOptional() + @IsEnum(PayoutStatus) + status?: PayoutStatus; +} diff --git a/src/organizer/dto/request-payout.dto.ts b/src/organizer/dto/request-payout.dto.ts new file mode 100644 index 0000000..46249a1 --- /dev/null +++ b/src/organizer/dto/request-payout.dto.ts @@ -0,0 +1,11 @@ +import { IsUUID } from 'class-validator'; + +/** + * Body for `POST /organizer/payouts/request`. Crypto-only in this slice: + * the payout withdraws the event's on-chain USDC escrow, so no provider + * or destination is taken — funds go to the organizer's Circle wallet. + */ +export class RequestPayoutDto { + @IsUUID() + eventId: string; +} diff --git a/src/organizer/organizer.controller.ts b/src/organizer/organizer.controller.ts index fdac5fd..ce71182 100644 --- a/src/organizer/organizer.controller.ts +++ b/src/organizer/organizer.controller.ts @@ -23,10 +23,13 @@ import { EnableMonnifyDto } from './dto/enable-monnify.dto'; import { EnablePaystackDto } from './dto/enable-paystack.dto'; import { QueryOrganizerEventsDto } from './dto/query-organizer-events.dto'; import { QueryAttendeesDto } from './dto/query-attendees.dto'; +import { QueryPayoutsDto } from './dto/query-payouts.dto'; +import { RequestPayoutDto } from './dto/request-payout.dto'; import { UpdateBankDetailsDto } from './dto/update-bank-details.dto'; import { TicketAdminsDto } from './dto/ticket-admins.dto'; import { UpdateTicketFeeDto } from './dto/update-ticket-fee.dto'; import { OrganizerService } from './organizer.service'; +import { PayoutsService } from './payouts.service'; import { TicketAdminsService } from './ticket-admins.service'; import { TicketFeesService } from './ticket-fees.service'; import { OnchainReadsService } from './onchain-reads.service'; @@ -45,6 +48,7 @@ import { OnchainReadsService } from './onchain-reads.service'; export class OrganizerController { constructor( private readonly organizer: OrganizerService, + private readonly payouts: PayoutsService, private readonly ticketAdmins: TicketAdminsService, private readonly ticketFees: TicketFeesService, private readonly onchainReads: OnchainReadsService, @@ -201,6 +205,29 @@ export class OrganizerController { return this.onchainReads.getCheckinsForDay(userId, id, day); } + @Post('payouts/request') + @Roles(UserRole.ORGANIZER) + @HttpCode(HttpStatus.CREATED) + @ApiOperation({ + summary: 'Request an event payout (withdraw on-chain USDC escrow)', + }) + requestPayout( + @CurrentUser('id') userId: string, + @Body() dto: RequestPayoutDto, + ) { + return this.payouts.requestPayout(userId, dto.eventId); + } + + @Get('payouts') + @Roles(UserRole.ORGANIZER) + @ApiOperation({ summary: 'Payout history with summary (filter, paginate)' }) + getPayouts( + @CurrentUser('id') userId: string, + @Query() query: QueryPayoutsDto, + ) { + return this.payouts.getPayoutHistory(userId, query); + } + @Post('providers/paystack/enable') @Roles(UserRole.ORGANIZER) @HttpCode(HttpStatus.OK) diff --git a/src/organizer/organizer.module.ts b/src/organizer/organizer.module.ts index 7ef347a..903f757 100644 --- a/src/organizer/organizer.module.ts +++ b/src/organizer/organizer.module.ts @@ -5,6 +5,7 @@ import { BlockchainModule } from '../blockchain/blockchain.module'; import { WalletsModule } from '../wallets/wallets.module'; import { OrganizerController } from './organizer.controller'; import { OrganizerService } from './organizer.service'; +import { PayoutsService } from './payouts.service'; import { TicketAdminsService } from './ticket-admins.service'; import { TicketFeesService } from './ticket-fees.service'; import { OnchainReadsService } from './onchain-reads.service'; @@ -20,6 +21,7 @@ import { OnchainReadsService } from './onchain-reads.service'; controllers: [OrganizerController], providers: [ OrganizerService, + PayoutsService, TicketAdminsService, TicketFeesService, OnchainReadsService, diff --git a/src/organizer/payouts.service.spec.ts b/src/organizer/payouts.service.spec.ts new file mode 100644 index 0000000..f6a6abd --- /dev/null +++ b/src/organizer/payouts.service.spec.ts @@ -0,0 +1,200 @@ +import { BadRequestException, ForbiddenException } from '@nestjs/common'; +import { + EventStatus, + PaymentProvider, + PayoutStatus, + Prisma, +} from '@prisma/client'; +import { PayoutsService } from './payouts.service'; + +const ORG = 'org-1'; +const EVENT = 'evt-1'; +const CHAIN = 'BASE-SEPOLIA'; +const PAST = new Date('2020-01-01T00:00:00Z'); +const REFUND_PERIOD = 259_200n; // 3 days in seconds + +function eventRow(overrides: Record = {}) { + return { + id: EVENT, + name: 'Lagos Tech Summit', + organizerId: ORG, + status: EventStatus.COMPLETED, + endTime: PAST, + chain: CHAIN, + ticketTypes: [ + { id: 'tt-1', onChainTicketId: 7n }, + { id: 'tt-2', onChainTicketId: 8n }, + ], + ...overrides, + }; +} + +function payoutRow(overrides: Record = {}) { + return { + id: 'p-1', + eventId: EVENT, + amount: new Prisma.Decimal('1'), + currency: 'USDC', + provider: PaymentProvider.CRYPTO, + status: PayoutStatus.PROCESSING, + providerReference: null, + scheduledDate: PAST, + processedAt: null, + createdAt: PAST, + event: { name: 'Lagos Tech Summit' }, + ...overrides, + }; +} + +function setup( + opts: { + event?: Record | null; + refundPeriod?: bigint; + duplicate?: unknown; + balances?: Record; + } = {}, +) { + const { + event = eventRow(), + refundPeriod = REFUND_PERIOD, + duplicate = null, + balances = { '7': 1_000_000n, '8': 0n }, + } = opts; + + const findEvent = jest.fn().mockResolvedValue(event); + const findFirst = jest.fn().mockResolvedValue(duplicate); + const create = jest + .fn() + .mockResolvedValue(payoutRow({ status: PayoutStatus.PENDING })); + const update = jest.fn().mockResolvedValue(payoutRow()); + const getRefundPeriod = jest.fn().mockResolvedValue(refundPeriod); + const getTicketBalance = jest.fn((_c: string, id: bigint) => + Promise.resolve(balances[id.toString()] ?? 0n), + ); + const enqueuePayout = jest.fn().mockResolvedValue(undefined); + + const prisma = { + event: { findUnique: findEvent }, + payout: { findFirst, create, update }, + }; + const read = { getRefundPeriod, getTicketBalance }; + + const service = new PayoutsService( + prisma as never, + read as never, + { enqueuePayout } as never, + ); + return { service, enqueuePayout, create, update, getTicketBalance }; +} + +describe('PayoutsService.requestPayout', () => { + it('withdraws escrow: enqueues per ticket type with a balance, returns PROCESSING', async () => { + const m = setup(); + + const res = await m.service.requestPayout(ORG, EVENT); + + // Only tt-1 has a balance; tt-2 (zero) is skipped. + expect(m.enqueuePayout).toHaveBeenCalledTimes(1); + expect(m.enqueuePayout).toHaveBeenCalledWith('tt-1', EVENT); + // Amount is escrow base units / 1e6. + expect(m.create).toHaveBeenCalledWith( + expect.objectContaining({ + data: expect.objectContaining({ + currency: 'USDC', + provider: PaymentProvider.CRYPTO, + }), + }), + ); + expect(res.status).toBe(PayoutStatus.PROCESSING); + expect(res.amount).toBe('1'); + }); + + it('rejects a non-owner', async () => { + const m = setup({ event: eventRow({ organizerId: 'someone-else' }) }); + await expect(m.service.requestPayout(ORG, EVENT)).rejects.toBeInstanceOf( + ForbiddenException, + ); + expect(m.enqueuePayout).not.toHaveBeenCalled(); + }); + + it('rejects a draft event', async () => { + const m = setup({ event: eventRow({ status: EventStatus.DRAFT }) }); + await expect(m.service.requestPayout(ORG, EVENT)).rejects.toBeInstanceOf( + BadRequestException, + ); + }); + + it('rejects before the refund period elapses', async () => { + const m = setup({ event: eventRow({ endTime: new Date() }) }); + await expect(m.service.requestPayout(ORG, EVENT)).rejects.toThrow( + /Refund period/, + ); + expect(m.enqueuePayout).not.toHaveBeenCalled(); + }); + + it('rejects a duplicate pending/processing payout', async () => { + const m = setup({ duplicate: { id: 'existing' } }); + await expect(m.service.requestPayout(ORG, EVENT)).rejects.toThrow( + /already pending/, + ); + expect(m.enqueuePayout).not.toHaveBeenCalled(); + }); + + it('rejects when there is no withdrawable escrow', async () => { + const m = setup({ balances: { '7': 0n, '8': 0n } }); + await expect(m.service.requestPayout(ORG, EVENT)).rejects.toThrow( + /No withdrawable/, + ); + expect(m.create).not.toHaveBeenCalled(); + }); +}); + +describe('PayoutsService.getPayoutHistory', () => { + function historyService(rows: unknown[], total: number) { + const $transaction = jest + .fn() + .mockResolvedValue([ + rows, + total, + { _sum: { amount: new Prisma.Decimal('8.5') } }, + { _sum: { amount: new Prisma.Decimal('1.25') } }, + total, + ]); + const prisma = { + payout: { + findMany: jest.fn(), + count: jest.fn(), + aggregate: jest.fn(), + }, + $transaction, + }; + return new PayoutsService(prisma as never, {} as never, {} as never); + } + + it('returns mapped payouts, summary, and pagination', async () => { + const service = historyService( + [payoutRow({ status: PayoutStatus.COMPLETED })], + 1, + ); + + const res = await service.getPayoutHistory(ORG, { + page: 1, + limit: 10, + skip: 0, + } as never); + + expect(res.payouts).toHaveLength(1); + expect(res.payouts[0].eventName).toBe('Lagos Tech Summit'); + expect(res.summary).toEqual({ + totalPaid: '8.5', + pendingAmount: '1.25', + totalPayouts: 1, + }); + expect(res.pagination).toEqual({ + page: 1, + limit: 10, + total: 1, + totalPages: 1, + }); + }); +}); diff --git a/src/organizer/payouts.service.ts b/src/organizer/payouts.service.ts new file mode 100644 index 0000000..f9264f4 --- /dev/null +++ b/src/organizer/payouts.service.ts @@ -0,0 +1,237 @@ +import { + BadRequestException, + ForbiddenException, + Injectable, + Logger, + NotFoundException, +} from '@nestjs/common'; +import { + EventStatus, + PaymentProvider, + PayoutStatus, + Prisma, +} from '@prisma/client'; +import { PrismaService } from '../prisma/prisma.service'; +import { BlockchainReadService } from '../blockchain/blockchain-read.service'; +import { PayoutQueueService } from '../blockchain/payout-queue.service'; +import { FEE_TYPE_USDC, USDC_DECIMALS } from '../blockchain/onchain-fees'; +import { QueryPayoutsDto } from './dto/query-payouts.dto'; + +const USDC_SCALE = new Prisma.Decimal(10).pow(USDC_DECIMALS); + +/** Statuses that block a new payout request / count as outstanding. */ +const ACTIVE_PAYOUT_STATUSES: PayoutStatus[] = [ + PayoutStatus.PENDING, + PayoutStatus.PROCESSING, +]; + +type PayoutWithEvent = Prisma.PayoutGetPayload<{ + include: { event: { select: { name: true } } }; +}>; + +/** + * Organizer payouts (#46). Crypto-only in this slice: a payout withdraws + * the event's on-chain USDC escrow to the organizer's Circle wallet via + * the #37 payout engine. + * + * Fiat revenue is already settled to the organizer's bank at purchase + * time via Paystack/Monnify split subaccounts, so it never surfaces here + * — only escrowed refundable-crypto revenue is withdrawable. + */ +@Injectable() +export class PayoutsService { + private readonly logger = new Logger(PayoutsService.name); + + constructor( + private readonly prisma: PrismaService, + private readonly read: BlockchainReadService, + private readonly payoutQueue: PayoutQueueService, + ) {} + + /** + * Request a payout for an event. Validates ownership, status, the + * on-chain refund window, and duplicate requests, then fans out one + * on-chain withdraw per ticket type that still holds escrow. + */ + async requestPayout(organizerId: string, eventId: string) { + const event = await this.prisma.event.findUnique({ + where: { id: eventId }, + select: { + id: true, + name: true, + organizerId: true, + status: true, + endTime: true, + chain: true, + ticketTypes: { + select: { id: true, onChainTicketId: true }, + orderBy: { createdAt: 'asc' }, + }, + }, + }); + + if (!event) { + throw new NotFoundException('Event not found'); + } + if (event.organizerId !== organizerId) { + throw new ForbiddenException('You do not own this event'); + } + if ( + event.status !== EventStatus.PUBLISHED && + event.status !== EventStatus.COMPLETED + ) { + throw new BadRequestException( + 'Payouts are only available for published or completed events', + ); + } + + // Refund/withdraw window: the Diamond rejects withdrawTicketBalance + // until the refund period after event end has elapsed. Pre-check here + // so the organizer gets a clear error instead of a failed job. + const refundPeriodSecs = await this.read.getRefundPeriod(event.chain); + const withdrawableAt = new Date( + event.endTime.getTime() + Number(refundPeriodSecs) * 1000, + ); + if (Date.now() < withdrawableAt.getTime()) { + throw new BadRequestException( + `Refund period has not elapsed; payout available after ${withdrawableAt.toISOString()}`, + ); + } + + const existing = await this.prisma.payout.findFirst({ + where: { eventId, status: { in: ACTIVE_PAYOUT_STATUSES } }, + }); + if (existing) { + throw new BadRequestException( + 'A payout for this event is already pending or processing', + ); + } + + // Read live escrow per published ticket type; only withdraw the ones + // that actually hold a balance. + const onchainTypes = event.ticketTypes.filter( + (t): t is { id: string; onChainTicketId: bigint } => + t.onChainTicketId !== null, + ); + const withdrawable: { ticketTypeId: string; balance: bigint }[] = []; + let totalRaw = 0n; + for (const t of onchainTypes) { + const balance = await this.read.getTicketBalance( + event.chain, + t.onChainTicketId, + FEE_TYPE_USDC, + ); + if (balance > 0n) { + withdrawable.push({ ticketTypeId: t.id, balance }); + totalRaw += balance; + } + } + + if (totalRaw === 0n) { + throw new BadRequestException( + 'No withdrawable on-chain balance for this event (funds may have already been settled at purchase)', + ); + } + + const amount = new Prisma.Decimal(totalRaw.toString()).div(USDC_SCALE); + + // Create the payout record, fan out the withdraws, then flip to + // PROCESSING. The finalizer closes it to COMPLETED once escrow is + // fully drained. + const payout = await this.prisma.payout.create({ + data: { + organizerId, + eventId, + amount, + currency: 'USDC', + provider: PaymentProvider.CRYPTO, + status: PayoutStatus.PENDING, + scheduledDate: new Date(), + }, + include: { event: { select: { name: true } } }, + }); + + for (const w of withdrawable) { + await this.payoutQueue.enqueuePayout(w.ticketTypeId, eventId); + } + + const processing = await this.prisma.payout.update({ + where: { id: payout.id }, + data: { status: PayoutStatus.PROCESSING }, + include: { event: { select: { name: true } } }, + }); + + this.logger.log( + `Payout requested (event=${eventId}, payout=${payout.id}, amount=${amount.toString()} USDC, withdraws=${withdrawable.length})`, + ); + + return { + ...this.toDto(processing), + message: 'Payout request submitted. Withdrawing on-chain now.', + }; + } + + /** Paginated payout history for the organizer, with an all-time summary. */ + async getPayoutHistory(organizerId: string, query: QueryPayoutsDto) { + const where: Prisma.PayoutWhereInput = { + organizerId, + ...(query.status ? { status: query.status } : {}), + }; + + const [payouts, total, paidAgg, pendingAgg, totalPayouts] = + await this.prisma.$transaction([ + this.prisma.payout.findMany({ + where, + include: { event: { select: { name: true } } }, + orderBy: { createdAt: 'desc' }, + skip: query.skip, + take: query.limit, + }), + this.prisma.payout.count({ where }), + this.prisma.payout.aggregate({ + where: { organizerId, status: PayoutStatus.COMPLETED }, + _sum: { amount: true }, + }), + this.prisma.payout.aggregate({ + where: { organizerId, status: { in: ACTIVE_PAYOUT_STATUSES } }, + _sum: { amount: true }, + }), + this.prisma.payout.count({ where: { organizerId } }), + ]); + + return { + payouts: payouts.map((p) => this.toDto(p)), + summary: { + totalPaid: (paidAgg._sum.amount ?? new Prisma.Decimal(0)).toString(), + pendingAmount: ( + pendingAgg._sum.amount ?? new Prisma.Decimal(0) + ).toString(), + totalPayouts, + }, + pagination: { + page: query.page, + limit: query.limit, + total, + totalPages: Math.ceil(total / query.limit), + }, + }; + } + + // ---------- internals ---------- + + private toDto(p: PayoutWithEvent) { + return { + id: p.id, + eventId: p.eventId, + eventName: p.event.name, + amount: p.amount.toString(), + currency: p.currency, + provider: p.provider, + status: p.status, + providerReference: p.providerReference, + scheduledDate: p.scheduledDate, + processedAt: p.processedAt, + createdAt: p.createdAt, + }; + } +}