From ca9e40d5acc3e270b725717e3e97f64c91cc0835 Mon Sep 17 00:00:00 2001 From: Kaustav Ghosh <44238657+teetangh@users.noreply.github.com> Date: Mon, 24 Aug 2026 01:31:16 +0530 Subject: [PATCH] fix(money): consolidated review-triage follow-ups across all session PRs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Legit-pending findings from the CodeRabbit sweep over #1205/#1218/#1219/ #1220/#1225/#1232, each verified against dev HEAD before fixing: - sweeper claim split onto a dedicated WebhookEvent.claimedAt column — bumping receivedAt on every re-drive let permanently-failing rows dodge the 7-day give-up cap forever; lost-claim + CAS-shape tests added. - with-cron-lock: a null acquire now RE-probes Redis health (the breaker's first-four-failures window returns null while isRedisCircuitOpen is still false — misreported as a clean held-skip); circuit-open and mid-window outage tests added. - wallet freeze/unfreeze writes propagate DB failures (recordSystemEvent swallows them by design); unfreeze check+write is atomic under Serializable so a concurrent re-freeze aborts instead of being cleared; route reports 409 on raced no-ops. - invoice-refund replay guard keyed on the ledger journal instead of the credit note (CN legitimately null for DRAFT invoices → redelivery double-credited the wallet). - razorpay SDK timeout helper clears its timer on synchronous throws. - refund adopt path merges Phase 1 audit metadata onto the surviving row. - org-rail redrive routes PERMANENT_4XX/validation rejections to markOrgPayoutFailed instead of hourly retries; 'Serializable' comment de-overclaimed to READ COMMITTED + CAS rationale. - reconcile pass-2 excludes ALL synthetic prefixes (internal_/credits_); bind helper takes metadata as a param (drops an extra round-trip). - user hard-delete gate counts consultant-side earnings/payouts/TDS via ConsultantProfile (payer-only counts let consultant history hit the Restrict 500). - signedDeltaPaise validates the RUNNING total (mid-sum overflow could resettle back into range). - stuck-payout permanent-FAIL claims PROCESSING via CAS inside the same tx as the earnings release. - checkout reuse v2: CONSULTATION/SUBSCRIPTION candidates gated on the slot window; amount-parity gate supersedes stale-priced holds (EXPIRED + superseded-reprice) instead of charging the old number; payment row id replaces raw userId in logs; discriminating unit tests for all gates; lock-TTL test asserts imported constants. - repair-SQL header documents payout-writer stop-the-world requirement. - trigger extension: Payment.amount UPDATEs validate their legs; leg reparenting validates BOTH payments. - free-credit rail: zero-value settlement posts no journal instead of throwing; org-clawback/TDS branches now have fixture coverage. --- .../sweep-stuck-webhook-events.test.ts | 33 ++++ __tests__/enterprise/with-cron-lock.test.ts | 39 ++++ __tests__/payments/checkout-lock-ttl.test.ts | 26 ++- .../checkout-open-order-reuse.test.ts | 141 +++++++++++++- __tests__/payments/free-credit-refund.test.ts | 115 +++++++++++ .../payments/razorpay-test-key-guard.test.ts | 4 + .../[billingAccountId]/unfreeze/route.ts | 11 +- app/api/user/[id]/route.ts | 20 +- app/api/webhooks/utils.ts | 45 +++-- lib/api/organizations/wallet.ts | 17 +- lib/cron/with-cron-lock.ts | 21 +- lib/payments/core/razorpay.ts | 27 ++- lib/payments/operations/booking-refund.ts | 10 + lib/payments/operations/checkout.ts | 179 ++++++++++++++++-- lib/payments/operations/refund.ts | 25 +++ lib/payments/payouts/org-payout-service.ts | 19 +- lib/payments/wallet-freeze.ts | 101 ++++++---- prisma/schema.prisma | 5 + ...epair-ready-earnings-welded-to-payouts.sql | 7 + prisma/sql/payment-legs-triggers.sql | 58 ++++-- scripts/cleanup/sweep-stuck-webhook-events.ts | 15 +- scripts/payouts/handle-stuck-payouts.ts | 45 +++-- scripts/refunds/reconcile-pending-refunds.ts | 19 +- 23 files changed, 824 insertions(+), 158 deletions(-) diff --git a/__tests__/enterprise/sweep-stuck-webhook-events.test.ts b/__tests__/enterprise/sweep-stuck-webhook-events.test.ts index 310781442..2feeee590 100644 --- a/__tests__/enterprise/sweep-stuck-webhook-events.test.ts +++ b/__tests__/enterprise/sweep-stuck-webhook-events.test.ts @@ -50,6 +50,8 @@ const mockWe = ( findMany: jest.Mock; findUnique: jest.Mock; update: jest.Mock; + // #1205-triage — the sweeper's claim CAS before each re-drive. + updateMany: jest.Mock; }; } ).webhookEvent; @@ -60,6 +62,7 @@ const stuckRow = (over: Record = {}) => ({ eventType: "payment.captured", payload: { payment: { entity: { id: "pay_1" } } }, receivedAt: new Date("2026-06-01T00:00:00Z"), + claimedAt: null as Date | null | undefined, ...over, }); @@ -69,6 +72,36 @@ beforeEach(() => { }); describe("sweepStuckWebhookEvents (#785)", () => { + it("a LOST claim (claimedAt raced) skips the re-drive entirely (#1205-triage)", async () => { + const ev = stuckRow(); + (mockWe.findMany as jest.Mock).mockResolvedValue([ev]); + // Another driver claimed between selection and claim: CAS misses. + (mockWe.updateMany as jest.Mock).mockResolvedValue({ count: 0 }); + + const result = await sweepStuckWebhookEvents({ staleMinutes: 6 }); + + expect(processRazorpayWebhookEvent).not.toHaveBeenCalled(); + expect(result.recovered).toBe(0); + }); + + it("the claim CAS keys on claimedAt, not receivedAt (age must survive re-drives)", async () => { + const ev = stuckRow(); + (mockWe.findMany as jest.Mock).mockResolvedValue([ev]); + (mockWe.updateMany as jest.Mock).mockResolvedValue({ count: 1 }); + + await sweepStuckWebhookEvents({ staleMinutes: 6 }); + + const [claim] = (mockWe.updateMany as jest.Mock).mock.calls; + expect(claim[0].where).toMatchObject({ + eventId: ev.eventId, + OR: [{ claimedAt: null }, { claimedAt: ev.claimedAt }], + }); + expect(claim[0].data.claimedAt).toBeInstanceOf(Date); + // receivedAt untouched — the give-up cap ages on it. + expect(claim[0].where.receivedAt).toBeUndefined(); + expect(claim[0].data.receivedAt).toBeUndefined(); + }); + it("re-drives a stuck event and reconstructs the full envelope", async () => { mockWe.findMany.mockResolvedValue([stuckRow()]); mockProcess.mockResolvedValue(undefined); diff --git a/__tests__/enterprise/with-cron-lock.test.ts b/__tests__/enterprise/with-cron-lock.test.ts index 6869d95f9..ebe39b4d7 100644 --- a/__tests__/enterprise/with-cron-lock.test.ts +++ b/__tests__/enterprise/with-cron-lock.test.ts @@ -118,6 +118,45 @@ describe("withCronLock", () => { expect(fn).not.toHaveBeenCalled(); }); + it("#1205-triage: breaker OPEN at acquire → CronLockUnavailableError (pages), not a held skip", async () => { + mockAcquire.mockResolvedValue(null); + mockHealth.mockResolvedValue(true); // pre-acquire health passes + const { isRedisCircuitOpen } = jest.requireMock("../../lib/redis") as { + isRedisCircuitOpen: jest.Mock; + }; + isRedisCircuitOpen.mockReturnValue(true); + + const fn = jest.fn().mockResolvedValue("done"); + const err = await withCronLock("dunning", { failMode: "closed" }, fn).catch( + (e: unknown) => e, + ); + expect(err).toBeInstanceOf(CronLockUnavailableError); + expect(fn).not.toHaveBeenCalled(); + }); + + it("#1205-triage: null acquire + Redis downed AFTER a healthy gate pages too", async () => { + // The first-four-failures window: breaker CLOSED, but every op fails — + // acquire returns null via the error fallback while isRedisCircuitOpen() + // is false. Only the fresh health probe distinguishes this from "held". + // Healthy at the pre-acquire gate (Redis reachable then), down by the + // post-null re-probe — exactly the mid-window failure the old code + // misclassified as CronLockHeldError. + mockHealth.mockResolvedValueOnce(true).mockResolvedValueOnce(false); + mockAcquire.mockResolvedValue(null); + const { isRedisCircuitOpen } = jest.requireMock("../../lib/redis") as { + isRedisCircuitOpen: jest.Mock; + }; + isRedisCircuitOpen.mockReturnValue(false); + const fn = jest.fn().mockResolvedValue("done"); + + const err = await withCronLock("dunning", { failMode: "closed" }, fn).catch( + (e: unknown) => e, + ); + expect(err).toBeInstanceOf(CronLockUnavailableError); + // Two probes total: the pre-acquire gate + the post-null re-probe. + expect(mockHealth).toHaveBeenCalledTimes(2); + }); + it("fail-closed: refuses to run when Redis is unhealthy (circuit open)", async () => { mockHealth.mockResolvedValue(false); await expect( diff --git a/__tests__/payments/checkout-lock-ttl.test.ts b/__tests__/payments/checkout-lock-ttl.test.ts index 69556227d..1551c6c79 100644 --- a/__tests__/payments/checkout-lock-ttl.test.ts +++ b/__tests__/payments/checkout-lock-ttl.test.ts @@ -7,31 +7,27 @@ * serverless-freeze worst case for CLASS: the platform can suspend the lock * holder AFTER the single checked renewal while Redis keeps counting the TTL * down, so the old 300s CLASS budget could expire mid-checkout and admit a - * second instance. Source-contract pins (same idiom as - * idempotency-minting.test.ts) freeze the values until someone deliberately - * re-litigates them. + * second instance. + * + * Asserts the IMPORTED constant (CodeRabbit #1220 triage): source-text regexes + * break on formatter changes (600_000 ↔ 600000) and pass through duplicate-key + * overrides — neither can lie here. */ -import fs from "fs"; -import path from "path"; - -const read = (rel: string) => - fs.readFileSync(path.join(process.cwd(), rel), "utf8"); +import { CHECKOUT_LOCK_TTL_MS } from "../../utils/appointmentlock"; describe("CHECKOUT_LOCK_TTL_MS (#832 serverless-freeze worst case)", () => { - const src = read("utils/appointmentlock.ts"); - it("raises CLASS to the documented freeze worst case: 600s", () => { - expect(src).toMatch(/CLASS:\s*600_000,/); + expect(CHECKOUT_LOCK_TTL_MS.CLASS).toBe(600_000); }); it("no longer carries the insufficient 300s CLASS budget", () => { - expect(src).not.toMatch(/CLASS:\s*300_000/); + expect(CHECKOUT_LOCK_TTL_MS.CLASS).not.toBe(300_000); }); it("keeps the smaller shapes on their sized #832 budgets", () => { - expect(src).toMatch(/CONSULTATION:\s*60_000,/); - expect(src).toMatch(/SUBSCRIPTION:\s*120_000,/); - expect(src).toMatch(/WEBINAR:\s*120_000,/); + expect(CHECKOUT_LOCK_TTL_MS.CONSULTATION).toBe(60_000); + expect(CHECKOUT_LOCK_TTL_MS.SUBSCRIPTION).toBe(120_000); + expect(CHECKOUT_LOCK_TTL_MS.WEBINAR).toBe(120_000); }); }); diff --git a/__tests__/payments/checkout-open-order-reuse.test.ts b/__tests__/payments/checkout-open-order-reuse.test.ts index 54bd2d322..4a83b25b2 100644 --- a/__tests__/payments/checkout-open-order-reuse.test.ts +++ b/__tests__/payments/checkout-open-order-reuse.test.ts @@ -23,9 +23,13 @@ jest.mock("../../lib/prisma", () => ({ default: { $transaction: jest.fn(async (fn: any, _opts?: unknown) => fn(txClient)), payment: { - findFirst: jest.fn(async ({ where }: any) => - reuseState.rows.find((row) => matchesReuseWhere(row, where)) ?? null, + // #1220-triage — the reuse lookup fetches the newest scope-matching + // candidates; window/amount gates run in-code and rejects get + // superseded via updateMany. + findMany: jest.fn(async ({ where }: any) => + reuseState.rows.filter((row) => matchesReuseWhere(row, where)), ), + updateMany: jest.fn(async () => ({ count: 1 })), }, webinar: { findUnique: jest.fn(async () => webinarRow()), @@ -239,7 +243,14 @@ function openSibling(overrides: Record = {}) { amount: 100000, currency: "INR", appointmentId: "appt-w", - appointment: { webinarId: "evt-1" }, + appointment: { + webinarId: "evt-1", + // Window gate reads the first slot (WEBINAR flow skips it, but keep the + // shape faithful for the direct unit cases below). + slotsOfAppointment: [ + { startsAt: new Date("2026-09-01T10:00:00Z"), endsAt: new Date("2026-09-01T11:00:00Z") }, + ], + }, ...overrides, }; } @@ -327,7 +338,7 @@ describe("rec C — checkout adopts an open PENDING order across remounts", () = // The lookup must be scoped tightly — user + PENDING + fresh window + // org equality + gateway + this event's appointment join. - const where = (prisma.payment.findFirst as jest.Mock).mock.calls[0][0] + const where = (prisma.payment.findMany as jest.Mock).mock.calls[0][0] .where; expect(where).toMatchObject({ userId: "user-1", @@ -378,3 +389,125 @@ describe("rec C — checkout adopts an open PENDING order across remounts", () = expect(res.paymentIntent?.id).toBe("order_NEW"); }); }); + +// --------------------------------------------------------------------------- +// #1220-triage — the gates themselves, exercised directly (the webinar flow +// above cannot discriminate them: eventId already pins scope and its fixtures +// share one price). +// --------------------------------------------------------------------------- +import { + findReusablePendingOrderPayment, +} from "../../lib/payments/operations/checkout"; + +const SLOT = { startsAt: new Date("2026-09-01T10:00:00Z"), endsAt: new Date("2026-09-01T11:00:00Z") }; +const OTHER_SLOT = { startsAt: new Date("2026-09-02T10:00:00Z"), endsAt: new Date("2026-09-02T11:00:00Z") }; + +function gateDb(rows: Array>) { + return { payment: { findMany: async () => rows } }; +} + +describe("#1220-triage — reuse gates", () => { + test("CONSULTATION: a different slot time is superseded, never resumed", async () => { + const row = openSibling({ + appointment: { consultationId: "cons_1", slotsOfAppointment: [OTHER_SLOT] }, + }); + const { reusable, supersede } = await findReusablePendingOrderPayment( + gateDb([row]) as never, + { + userId: "user-1", + appointmentType: "CONSULTATION", + planId: "plan-1", + organizationId: null, + paymentGateway: "RAZORPAY" as never, + expectedAmountPaise: 100_000, + slotWindow: SLOT, + }, + ); + expect(reusable).toBeNull(); + expect(supersede).toEqual([{ id: "pay-open", reason: "slot-window-mismatch" }]); + }); + + test("CONSULTATION: identical slot window resumes", async () => { + const row = openSibling({ + appointment: { consultationId: "cons_1", slotsOfAppointment: [SLOT] }, + }); + const { reusable, supersede } = await findReusablePendingOrderPayment( + gateDb([row]) as never, + { + userId: "user-1", + appointmentType: "CONSULTATION", + planId: "plan-1", + organizationId: null, + paymentGateway: "RAZORPAY" as never, + expectedAmountPaise: 100_000, + slotWindow: SLOT, + }, + ); + expect(reusable?.id).toBe("pay-open"); + expect(supersede).toEqual([]); + }); + + test("amount parity gate: a stale frozen total is superseded, not resumed", async () => { + const row = openSibling({ amount: 80_000 }); // coupon changed since mint + const { reusable, supersede } = await findReusablePendingOrderPayment( + gateDb([row]) as never, + { + userId: "user-1", + appointmentType: "WEBINAR", + planId: "plan-1", + eventId: "evt-1", + organizationId: null, + paymentGateway: "RAZORPAY" as never, + expectedAmountPaise: 100_000, + }, + ); + expect(reusable).toBeNull(); + expect(supersede).toEqual([{ id: "pay-open", reason: "amount-mismatch" }]); + }); + + test("SUBSCRIPTION: period mismatch supersedes; both-null request matches both-null rows only", async () => { + const withPeriod = openSibling({ + id: "pay-period", + appointment: { + subscriptionId: "sub_1", + slotsOfAppointment: [SLOT], + }, + }); + const withoutPeriod = openSibling({ + id: "pay-noperiod", + appointment: { subscriptionId: "sub_2", slotsOfAppointment: [] }, + }); + + // Request WITH a period must not resume a period-less hold. + const gated = await findReusablePendingOrderPayment( + gateDb([withoutPeriod, withPeriod]) as never, + { + userId: "user-1", + appointmentType: "SUBSCRIPTION", + planId: "plan-1", + organizationId: null, + paymentGateway: "RAZORPAY" as never, + expectedAmountPaise: 100_000, + schedulingPeriod: SLOT, + }, + ); + expect(gated.reusable?.id).toBe("pay-period"); + expect(gated.supersede.map((s) => s.id)).toEqual(["pay-noperiod"]); + + // Keyless request must not resume a hold that carries a period. + const keyless = await findReusablePendingOrderPayment( + gateDb([withPeriod]) as never, + { + userId: "user-1", + appointmentType: "SUBSCRIPTION", + planId: "plan-1", + organizationId: null, + paymentGateway: "RAZORPAY" as never, + expectedAmountPaise: 100_000, + schedulingPeriod: null, + }, + ); + expect(keyless.reusable).toBeNull(); + expect(keyless.supersede.map((s) => s.id)).toEqual(["pay-period"]); + }); +}); diff --git a/__tests__/payments/free-credit-refund.test.ts b/__tests__/payments/free-credit-refund.test.ts index 071b8ef94..7e7cb76bf 100644 --- a/__tests__/payments/free-credit-refund.test.ts +++ b/__tests__/payments/free-credit-refund.test.ts @@ -339,3 +339,118 @@ describe("refundBookingPayment — free_ credit rail (#1161)", () => { ); }); }); + +// --------------------------------------------------------------------------- +// #1218-triage — the org-clawback and TDS branches of +// reverseFreeCreditSettlement were never exercised (all fixtures had empty +// organizationEarnings and payoutId: null). +// --------------------------------------------------------------------------- +describe("free_ credit rail — org clawback + TDS reversal branches", () => { + beforeEach(() => { + mockPaymentFindUnique + .mockResolvedValueOnce({ paymentIntent: "free_1730000000_abc" }) + .mockResolvedValueOnce({ + id: PAYMENT_ID, + currency: "INR", + paymentStatus: "SUCCEEDED", + paymentGateway: "RAZORPAY", + }); + }); + + it("nets PAID-out consultant earnings and reverses their TDS", async () => { + tx.payment.findUniqueOrThrow.mockResolvedValue({ + ...freeCreditSettlement(), + earnings: [ + { + id: "ce-paid", + consultantProfileId: "cp-1", + consultantSharePaise: 80_000, + refundedShareAmount: 0, + status: "PAID", + payoutId: "payout-9", + }, + ], + }); + + await refundBookingPayment({ paymentId: PAYMENT_ID, reason: "cancellation" }); + + // The paid share nets to REFUNDED… + const earningUpdate = tx.consultantEarnings.update.mock.calls.find( + ([arg]: [{ where: { id: string } }]) => arg.where.id === "ce-paid", + ); + // Cumulative-set semantics on this rail (not {increment}). + expect(earningUpdate[0].data).toMatchObject({ + status: "REFUNDED", + refundedShareAmount: 80_000, + }); + // …and the withholding against its payout is reversed. + expect(mockRecordTdsReversal).toHaveBeenCalledWith( + expect.anything(), // tx client + expect.objectContaining({ + payoutId: "payout-9", + earningsId: "ce-paid", + }), + ); + // Journal still balances with the paid-share debit included. + const postings = mockPostLedgerTxn.mock.calls[0][1].postings; + expect(sum(postings, "DEBIT")).toBe(sum(postings, "CREDIT")); + }); + + it("claws back a COMPLETED org payout and writes the audit row", async () => { + tx.payment.findUniqueOrThrow.mockResolvedValue({ + ...freeCreditSettlement(), + organizationEarnings: [ + { + id: "oe-1", + organizationId: "org-1", + orgSharePaise: 20_000, + refundedAmountPaise: 0, + status: "PAID", + orgPayoutId: "opayout-7", + orgPayout: { status: "COMPLETED", clawbackInitiatedAt: null }, + }, + ], + // No consultant side — org-collaborator-only settlement. + earnings: [], + }); + const orgEarningUpdates: Array<{ data: Record }> = []; + tx.organizationEarnings.update.mockImplementation( + async ({ data }: { data: Record }) => { + orgEarningUpdates.push({ data }); + return {}; + }, + ); + + await refundBookingPayment({ paymentId: PAYMENT_ID, reason: "cancellation" }); + + // Org share flips to REFUNDED with the full proration. + expect(orgEarningUpdates[0]?.data).toMatchObject({ + status: "REFUNDED", + refundedAmountPaise: 20_000, // cumulative-set + }); + // Clawback recorded on the COMPLETED payout — exactly once stamped. + const clawback = tx.organizationPayout.update.mock.calls.find( + ([arg]: [{ data?: { clawbackAmountPaise?: unknown } }]) => + !!arg.data?.clawbackAmountPaise, + ); + expect(clawback?.[0].data.clawbackAmountPaise).toEqual({ + increment: 20_000, + }); + expect(tx.orgAuditLog.create).toHaveBeenCalledWith( + expect.objectContaining({ + data: expect.objectContaining({ category: "PAYOUT" }), + }), + ); + // And the journal balances with the ORG_PAYABLE debit present. + const postings = + mockPostLedgerTxn.mock.calls[mockPostLedgerTxn.mock.calls.length - 1][1] + .postings; + expect( + postings.some( + (p: { account: { kind: string }; direction: string }) => + p.account.kind === "ORG_PAYABLE" && p.direction === "DEBIT", + ), + ).toBe(true); + expect(sum(postings, "DEBIT")).toBe(sum(postings, "CREDIT")); + }); +}); diff --git a/__tests__/payments/razorpay-test-key-guard.test.ts b/__tests__/payments/razorpay-test-key-guard.test.ts index b5a84fc84..7eddb26ec 100644 --- a/__tests__/payments/razorpay-test-key-guard.test.ts +++ b/__tests__/payments/razorpay-test-key-guard.test.ts @@ -190,6 +190,10 @@ describe("RazorpayX payouts client (lib/payments/payouts/razorpay-payouts.ts)", process.env.ENABLE_LIVE_PAYOUTS = "true"; process.env.NEXT_PHASE = "phase-production-build"; process.env.RAZORPAYX_KEY_ID = "rzp_test_buildx"; + // #1219-triage — a missing secret must not be the reason this passes: + // construct against a complete credential set so the exemption (not an + // unrelated config throw) is what's under test. + process.env.RAZORPAYX_KEY_SECRET = "xsecret"; process.env.RAZORPAYX_ACCOUNT_NUMBER = "acc_1"; const mod = requirePayoutsFactory(); expect(() => mod.getRazorpayPayoutsService()).not.toThrow(); diff --git a/app/api/admin/billing-accounts/[billingAccountId]/unfreeze/route.ts b/app/api/admin/billing-accounts/[billingAccountId]/unfreeze/route.ts index 0aac1a797..14a5ed427 100644 --- a/app/api/admin/billing-accounts/[billingAccountId]/unfreeze/route.ts +++ b/app/api/admin/billing-accounts/[billingAccountId]/unfreeze/route.ts @@ -60,12 +60,21 @@ export async function POST( ); } - await unfreezeWalletSpend({ + const applied = await unfreezeWalletSpend({ billingAccountId: account.id, organizationId: account.ownerOrgId, actorUserId: auth.session.user.id, reason: body.data.reason, }); + // false ⇒ not frozen (raced with another actor or a concurrent re-freeze + // abort) — surface as conflict, never report success for a no-op. + if (!applied) { + return NextResponse.json( + { error: "Wallet spend is not frozen on this account" }, + { status: 409 }, + ); + } + return NextResponse.json({ status: "unfrozen", billingAccountId: account.id }); } diff --git a/app/api/user/[id]/route.ts b/app/api/user/[id]/route.ts index 32d556325..95be74c59 100644 --- a/app/api/user/[id]/route.ts +++ b/app/api/user/[id]/route.ts @@ -236,11 +236,27 @@ export async function DELETE( // says are retained per IT Act 5–7y obligations. Users whose money ever // moved get the §12 erasure scrub instead: PII pseudonymised, erasedAt // tombstone set, financial rows intact. - const [paymentCount, referralCreditCount] = await Promise.all([ + // Consultant-side money lives on ConsultantProfile (earnings/payouts/TDS + // Restrict-delete through it), not on Payment — a consultant with payout + // history but no payer-side rows must also take the scrub path, or the + // hard delete 500s on the first Restrict (#1205-triage). + const [paymentCount, referralCreditCount, profile] = await Promise.all([ prisma.payment.count({ where: { userId: id } }), prisma.referralCredit.count({ where: { userId: id } }), + prisma.consultantProfile.findFirst({ + where: { userId: id }, + select: { + _count: { select: { earnings: true, payouts: true, tdsRecords: true } }, + }, + }), ]); - const hasMoneyHistory = paymentCount + referralCreditCount > 0; + const consultantMoneyCount = profile + ? profile._count.earnings + + profile._count.payouts + + profile._count.tdsRecords + : 0; + const hasMoneyHistory = + paymentCount + referralCreditCount + consultantMoneyCount > 0; if (hasMoneyHistory) { await scrubUser(prisma, id); diff --git a/app/api/webhooks/utils.ts b/app/api/webhooks/utils.ts index 42bcee641..bc304499b 100644 --- a/app/api/webhooks/utils.ts +++ b/app/api/webhooks/utils.ts @@ -794,26 +794,31 @@ export async function handleRefundCreated( status: true, }, }); - if (invoice) { - const mapped = mapGatewayRefundStatus(status); - if (mapped === "SUCCEEDED") { - // Per-refund idempotency: the GST credit note is unique on - // refundId, so its presence means THIS gateway refund was already - // booked end-to-end. (The old invoice-status guard collapsed - // distinct refunds: the first partial flipped the invoice REFUNDED - // and every later partial was skipped wholesale — real cash left - // via the gateway with no credit note, no wallet credit, no - // journal.) - const existingCreditNote = await tx.creditNote.findUnique({ - where: { refundId }, - select: { id: true }, - }); - if (existingCreditNote) { - console.log( - `💸 Invoice refund ${refundId} already booked, skipping`, - ); - return; - } + if (invoice) { + const mapped = mapGatewayRefundStatus(status); + if (mapped === "SUCCEEDED") { + // Per-refund idempotency — keyed on the LEDGER JOURNAL, not the + // credit note. The journal (`invoice-refund:`) is the + // one write that happens for EVERY booked refund, while + // mintInvoiceRefundCreditNote legitimately returns null for DRAFT/ + // unissued invoices — a CN-only probe let redeliveries of those + // re-run the audit log and (pre-#1128-fix) double the wallet + // credit. postLedgerTxn's own idempotency stays as the second + // layer; this probe just short-circuits before any side effects. + // (The old invoice-status guard collapsed distinct refunds: the + // first partial flipped the invoice REFUNDED and every later + // partial was skipped wholesale — real cash left via the gateway + // with no credit note, no wallet credit, no journal.) + const alreadyBooked = await tx.ledgerTransaction.findUnique({ + where: { idempotencyKey: `invoice-refund:${refundId}` }, + select: { id: true }, + }); + if (alreadyBooked) { + console.log( + `💸 Invoice refund ${refundId} already booked, skipping`, + ); + return; + } // #776 / PR#785 review — mint the GST credit note (Sec 34) for the // refunded invoice. One per gateway refund, idempotent on refundId. diff --git a/lib/api/organizations/wallet.ts b/lib/api/organizations/wallet.ts index 77b13e501..8ae597c88 100644 --- a/lib/api/organizations/wallet.ts +++ b/lib/api/organizations/wallet.ts @@ -314,7 +314,8 @@ export function withRunningBalance( export function signedDeltaPaise( rows: readonly { direction: string; amountPaise: bigint | number }[], ): number { - const total = rows.reduce((acc, r) => { + let total = 0; + for (const r of rows) { const amount = typeof r.amountPaise === "bigint" ? Number(r.amountPaise) @@ -324,12 +325,14 @@ export function signedDeltaPaise( `signedDeltaPaise: amount ${r.amountPaise} exceeds the safe integer range`, ); } - return acc + (r.direction === "CREDIT" ? amount : -amount); - }, 0); - if (!Number.isSafeInteger(total)) { - throw new Error( - `signedDeltaPaise: summed delta ${total} exceeds the safe integer range`, - ); + total += r.direction === "CREDIT" ? amount : -amount; + // Validate the RUNNING total too: a later debit can bring an imprecise + // sum back into range after precision was already lost mid-accumulation. + if (!Number.isSafeInteger(total)) { + throw new Error( + `signedDeltaPaise: intermediate total ${total} exceeds the safe integer range`, + ); + } } return total; } diff --git a/lib/cron/with-cron-lock.ts b/lib/cron/with-cron-lock.ts index 4b20eae6f..896f72fd7 100644 --- a/lib/cron/with-cron-lock.ts +++ b/lib/cron/with-cron-lock.ts @@ -138,14 +138,19 @@ export async function withCronLock( const token = await acquireLock(key, opts.ttlMs ?? DEFAULT_TTL_MS); if (!token) { - // acquireLock returns null for BOTH "held" and "circuit open" — the - // breaker's fallback short-circuits to null without touching Redis. The - // health check above ran BEFORE the acquire, so a breaker that tripped - // in between (or was already open from unrelated Redis ops on this warm - // instance) used to be misreported as a clean "held" skip: exit 0, no - // page, money job silently frozen for the breaker's reset window. - if (opts.failMode === "closed" && isRedisCircuitOpen()) { - throw new CronLockUnavailableError(jobName); + // acquireLock returns null for BOTH "held" and "Redis trouble": the + // breaker fallback short-circuits to null while OPEN, and during the + // FIRST FOUR consecutive failures the breaker is still CLOSED while every + // acquire already fails. The pre-acquire health check cannot see either + // window (it ran earlier, and it deliberately bypasses the breaker). So + // on a null token for a fail-closed job we probe Redis AGAIN, right now: + // unhealthy ⇒ page (CronLockUnavailableError); reachable ⇒ someone really + // holds the lock (clean CronLockHeldError skip). + if (opts.failMode === "closed") { + const healthyNow = await checkRedisHealth(); + if (!healthyNow || isRedisCircuitOpen()) { + throw new CronLockUnavailableError(jobName); + } } throw new CronLockHeldError(jobName); } diff --git a/lib/payments/core/razorpay.ts b/lib/payments/core/razorpay.ts index d5026ab5b..1f6923f60 100644 --- a/lib/payments/core/razorpay.ts +++ b/lib/payments/core/razorpay.ts @@ -111,16 +111,23 @@ export function withRazorpaySdkTimeout( ), SDK_CALL_TIMEOUT_MS, ); - call().then( - (value) => { - clearTimeout(timer); - resolve(value); - }, - (err) => { - clearTimeout(timer); - reject(err); - }, - ); + try { + call().then( + (value) => { + clearTimeout(timer); + resolve(value); + }, + (err) => { + clearTimeout(timer); + reject(err); + }, + ); + } catch (err) { + // call() threw SYNCHRONOUSLY — .then never attached, so the timer + // would linger for the full window. Clear and propagate. + clearTimeout(timer); + throw err; + } }); } diff --git a/lib/payments/operations/booking-refund.ts b/lib/payments/operations/booking-refund.ts index 18f6eabd0..cfa623aed 100644 --- a/lib/payments/operations/booking-refund.ts +++ b/lib/payments/operations/booking-refund.ts @@ -461,6 +461,16 @@ async function reverseFreeCreditSettlement( const debitTotal = debits.reduce((s, d) => s + d.amountPaise, 0); const creditTotal = credits.reduce((s, c) => s + c.amountPaise, 0); + // #1218-triage — a fully-zero settlement (zero-share earnings rows only) + // produces empty postings on BOTH sides: that is a legitimate nothing-to- + // journal outcome, not an imbalance. Only a one-sided non-empty total is + // a real logic bug worth rolling back for. + if (debits.length === 0 && credits.length === 0) { + console.log( + `ℹ️ free-credit reversal for payment ${input.paymentId}: zero-value settlement — no journal posted`, + ); + return; + } if (debitTotal === 0 || debitTotal !== creditTotal) { // Unreachable given the plug — a mismatch is a real logic bug. Throwing // rolls back the whole Serializable tx (Refund row included) for a clean diff --git a/lib/payments/operations/checkout.ts b/lib/payments/operations/checkout.ts index f00eb18ba..f2c96e5ec 100644 --- a/lib/payments/operations/checkout.ts +++ b/lib/payments/operations/checkout.ts @@ -193,6 +193,20 @@ function buildPaymentMetadata( * observes the winner's COMMITTED row; same-key duplicates stay covered by * the clientIdempotencyKey unique (#828 P2002 replay). */ +/** A resumable open order — the fields the resume response needs. */ +interface ReusableOrder { + id: string; + paymentIntent: string; + amount: number; + currency: string; + isMockPayment: boolean; + /** First slot of the booked window (consultation/class shape); empty for + * subscription placeholders whose period lives on the slot rows too. */ + appointment?: { + slotsOfAppointment: Array<{ startsAt: Date; endsAt: Date }>; + } | null; +} + export async function findReusablePendingOrderPayment( db: Pick, params: { @@ -202,8 +216,26 @@ export async function findReusablePendingOrderPayment( eventId?: string; organizationId: string | null; paymentGateway: PaymentGateway; + /** + * #1220-triage — the CURRENT request's computed total. A candidate whose + * frozen amount differs (changed coupon, credit balance moved) is + * superseded instead of resumed, so a stale price can never be charged. + */ + expectedAmountPaise: number; + /** Slot window for direct bookings — a resume must be for THIS time, + * not just this plan (#1220-triage critical finding). */ + slotWindow?: { startsAt: Date; endsAt: Date }; + /** Subscription billing-period window; both-null rows only match a + * both-null request. */ + schedulingPeriod?: { startsAt: Date; endsAt: Date } | null; }, -) { +): Promise<{ + reusable: ReusableOrder | null; + /** Scope/freshness matches rejected by the window/amount gates — the caller + * EXPIRES these ("superseded") so they can neither be resumed nor re-minted + * into a parallel charge. */ + supersede: Array<{ id: string; reason: string }>; +}> { // Plan identity per type — events are identified by their event row (the // plan is 1:1 with it); direct bookings by their plan id. An undefined // eventId would make Prisma drop the filter entirely (matches ANY @@ -225,9 +257,9 @@ export async function findReusablePendingOrderPayment( default: planScope = null; } - if (!planScope) return null; + if (!planScope) return { reusable: null, supersede: [] }; - return db.payment.findFirst({ + const candidates = await db.payment.findMany({ where: { userId: params.userId, paymentStatus: PaymentStatus.PENDING, @@ -240,14 +272,86 @@ export async function findReusablePendingOrderPayment( appointment: planScope, }, orderBy: { createdAt: "desc" }, + take: 5, // bounded: newest attempts first; older ones get superseded below select: { id: true, paymentIntent: true, amount: true, currency: true, isMockPayment: true, + appointment: { + select: { + slotsOfAppointment: { + select: { startsAt: true, endsAt: true }, + orderBy: { startsAt: "asc" as const }, + take: 1, + }, + }, + }, }, }); + + const reusable: ReusableOrder[] = []; + const supersede: Array<{ id: string; reason: string }> = []; + + for (const candidate of candidates) { + const appt = candidate.appointment as + | { slotsOfAppointment: Array<{ startsAt: Date; endsAt: Date }> } + | null; + + // Gate 1 — slot window (#1220-triage Critical): a second checkout for a + // DIFFERENT appointment time must never resume the first attempt's order. + if ( + params.appointmentType === "CONSULTATION" + ) { + const slot = appt?.slotsOfAppointment?.[0]; + if (!params.slotWindow || !slot) { + supersede.push({ id: candidate.id, reason: "window-unmatchable" }); + continue; + } + if ( + slot.startsAt.getTime() !== params.slotWindow.startsAt.getTime() || + slot.endsAt.getTime() !== params.slotWindow.endsAt.getTime() + ) { + supersede.push({ id: candidate.id, reason: "slot-window-mismatch" }); + continue; + } + } + if (params.appointmentType === "SUBSCRIPTION") { + const reqPeriod = params.schedulingPeriod ?? null; + // Subscription windows ride the SAME slot rows as consultations — the + // minted placeholder's slot carries the scheduling-period bounds. + const subSlot = appt?.slotsOfAppointment?.[0]; + const rowPeriod = + subSlot ? { startsAt: subSlot.startsAt, endsAt: subSlot.endsAt } : null; + if (!!reqPeriod !== !!rowPeriod) { + supersede.push({ id: candidate.id, reason: "period-mismatch" }); + continue; + } + if ( + reqPeriod && + rowPeriod && + (rowPeriod.startsAt.getTime() !== reqPeriod.startsAt.getTime() || + rowPeriod.endsAt.getTime() !== reqPeriod.endsAt.getTime()) + ) { + supersede.push({ id: candidate.id, reason: "period-mismatch" }); + continue; + } + } + + // Gate 2 — priced-input parity (#1220-triage Major): a changed coupon / + // credit balance / tax profile computes a DIFFERENT total for this + // request; resuming would charge the stale number. Expire-and-fresh. + if (candidate.amount !== params.expectedAmountPaise) { + supersede.push({ id: candidate.id, reason: "amount-mismatch" }); + continue; + } + + reusable.push(candidate); + break; // newest match wins + } + + return { reusable: reusable[0] ?? null, supersede }; } // ============================================================================ @@ -2368,21 +2472,72 @@ export async function handleCheckout( // a remount/new tab converges on the first attempt's order instead of // charging in parallel. Mock/zero-amount/org-sponsored flows are never // PENDING, so the lookup can only ever match a real gateway hold. - const reusableOrder = await findReusablePendingOrderPayment(prisma, { - userId, - appointmentType, - planId: validatedData.planId, - eventId: validatedData.eventId, - organizationId, - paymentGateway: validatedData.paymentGateway, - }); + // + // #1220-triage — candidates are additionally gated on the slot window + // (a different appointment time must never resume) and on priced-input + // parity (amount must equal THIS request's computation). Rejections are + // superseded to EXPIRED so they can neither be resumed later nor re-minted + // into a parallel charge by a third tab. + const { reusable: reusableOrder, supersede: supersededOrders } = + await findReusablePendingOrderPayment(prisma, { + userId, + appointmentType, + planId: validatedData.planId, + eventId: validatedData.eventId, + organizationId, + paymentGateway: validatedData.paymentGateway, + expectedAmountPaise: amount, + ...(appointmentType === "CONSULTATION" && + validatedData.startsAt && + validatedData.endsAt + ? { + slotWindow: { + startsAt: new Date(validatedData.startsAt), + endsAt: new Date(validatedData.endsAt), + }, + } + : {}), + ...(appointmentType === "SUBSCRIPTION" + ? { + schedulingPeriod: + validatedData.schedulingPeriodStartsAt && + validatedData.schedulingPeriodEndsAt + ? { + startsAt: new Date( + validatedData.schedulingPeriodStartsAt, + ), + endsAt: new Date(validatedData.schedulingPeriodEndsAt), + } + : null, + } + : {}), + }); + if (supersededOrders.length > 0) { + await prisma.payment.updateMany({ + where: { id: { in: supersededOrders.map((s) => s.id) } }, + data: { + paymentStatus: PaymentStatus.EXPIRED, + expiresAt: new Date(), + }, + }); + console.log( + JSON.stringify({ + event: "checkout_open_order_superseded", + appointmentType, + count: supersededOrders.length, + reason: supersededOrders[0].reason, + timestamp: new Date().toISOString(), + }), + ); + } if (reusableOrder) { console.log( JSON.stringify({ event: "checkout_open_order_reused", appointmentType, orderId: reusableOrder.paymentIntent, - userId, + // #1220-triage — payment row id, not raw userId (no PII pairing in logs). + paymentRowId: reusableOrder.id, timestamp: new Date().toISOString(), }), ); diff --git a/lib/payments/operations/refund.ts b/lib/payments/operations/refund.ts index c9e085afe..8bc2b008e 100644 --- a/lib/payments/operations/refund.ts +++ b/lib/payments/operations/refund.ts @@ -504,6 +504,31 @@ export async function refundPayment(input: RefundInput): Promise { // Retire our placeholder: it is a pure reservation (cascadedAt null, no // legs reference it), and leaving it PENDING would double-count against // the refundable balance until the reconciler failed it at 24h. + // #1205-triage — carry Phase 1's audit keys onto the surviving row: + // the webhook that minted it had no knowledge of initiatedByUserId/ + // source, and without this merge the adopt path is the one bind path + // that loses them. + const winnerFull = await prisma.refund.findUniqueOrThrow({ + where: { id: winner.id }, + select: { metadata: true }, + }); + await prisma.refund.update({ + where: { id: winner.id }, + data: { + metadata: { + ...(winnerFull.metadata && + typeof winnerFull.metadata === "object" && + !Array.isArray(winnerFull.metadata) + ? winnerFull.metadata + : {}), + ...(reserved.metadata && + typeof reserved.metadata === "object" && + !Array.isArray(reserved.metadata) + ? reserved.metadata + : {}), + } as Prisma.InputJsonValue, + }, + }); await prisma.refund.delete({ where: { id: reserved.id } }); boundRefundRowId = winner.id; reportSentryMessage( diff --git a/lib/payments/payouts/org-payout-service.ts b/lib/payments/payouts/org-payout-service.ts index a31fbfaf3..d032ddb25 100644 --- a/lib/payments/payouts/org-payout-service.ts +++ b/lib/payments/payouts/org-payout-service.ts @@ -551,8 +551,10 @@ export async function processOrgPayout( } // #1020 — a payout whose earnings sit on a disputed payment must not - // leave the building. Checked INSIDE this Serializable tx (free with - // the isolation we're already paying for); returning unclaimed keeps + // leave the building. Checked inside the claim tx (READ COMMITTED — + // race-safety comes from the CAS claim below per ADR 13, and the + // residual window to gateway submit is backstopped by the LOST + // clawback); returning unclaimed keeps // the row PENDING so a later cron run advances it once the dispute // resolves. Residual window to gateway submit is backstopped by the // LOST-handler clawback (#1020-2). @@ -971,8 +973,19 @@ async function redriveStaleProcessingOrgPayouts(): Promise break; } } catch (err) { - reportSentryError(err, { subsystem: "payments" }); + // #1205-triage — PERMANENT rejections must terminate the row, not sit + // in PROCESSING forever: the redrive would retry a data rejection + // every hour indefinitely while its BATCHED earnings stay locked. const message = err instanceof Error ? err.message : String(err); + if ( + classifyGatewaySubmissionError(err) === "PERMANENT_4XX" || + err instanceof PayoutValidationError + ) { + await markOrgPayoutFailed(p.id, `redrive rejected: ${message}`); + result.advanced++; + continue; + } + reportSentryError(err, { subsystem: "payments" }); result.errors.push(`OrgPayout redrive ${p.id}: ${message}`); } } diff --git a/lib/payments/wallet-freeze.ts b/lib/payments/wallet-freeze.ts index 668557f46..363d82132 100644 --- a/lib/payments/wallet-freeze.ts +++ b/lib/payments/wallet-freeze.ts @@ -15,13 +15,19 @@ * drift is fixed. The check is a single indexed DB read inside the caller's tx, * so it fails CLOSED — an unreadable log blocks the spend rather than leaking it. * + * #1205-triage — these writes are ERROR-PROPAGATING, unlike recordSystemEvent + * (which swallows insert failures for operator surfaces). The freeze/unfreeze + * pair IS the kill-switch state: a silently-dropped FREEZE leaves spend open; + * a dropped UNFREEZE told the admin the account was released when it wasn't. + * * Deliberately gates only discretionary SPEND (the checkout booking debit), not * chargeback recovery (app/api/webhooks/utils.ts): a lost-dispute debit recovers * money the bank already pulled and must not be stranded on a drift freeze. */ -import prisma, { type PrismaLike } from "@/lib/prisma"; -import { recordSystemEvent } from "@/lib/enterprise/system-events"; +import { Prisma } from "@prisma/client"; + +import prisma from "@/lib/prisma"; const FREEZE_CATEGORY = "WALLET_FREEZE"; const UNFREEZE_CATEGORY = "WALLET_UNFREEZE"; @@ -40,7 +46,7 @@ export class WalletFrozenError extends Error { /** True when the account's latest freeze event is a FREEZE. Reads inside the * caller's tx/client so the check shares the spend's snapshot. */ export async function isWalletFrozen( - db: PrismaLike, + db: Pick, billingAccountId: string, ): Promise { const latest = await db.systemEvent.findFirst({ @@ -54,22 +60,30 @@ export async function isWalletFrozen( return latest?.category === FREEZE_CATEGORY; } -/** Freeze wallet spend for one account. Idempotent — no-op if already frozen. - * Returns true when it wrote a new freeze event. */ -export async function freezeWalletSpend(params: { +interface FreezeWriteParams { billingAccountId: string; organizationId?: string | null; - reason: string; -}): Promise { - if (await isWalletFrozen(prisma, params.billingAccountId)) return false; - await recordSystemEvent({ +} + +function freezeEventData(params: FreezeWriteParams & { reason: string }) { + return { organizationId: params.organizationId ?? null, category: FREEZE_CATEGORY, - severity: "ERROR", + severity: "ERROR" as const, message: `Wallet spend FROZEN for billing account ${params.billingAccountId}: ${params.reason}`, - context: { billingAccountId: params.billingAccountId, reason: params.reason }, + context: { billingAccountId: params.billingAccountId, reason: params.reason } as Prisma.InputJsonObject, correlationId: freezeKey(params.billingAccountId), - }); + }; +} + +/** Freeze wallet spend for one account. Idempotent — no-op if already frozen. + * Write failures PROPAGATE (see #1205-triage note above); the reconcile job's + * own error handling pages on them. Returns true when it wrote a new event. */ +export async function freezeWalletSpend(params: FreezeWriteParams & { + reason: string; +}): Promise { + if (await isWalletFrozen(prisma, params.billingAccountId)) return false; + await prisma.systemEvent.create({ data: freezeEventData(params) }); return true; } @@ -77,26 +91,49 @@ export async function freezeWalletSpend(params: { * above documents. Until this writer existed, a freeze was unrecoverable * except by hand-inserting a SystemEvent row via raw SQL: the fail-closed * read blocked every checkout for the org forever. Admin-gated route at - * app/api/admin/billing-accounts/[billingAccountId]/unfreeze. Idempotent — - * no-op when not frozen. Returns true when it wrote an unfreeze event. */ -export async function unfreezeWalletSpend(params: { - billingAccountId: string; - organizationId?: string | null; + * app/api/admin/billing-accounts/[billingAccountId]/unfreeze. + * + * #1205-triage — atomic under Serializable: the frozen-check and the + * UNFREEZE write share one tx, so a reconciler re-freezing concurrently + * (fresh drift discovered between our read and write) aborts us instead of + * being silently cleared by a later-dated UNFREEZE. Returns false (409 at + * the route) when not frozen or when the concurrent-write abort fires. + * Write failures propagate. */ +export async function unfreezeWalletSpend(params: FreezeWriteParams & { actorUserId: string; reason: string; }): Promise { - if (!(await isWalletFrozen(prisma, params.billingAccountId))) return false; - await recordSystemEvent({ - organizationId: params.organizationId ?? null, - category: UNFREEZE_CATEGORY, - severity: "INFO", - message: `Wallet spend UNFROZEN for billing account ${params.billingAccountId} by ${params.actorUserId}: ${params.reason}`, - context: { - billingAccountId: params.billingAccountId, - reason: params.reason, - unfrozenBy: params.actorUserId, - }, - correlationId: freezeKey(params.billingAccountId), - }); - return true; + try { + return await prisma.$transaction( + async (tx) => { + if (!(await isWalletFrozen(tx, params.billingAccountId))) return false; + await tx.systemEvent.create({ + data: { + organizationId: params.organizationId ?? null, + category: UNFREEZE_CATEGORY, + severity: "INFO", + message: `Wallet spend UNFROZEN for billing account ${params.billingAccountId} by ${params.actorUserId}: ${params.reason}`, + context: { + billingAccountId: params.billingAccountId, + reason: params.reason, + unfrozenBy: params.actorUserId, + } as Prisma.InputJsonObject, + correlationId: freezeKey(params.billingAccountId), + }, + }); + return true; + }, + { isolationLevel: Prisma.TransactionIsolationLevel.Serializable }, + ); + } catch (err) { + // SSI abort = a concurrent freeze/unfreeze interleaved with us. Surface + // as not-applied rather than throwing out of an admin route. + if ( + err instanceof Prisma.PrismaClientKnownRequestError && + err.code === "P2034" + ) { + return false; + } + throw err; + } } diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 0688c342a..e154406c4 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -4984,6 +4984,11 @@ model WebhookEvent { processedAt DateTime? error String? receivedAt DateTime @default(now()) + // #1205-triage — sweeper claim stamp, SEPARATE from receivedAt: bumping + // receivedAt on every re-drive used to reset the event's age and let a + // permanently-failing row dodge the give-up cap forever. The stale selector + // reads (receivedAt, claimedAt) together; giveUp aging stays on receivedAt. + claimedAt DateTime? @@index([provider]) @@index([processed]) diff --git a/prisma/sql/one-off/2026-08-21-repair-ready-earnings-welded-to-payouts.sql b/prisma/sql/one-off/2026-08-21-repair-ready-earnings-welded-to-payouts.sql index 0c3ce9763..c8850cbd3 100644 --- a/prisma/sql/one-off/2026-08-21-repair-ready-earnings-welded-to-payouts.sql +++ b/prisma/sql/one-off/2026-08-21-repair-ready-earnings-welded-to-payouts.sql @@ -16,6 +16,13 @@ -- create-payout-batch.ts now writes BATCHED itself; this script repairs the -- historical rows. Run once per environment AFTER deploying the code fix. -- +-- ⚠️ OPS GATING (#1205 review): run with payout WRITERS stopped — pause the +-- process-payouts workflow and hold the admin approve/process routes for the +-- duration. The three UPDATEs below are not one locked payout-state snapshot: +-- a payout completing between statement 1 and 3 would leave its earnings +-- READY while its new COMPLETED status expects PAID, and the completion +-- handler only promotes BATCHED rows. +-- -- Repair semantics (mirroring what the canonical paths would have done): -- * payout COMPLETED → earnings PAID, paidAt = payout processedAt -- (fallback updatedAt), idempotent guard on diff --git a/prisma/sql/payment-legs-triggers.sql b/prisma/sql/payment-legs-triggers.sql index 3e1bd7e47..eb421a0a8 100644 --- a/prisma/sql/payment-legs-triggers.sql +++ b/prisma/sql/payment-legs-triggers.sql @@ -19,56 +19,67 @@ -- Postgres constraint-trigger rules; each fired row re-checks its whole -- payment, which is idempotent when several rows commit together. -CREATE OR REPLACE FUNCTION assert_payment_legs_sum_to_amount() RETURNS trigger AS $$ +CREATE OR REPLACE FUNCTION assert_payment_legs_ok(p_payment_id TEXT) RETURNS void AS $$ DECLARE - v_payment_id TEXT; v_amount BIGINT; v_funding_sum BIGINT; v_sibling_sum BIGINT; r RECORD; BEGIN - v_payment_id := COALESCE(NEW.paymentId, OLD.paymentId); - - SELECT "amount" INTO v_amount FROM "Payment" WHERE "id" = v_payment_id; + SELECT "amount" INTO v_amount FROM "Payment" WHERE "id" = p_payment_id; IF NOT FOUND THEN - RETURN NULL; -- payment already gone (cascade delete) — nothing to guard + RETURN; -- payment already gone (cascade delete) — nothing to guard END IF; SELECT COALESCE(SUM("amountPaise"), 0) INTO v_funding_sum FROM "PaymentLeg" - WHERE "paymentId" = v_payment_id + WHERE "paymentId" = p_payment_id AND RIGHT("source"::text, 9) <> '_REVERSAL'; IF v_funding_sum <> v_amount THEN RAISE EXCEPTION 'payment_legs_sum_to_amount violated for payment %: legs sum to % but Payment.amount is %', - v_payment_id, v_funding_sum, v_amount + p_payment_id, v_funding_sum, v_amount USING ERRCODE = 'check_violation'; END IF; FOR r IN SELECT "source", "amountPaise" FROM "PaymentLeg" - WHERE "paymentId" = v_payment_id + WHERE "paymentId" = p_payment_id AND RIGHT("source"::text, 9) = '_REVERSAL' LOOP IF r."amountPaise" >= 0 THEN RAISE EXCEPTION 'payment_legs_reversal_pair violated for payment %: reversal leg % carries non-negative %', - v_payment_id, r."source", r."amountPaise" + p_payment_id, r."source", r."amountPaise" USING ERRCODE = 'check_violation'; END IF; SELECT COALESCE(SUM("amountPaise"), 0) INTO v_sibling_sum FROM "PaymentLeg" - WHERE "paymentId" = v_payment_id + WHERE "paymentId" = p_payment_id AND "source"::text = LEFT(r."source"::text, LENGTH(r."source"::text) - 9); IF -r."amountPaise" > v_sibling_sum THEN RAISE EXCEPTION 'payment_legs_reversal_pair violated for payment %: reversal % (%) exceeds original sibling sum %', - v_payment_id, r."source", -r."amountPaise", v_sibling_sum + p_payment_id, r."source", -r."amountPaise", v_sibling_sum USING ERRCODE = 'check_violation'; END IF; END LOOP; - +END; +$$ LANGUAGE plpgsql; +-- SPLIT +-- #1205-triage — a leg RE-PARENTING must validate BOTH payments: moving one +-- leg from payment A to B can leave A under-funded while B validates clean. +CREATE OR REPLACE FUNCTION assert_payment_legs_on_leg_write() RETURNS trigger AS $$ +BEGIN + IF TG_OP = 'DELETE' THEN + PERFORM assert_payment_legs_ok(OLD.paymentId); + ELSIF TG_OP = 'UPDATE' AND NEW.paymentId IS DISTINCT FROM OLD.paymentId THEN + PERFORM assert_payment_legs_ok(OLD.paymentId); + PERFORM assert_payment_legs_ok(NEW.paymentId); + ELSE + PERFORM assert_payment_legs_ok(NEW.paymentId); + END IF; RETURN NULL; END; $$ LANGUAGE plpgsql; @@ -79,4 +90,23 @@ CREATE CONSTRAINT TRIGGER payment_legs_sum_to_amount AFTER INSERT OR UPDATE OR DELETE ON "PaymentLeg" DEFERRABLE INITIALLY DEFERRED FOR EACH ROW - EXECUTE FUNCTION assert_payment_legs_sum_to_amount(); + EXECUTE FUNCTION assert_payment_legs_on_leg_write(); +-- SPLIT +-- #1205-triage — a DIRECT Payment.amount UPDATE escapes the leg-side trigger +-- entirely (it only fires on PaymentLeg writes). Guard the parent too. +-- SPLIT +-- Parent-side validator: same invariant from a Payment.amount write. +CREATE OR REPLACE FUNCTION assert_payment_legs_on_payment_update() RETURNS trigger AS $$ +BEGIN + PERFORM assert_payment_legs_ok(NEW.id); + RETURN NULL; +END; +$$ LANGUAGE plpgsql; +-- SPLIT +DROP TRIGGER IF EXISTS payment_amount_vs_legs ON "Payment"; +-- SPLIT +CREATE CONSTRAINT TRIGGER payment_amount_vs_legs + AFTER UPDATE OF "amount" ON "Payment" + DEFERRABLE INITIALLY DEFERRED + FOR EACH ROW + EXECUTE FUNCTION assert_payment_legs_on_payment_update(); diff --git a/scripts/cleanup/sweep-stuck-webhook-events.ts b/scripts/cleanup/sweep-stuck-webhook-events.ts index 95aa79342..63ea9629f 100644 --- a/scripts/cleanup/sweep-stuck-webhook-events.ts +++ b/scripts/cleanup/sweep-stuck-webhook-events.ts @@ -125,6 +125,14 @@ async function sweepStuckWebhookEventsUnlocked( // no redelivery to rescue it. This sweep is the only thing that will. provider: { in: ["razorpay", "stream"] }, receivedAt: { lt: staleBefore }, + // #1205-triage — claim freshness rides the dedicated claimedAt column; + // receivedAt stays untouched so give-up aging cannot be reset by our + // own re-drives. + AND: [ + { + OR: [{ claimedAt: null }, { claimedAt: { lt: staleBefore } }], + }, + ], OR: [ // Crashed before recording anything. { processed: false, error: null }, @@ -170,8 +178,11 @@ async function sweepStuckWebhookEventsUnlocked( // 30s per call, so "still alive past the staleness window" is rare — // but a claim makes sweep-vs-sweep double-drive impossible outright. const claimed = await prisma.webhookEvent.updateMany({ - where: { eventId: ev.eventId, receivedAt: ev.receivedAt }, - data: { receivedAt: new Date() }, + where: { + eventId: ev.eventId, + OR: [{ claimedAt: null }, { claimedAt: ev.claimedAt }], + }, + data: { claimedAt: new Date() }, }); if (claimed.count === 0) { console.log( diff --git a/scripts/payouts/handle-stuck-payouts.ts b/scripts/payouts/handle-stuck-payouts.ts index 745eef917..feacbdb9b 100644 --- a/scripts/payouts/handle-stuck-payouts.ts +++ b/scripts/payouts/handle-stuck-payouts.ts @@ -252,26 +252,35 @@ async function handleStuckPayoutsUnlocked(): Promise { console.log(` No provider payout ID - marking as FAILED`); if (payout.retryCount >= MAX_RETRIES) { - await prisma.consultantPayout.update({ - where: { id: payout.id }, - data: { - status: PayoutStatus.FAILED, - failureReason: - "Payout never sent to gateway after multiple attempts", - }, - }); - // Release the earnings like the webhook FAILED path does — without - // this, BATCHED earnings stayed welded to a permanently-FAILED - // payout: excluded from future batches by `payoutId: null`, never - // released, money held hostage with no actor. - const released = await prisma.consultantEarnings.updateMany({ - where: { payoutId: payout.id, status: EarningStatus.BATCHED }, - data: { payoutId: null, status: EarningStatus.READY }, + // #1205-triage — CAS the terminal flip inside the same tx as the + // earnings release: without the PROCESSING guard, a concurrently + // completing gateway webhook could be overwritten by this FAILED. + const cas = await prisma.$transaction(async (tx) => { + const claimed = await tx.consultantPayout.updateMany({ + where: { id: payout.id, status: PayoutStatus.PROCESSING }, + data: { + status: PayoutStatus.FAILED, + failureReason: + "Payout never sent to gateway after multiple attempts", + }, + }); + if (claimed.count === 0) return { released: 0, claimed: false }; + const released = await tx.consultantEarnings.updateMany({ + where: { payoutId: payout.id, status: EarningStatus.BATCHED }, + data: { payoutId: null, status: EarningStatus.READY }, + }); + return { released: released.count, claimed: true }; }); failedCount++; - console.log( - ` Marked as permanently FAILED (max retries reached); released ${released.count} earning(s)`, - ); + if (!cas.claimed) { + console.log( + ` Skipped — payout left PROCESSING concurrently (webhook won)`, + ); + } else { + console.log( + ` Marked as permanently FAILED (max retries reached); released ${cas.released} earning(s)`, + ); + } } else { // Reset to APPROVED for retry await prisma.consultantPayout.update({ diff --git a/scripts/refunds/reconcile-pending-refunds.ts b/scripts/refunds/reconcile-pending-refunds.ts index 4e2d657c3..7620f703d 100644 --- a/scripts/refunds/reconcile-pending-refunds.ts +++ b/scripts/refunds/reconcile-pending-refunds.ts @@ -145,6 +145,7 @@ async function reconcilePendingRefundsUnlocked(): Promise/credits_ synthetic ids + // alongside the pending_ placeholders — none exist at any gateway. + AND: SYNTHETIC_PREFIXES.map((prefix) => ({ + refundId: { not: { startsWith: prefix } }, + })), createdAt: { lt: thresholdDate }, }, include: { payment: { select: { paymentGateway: true } } }, @@ -307,10 +313,11 @@ async function reconcilePendingRefundsUnlocked(): Promise, ): Promise<"bound" | "superseded"> { const nextStatus = mapGatewayRefundStatus(gatewayRefund.status); const mergedMetadata = { - ...(prismaMetadataObject(await readRefundMetadata(placeholderRowId))), + ...existingMetadata, ...(gatewayRefund.metadata ?? {}), reconciled_at: new Date().toISOString(), } as Prisma.InputJsonValue; @@ -354,14 +361,6 @@ async function bindGatewayRefundToPlaceholder( return "superseded"; } -async function readRefundMetadata(refundRowId: string): Promise { - const row = await prisma.refund.findUnique({ - where: { id: refundRowId }, - select: { metadata: true }, - }); - return row?.metadata; -} - function prismaMetadataObject(metadata: unknown): Record { return metadata && typeof metadata === "object" && !Array.isArray(metadata) ? (metadata as Record)