Feat/organizer functions

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,
   ],
 })

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,

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

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();
+  });
+});

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<void> {
+    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<void> {
+    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,
+    };
+  }
+}