From 171bd815ab822db0ed9a8fc5bc092e817bdd6118 Mon Sep 17 00:00:00 2001 From: Volodymyr Zakhovaiko Date: Fri, 21 Aug 2026 14:52:24 +0200 Subject: [PATCH 1/2] feat(wallet): reconciliation job, admin routes, and findings Diffs the ledger against each provider's transaction list over a window, files findings instead of ever auto-crediting, and exposes the report plus a resolve flow behind the `wallet-reconciliation` permission. Sweeps are excluded from the diff before it runs, or every internal transfer would file a `missing_deposit` and bury the real findings. Stacks on the custody sweep rather than sitting beside it, so the two are reconciled here instead of at merge time. With both in one tree the last stub is gone, so `notImplemented` and `NOT_IMPLEMENTED_YET` are deleted - reaching zero was the completion check for the whole stack. The test file covering those routes is renamed off "stubs" for the same reason. Includes the review-gate fixes: - The live-webhook currency mismatch is tagged `currency_mismatch`, the same kind the polled path already uses, so filtering by kind sees both. - The window is anchored to the last completed run rather than a fixed `now - lookbackHours`. Repeatable jobs are delayed jobs, so a tick missed while no worker was up is skipped and never backfilled - any outage longer than the lookback left a span nothing reconciled and nothing reported. Catch-up is capped and a capped window is reported. - Stale `pending` sweeps are scanned alongside `unknown` ones. A worker dying between the claim insert and the vendor call left a row nothing polls, while the partial unique index still counted it in-flight, so that container would never sweep again. - Stuck-sweep detection no longer switches itself off when the `wallet.sweep` config block is absent. - A failed run gets its own audit action so the log filters by outcome. --- docs/catalog.json | 1 + packages/core/src/audit/plugin.ts | 14 + packages/core/src/contracts/adapters/audit.ts | 5 +- packages/core/src/contracts/schemas/events.ts | 9 + packages/core/src/testing/mock.ts | 12 + .../__tests__/reconciliation-diff.test.ts | 63 ++ .../reconciliation.service.int.test.ts | 468 ++++++++++++ .../__tests__/wallet-asset.router.int.test.ts | 9 +- .../wallet-auto-rule.router.int.test.ts | 16 +- ...-auto-withdrawal-config.router.int.test.ts | 15 +- ...s => wallet-custody-routes.router.test.ts} | 95 ++- .../wallet-webhook.router.int.test.ts | 49 +- .../__tests__/wallet.router.int.test.ts | 9 +- .../__tests__/wallet.service.int.test.ts | 71 +- packages/core/src/wallet/contract/index.ts | 15 +- packages/core/src/wallet/plugin.ts | 117 ++- packages/core/src/wallet/router/index.ts | 59 +- .../service/reconciliation-finding.service.ts | 63 ++ .../wallet/service/reconciliation.service.ts | 705 ++++++++++++++++++ .../core/src/wallet/service/wallet.service.ts | 58 +- 20 files changed, 1721 insertions(+), 132 deletions(-) create mode 100644 packages/core/src/wallet/__tests__/reconciliation-diff.test.ts create mode 100644 packages/core/src/wallet/__tests__/reconciliation.service.int.test.ts rename packages/core/src/wallet/__tests__/{wallet-custody-stubs.router.test.ts => wallet-custody-routes.router.test.ts} (52%) create mode 100644 packages/core/src/wallet/service/reconciliation-finding.service.ts create mode 100644 packages/core/src/wallet/service/reconciliation.service.ts diff --git a/docs/catalog.json b/docs/catalog.json index 47b25070..1e013fee 100644 --- a/docs/catalog.json +++ b/docs/catalog.json @@ -955,6 +955,7 @@ "tag.rule.upserted", "wallet.deposit.completed", "wallet.manual_adjustment.created", + "wallet.reconciliation.alert", "wallet.withdrawal.approved", "wallet.withdrawal.completed", "wallet.withdrawal.failed", diff --git a/packages/core/src/audit/plugin.ts b/packages/core/src/audit/plugin.ts index 92796aca..99ea9d70 100644 --- a/packages/core/src/audit/plugin.ts +++ b/packages/core/src/audit/plugin.ts @@ -580,6 +580,19 @@ export async function mapEventToRecord( }; } + // System-driven: a reconciliation run's open findings exceeded the configured + // threshold. resource = the run itself; after carries counts only, NEVER a finding's + // payload (an address or tx hash must never reach the audit log through this event). + if (topic === 'wallet.reconciliation.alert') { + return { + ...base, + actorType: 'system', + resourceType: 'wallet_job_run', + resourceId: str(p['runId']), + after: { openFindings: p['openFindings'] ?? null, threshold: p['threshold'] ?? null }, + }; + } + // Wallet events carry the txn ref in transactionId; surface it as resourceId so // a transaction reference is searchable (it otherwise stays buried in `after`). // actorId = the resolved playerId (the wallet owner). @@ -656,6 +669,7 @@ const SUBSCRIBED_TOPICS: DomainEventName[] = [ 'wallet.withdrawal.approved', 'wallet.withdrawal.rejected', 'wallet.withdrawal.failed', + 'wallet.reconciliation.alert', 'gaming.round.started', 'gaming.round.ended', 'chat.user.blocked', diff --git a/packages/core/src/contracts/adapters/audit.ts b/packages/core/src/contracts/adapters/audit.ts index 76bae753..15f43029 100644 --- a/packages/core/src/contracts/adapters/audit.ts +++ b/packages/core/src/contracts/adapters/audit.ts @@ -25,7 +25,10 @@ export type DirectAuditAction = | 'wallet.auto_withdrawal_rule.deleted' | 'wallet.auto_withdrawal_config.set' | 'wallet.manual_adjustment.created' - | 'wallet.custody.sweep_cycle'; + | 'wallet.custody.sweep_cycle' + | 'wallet.reconciliation_run.completed' + | 'wallet.reconciliation_run.failed' + | 'wallet.reconciliation_finding.resolved'; /** * Every value the audit `action` column legitimately holds: a cross-module domain diff --git a/packages/core/src/contracts/schemas/events.ts b/packages/core/src/contracts/schemas/events.ts index c4a02697..ef3beed1 100644 --- a/packages/core/src/contracts/schemas/events.ts +++ b/packages/core/src/contracts/schemas/events.ts @@ -174,6 +174,15 @@ export const domainEventSchemas = { reason: z.string(), }) .extend(authContextBase.shape), + // A reconciliation run's open-findings count exceeded the operator's configured + // threshold. System-driven (no player/admin actor) - the resource is the run itself, + // never a finding's payload (an address or tx hash must never reach the audit log + // through this event - see docs/standards/audit.md). + 'wallet.reconciliation.alert': z.object({ + runId: UuidSchema, + openFindings: z.number().int().nonnegative(), + threshold: z.number().int().nonnegative(), + }), 'gaming.round.started': z.object({ roundId: UuidSchema, diff --git a/packages/core/src/testing/mock.ts b/packages/core/src/testing/mock.ts index b7e1ba3b..d3718a33 100644 --- a/packages/core/src/testing/mock.ts +++ b/packages/core/src/testing/mock.ts @@ -12,6 +12,7 @@ import { type AuditWritePort, type ClientMeta, type IdentityReader, + type JobQueueAdapter, type PaymentAdapter, type PaymentProviderRegistry, type PaymentWebhookVerifier, @@ -134,6 +135,17 @@ export const makePaymentProviderRegistry = ( }; }; +/** JobQueueAdapter double whose `enqueue` is a vitest mock, for a router test that only + * needs to assert a job was enqueued, never that it actually ran. */ +export const makeJobQueue = (): JobQueueAdapter & { enqueue: Mock } => + mock({ + enqueue: vi.fn(async () => ({ id: 'test-job' })), + schedule: vi.fn(async () => undefined), + unschedule: vi.fn(async () => undefined), + registerWorker: vi.fn(), + close: vi.fn(async () => undefined), + }); + export const makeIdentityReader = (): IdentityReader => mock({ getLastLoginAt: vi.fn().mockResolvedValue(null), diff --git a/packages/core/src/wallet/__tests__/reconciliation-diff.test.ts b/packages/core/src/wallet/__tests__/reconciliation-diff.test.ts new file mode 100644 index 00000000..548003fa --- /dev/null +++ b/packages/core/src/wallet/__tests__/reconciliation-diff.test.ts @@ -0,0 +1,63 @@ +import { describe, it, expect } from 'vitest'; +import { diffDeposit } from '../service/reconciliation.service.js'; + +const depositEvent = (over: Partial[0]> = {}) => ({ + kind: 'deposit' as const, + address: 'bc1qxyz', + amount: '1', + currency: 'BTC', + txHash: '0xabc', + externalId: 'vendor-ext-1', + ...over, +}); + +const ledgerRow = ( + over: Partial<{ id: string; currency: string; amount: string; network: string | null }> = {}, +) => ({ + id: 'tx-1', + currency: 'BTC', + amount: '1', + network: null, + ...over, +}); + +describe('diffDeposit', () => { + it('flags missing_deposit when no ledger row matches the vendor event', () => { + expect(diffDeposit(depositEvent(), undefined)).toBe('missing_deposit'); + }); + + it('flags currency_mismatch when the ledger row settled in a different currency', () => { + const event = depositEvent({ currency: 'ETH' }); + const tx = ledgerRow({ currency: 'BTC' }); + + expect(diffDeposit(event, tx)).toBe('currency_mismatch'); + }); + + it('flags currency_mismatch case-insensitively as a genuine mismatch, not a false positive', () => { + const event = depositEvent({ currency: 'btc' }); + const tx = ledgerRow({ currency: 'BTC' }); + + expect(diffDeposit(event, tx)).toBeNull(); + }); + + it('flags amount_mismatch when the currencies agree but the amounts differ', () => { + const event = depositEvent({ amount: '2' }); + const tx = ledgerRow({ amount: '1' }); + + expect(diffDeposit(event, tx)).toBe('amount_mismatch'); + }); + + it('never routes an amount compare through float - "1" and "1.00" reconcile exactly', () => { + const event = depositEvent({ amount: '1.00' }); + const tx = ledgerRow({ amount: '1' }); + + expect(diffDeposit(event, tx)).toBeNull(); + }); + + it('reconciles (returns null) when currency and amount both match', () => { + const event = depositEvent(); + const tx = ledgerRow(); + + expect(diffDeposit(event, tx)).toBeNull(); + }); +}); diff --git a/packages/core/src/wallet/__tests__/reconciliation.service.int.test.ts b/packages/core/src/wallet/__tests__/reconciliation.service.int.test.ts new file mode 100644 index 00000000..c264a7a9 --- /dev/null +++ b/packages/core/src/wallet/__tests__/reconciliation.service.int.test.ts @@ -0,0 +1,468 @@ +import { describe, it, expect, vi, beforeAll, afterAll, beforeEach } from 'vitest'; +import { randomUUID } from 'node:crypto'; +import { eq, sql } from 'drizzle-orm'; +import { findOneOrThrow } from '@openora/core/server'; +import type { + IdentityReader, + PaymentAdapter, + PaymentWebhookEvent, + PlatformConfig, +} from '@openora/core/contracts'; +import { createTestDb, type TestDb } from '@openora/core/testing'; +import { + mock, + makeEventBus, + makeIdentityReader, + makeAuditWriter, + makePaymentProviderRegistry, +} from '../../testing/mock.js'; +import { migrate } from '../migrate.js'; +import { + wallet, + walletBalance, + walletTransaction, + walletCustodySweep, + walletJobRun, + walletReconciliationFinding, +} from '../schema/index.js'; +import { WalletService } from '../service/wallet.service.js'; +import { + ReconciliationService, + ReconciliationCreditMismatchError, + ReconciliationFindingNotFoundError, +} from '../service/reconciliation.service.js'; + +let db: TestDb; + +beforeAll(async () => { + db = await createTestDb([migrate]); +}); + +afterAll(async () => { + await db.drop(); +}); + +beforeEach(async () => { + await db.drizzle.db.execute( + sql`TRUNCATE ${walletTransaction}, ${walletCustodySweep}, ${walletJobRun}, ${walletReconciliationFinding}, ${wallet} RESTART IDENTITY CASCADE`, + ); +}); + +const RECONCILIATION_CONFIG: NonNullable['reconciliation'] = { + cron: '0 * * * *', + lookbackHours: 24, + stuckAfterMinutes: 60, + staleRunAfterMinutes: 30, + alertThreshold: 10, +}; + +function makeServices( + payment: PaymentAdapter, + overrides: { + platformConfig?: Partial; + identityReader?: ReturnType; + } = {}, +) { + const paymentProviders = makePaymentProviderRegistry({ adapter: payment }); + const audit = makeAuditWriter(); + const events = makeEventBus(); + const platformConfig = mock({ + wallet: { reconciliation: RECONCILIATION_CONFIG, ...overrides.platformConfig }, + }); + const wallet = new WalletService({ + drizzle: db.drizzle, + events: makeEventBus(), + payment, + paymentProviders, + audit, + identityReader: overrides.identityReader ?? makeIdentityReader(), + platformConfig, + }); + const reconciliation = new ReconciliationService({ + drizzle: db.drizzle, + events, + wallet, + paymentProviders, + audit, + platformConfig, + }); + return { wallet, reconciliation, audit, events }; +} + +async function seedWallet(currency = 'BTC', balance = '0') { + const row = findOneOrThrow( + await db.drizzle.db.insert(wallet).values({ userId: randomUUID(), currency }).returning(), + new Error('seedWallet: query returned no row'), + ); + await db.drizzle.db.insert(walletBalance).values({ walletId: row.id, currency, amount: balance }); + return row; +} + +async function balanceOf(walletId: string) { + const [row] = await db.drizzle.db + .select() + .from(walletBalance) + .where(eq(walletBalance.walletId, walletId)); + return row?.amount ?? '0'; +} + +async function findingRows() { + return db.drizzle.db.select().from(walletReconciliationFinding); +} + +function listTransactionsReturning(events: PaymentWebhookEvent[]): PaymentAdapter { + return mock({ listTransactions: vi.fn(async () => events) }); +} + +describe('ReconciliationService.runCycle - internal transfer exclusion', () => { + it('produces zero findings for a sweep that appears in the vendor transaction list', async () => { + const w = await seedWallet(); + const sweepExternalId = randomUUID(); + await db.drizzle.db.insert(walletCustodySweep).values({ + userId: w.userId, + providerName: 'default', + currency: 'BTC', + network: 'BTC', + amount: '5', + estimatedFee: '0.0001', + externalId: sweepExternalId, + status: 'completed', + }); + // The vendor's own ledger reports the sweep back as a transaction with no ledger + // row on our side - exactly the shape that would file a false missing_deposit if + // the exclusion were skipped. + const payment = listTransactionsReturning([ + { + kind: 'deposit', + address: 'pool-address', + amount: '5', + currency: 'BTC', + txHash: '0xsweep', + externalId: sweepExternalId, + }, + ]); + const { reconciliation } = makeServices(payment); + + const result = await reconciliation.runCycle(); + + expect(result).not.toBeNull(); + expect(await findingRows()).toHaveLength(0); + }); +}); + +describe('ReconciliationService.runCycle - missing deposit', () => { + it('produces a finding and zero balance change', async () => { + const w = await seedWallet('BTC', '2'); + const externalId = randomUUID(); + const payment = listTransactionsReturning([ + { + kind: 'deposit', + address: 'bc1qmissing', + amount: '1', + currency: 'BTC', + txHash: '0xmissing', + externalId, + }, + ]); + const { reconciliation } = makeServices(payment); + + await reconciliation.runCycle(); + + const findings = await findingRows(); + expect(findings).toHaveLength(1); + expect(findings[0]).toMatchObject({ kind: 'missing_deposit', status: 'open', externalId }); + expect(await balanceOf(w.id)).toBe('2.000000000000000000'); + expect(await db.drizzle.db.select().from(walletTransaction)).toHaveLength(0); + }); + + it('does not duplicate the finding when the same window is reconciled again', async () => { + const externalId = randomUUID(); + const payment = listTransactionsReturning([ + { + kind: 'deposit', + address: 'bc1qmissing', + amount: '1', + currency: 'BTC', + txHash: '0xmissing', + externalId, + }, + ]); + const { reconciliation } = makeServices(payment); + + await reconciliation.runCycle(); + await reconciliation.runCycle(); + + const findings = await findingRows(); + expect(findings).toHaveLength(1); + }); +}); + +describe('ReconciliationService.runCycle - amount and currency mismatch', () => { + it('flags amount_mismatch when the ledger amount differs from the vendor report', async () => { + const w = await seedWallet(); + const externalId = randomUUID(); + const tx = findOneOrThrow( + await db.drizzle.db + .insert(walletTransaction) + .values({ + walletId: w.id, + type: 'deposit', + amount: '1', + currency: 'BTC', + status: 'completed', + rail: 'crypto', + providerRefId: externalId, + }) + .returning(), + new Error('seed: insert returned no row'), + ); + + const payment = listTransactionsReturning([ + { + kind: 'deposit', + address: 'bc1qamount', + amount: '2', + currency: 'BTC', + txHash: '0xamount', + externalId, + }, + ]); + const { reconciliation } = makeServices(payment); + + await reconciliation.runCycle(); + + const findings = await findingRows(); + expect(findings).toHaveLength(1); + expect(findings[0]).toMatchObject({ kind: 'amount_mismatch', transactionId: tx.id }); + }); +}); + +describe('ReconciliationService.runCycle - stuck withdrawals', () => { + async function seedStuckWithdrawal(providerRefId: string) { + const w = await seedWallet('BTC', '5'); + const oldDate = new Date(Date.now() - 2 * 60 * 60 * 1000); + const tx = findOneOrThrow( + await db.drizzle.db + .insert(walletTransaction) + .values({ + walletId: w.id, + type: 'withdrawal', + amount: '1', + currency: 'BTC', + status: 'processing', + rail: 'crypto', + providerName: 'default', + providerRefId, + createdAt: oldDate, + }) + .returning(), + new Error('seedStuckWithdrawal: query returned no row'), + ); + return { w, tx }; + } + + it('finalizes through reconcileWithdrawalStatus and refunds exactly once when it comes back failed', async () => { + const providerRefId = randomUUID(); + const { w } = await seedStuckWithdrawal(providerRefId); + const payment = mock({ + listTransactions: vi.fn(async () => []), + getWithdrawalStatus: vi.fn(async () => ({ status: 'failed' as const })), + }); + const { reconciliation } = makeServices(payment); + + await reconciliation.runCycle(); + + expect(await balanceOf(w.id)).toBe('6.000000000000000000'); + const [tx] = await db.drizzle.db + .select() + .from(walletTransaction) + .where(eq(walletTransaction.walletId, w.id)); + expect(tx?.status).toBe('failed'); + + // Re-running must not refund a second time: the withdrawal is no longer + // `processing`, so it no longer matches the stuck-withdrawal query at all. + await reconciliation.runCycle(); + expect(await balanceOf(w.id)).toBe('6.000000000000000000'); + }); + + it('files an unknown_at_provider finding when the vendor has no record', async () => { + const providerRefId = randomUUID(); + await seedStuckWithdrawal(providerRefId); + const payment = mock({ + listTransactions: vi.fn(async () => []), + getWithdrawalStatus: vi.fn(async () => null), + }); + const { reconciliation } = makeServices(payment); + + await reconciliation.runCycle(); + + const findings = await findingRows(); + expect(findings).toHaveLength(1); + expect(findings[0]).toMatchObject({ kind: 'unknown_at_provider', externalId: providerRefId }); + }); +}); + +describe('ReconciliationService.runCycle - claim concurrency', () => { + it('the second of two concurrent cycles returns immediately', async () => { + // A near-instant cycle can fully finish (clearing the claim) before the sibling's + // own claim attempt is even issued, which would let both legitimately succeed as + // two SEPARATE runs rather than actually racing. Hold the first cycle's claim open + // long enough for the second's insert to land while finishedAt is still NULL - that + // is the real race this test exists to prove. + const payment = mock({ + listTransactions: vi.fn(() => new Promise((resolve) => setTimeout(() => resolve([]), 100))), + }); + const { reconciliation } = makeServices(payment); + + const [first, second] = await Promise.all([ + reconciliation.runCycle(), + reconciliation.runCycle(), + ]); + + const results = [first, second]; + expect(results.filter((r) => r !== null)).toHaveLength(1); + expect(results.filter((r) => r === null)).toHaveLength(1); + }); +}); + +describe('ReconciliationService.runCycle - audit', () => { + it('writes exactly one audit entry per run, with no address or tx hash in its payload', async () => { + const externalId = randomUUID(); + const payment = listTransactionsReturning([ + { + kind: 'deposit', + address: 'bc1qaudited', + amount: '1', + currency: 'BTC', + txHash: '0xaudited', + externalId, + }, + ]); + const { reconciliation, audit } = makeServices(payment); + + await reconciliation.runCycle(); + + expect(audit.record).toHaveBeenCalledTimes(1); + const [entry] = (audit.record as ReturnType).mock.calls[0] as [ + Record, + ]; + expect(entry.action).toBe('wallet.reconciliation_run.completed'); + const payload = JSON.stringify(entry); + expect(payload).not.toContain('bc1qaudited'); + expect(payload).not.toContain('0xaudited'); + }); +}); + +describe('ReconciliationService.resolveFinding', () => { + async function seedFinding(kind: 'missing_deposit' = 'missing_deposit') { + const runId = randomUUID(); + await db.drizzle.db.insert(walletJobRun).values({ jobName: 'wallet-reconciliation', runId }); + const [row] = await db.drizzle.db + .insert(walletReconciliationFinding) + .values({ + runId, + providerName: 'default', + kind, + currency: 'BTC', + amount: '1', + externalId: randomUUID(), + }) + .returning(); + return findOneOrThrow([row], new Error('seedFinding: insert returned no row')); + } + + it('rejects a credited resolution whose transaction amount does not match the finding', async () => { + const finding = await seedFinding(); + const w = await seedWallet('BTC', '0'); + const payment = listTransactionsReturning([]); + const { wallet: walletSvc, reconciliation } = makeServices(payment, { + identityReader: mock({ + ...makeIdentityReader(), + getPlayerIdByUserId: vi.fn().mockResolvedValue(randomUUID()), + }), + }); + const creditTx = await walletSvc.manualAdjust({ + adminId: randomUUID(), + userId: w.userId, + direction: 'credit', + amount: '999', + currency: 'BTC', + reason: 'test', + idempotencyKey: randomUUID(), + ip: null, + userAgent: null, + }); + + await expect( + reconciliation.resolveFinding(randomUUID(), finding.id, { + outcome: 'credited', + transactionId: creditTx.transactionId, + }), + ).rejects.toBeInstanceOf(ReconciliationCreditMismatchError); + }); + + it('accepts a credited resolution whose manual-credit transaction matches exactly', async () => { + const finding = await seedFinding(); + const w = await seedWallet('BTC', '0'); + const payment = listTransactionsReturning([]); + const { wallet: walletSvc, reconciliation } = makeServices(payment, { + identityReader: mock({ + ...makeIdentityReader(), + getPlayerIdByUserId: vi.fn().mockResolvedValue(randomUUID()), + }), + }); + const creditTx = await walletSvc.manualAdjust({ + adminId: randomUUID(), + userId: w.userId, + direction: 'credit', + amount: '1', + currency: 'BTC', + reason: 'reconciliation credit', + idempotencyKey: randomUUID(), + ip: null, + userAgent: null, + }); + + const resolved = await reconciliation.resolveFinding(randomUUID(), finding.id, { + outcome: 'credited', + transactionId: creditTx.transactionId, + }); + + expect(resolved.status).toBe('resolved'); + expect(resolved.transactionId).toBe(creditTx.transactionId); + }); + + it('a double-resolve is a no-op: second call does not write a second audit entry', async () => { + const finding = await seedFinding(); + const payment = listTransactionsReturning([]); + const { reconciliation, audit } = makeServices(payment); + const adminId = randomUUID(); + + await reconciliation.resolveFinding(adminId, finding.id, { + outcome: 'dismissed', + note: 'confirmed non-issue', + }); + expect(audit.record).not.toHaveBeenCalled(); + expect(audit.recordInTransaction).toHaveBeenCalledTimes(1); + + const second = await reconciliation.resolveFinding(adminId, finding.id, { + outcome: 'dismissed', + note: 'a different note this time', + }); + + expect(audit.recordInTransaction).toHaveBeenCalledTimes(1); + expect(second.resolutionNote).toBe('confirmed non-issue'); + }); + + it('404s resolving a finding that does not exist', async () => { + const payment = listTransactionsReturning([]); + const { reconciliation } = makeServices(payment); + + await expect( + reconciliation.resolveFinding(randomUUID(), randomUUID(), { + outcome: 'dismissed', + note: 'n/a', + }), + ).rejects.toBeInstanceOf(ReconciliationFindingNotFoundError); + }); +}); diff --git a/packages/core/src/wallet/__tests__/wallet-asset.router.int.test.ts b/packages/core/src/wallet/__tests__/wallet-asset.router.int.test.ts index 34d8d8f3..6897440e 100644 --- a/packages/core/src/wallet/__tests__/wallet-asset.router.int.test.ts +++ b/packages/core/src/wallet/__tests__/wallet-asset.router.int.test.ts @@ -3,7 +3,7 @@ import { findOneOrThrow } from '@openora/core/server'; import { randomUUID } from 'node:crypto'; import { call, ORPCError } from '@orpc/server'; import type { AdminGuard } from '@openora/core/server'; -import type { PaymentAdapter } from '@openora/core/contracts'; +import { queue, type PaymentAdapter } from '@openora/core/contracts'; import { createTestDb, type TestDb } from '@openora/core/testing'; import { migrate as migrateProfile } from '@openora/core/pam/migrate/profile'; import { @@ -13,12 +13,16 @@ import { makeAuditWriter, makeAdminGuard, makeIdentityReader, + makeJobQueue, makePaymentProviderRegistry, } from '../../testing/mock.js'; import { migrate } from '../migrate.js'; import { wallet, walletBalance, walletTransaction, walletAsset } from '../schema/index.js'; import { createWalletRouter } from '../router/index.js'; import { WalletService } from '../service/wallet.service.js'; +import type { ReconciliationService } from '../service/reconciliation.service.js'; + +const RECONCILIATION_QUEUE = queue('wallet-reconciliation'); const CTX = testContext(); const CALLER_ID = '9a2f7c11-0000-4000-8000-0000000000cc'; @@ -84,6 +88,9 @@ function routerWith( adminGuard: guard, audit, paymentProviders, + reconciliation: mock({}), + jobQueue: makeJobQueue(), + reconciliationQueue: RECONCILIATION_QUEUE, }); return { router, audit, service }; } diff --git a/packages/core/src/wallet/__tests__/wallet-auto-rule.router.int.test.ts b/packages/core/src/wallet/__tests__/wallet-auto-rule.router.int.test.ts index d8e24d13..29b92fba 100644 --- a/packages/core/src/wallet/__tests__/wallet-auto-rule.router.int.test.ts +++ b/packages/core/src/wallet/__tests__/wallet-auto-rule.router.int.test.ts @@ -2,7 +2,7 @@ import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; import { eq } from 'drizzle-orm'; import { call, ORPCError } from '@orpc/server'; import type { AdminGuard } from '@openora/core/server'; -import type { PaymentAdapter } from '@openora/core/contracts'; +import { queue, type PaymentAdapter } from '@openora/core/contracts'; import { createTestDb, type TestDb } from '@openora/core/testing'; import { mock, @@ -11,12 +11,16 @@ import { makeAuditWriter, makeAdminGuard, makeIdentityReader, + makeJobQueue, makePaymentProviderRegistry, } from '../../testing/mock.js'; import { migrate } from '../migrate.js'; import { autoWithdrawalRule } from '../schema/index.js'; import { createWalletRouter } from '../router/index.js'; import { WalletService } from '../service/wallet.service.js'; +import type { ReconciliationService } from '../service/reconciliation.service.js'; + +const RECONCILIATION_QUEUE = queue('wallet-reconciliation'); const CTX = testContext(); const USER_ID = '63d3c264-3bf4-4d08-9b92-ea3eaf40a440'; @@ -55,7 +59,15 @@ function routerWith(adminGuard: AdminGuard) { audit, identityReader: makeIdentityReader(), }); - const router = createWalletRouter({ wallet: service, adminGuard, audit, paymentProviders }); + const router = createWalletRouter({ + wallet: service, + adminGuard, + audit, + paymentProviders, + reconciliation: mock({}), + jobQueue: makeJobQueue(), + reconciliationQueue: RECONCILIATION_QUEUE, + }); return { router, audit }; } diff --git a/packages/core/src/wallet/__tests__/wallet-auto-withdrawal-config.router.int.test.ts b/packages/core/src/wallet/__tests__/wallet-auto-withdrawal-config.router.int.test.ts index ee2c022a..71860eea 100644 --- a/packages/core/src/wallet/__tests__/wallet-auto-withdrawal-config.router.int.test.ts +++ b/packages/core/src/wallet/__tests__/wallet-auto-withdrawal-config.router.int.test.ts @@ -10,6 +10,7 @@ import type { PlatformConfig, PlayerTags, } from '@openora/core/contracts'; +import { queue } from '@openora/core/contracts'; import { createTestDb, type TestDb } from '@openora/core/testing'; import { migrate as migrateProfile } from '@openora/core/pam/migrate/profile'; import { @@ -19,6 +20,7 @@ import { makeAuditWriter, makeAdminGuard, makeIdentityReader, + makeJobQueue, makePaymentProviderRegistry, NO_CLIENT_META, } from '../../testing/mock.js'; @@ -31,6 +33,9 @@ import { } from '../schema/index.js'; import { createWalletRouter } from '../router/index.js'; import { WalletService } from '../service/wallet.service.js'; +import type { ReconciliationService } from '../service/reconciliation.service.js'; + +const RECONCILIATION_QUEUE = queue('wallet-reconciliation'); const CTX = testContext(); const CALLER_ID = '9a2f7c11-0000-4000-8000-0000000000bb'; @@ -106,7 +111,15 @@ function routerWith(adminGuard: AdminGuard, platformConfig?: Partial(platformConfig) : undefined, riskTags, }); - const router = createWalletRouter({ wallet: service, adminGuard, audit, paymentProviders }); + const router = createWalletRouter({ + wallet: service, + adminGuard, + audit, + paymentProviders, + reconciliation: mock({}), + jobQueue: makeJobQueue(), + reconciliationQueue: RECONCILIATION_QUEUE, + }); return { router, audit, service }; } diff --git a/packages/core/src/wallet/__tests__/wallet-custody-stubs.router.test.ts b/packages/core/src/wallet/__tests__/wallet-custody-routes.router.test.ts similarity index 52% rename from packages/core/src/wallet/__tests__/wallet-custody-stubs.router.test.ts rename to packages/core/src/wallet/__tests__/wallet-custody-routes.router.test.ts index e489eb92..fbc16a88 100644 --- a/packages/core/src/wallet/__tests__/wallet-custody-stubs.router.test.ts +++ b/packages/core/src/wallet/__tests__/wallet-custody-routes.router.test.ts @@ -1,7 +1,7 @@ import { describe, it, expect, vi } from 'vitest'; import { call, ORPCError } from '@orpc/server'; import type { AdminGuard } from '@openora/core/server'; -import type { JobQueueAdapter, PaymentAdapter } from '@openora/core/contracts'; +import { queue, type JobQueueAdapter, type PaymentAdapter } from '@openora/core/contracts'; import { mock, makeDrizzle, @@ -10,18 +10,24 @@ import { makeAuditWriter, makeAdminGuard, makeIdentityReader, + makeJobQueue, makePaymentProviderRegistry, } from '../../testing/mock.js'; import { createWalletRouter } from '../router/index.js'; import { WalletService } from '../service/wallet.service.js'; +import type { ReconciliationService } from '../service/reconciliation.service.js'; const CTX = testContext(); const CALLER_ID = '9a2f7c11-0000-4000-8000-0000000000dd'; - -// The reconciliation routes below only ever run adminGuard.assert() then throw -// NOT_IMPLEMENTED - no DB row is ever read or written, so a mocked DrizzleService is -// enough (no real Postgres). The custody sweep route now enqueues instead of throwing. -function routerWith(guard: AdminGuard, jobQueue?: JobQueueAdapter) { +const RECONCILIATION_QUEUE = queue('wallet-reconciliation'); + +// Neither route reads or writes a row directly - they authorize, then delegate to a +// service or enqueue a job - so a mocked DrizzleService is enough (no real Postgres). +// The collaborators are per-test doubles so each case can assert what it delegated. +function routerWith( + guard: AdminGuard, + overrides: { reconciliation?: ReconciliationService; jobQueue?: JobQueueAdapter } = {}, +) { const paymentProviders = makePaymentProviderRegistry(); const service = new WalletService({ drizzle: makeDrizzle(), @@ -36,7 +42,9 @@ function routerWith(guard: AdminGuard, jobQueue?: JobQueueAdapter) { adminGuard: guard, audit: makeAuditWriter(), paymentProviders, - jobQueue, + reconciliation: overrides.reconciliation ?? mock({}), + jobQueue: overrides.jobQueue ?? makeJobQueue(), + reconciliationQueue: RECONCILIATION_QUEUE, }); } @@ -45,12 +53,13 @@ const authorizedGuard = () => makeAdminGuard({ caller: { userId: CALLER_ID, role const deniedGuard = (deny: readonly string[]) => makeAdminGuard({ deny, caller: { userId: CALLER_ID, role: 'support' } }); -describe('wallet custody/reconciliation stub routes', () => { +describe('wallet custody and reconciliation routes', () => { describe('POST /wallet/custody/sweep/run', () => { it('enqueues a sweep cycle and returns its runId for an authorized caller', async () => { const enqueue = vi.fn().mockResolvedValue({ id: 'job-1' }); - const jobQueue = mock({ enqueue }); - const router = routerWith(authorizedGuard(), jobQueue); + const router = routerWith(authorizedGuard(), { + jobQueue: mock({ enqueue }), + }); const result = await call(router.custody.sweep.run, {}, { context: CTX }); @@ -77,12 +86,16 @@ describe('wallet custody/reconciliation stub routes', () => { describe('GET /wallet/reconciliation', () => { const input = { page: 1, limit: 20 }; - it('501s for an authorized caller - NEVER a success shape, empty or otherwise', async () => { - const router = routerWith(authorizedGuard()); + it('delegates to the reconciliation service for an authorized caller', async () => { + const listFindings = vi.fn(async () => ({ items: [], total: 0, page: 1, limit: 20 })); + const router = routerWith(authorizedGuard(), { + reconciliation: mock({ listFindings }), + }); - await expect(call(router.reconciliation.list, input, { context: CTX })).rejects.toMatchObject( - { code: 'NOT_IMPLEMENTED' }, - ); + const result = await call(router.reconciliation.list, input, { context: CTX }); + + expect(result).toEqual({ items: [], total: 0, page: 1, limit: 20 }); + expect(listFindings).toHaveBeenCalledWith(input); }); it('403s for a caller missing wallet-reconciliation:view', async () => { @@ -99,13 +112,42 @@ describe('wallet custody/reconciliation stub routes', () => { id: '63d3c264-3bf4-4d08-9b92-ea3eaf40a440', resolution: { outcome: 'dismissed' as const, note: 'confirmed non-issue' }, }; + const resolved = { + id: input.id, + runId: '00000000-0000-0000-0000-000000000000', + providerName: 'default', + kind: 'missing_deposit' as const, + currency: 'BTC', + network: null, + amount: '1', + address: null, + tag: null, + txHash: null, + externalId: 'vendor-ext-1', + transactionId: null, + detail: null, + status: 'resolved' as const, + resolvedBy: CALLER_ID, + resolvedAt: new Date().toISOString(), + resolutionNote: input.resolution.note, + createdAt: new Date().toISOString(), + }; - it('501s for an authorized caller', async () => { - const router = routerWith(authorizedGuard()); + it('delegates to the reconciliation service for an authorized caller', async () => { + const resolveFinding = vi.fn(async () => resolved); + const router = routerWith(authorizedGuard(), { + reconciliation: mock({ resolveFinding }), + }); - await expect( - call(router.reconciliation.resolve, input, { context: CTX }), - ).rejects.toMatchObject({ code: 'NOT_IMPLEMENTED' }); + const result = await call(router.reconciliation.resolve, input, { context: CTX }); + + expect(result).toEqual(resolved); + expect(resolveFinding).toHaveBeenCalledWith( + CALLER_ID, + input.id, + input.resolution, + expect.objectContaining({}), + ); }); it('403s for a caller missing wallet-reconciliation:resolve', async () => { @@ -118,12 +160,15 @@ describe('wallet custody/reconciliation stub routes', () => { }); describe('POST /wallet/reconciliation/run', () => { - it('501s for an authorized caller', async () => { - const router = routerWith(authorizedGuard()); + it('enqueues the job and returns a runId for an authorized caller, never running inline', async () => { + const jobQueue = makeJobQueue(); + const router = routerWith(authorizedGuard(), { jobQueue }); - await expect(call(router.reconciliation.run, {}, { context: CTX })).rejects.toMatchObject({ - code: 'NOT_IMPLEMENTED', - }); + const result = await call(router.reconciliation.run, {}, { context: CTX }); + + expect(result.runId).toEqual(expect.any(String)); + expect(jobQueue.enqueue).toHaveBeenCalledTimes(1); + expect(jobQueue.enqueue).toHaveBeenCalledWith(RECONCILIATION_QUEUE, { runId: result.runId }); }); it('403s for a caller missing wallet-reconciliation:run', async () => { diff --git a/packages/core/src/wallet/__tests__/wallet-webhook.router.int.test.ts b/packages/core/src/wallet/__tests__/wallet-webhook.router.int.test.ts index 793275e4..fc7da211 100644 --- a/packages/core/src/wallet/__tests__/wallet-webhook.router.int.test.ts +++ b/packages/core/src/wallet/__tests__/wallet-webhook.router.int.test.ts @@ -4,13 +4,14 @@ import { randomUUID } from 'node:crypto'; import { eq } from 'drizzle-orm'; import { call, ORPCError } from '@orpc/server'; import type { AdminGuard } from '@openora/core/server'; -import type { - PaymentAdapter, - PaymentProviderRegistry, - PaymentWebhookEvent, - PaymentWebhookVerifier, - RateLimiterAdapter, - RateLimitKey, +import { + queue, + type PaymentAdapter, + type PaymentProviderRegistry, + type PaymentWebhookEvent, + type PaymentWebhookVerifier, + type RateLimiterAdapter, + type RateLimitKey, } from '@openora/core/contracts'; import { createTestDb, type TestDb } from '@openora/core/testing'; import { migrate as migrateProfile } from '@openora/core/pam/migrate/profile'; @@ -20,15 +21,24 @@ import { makeIdentityReader, testContext, makeAuditWriter, + makeJobQueue, makePaymentProviderRegistry, } from '../../testing/mock.js'; import { migrate } from '../migrate.js'; -import { wallet, walletBalance, walletTransaction, walletDepositAddress } from '../schema/index.js'; +import { + wallet, + walletBalance, + walletTransaction, + walletDepositAddress, + walletReconciliationFinding, +} from '../schema/index.js'; import { createWalletRouter } from '../router/index.js'; import { WalletService } from '../service/wallet.service.js'; +import type { ReconciliationService } from '../service/reconciliation.service.js'; const USER_ID = '63d3c264-3bf4-4d08-9b92-ea3eaf40a440'; const DEPOSIT_ADDRESS = 'bc1qxyz'; +const RECONCILIATION_QUEUE = queue('wallet-reconciliation'); let db: TestDb; @@ -41,6 +51,7 @@ afterAll(async () => { }); beforeEach(async () => { + await db.drizzle.db.delete(walletReconciliationFinding); await db.drizzle.db.delete(walletTransaction); await db.drizzle.db.delete(walletDepositAddress); await db.drizzle.db.delete(walletBalance); @@ -77,10 +88,20 @@ function routerWithProviders( adminGuard: mock({ assert: vi.fn() }), audit: makeAuditWriter(), paymentProviders, + reconciliation: mock({}), + jobQueue: makeJobQueue(), + reconciliationQueue: RECONCILIATION_QUEUE, limiter, }); } +async function findingsFor(externalId: string) { + return db.drizzle.db + .select() + .from(walletReconciliationFinding) + .where(eq(walletReconciliationFinding.externalId, externalId)); +} + // A registry with two DISTINCTLY-behaving named providers, to test that a webhook is // never verified with one vendor's key and parsed with another's format. function twoProviderRegistry( @@ -260,7 +281,7 @@ describe('wallet webhook route (M2M, no admin session)', () => { expect(await ledgerFor(w.id)).toHaveLength(0); }); - it('returns ok and credits nothing when the deposit address is unknown', async () => { + it('returns ok, credits nobody, and files an unattributed_deposit finding when the deposit address is unknown', async () => { const w = await seedWallet(); const router = routerWith(paymentParsing(depositEvent), verifierReturning(true)); @@ -268,6 +289,16 @@ describe('wallet webhook route (M2M, no admin session)', () => { expect(result).toEqual({ ok: true }); expect(await ledgerFor(w.id)).toHaveLength(0); + const findings = await findingsFor(depositEvent.externalId); + expect(findings).toHaveLength(1); + expect(findings[0]).toMatchObject({ + kind: 'unattributed_deposit', + providerName: 'default', + currency: 'BTC', + amount: '0.500000000000000000', + address: DEPOSIT_ADDRESS, + status: 'open', + }); }); }); diff --git a/packages/core/src/wallet/__tests__/wallet.router.int.test.ts b/packages/core/src/wallet/__tests__/wallet.router.int.test.ts index e3ad8f8c..f2c156b5 100644 --- a/packages/core/src/wallet/__tests__/wallet.router.int.test.ts +++ b/packages/core/src/wallet/__tests__/wallet.router.int.test.ts @@ -3,7 +3,7 @@ import { findOneOrThrow } from '@openora/core/server'; import { randomUUID } from 'node:crypto'; import { call, ORPCError } from '@orpc/server'; import type { AdminGuard } from '@openora/core/server'; -import type { PaymentAdapter } from '@openora/core/contracts'; +import { queue, type PaymentAdapter } from '@openora/core/contracts'; import { createTestDb, type TestDb } from '@openora/core/testing'; import { mock, @@ -12,12 +12,16 @@ import { makeAuditWriter, makeAdminGuard, makeIdentityReader, + makeJobQueue, makePaymentProviderRegistry, } from '../../testing/mock.js'; import { migrate } from '../migrate.js'; import { wallet, walletTransaction } from '../schema/index.js'; import { createWalletRouter } from '../router/index.js'; import { WalletService } from '../service/wallet.service.js'; +import type { ReconciliationService } from '../service/reconciliation.service.js'; + +const RECONCILIATION_QUEUE = queue('wallet-reconciliation'); const CTX = testContext(); const USER_ID = '63d3c264-3bf4-4d08-9b92-ea3eaf40a440'; @@ -54,6 +58,9 @@ function routerWith(adminGuard: AdminGuard) { adminGuard, audit: makeAuditWriter(), paymentProviders: makePaymentProviderRegistry(), + reconciliation: mock({}), + jobQueue: makeJobQueue(), + reconciliationQueue: RECONCILIATION_QUEUE, }); } diff --git a/packages/core/src/wallet/__tests__/wallet.service.int.test.ts b/packages/core/src/wallet/__tests__/wallet.service.int.test.ts index eacf1220..3f7500e2 100644 --- a/packages/core/src/wallet/__tests__/wallet.service.int.test.ts +++ b/packages/core/src/wallet/__tests__/wallet.service.int.test.ts @@ -26,6 +26,7 @@ import { walletTransaction, walletDepositAddress, walletAsset, + walletReconciliationFinding, } from '../schema/index.js'; import { WalletService, @@ -35,7 +36,6 @@ import { WithdrawalNotPendingError, InsufficientBalanceError, AmbiguousDepositAddressError, - CurrencyMismatchError, IdempotencyKeyReuseError, DepositAddressUnsupportedError, DestinationAddressRequiredError, @@ -170,7 +170,7 @@ afterAll(async () => { beforeEach(async () => { await db.drizzle.db.execute( - sql`TRUNCATE ${walletTransaction}, ${walletDepositAddress}, ${walletAsset}, ${wallet} RESTART IDENTITY CASCADE`, + sql`TRUNCATE ${walletTransaction}, ${walletDepositAddress}, ${walletAsset}, ${walletReconciliationFinding}, ${wallet} RESTART IDENTITY CASCADE`, ); }); @@ -1394,19 +1394,33 @@ describe('WalletService.creditDepositByAddress (real PG)', () => { expect(emittedTopics(events)).toEqual(['wallet.deposit.completed']); }); - it('logs and no-ops when the address is unknown', async () => { + it('credits nobody and files an unattributed_deposit finding when the address is unknown', async () => { const { svc, events } = makeService(); + const externalId = randomUUID(); - await svc.creditDepositByAddress({ - kind: 'deposit', - address: 'bc1qunknown', - amount: '1', - currency: 'BTC', - externalId: randomUUID(), - txHash: '0xunknown', - }); + await svc.creditDepositByAddress( + { + kind: 'deposit', + address: 'bc1qunknown', + amount: '1', + currency: 'BTC', + externalId, + txHash: '0xunknown', + }, + 'default', + ); expect(events.emit).not.toHaveBeenCalled(); + const findings = await db.drizzle.db + .select() + .from(walletReconciliationFinding) + .where(eq(walletReconciliationFinding.externalId, externalId)); + expect(findings).toHaveLength(1); + expect(findings[0]).toMatchObject({ + kind: 'unattributed_deposit', + providerName: 'default', + status: 'open', + }); }); it('is idempotent on a replayed externalId', async () => { @@ -1431,21 +1445,32 @@ describe('WalletService.creditDepositByAddress (real PG)', () => { expect(emittedTopics(events)).toEqual(['wallet.deposit.completed']); }); - it('throws CurrencyMismatchError when the event currency differs from the address', async () => { - const { svc } = makeService(); + it('credits nobody and files a currency_mismatch finding on a currency mismatch', async () => { + const { svc, events } = makeService(); const w = await seedWallet({ currency: 'BTC' }); await seedAddress(w.userId, 'bc1qmismatch'); + const externalId = randomUUID(); - await expect( - svc.creditDepositByAddress({ - kind: 'deposit', - address: 'bc1qmismatch', - amount: '1', - currency: 'ETH', - externalId: randomUUID(), - txHash: '0xmismatch', - }), - ).rejects.toBeInstanceOf(CurrencyMismatchError); + await svc.creditDepositByAddress({ + kind: 'deposit', + address: 'bc1qmismatch', + amount: '1', + currency: 'ETH', + externalId, + txHash: '0xmismatch', + }); + + expect(events.emit).not.toHaveBeenCalled(); + expect(await balanceOf(w.userId)).toBe(0); + const findings = await db.drizzle.db + .select() + .from(walletReconciliationFinding) + .where(eq(walletReconciliationFinding.externalId, externalId)); + expect(findings).toHaveLength(1); + // Its own taxonomy, not `unattributed_deposit`: the address resolved fine, the + // currency did not. The polled path already tags this case `currency_mismatch`, + // and an admin filtering findings by kind has to see both paths alike. + expect(findings[0]).toMatchObject({ kind: 'currency_mismatch', status: 'open' }); }); it('credits a token that shares the EVM address issued for another currency', async () => { diff --git a/packages/core/src/wallet/contract/index.ts b/packages/core/src/wallet/contract/index.ts index 9a4fdd9a..4743d99f 100644 --- a/packages/core/src/wallet/contract/index.ts +++ b/packages/core/src/wallet/contract/index.ts @@ -24,11 +24,6 @@ export { WalletReconciliationFindingStatusSchema, }; -// Stub routes not implemented in this PR - kept as one string so `grep 'Not implemented yet'` -// (matching the router's `notImplemented()` helper) finds every remaining stub; the count -// reaching zero is the definition of done for the custody/reconciliation feature set. -const NOT_IMPLEMENTED_YET = 'Not implemented yet'; - // Deposit/withdraw amounts must be strictly positive; balances/thresholds may be zero. const PositiveMoneyAmountSchema = MoneyAmountSchema.refine((v) => Number(v) > 0, { message: 'must be greater than zero', @@ -520,7 +515,11 @@ export const walletContract = { reconciliation: { list: oc - .route({ method: 'GET', path: '/wallet/reconciliation', summary: NOT_IMPLEMENTED_YET }) + .route({ + method: 'GET', + path: '/wallet/reconciliation', + summary: 'List reconciliation findings', + }) .input(ListReconciliationFindingsInputSchema) .output(paginated(WalletReconciliationFindingSchema)), @@ -528,7 +527,7 @@ export const walletContract = { .route({ method: 'POST', path: '/wallet/reconciliation/{id}/resolve', - summary: NOT_IMPLEMENTED_YET, + summary: 'Resolve a reconciliation finding as credited or dismissed', }) .input(ResolveReconciliationFindingInputSchema) .output(WalletReconciliationFindingSchema), @@ -537,7 +536,7 @@ export const walletContract = { .route({ method: 'POST', path: '/wallet/reconciliation/run', - summary: NOT_IMPLEMENTED_YET, + summary: 'Enqueue a reconciliation run', }) .output(JobRunResultSchema), }, diff --git a/packages/core/src/wallet/plugin.ts b/packages/core/src/wallet/plugin.ts index 481b8615..a41ec91a 100644 --- a/packages/core/src/wallet/plugin.ts +++ b/packages/core/src/wallet/plugin.ts @@ -19,6 +19,8 @@ import { PLAY_ELIGIBILITY, AUDIT_WRITER, JOB_QUEUE, + UuidSchema, + queue, } from '@openora/core/contracts'; import { WalletService } from './service/wallet.service.js'; import { WalletCommandsService } from './service/wallet-commands.service.js'; @@ -33,6 +35,7 @@ import { DrizzleAdminWalletReporting } from './admin-reporting.js'; import { createWalletRouter } from './router/index.js'; import { MockPaymentAdapter } from './adapters/mock/mock-payment-adapter.js'; import { HmacPaymentWebhookVerifier } from './adapters/hmac-payment-webhook-verifier.js'; +import { ReconciliationService } from './service/reconciliation.service.js'; const logger = createLogger('wallet'); @@ -41,6 +44,9 @@ const logger = createLogger('wallet'); // (CustodySweepService.runCycle no-ops every tick until sweep policy is configured). const DEFAULT_SWEEP_CRON = '*/15 * * * *'; +const WALLET_RECONCILIATION_QUEUE = queue('wallet-reconciliation'); +const WalletReconciliationJobSchema = z.object({ runId: UuidSchema.optional() }); + export default { // NOT dependsOn 'tag': that would cycle (tag hard-depends on wallet's WALLET_READER). // wallet's use of tag's PLAYER_TAGS / TAG_EVALUATION_COMMANDS is optional and resolved @@ -49,6 +55,39 @@ export default { id: 'wallet', dependsOn: ['identity', 'audit'], register(ctx) { + // Set inside the router factory (container access); the job worker's handler + // closes over this ref, same shape as pam/tag's evalSvc - subscriptions/workers + // wire before router factories run, but are set before any real job arrives. + let reconciliationRef: ReconciliationService | null = null; + + // One memoized instance backs the cron worker and the router factory below - lazily + // constructed (subscriptions/workers wire before router factories run), matching + // pam/tag's and compliance's job-worker shape. + let sweepSvc: CustodySweepService | null = null; + const custodySweepService = (c: TypedContainer) => + (sweepSvc ??= new CustodySweepService({ + drizzle: c.get(DRIZZLE), + paymentProviders: c.get(PAYMENT_PROVIDERS), + audit: c.get(AUDIT_WRITER), + platformConfig: c.has(PLATFORM_CONFIG) ? c.get(PLATFORM_CONFIG) : undefined, + })); + + ctx.jobs.worker({ + queue: CUSTODY_SWEEP_QUEUE, + schema: CustodySweepJobPayloadSchema, + handler: async ({ payload }) => { + await sweepSvc?.runCycle(payload.runId); + }, + }); + + ctx.jobs.worker({ + queue: WALLET_RECONCILIATION_QUEUE, + schema: WalletReconciliationJobSchema, + handler: async ({ payload }) => { + await reconciliationRef?.runCycle(payload.runId); + }, + }); + ctx.provide(PAYMENT_ADAPTER, () => new MockPaymentAdapter()); ctx.provide(PAYMENT_WEBHOOK_VERIFIER, () => { const webhookSecret = z @@ -86,60 +125,66 @@ export default { // Operator-editable currency/network config, readable by a payment adapter without // importing wallet tables. Overlay-rebindable, but bound here so it always works. ctx.provide(WALLET_ASSET_CATALOG, (c) => new WalletAssetCatalogService(c.get(DRIZZLE))); - - // One memoized instance backs the cron worker and the router factory below - lazily - // constructed (subscriptions/workers wire before router factories run), matching - // pam/tag's and compliance's job-worker shape. - let sweepSvc: CustodySweepService | null = null; - const custodySweepService = (c: TypedContainer) => - (sweepSvc ??= new CustodySweepService({ + ctx.routers.add('wallet', (c) => { + const platformConfig = c.has(PLATFORM_CONFIG) ? c.get(PLATFORM_CONFIG) : undefined; + const walletService = new WalletService({ drizzle: c.get(DRIZZLE), + events: c.get(EVENT_BUS), + payment: c.get(PAYMENT_ADAPTER), paymentProviders: c.get(PAYMENT_PROVIDERS), + identityReader: c.get(IDENTITY_READER), + directory: c.get(ADMIN_USER_DIRECTORY), + platformConfig, + limiter: c.get(RATE_LIMITER), + riskTags: c.has(PLAYER_TAGS) ? c.get(PLAYER_TAGS) : undefined, + tagEvaluationCommands: c.has(TAG_EVALUATION_COMMANDS) + ? c.get(TAG_EVALUATION_COMMANDS) + : undefined, audit: c.get(AUDIT_WRITER), - platformConfig: c.has(PLATFORM_CONFIG) ? c.get(PLATFORM_CONFIG) : undefined, - })); + }); - ctx.jobs.worker({ - queue: CUSTODY_SWEEP_QUEUE, - schema: CustodySweepJobPayloadSchema, - handler: async ({ payload }) => { - await sweepSvc?.runCycle(payload.runId); - }, - }); + const reconciliation = new ReconciliationService({ + drizzle: c.get(DRIZZLE), + events: c.get(EVENT_BUS), + wallet: walletService, + paymentProviders: c.get(PAYMENT_PROVIDERS), + audit: c.get(AUDIT_WRITER), + platformConfig, + }); + reconciliationRef = reconciliation; - ctx.routers.add('wallet', (c) => { const jobQueue = c.get(JOB_QUEUE); + custodySweepService(c); // ensure sweepSvc is constructed for the job worker above - const sweepCron = - (c.has(PLATFORM_CONFIG) ? c.get(PLATFORM_CONFIG) : undefined)?.wallet?.sweep?.cron ?? - DEFAULT_SWEEP_CRON; + const sweepCron = platformConfig?.wallet?.sweep?.cron ?? DEFAULT_SWEEP_CRON; // Idempotent schedule (keyed by scheduleId). If platformConfig.wallet.sweep is // absent, this still registers the tick - the handler just no-ops every time. void jobQueue .schedule(CUSTODY_SWEEP_QUEUE, 'wallet-custody-sweep.cron', {}, { cron: sweepCron }) .catch((err) => logger.error({ err }, 'wallet-custody-sweep schedule failed')); + // Absent config means the cycle no-ops - see ReconciliationService.runCycle - but + // there is also nothing to schedule at all. + if (platformConfig?.wallet?.reconciliation) { + void jobQueue + .schedule( + WALLET_RECONCILIATION_QUEUE, + 'wallet-reconciliation.cron', + {}, + { cron: platformConfig.wallet.reconciliation.cron }, + ) + .catch((err) => logger.error({ err }, 'wallet-reconciliation schedule failed')); + } + return createWalletRouter({ - wallet: new WalletService({ - drizzle: c.get(DRIZZLE), - events: c.get(EVENT_BUS), - payment: c.get(PAYMENT_ADAPTER), - paymentProviders: c.get(PAYMENT_PROVIDERS), - identityReader: c.get(IDENTITY_READER), - directory: c.get(ADMIN_USER_DIRECTORY), - platformConfig: c.has(PLATFORM_CONFIG) ? c.get(PLATFORM_CONFIG) : undefined, - limiter: c.get(RATE_LIMITER), - riskTags: c.has(PLAYER_TAGS) ? c.get(PLAYER_TAGS) : undefined, - tagEvaluationCommands: c.has(TAG_EVALUATION_COMMANDS) - ? c.get(TAG_EVALUATION_COMMANDS) - : undefined, - audit: c.get(AUDIT_WRITER), - }), + wallet: walletService, adminGuard: c.get(ADMIN_GUARD), audit: c.get(AUDIT_WRITER), paymentProviders: c.get(PAYMENT_PROVIDERS), - limiter: c.get(RATE_LIMITER), + reconciliation, jobQueue, + reconciliationQueue: WALLET_RECONCILIATION_QUEUE, + limiter: c.get(RATE_LIMITER), }); }); }, diff --git a/packages/core/src/wallet/router/index.ts b/packages/core/src/wallet/router/index.ts index af76795a..bb4c950c 100644 --- a/packages/core/src/wallet/router/index.ts +++ b/packages/core/src/wallet/router/index.ts @@ -15,6 +15,7 @@ import { type JobQueueAdapter, type PaymentProviderRegistry, type PaymentWebhookEvent, + type QueueName, type RateLimiterAdapter, type RateLimitKey, } from '@openora/core/contracts'; @@ -26,7 +27,6 @@ import { WithdrawalNotFoundError, WithdrawalNotPendingError, InsufficientBalanceError, - CurrencyMismatchError, KycRequiredError, IdempotencyKeyReuseError, DepositAddressUnsupportedError, @@ -44,12 +44,12 @@ import { BelowMinimumWithdrawalError, PlayerNotFoundError, } from '../service/wallet.service.js'; - -// One helper so `grep notImplemented` finds every remaining stub; the count reaching -// zero is the definition of done for the custody sweep / reconciliation feature set. -const notImplemented = (): never => { - throw new ORPCError('NOT_IMPLEMENTED'); -}; +import { + ReconciliationService, + ReconciliationFindingNotFoundError, + ReconciliationCreditTransactionNotFoundError, + ReconciliationCreditMismatchError, +} from '../service/reconciliation.service.js'; // Unauthenticated route: it costs a signature verification (and, for a custody vendor, // a DB lookup) per request with nothing else gating it. Keyed on client IP, not a @@ -83,7 +83,7 @@ async function dispatchWebhook( ); if (event) { if (event.kind === 'deposit') { - await wallet.creditDepositByAddress(event); + await wallet.creditDepositByAddress(event, providerName); } else { await wallet.reconcileWithdrawalStatus(event); } @@ -102,8 +102,10 @@ export type WalletRouterDeps = { adminGuard: AdminGuard; audit: AuditWritePort; paymentProviders: PaymentProviderRegistry; + reconciliation: ReconciliationService; + jobQueue: JobQueueAdapter; + reconciliationQueue: QueueName; limiter?: RateLimiterAdapter; - jobQueue?: JobQueueAdapter; }; export function createWalletRouter({ @@ -111,8 +113,10 @@ export function createWalletRouter({ adminGuard, audit, paymentProviders, - limiter, + reconciliation, jobQueue, + reconciliationQueue, + limiter, }: WalletRouterDeps) { const os = implement(walletContract).$context(); @@ -135,7 +139,7 @@ export function createWalletRouter({ ), deposit: os.deposit.handler(({ input, context }) => - mapErrors({ BAD_REQUEST: CurrencyMismatchError, CONFLICT: IdempotencyKeyReuseError }, () => + mapErrors({ CONFLICT: IdempotencyKeyReuseError }, () => wallet.deposit({ userId: getUserId(context), amount: input.amount, @@ -152,7 +156,6 @@ export function createWalletRouter({ NOT_FOUND: WalletNotFoundError, BAD_REQUEST: [ InsufficientBalanceError, - CurrencyMismatchError, AmbiguousNetworkError, UnsupportedNetworkError, BelowMinimumWithdrawalError, @@ -430,19 +433,39 @@ export function createWalletRouter({ }, reconciliation: { - list: os.reconciliation.list.handler(async ({ context }) => { + list: os.reconciliation.list.handler(async ({ input, context }) => { await adminGuard.assert(context, 'wallet-reconciliation', 'view'); - return notImplemented(); + return reconciliation.listFindings(input); }), - resolve: os.reconciliation.resolve.handler(async ({ context }) => { - await adminGuard.assert(context, 'wallet-reconciliation', 'resolve'); - return notImplemented(); + resolve: os.reconciliation.resolve.handler(async ({ input, context }) => { + const { + userId: adminId, + ip, + userAgent, + } = await adminGuard.assert(context, 'wallet-reconciliation', 'resolve'); + return mapErrors( + { + NOT_FOUND: [ + ReconciliationFindingNotFoundError, + ReconciliationCreditTransactionNotFoundError, + ], + CONFLICT: ReconciliationCreditMismatchError, + }, + () => + reconciliation.resolveFinding(adminId, input.id, input.resolution, { ip, userAgent }), + ); }), + // Enqueues the job and returns the runId immediately - never runs inline. The + // eventual worker's own claim (wallet_job_run's partial unique index) stays the + // single concurrency authority; this route does not know or care whether its + // claim will win. run: os.reconciliation.run.handler(async ({ context }) => { await adminGuard.assert(context, 'wallet-reconciliation', 'run'); - return notImplemented(); + const runId = randomUUID(); + await jobQueue.enqueue(reconciliationQueue, { runId }); + return { runId }; }), }, }); diff --git a/packages/core/src/wallet/service/reconciliation-finding.service.ts b/packages/core/src/wallet/service/reconciliation-finding.service.ts new file mode 100644 index 00000000..339109c1 --- /dev/null +++ b/packages/core/src/wallet/service/reconciliation-finding.service.ts @@ -0,0 +1,63 @@ +import type { DrizzleDb, DrizzleTx } from '@openora/core/server'; +import type { WalletReconciliationFindingKind } from '@openora/core/contracts'; +import { walletReconciliationFinding } from '../schema/index.js'; + +/** + * Sentinel `runId` for a finding produced OUTSIDE any reconciliation job cycle - the + * live webhook path (`WalletService.creditDepositByAddress`) hits an unattributable + * deposit in real time, not on a poll. `runId` is NOT NULL on the table, so this fixed + * zero-uuid documents "no job run owns this row", mirroring the audit plugin's + * SYSTEM_ACTOR sentinel for "no admin acted here". + */ +export const LIVE_WEBHOOK_RUN_ID = '00000000-0000-0000-0000-000000000000'; + +export type ReconciliationFindingInput = { + runId: string; + providerName: string; + kind: WalletReconciliationFindingKind; + currency?: string | null; + network?: string | null; + amount?: string | null; + address?: string | null; + tag?: string | null; + txHash?: string | null; + /** + * The dedup key: a partial unique index on (kind, externalId) makes a duplicate + * insert for the same underlying vendor event or job re-run a silent no-op. Pass a + * stable stand-in (eg the source row's own id) when no vendor externalId exists, so + * a finding with no natural external reference still dedupes across re-runs. + */ + externalId?: string | null; + transactionId?: string | null; + detail?: string | null; +}; + +/** + * The single write path for a reconciliation finding - never a raw `insert` scattered + * across WalletService and ReconciliationService (both import this, neither imports the + * other, so there is no cycle). A finding is a report, never a credit instruction (see + * the schema's own comment): this never touches a balance or a transaction row, only + * records that something needs a human look. + */ +export async function recordReconciliationFinding( + db: DrizzleDb | DrizzleTx, + input: ReconciliationFindingInput, +): Promise { + await db + .insert(walletReconciliationFinding) + .values({ + runId: input.runId, + providerName: input.providerName, + kind: input.kind, + currency: input.currency ?? null, + network: input.network ?? null, + amount: input.amount ?? null, + address: input.address ?? null, + tag: input.tag ?? null, + txHash: input.txHash ?? null, + externalId: input.externalId ?? null, + transactionId: input.transactionId ?? null, + detail: input.detail ?? null, + }) + .onConflictDoNothing(); +} diff --git a/packages/core/src/wallet/service/reconciliation.service.ts b/packages/core/src/wallet/service/reconciliation.service.ts new file mode 100644 index 00000000..4a2c3de5 --- /dev/null +++ b/packages/core/src/wallet/service/reconciliation.service.ts @@ -0,0 +1,705 @@ +import { randomUUID } from 'node:crypto'; +import { and, count, desc, eq, gte, inArray, isNull, lt } from 'drizzle-orm'; +import { + type DrizzleService, + type EventBus, + createLogger, + findOneOrThrow, + makeConflictError, + makeNotFoundError, + moneyEquals, + pageToOffset, + serializeRow, +} from '@openora/core/server'; +import { + DEFAULT_PAYMENT_PROVIDER, + type AuditWritePort, + type ClientMeta, + type PaymentProviderRegistry, + type PaymentWebhookEvent, + type PlatformConfig, + type User, + type WalletJobRunStatus, +} from '@openora/core/contracts'; +import { + walletCustodySweep, + walletJobRun, + walletReconciliationFinding, + walletTransaction, + type WalletJobRun, + type WalletReconciliationFinding as WalletReconciliationFindingRow, + type WalletTransaction, +} from '../schema/index.js'; +import type { + ListReconciliationFindingsInput, + ReconciliationResolution, + WalletReconciliationFinding as WalletReconciliationFindingDto, +} from '../contract/index.js'; +import type { WalletService } from './wallet.service.js'; +import { recordReconciliationFinding } from './reconciliation-finding.service.js'; + +const logger = createLogger('wallet-reconciliation'); + +export const ReconciliationFindingNotFoundError = makeNotFoundError('ReconciliationFinding'); + +export const ReconciliationCreditTransactionNotFoundError = makeNotFoundError( + 'ReconciliationCreditTransaction', +); + +export const ReconciliationCreditMismatchError = makeConflictError( + 'ReconciliationCreditMismatchError', + "The referenced transaction is not a manual credit matching this finding's currency and amount", +); + +type LedgerMatch = Pick; + +/** + * Pure diff of one vendor deposit event against its (maybe absent) ledger row - the + * whole decision reconciliation makes for a deposit, extracted so it is unit-testable + * without a database. Never routes through moneyToNumber (see money.md): an exact + * decimal compare, not a float one. + */ +export function diffDeposit( + event: Extract, + tx: LedgerMatch | undefined, +): 'missing_deposit' | 'currency_mismatch' | 'amount_mismatch' | null { + if (!tx) { + return 'missing_deposit'; + } + if (tx.currency.toUpperCase() !== event.currency.toUpperCase()) { + return 'currency_mismatch'; + } + if (!moneyEquals(tx.amount, event.amount)) { + return 'amount_mismatch'; + } + return null; +} + +function chunk(items: readonly T[], size: number): T[][] { + const chunks: T[][] = []; + for (let i = 0; i < items.length; i += size) { + chunks.push(items.slice(i, i + size)); + } + return chunks; +} + +// Vendor listTransactions batches are thousands of rows; a well-under-the-parameter- +// ceiling chunk size keeps every lookup a single inArray(...) round trip per chunk, +// never one query per vendor transaction. +const PROVIDER_REF_LOOKUP_CHUNK_SIZE = 1000; + +// A claimed run stuck this long is presumed crashed (the process died mid-cycle) rather +// than still working, so the next tick may mark it `abandoned` and take the slot over. +// Distinct from the vendor-side ambiguity thresholds: this one is about our own worker. +const DEFAULT_STALE_RUN_AFTER_MINUTES = 30; + +/** Fallback when reconciliation runs without a `wallet.sweep` config block. */ +const DEFAULT_UNKNOWN_AFTER_MINUTES = 60; + +const JOB_NAME = 'wallet-reconciliation'; + +/** + * Ceiling on catching up after an outage, as a multiple of `lookbackHours`. Recovering + * from a week of downtime must not ask a vendor for one unbounded page of history. + */ +const MAX_CATCH_UP_MULTIPLE = 7; + +function toFindingDto(row: WalletReconciliationFindingRow): WalletReconciliationFindingDto { + return serializeRow(row, { dateFields: ['createdAt', 'resolvedAt'], decimalFields: [] }); +} + +export type ReconciliationServiceDeps = { + drizzle: DrizzleService; + events: EventBus; + wallet: WalletService; + paymentProviders: PaymentProviderRegistry; + audit: AuditWritePort; + platformConfig?: PlatformConfig; +}; + +/** + * Owns the reconciliation job cycle (claim -> per-provider diff -> stuck-withdrawal / + * stuck-sweep sweep -> finish + audit + alert) and the admin surface over its findings + * (list/resolve). Never credits or moves money itself - every finding is a report; the + * only paths that touch a balance are `WalletService.reconcileWithdrawalStatus` (already + * idempotent) and the pre-existing manual-adjustment path a `credited` resolution merely + * references. + */ +export class ReconciliationService { + private readonly drizzle: DrizzleService; + private readonly events: EventBus; + private readonly wallet: WalletService; + private readonly paymentProviders: PaymentProviderRegistry; + private readonly audit: AuditWritePort; + private readonly platformConfig?: PlatformConfig; + + constructor({ + drizzle, + events, + wallet, + paymentProviders, + audit, + platformConfig, + }: ReconciliationServiceDeps) { + this.drizzle = drizzle; + this.events = events; + this.wallet = wallet; + this.paymentProviders = paymentProviders; + this.audit = audit; + this.platformConfig = platformConfig; + } + + async listFindings(filter: ListReconciliationFindingsInput) { + const db = this.drizzle.db; + const conditions = []; + if (filter.status) { + conditions.push(eq(walletReconciliationFinding.status, filter.status)); + } + if (filter.kind) { + conditions.push(eq(walletReconciliationFinding.kind, filter.kind)); + } + if (filter.providerName) { + conditions.push(eq(walletReconciliationFinding.providerName, filter.providerName)); + } + const where = conditions.length > 0 ? and(...conditions) : undefined; + + const [rows, [{ n } = { n: 0 }]] = await Promise.all([ + db + .select() + .from(walletReconciliationFinding) + .where(where) + .orderBy(desc(walletReconciliationFinding.createdAt), desc(walletReconciliationFinding.id)) + .limit(filter.limit) + .offset(pageToOffset(filter.page, filter.limit)), + db.select({ n: count() }).from(walletReconciliationFinding).where(where), + ]); + + return { + items: rows.map(toFindingDto), + total: Number(n), + page: filter.page, + limit: filter.limit, + }; + } + + /** + * Conditional `UPDATE ... WHERE status = 'open'` is the whole double-resolve guard - + * a second resolve of an already-resolved finding matches zero rows and returns the + * row unchanged, with no second audit entry. `credited` requires the referenced + * transaction to already exist (the manual-adjustment path creates it - this never + * credits anything itself) and match the finding's currency/amount exactly. + */ + async resolveFinding( + adminId: User['id'], + id: WalletReconciliationFindingRow['id'], + resolution: ReconciliationResolution, + meta?: ClientMeta, + ): Promise { + return this.drizzle.db.transaction(async (txn) => { + let transactionId: string | undefined; + let resolutionNote: string | null = null; + + if (resolution.outcome === 'credited') { + const [finding] = await txn + .select() + .from(walletReconciliationFinding) + .where(eq(walletReconciliationFinding.id, id)); + if (!finding) { + throw new ReconciliationFindingNotFoundError(id); + } + const [tx] = await txn + .select() + .from(walletTransaction) + .where(eq(walletTransaction.id, resolution.transactionId)); + if (!tx) { + throw new ReconciliationCreditTransactionNotFoundError(resolution.transactionId); + } + if ( + tx.type !== 'manual_credit' || + finding.currency === null || + tx.currency.toUpperCase() !== finding.currency.toUpperCase() || + finding.amount === null || + !moneyEquals(tx.amount, finding.amount) + ) { + throw new ReconciliationCreditMismatchError(); + } + transactionId = tx.id; + } else { + resolutionNote = resolution.note; + } + + const updated = await txn + .update(walletReconciliationFinding) + .set({ + status: 'resolved', + resolvedBy: adminId, + resolvedAt: new Date(), + resolutionNote, + ...(transactionId ? { transactionId } : {}), + }) + .where( + and( + eq(walletReconciliationFinding.id, id), + eq(walletReconciliationFinding.status, 'open'), + ), + ) + .returning(); + + if (updated.length === 0) { + const [existing] = await txn + .select() + .from(walletReconciliationFinding) + .where(eq(walletReconciliationFinding.id, id)); + // Already resolved by a concurrent/earlier call: return it unchanged, no audit entry. + return toFindingDto( + findOneOrThrow(existing ? [existing] : [], new ReconciliationFindingNotFoundError(id)), + ); + } + + const row = findOneOrThrow(updated, new ReconciliationFindingNotFoundError(id)); + await this.audit.recordInTransaction(txn, { + actorId: adminId, + actorType: 'admin', + action: 'wallet.reconciliation_finding.resolved', + resourceType: 'wallet_reconciliation_finding', + resourceId: row.id, + before: { status: 'open' }, + after: { + status: row.status, + outcome: resolution.outcome, + transactionId: row.transactionId, + resolutionNote: row.resolutionNote, + }, + ...meta, + }); + return toFindingDto(row); + }); + } + + /** + * The window is anchored to the last completed run, not to the wall clock. BullMQ + * repeatable jobs are delayed jobs: a tick missed while no worker was running is + * skipped, never backfilled. Against a fixed `now - lookbackHours` window that means + * any outage longer than `lookbackHours` leaves a span nothing ever reconciles and + * nothing ever reports - the worst possible failure mode for a compensating control. + * Catching up is capped, and a capped window is reported rather than passed over. + */ + private async resolveWindow( + until: Date, + lookbackHours: number, + ): Promise<{ since: Date; unreconciledHours: number }> { + const lookbackStart = new Date(until.getTime() - lookbackHours * 3_600_000); + const [last] = await this.drizzle.db + .select({ finishedAt: walletJobRun.finishedAt }) + .from(walletJobRun) + .where(and(eq(walletJobRun.jobName, JOB_NAME), eq(walletJobRun.status, 'completed'))) + .orderBy(desc(walletJobRun.finishedAt)) + .limit(1); + + const anchor = last?.finishedAt; + if (!anchor || anchor >= lookbackStart) { + return { since: lookbackStart, unreconciledHours: 0 }; + } + + const floor = new Date(until.getTime() - lookbackHours * MAX_CATCH_UP_MULTIPLE * 3_600_000); + const since = anchor < floor ? floor : anchor; + return { + since, + unreconciledHours: Math.round((since.getTime() - anchor.getTime()) / 3_600_000), + }; + } + + /** + * Claim -> per-provider deposit/withdrawal diff -> stuck-withdrawal / stuck-sweep + * sweep -> finish + one audit entry + an alert past threshold. Returns null without + * doing any work when another cycle already owns the claim - the caller (cron tick or + * the on-demand route's worker) treats that as a normal, expected outcome. + */ + async runCycle( + runId: WalletJobRun['runId'] = randomUUID(), + ): Promise<{ runId: WalletJobRun['runId'] } | null> { + const claimed = await this.claimRun(runId); + if (!claimed) { + return null; + } + + const counts = { + missingDeposit: 0, + currencyMismatch: 0, + amountMismatch: 0, + withdrawalsReconciled: 0, + unknownAtProvider: 0, + stuckSweeps: 0, + unreconciledHours: 0, + }; + + try { + const cfg = this.platformConfig?.wallet?.reconciliation; + if (cfg) { + const until = new Date(); + const { since, unreconciledHours } = await this.resolveWindow(until, cfg.lookbackHours); + counts.unreconciledHours = unreconciledHours; + if (unreconciledHours > 0) { + logger.error( + { runId, unreconciledHours, since }, + 'wallet reconciliation: catch-up window capped, an earlier span stays unreconciled', + ); + } + + for (const providerName of this.paymentProviders.names()) { + await this.reconcileProvider(runId, providerName, since, until, counts); + } + + await this.reconcileStuckWithdrawals(runId, cfg.stuckAfterMinutes, counts); + + // Deliberately not gated on `wallet.sweep` being configured. An operator can + // adopt reconciliation without sweeping, and a stuck sweep is real player money + // parked vendor-side in an ambiguous state - the last finding class that should + // quietly switch itself off because a different feature's config block is absent. + await this.reconcileStuckSweeps( + runId, + this.platformConfig?.wallet?.sweep?.unknownAfterMinutes ?? DEFAULT_UNKNOWN_AFTER_MINUTES, + counts, + ); + } + + const openFindings = await this.countOpenFindings(); + await this.finishRun(runId, 'completed', { ...counts, openFindings }); + await this.audit.record({ + actorType: 'system', + action: 'wallet.reconciliation_run.completed', + resourceType: 'wallet_job_run', + resourceId: runId, + after: { runId, ...counts, openFindings }, + }); + + if (cfg && openFindings > cfg.alertThreshold) { + logger.error( + { runId, openFindings, threshold: cfg.alertThreshold }, + 'wallet reconciliation: open findings exceed threshold', + ); + this.events.emit('wallet.reconciliation.alert', { + runId, + openFindings, + threshold: cfg.alertThreshold, + }); + } + + return { runId }; + } catch (err) { + await this.finishRun(runId, 'failed', { + ...counts, + error: err instanceof Error ? err.message : String(err), + }); + await this.audit.record({ + actorType: 'system', + action: 'wallet.reconciliation_run.failed', + resourceType: 'wallet_job_run', + resourceId: runId, + after: { runId, ...counts }, + }); + throw err; + } + } + + private async reconcileProvider( + runId: WalletJobRun['runId'], + providerName: string, + since: Date, + until: Date, + counts: { + missingDeposit: number; + currencyMismatch: number; + amountMismatch: number; + withdrawalsReconciled: number; + }, + ): Promise { + const provider = this.paymentProviders.get(providerName); + if (!provider?.adapter.listTransactions) { + return; + } + const events = await provider.adapter.listTransactions({ since, until }); + + // Every sweep (an internal transfer of vendor-side funds, never a ledger row) shows + // up in the vendor's own transaction list too - exclude it BEFORE diffing, or every + // sweep files a false missing_deposit/unmatched-withdrawal finding. + const swept = await this.sweptExternalIds(providerName, since); + const relevant = events.filter((event) => !swept.has(event.externalId)); + + const deposits = relevant.filter( + (event): event is Extract => + event.kind === 'deposit', + ); + const withdrawals = relevant.filter( + (event): event is Extract => + event.kind === 'withdrawal', + ); + + await this.reconcileDeposits(runId, providerName, deposits, counts); + + for (const event of withdrawals) { + // Reconciliation is the same normalization a webhook produces, polled instead of + // pushed - already idempotent, already guards on status === 'processing'. + await this.wallet.reconcileWithdrawalStatus(event); + counts.withdrawalsReconciled += 1; + } + } + + private async sweptExternalIds(providerName: string, since: Date): Promise> { + const rows = await this.drizzle.db + .select({ externalId: walletCustodySweep.externalId }) + .from(walletCustodySweep) + .where( + and( + eq(walletCustodySweep.providerName, providerName), + gte(walletCustodySweep.createdAt, since), + ), + ); + return new Set(rows.map((row) => row.externalId).filter((id): id is string => id !== null)); + } + + private async reconcileDeposits( + runId: WalletJobRun['runId'], + providerName: string, + deposits: readonly Extract[], + counts: { missingDeposit: number; currencyMismatch: number; amountMismatch: number }, + ): Promise { + if (deposits.length === 0) { + return; + } + const byExternalId = await this.batchLookupByProviderRefId(deposits.map((d) => d.externalId)); + + for (const event of deposits) { + const tx = byExternalId.get(event.externalId); + const result = diffDeposit(event, tx); + if (result === null) { + continue; + } + counts[ + result === 'missing_deposit' + ? 'missingDeposit' + : result === 'currency_mismatch' + ? 'currencyMismatch' + : 'amountMismatch' + ] += 1; + await recordReconciliationFinding(this.drizzle.db, { + runId, + providerName, + kind: result, + currency: event.currency, + network: event.network ?? tx?.network ?? null, + amount: event.amount, + address: event.address, + tag: event.tag ?? null, + txHash: event.txHash, + externalId: event.externalId, + transactionId: tx?.id ?? null, + detail: + result === 'currency_mismatch' + ? `ledger currency ${tx?.currency}, vendor reported ${event.currency}` + : result === 'amount_mismatch' + ? `ledger amount ${tx?.amount}, vendor reported ${event.amount}` + : null, + }); + } + } + + // Single chunked inArray(...) batch read (chunks well under the parameter ceiling) - + // never one query per vendor transaction over a window that is routinely thousands of rows. + private async batchLookupByProviderRefId( + externalIds: string[], + ): Promise> { + const byExternalId = new Map(); + for (const batch of chunk(externalIds, PROVIDER_REF_LOOKUP_CHUNK_SIZE)) { + const rows = await this.drizzle.db + .select() + .from(walletTransaction) + .where(inArray(walletTransaction.providerRefId, batch)); + for (const row of rows) { + if (row.providerRefId) { + byExternalId.set(row.providerRefId, row); + } + } + } + return byExternalId; + } + + // Covered exactly by wallet_transaction_status_type_created_at_idx - equality on + // status and type, then a range on createdAt, in that column order. + private async reconcileStuckWithdrawals( + runId: WalletJobRun['runId'], + stuckAfterMinutes: number, + counts: { unknownAtProvider: number }, + ): Promise { + const cutoff = new Date(Date.now() - stuckAfterMinutes * 60 * 1000); + const stuck = await this.drizzle.db + .select() + .from(walletTransaction) + .where( + and( + eq(walletTransaction.status, 'processing'), + eq(walletTransaction.type, 'withdrawal'), + lt(walletTransaction.createdAt, cutoff), + ), + ); + + for (const tx of stuck) { + const providerName = tx.providerName ?? DEFAULT_PAYMENT_PROVIDER; + const provider = this.paymentProviders.get(providerName); + + // Nothing to look up at all (crashed before the vendor ever responded with a + // reference) - still worth a finding, deduped on the transaction's own id. + if (!tx.providerRefId) { + counts.unknownAtProvider += 1; + await recordReconciliationFinding(this.drizzle.db, { + runId, + providerName, + kind: 'unknown_at_provider', + currency: tx.currency, + network: tx.network, + amount: tx.amount, + transactionId: tx.id, + externalId: tx.id, + detail: 'withdrawal has no providerRefId to look up at the vendor', + }); + continue; + } + + const status = await provider?.adapter.getWithdrawalStatus?.(tx.providerRefId); + if (status) { + await this.wallet.reconcileWithdrawalStatus({ + kind: 'withdrawal', + externalId: tx.providerRefId, + status: status.status, + ...(status.txHash ? { txHash: status.txHash } : {}), + }); + continue; + } + + counts.unknownAtProvider += 1; + await recordReconciliationFinding(this.drizzle.db, { + runId, + providerName, + kind: 'unknown_at_provider', + currency: tx.currency, + network: tx.network, + amount: tx.amount, + transactionId: tx.id, + externalId: tx.providerRefId, + detail: 'vendor has no record of this withdrawal', + }); + } + } + + // Covered exactly by wallet_custody_sweep_status_created_at_idx. Human resolution + // only - this NEVER touches the sweep row's status or releases its in-flight guard. + private async reconcileStuckSweeps( + runId: WalletJobRun['runId'], + unknownAfterMinutes: number, + counts: { stuckSweeps: number }, + ): Promise { + const cutoff = new Date(Date.now() - unknownAfterMinutes * 60 * 1000); + const stuck = await this.drizzle.db + .select() + .from(walletCustodySweep) + .where( + and( + // `pending` belongs here alongside `unknown`. A worker that dies between the + // claim insert and the vendor call leaves a `pending` row with no externalId: + // resolveInFlightSweeps skips it (nothing to poll) and the partial unique + // index still counts it as in-flight, so that container would never sweep + // again and nothing would say why. Same lost-response ambiguity as `unknown`, + // so it files a finding for a human rather than releasing the guard. + inArray(walletCustodySweep.status, ['pending', 'unknown']), + lt(walletCustodySweep.createdAt, cutoff), + ), + ); + + for (const sweep of stuck) { + counts.stuckSweeps += 1; + await recordReconciliationFinding(this.drizzle.db, { + runId, + providerName: sweep.providerName, + kind: 'stuck_sweep', + currency: sweep.currency, + network: sweep.network, + amount: sweep.amount, + txHash: sweep.txHash, + externalId: sweep.externalId ?? sweep.id, + detail: `custody sweep ${sweep.id} has been ${sweep.status} since ${sweep.updatedAt.toISOString()}`, + }); + } + } + + private async countOpenFindings(): Promise { + const [row] = await this.drizzle.db + .select({ n: count() }) + .from(walletReconciliationFinding) + .where(eq(walletReconciliationFinding.status, 'open')); + return Number(row?.n ?? 0); + } + + /** + * The claim IS the insert - a partial unique index on (jobName) WHERE finishedAt IS + * NULL makes a conflicting insert the concurrency guard. A stale in-flight run (older + * than the configured staleness window - presumed crashed) is marked `abandoned` and its slot freed + * for this call to retry the claim once. + */ + private async claimRun(runId: WalletJobRun['runId']): Promise { + const inserted = await this.drizzle.db + .insert(walletJobRun) + .values({ jobName: JOB_NAME, runId }) + .onConflictDoNothing() + .returning({ id: walletJobRun.id }); + if (inserted.length > 0) { + return true; + } + + const [existing] = await this.drizzle.db + .select() + .from(walletJobRun) + .where(and(eq(walletJobRun.jobName, JOB_NAME), isNull(walletJobRun.finishedAt))); + if (!existing) { + // The blocking run finished between our insert conflict and this read - retry once. + return this.retryClaim(runId); + } + const staleAfterMs = + (this.platformConfig?.wallet?.reconciliation?.staleRunAfterMinutes ?? + DEFAULT_STALE_RUN_AFTER_MINUTES) * 60_000; + if (existing.startedAt.getTime() >= Date.now() - staleAfterMs) { + // A live run genuinely owns the claim - return immediately, no retry loop. + return false; + } + + const abandoned = await this.drizzle.db + .update(walletJobRun) + .set({ status: 'abandoned', finishedAt: new Date() }) + .where(and(eq(walletJobRun.id, existing.id), isNull(walletJobRun.finishedAt))) + .returning({ id: walletJobRun.id }); + if (abandoned.length === 0) { + // Someone else already resolved the stale run concurrently. + return false; + } + return this.retryClaim(runId); + } + + private async retryClaim(runId: WalletJobRun['runId']): Promise { + const inserted = await this.drizzle.db + .insert(walletJobRun) + .values({ jobName: JOB_NAME, runId }) + .onConflictDoNothing() + .returning({ id: walletJobRun.id }); + return inserted.length > 0; + } + + private async finishRun( + runId: WalletJobRun['runId'], + status: WalletJobRunStatus, + summary: Record, + ): Promise { + await this.drizzle.db + .update(walletJobRun) + .set({ finishedAt: new Date(), status, summary }) + .where(eq(walletJobRun.runId, runId)); + } +} diff --git a/packages/core/src/wallet/service/wallet.service.ts b/packages/core/src/wallet/service/wallet.service.ts index 1f7a727e..e4b7998f 100644 --- a/packages/core/src/wallet/service/wallet.service.ts +++ b/packages/core/src/wallet/service/wallet.service.ts @@ -56,6 +56,10 @@ import { type WalletAutoWithdrawalConfig as WalletAutoWithdrawalConfigRow, type WalletAssetRow, } from '../schema/index.js'; +import { + LIVE_WEBHOOK_RUN_ID, + recordReconciliationFinding, +} from './reconciliation-finding.service.js'; import type { TransactionResult, WithdrawalQueueItem, @@ -158,12 +162,6 @@ export const BelowMinimumWithdrawalError = createDomainError( const KYC_PASS_STATUSES: ReadonlySet = new Set(['approved', 'manually_overridden']); -export const CurrencyMismatchError = createDomainError( - 'CurrencyMismatchError', - (requested, walletCurrency) => - `Currency mismatch: requested ${requested}, wallet holds ${walletCurrency}`, -); - export const AmbiguousDepositAddressError = createDomainError( 'AmbiguousDepositAddressError', (address, network) => @@ -2106,22 +2104,68 @@ export class WalletService { return toDepositAddressResult(winner); } + /** + * `providerName` is the vendor the webhook route already resolved (verifier + adapter + * came from the same registry entry) - it is only used to label a finding when this + * deposit can't be attributed, never to look anything up. Defaults to the single + * default binding for callers (tests, a poller with no named provider) that don't care. + */ async creditDepositByAddress( event: Extract, + providerName: string = DEFAULT_PAYMENT_PROVIDER, ): Promise { const depositAddress = await this.findDepositAddressByAddress(event); if (!depositAddress) { + // A known vendor defect hits this live: a token deposit reported under a sibling + // token's asset id never resolves to a row. Never let it survive only as a log + // line - file a finding so an operator can attribute and manually credit it. logger.warn( { address: event.address, network: event.network, tag: event.tag }, 'payment webhook: no wallet_deposit_address for inbound deposit', ); + await recordReconciliationFinding(this.drizzle.db, { + runId: LIVE_WEBHOOK_RUN_ID, + providerName, + kind: 'unattributed_deposit', + currency: event.currency, + network: event.network ?? null, + amount: event.amount, + address: event.address, + tag: event.tag ?? null, + txHash: event.txHash, + externalId: event.externalId, + detail: 'no wallet_deposit_address matched this address/network/tag', + }); return; } if ( event.network === undefined && event.currency.toUpperCase() !== depositAddress.currency.toUpperCase() ) { - throw new CurrencyMismatchError(event.currency, depositAddress.currency); + // Same "do not silently drop" discipline as the no-address branch above - this is + // the push-path counterpart to the poll-path's currency_mismatch finding. + logger.warn( + { + address: event.address, + eventCurrency: event.currency, + addressCurrency: depositAddress.currency, + }, + 'payment webhook: currency mismatch for inbound deposit', + ); + await recordReconciliationFinding(this.drizzle.db, { + runId: LIVE_WEBHOOK_RUN_ID, + providerName: depositAddress.providerName, + kind: 'currency_mismatch', + currency: event.currency, + network: event.network ?? depositAddress.network, + amount: event.amount, + address: event.address, + tag: event.tag ?? null, + txHash: event.txHash, + externalId: event.externalId, + detail: `currency mismatch: event reported ${event.currency}, address issued for ${depositAddress.currency}`, + }); + return; } const { transactionId, replayed } = await this.drizzle.db.transaction(async (txn) => { From cf3fa1ae4f12d56aa14ebf9a7ddeda6d98061896 Mon Sep 17 00:00:00 2001 From: Volodymyr Zakhovaiko Date: Fri, 21 Aug 2026 19:09:15 +0200 Subject: [PATCH 2/2] fix(wallet): key the reconciliation run finish on the claimed row, not the runId BullMQ retries a job with the same payload, so a retried on-demand run carries the runId of the attempt that failed. `UPDATE ... WHERE run_id = $1` therefore reached back and stamped `completed` over the earlier row, erasing the only record that the first attempt failed at all. claimRun now returns the row it claimed and finishRun targets it. Also drops the router's dead job-queue guard - jobQueue stopped being optional when the deps became a named object - and routes the sweep's unconfigured_asset finding through recordReconciliationFinding, which this branch introduced as the single write path precisely to stop raw inserts drifting apart. --- .../reconciliation.service.int.test.ts | 24 ++++++++++++ packages/core/src/wallet/router/index.ts | 5 --- .../wallet/service/custody-sweep.service.ts | 38 +++++++++---------- .../wallet/service/reconciliation.service.ts | 38 +++++++++---------- 4 files changed, 60 insertions(+), 45 deletions(-) diff --git a/packages/core/src/wallet/__tests__/reconciliation.service.int.test.ts b/packages/core/src/wallet/__tests__/reconciliation.service.int.test.ts index c264a7a9..ce27b54c 100644 --- a/packages/core/src/wallet/__tests__/reconciliation.service.int.test.ts +++ b/packages/core/src/wallet/__tests__/reconciliation.service.int.test.ts @@ -325,6 +325,30 @@ describe('ReconciliationService.runCycle - claim concurrency', () => { }); }); +describe('ReconciliationService.runCycle - retried run', () => { + it('a queue retry carrying the same runId does not rewrite the failed attempt', async () => { + // BullMQ retries a job with the SAME payload, so both attempts share a runId. If + // the finish update keyed on runId instead of the claimed row, the retry's success + // would reach back and stamp `completed` over the first attempt - erasing the only + // record that reconciliation ever failed. + const runId = randomUUID(); + const listTransactions = vi + .fn() + .mockRejectedValueOnce(new Error('vendor 503')) + .mockResolvedValue([]); + const { reconciliation } = makeServices(mock({ listTransactions })); + + await expect(reconciliation.runCycle(runId)).rejects.toThrow('vendor 503'); + expect(await reconciliation.runCycle(runId)).toEqual({ runId }); + + const runs = await db.drizzle.db + .select() + .from(walletJobRun) + .where(eq(walletJobRun.runId, runId)); + expect(runs.map((r) => r.status).sort()).toEqual(['completed', 'failed']); + }); +}); + describe('ReconciliationService.runCycle - audit', () => { it('writes exactly one audit entry per run, with no address or tx hash in its payload', async () => { const externalId = randomUUID(); diff --git a/packages/core/src/wallet/router/index.ts b/packages/core/src/wallet/router/index.ts index bb4c950c..59eebf59 100644 --- a/packages/core/src/wallet/router/index.ts +++ b/packages/core/src/wallet/router/index.ts @@ -416,11 +416,6 @@ export function createWalletRouter({ sweep: { run: os.custody.sweep.run.handler(async ({ context }) => { await adminGuard.assert(context, 'wallet-custody', 'run'); - if (!jobQueue) { - throw new ORPCError('INTERNAL_SERVER_ERROR', { - message: 'custody sweep job queue is not wired', - }); - } // The run claim (a unique-index insert in wallet_job_run) is the sole // concurrency authority, not this request - enqueue and return immediately so // an HTTP timeout can never orphan a cycle. runId is minted here so the caller diff --git a/packages/core/src/wallet/service/custody-sweep.service.ts b/packages/core/src/wallet/service/custody-sweep.service.ts index 6688cdb9..c8a9db08 100644 --- a/packages/core/src/wallet/service/custody-sweep.service.ts +++ b/packages/core/src/wallet/service/custody-sweep.service.ts @@ -23,10 +23,10 @@ import { walletAsset, walletCustodySweep, walletJobRun, - walletReconciliationFinding, type WalletAssetRow, type WalletJobRun, } from '../schema/index.js'; +import { recordReconciliationFinding } from './reconciliation-finding.service.js'; const logger = createLogger('wallet-custody-sweep'); @@ -351,26 +351,22 @@ export class CustodySweepService { if (!asset) { // Funds sitting in a container the operator has no policy for must be visible, // not a log line. - await this.drizzle.db - .insert(walletReconciliationFinding) - .values({ - runId, - providerName, - kind: 'unconfigured_asset', - currency: balance.currency, - network: balance.network, - amount: balance.amount, - // The dedup key for the partial unique index on (kind, externalId). This - // finding has no vendor reference, and the condition recurs every single - // cycle until an operator configures the asset - without a stable stand-in - // one missing catalog row files a finding every cron tick, drowning the real - // findings and pinning the alert threshold permanently over the line. The - // asset, not the player, is the subject: one row to act on, not one per - // affected container. - externalId: `unconfigured:${providerName}:${balance.currency}:${balance.network}`, - detail: `no wallet_asset configured for ${balance.currency}/${balance.network}`, - }) - .onConflictDoNothing(); + await recordReconciliationFinding(this.drizzle.db, { + runId, + providerName, + kind: 'unconfigured_asset', + currency: balance.currency, + network: balance.network, + amount: balance.amount, + // No vendor reference exists for this finding, and the condition recurs every + // single cycle until an operator configures the asset - without a stable + // stand-in one missing catalog row files a finding every cron tick, drowning + // the real findings and pinning the alert threshold permanently over the line. + // The asset, not the player, is the subject: one row to act on, not one per + // affected container. + externalId: `unconfigured:${providerName}:${balance.currency}:${balance.network}`, + detail: `no wallet_asset configured for ${balance.currency}/${balance.network}`, + }); return; } diff --git a/packages/core/src/wallet/service/reconciliation.service.ts b/packages/core/src/wallet/service/reconciliation.service.ts index 4a2c3de5..f8d28e2f 100644 --- a/packages/core/src/wallet/service/reconciliation.service.ts +++ b/packages/core/src/wallet/service/reconciliation.service.ts @@ -318,8 +318,8 @@ export class ReconciliationService { async runCycle( runId: WalletJobRun['runId'] = randomUUID(), ): Promise<{ runId: WalletJobRun['runId'] } | null> { - const claimed = await this.claimRun(runId); - if (!claimed) { + const jobRunId = await this.claimRun(runId); + if (!jobRunId) { return null; } @@ -364,7 +364,7 @@ export class ReconciliationService { } const openFindings = await this.countOpenFindings(); - await this.finishRun(runId, 'completed', { ...counts, openFindings }); + await this.finishRun(jobRunId, 'completed', { ...counts, openFindings }); await this.audit.record({ actorType: 'system', action: 'wallet.reconciliation_run.completed', @@ -387,7 +387,7 @@ export class ReconciliationService { return { runId }; } catch (err) { - await this.finishRun(runId, 'failed', { + await this.finishRun(jobRunId, 'failed', { ...counts, error: err instanceof Error ? err.message : String(err), }); @@ -645,14 +645,10 @@ export class ReconciliationService { * than the configured staleness window - presumed crashed) is marked `abandoned` and its slot freed * for this call to retry the claim once. */ - private async claimRun(runId: WalletJobRun['runId']): Promise { - const inserted = await this.drizzle.db - .insert(walletJobRun) - .values({ jobName: JOB_NAME, runId }) - .onConflictDoNothing() - .returning({ id: walletJobRun.id }); - if (inserted.length > 0) { - return true; + private async claimRun(runId: WalletJobRun['runId']): Promise { + const claimed = await this.retryClaim(runId); + if (claimed) { + return claimed; } const [existing] = await this.drizzle.db @@ -668,7 +664,7 @@ export class ReconciliationService { DEFAULT_STALE_RUN_AFTER_MINUTES) * 60_000; if (existing.startedAt.getTime() >= Date.now() - staleAfterMs) { // A live run genuinely owns the claim - return immediately, no retry loop. - return false; + return null; } const abandoned = await this.drizzle.db @@ -678,28 +674,32 @@ export class ReconciliationService { .returning({ id: walletJobRun.id }); if (abandoned.length === 0) { // Someone else already resolved the stale run concurrently. - return false; + return null; } return this.retryClaim(runId); } - private async retryClaim(runId: WalletJobRun['runId']): Promise { - const inserted = await this.drizzle.db + private async retryClaim(runId: WalletJobRun['runId']): Promise { + const [inserted] = await this.drizzle.db .insert(walletJobRun) .values({ jobName: JOB_NAME, runId }) .onConflictDoNothing() .returning({ id: walletJobRun.id }); - return inserted.length > 0; + return inserted?.id ?? null; } + // Keyed on the claimed row's own id, never on runId: a job retried by the queue + // carries the same runId as the attempt that failed, so `WHERE run_id = ...` would + // reach back and rewrite that earlier row's `failed`/`abandoned` verdict to + // `completed` - erasing the only evidence that the first attempt went wrong. private async finishRun( - runId: WalletJobRun['runId'], + jobRunId: WalletJobRun['id'], status: WalletJobRunStatus, summary: Record, ): Promise { await this.drizzle.db .update(walletJobRun) .set({ finishedAt: new Date(), status, summary }) - .where(eq(walletJobRun.runId, runId)); + .where(eq(walletJobRun.id, jobRunId)); } }