From 4b09d4136f59b8e06339b13c539b6f8721bb3ccf Mon Sep 17 00:00:00 2001 From: siddharthd Date: Fri, 21 Aug 2026 18:23:51 +1000 Subject: [PATCH] intake: lock the purchase, not the batch, and hold on both write paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two different unresolved batches for one purchase could each create a transaction. ON CONFLICT (source, source_ref) cannot stop it — their source_refs differ — and both taps clear their match and duplicate checks before either inserts. The create now takes a transaction-scoped advisory lock keyed on (platform, merchant, date, total), re-checks inside it, and inserts; the loser reports `duplicate` against the winner's transaction, the same answer the sequential path gives. The lock has to be inside a real transaction, which this app did not have. queryRaw is $queryRawUnsafe on a POOLED client, so every call is its own implicit transaction and pg_advisory_xact_lock taken in a statement of its own is released before the next statement runs — the lock would have looked right and held nothing. withTransaction() in db.ts is the scope; the integration helper mirrors it with BEGIN/COMMIT on one pooled connection. Also: the "hold rather than replace an itemised receipt with an unpriced upload" guard lived only in routeFood, and createFromBatch's re-match walked straight past it — the same upload destroyed the same receipt depending on which button reached it. One function, both callers. --- src/lib/db.ts | 30 +++++++++ src/lib/intake-actions.ts | 2 +- src/lib/intake-router.ts | 126 +++++++++++++++++++++++++------------- 3 files changed, 113 insertions(+), 45 deletions(-) diff --git a/src/lib/db.ts b/src/lib/db.ts index 59359f1..6fc3b3f 100644 --- a/src/lib/db.ts +++ b/src/lib/db.ts @@ -20,3 +20,33 @@ export async function queryRow(sql: string, params: unknown[] = []): Promise< const rows = await prisma.$queryRawUnsafe(sql, ...params); return rows.length > 0 ? rows[0] : null; } + +/** The same two calls, scoped to one connection inside one transaction. */ +export type Queryable = { + queryRaw(sql: string, params?: unknown[]): Promise; + queryRow(sql: string, params?: unknown[]): Promise; +}; + +/** + * Run several statements as one transaction, on one connection. + * + * The module-level `queryRaw`/`queryRow` are `$queryRawUnsafe` on a POOLED + * client, so each call is its own implicit transaction on whichever connection + * the pool hands out. That makes a check-then-write pair non-atomic, and it + * makes `pg_advisory_xact_lock` useless outside this helper — the lock is + * released the instant the statement that took it commits, which is before the + * next statement runs. Anything that needs a lock to still be held on its next + * query has to be in here. + */ +export async function withTransaction(fn: (q: Queryable) => Promise): Promise { + return prisma.$transaction(async (tx) => { + const q: Queryable = { + queryRaw: (sql: string, params: unknown[] = []) => tx.$queryRawUnsafe(sql, ...params), + queryRow: async (sql: string, params: unknown[] = []) => { + const rows = await tx.$queryRawUnsafe(sql, ...params); + return rows.length > 0 ? rows[0] : null; + }, + }; + return fn(q); + }); +} diff --git a/src/lib/intake-actions.ts b/src/lib/intake-actions.ts index e30f9a6..3de2aef 100644 --- a/src/lib/intake-actions.ts +++ b/src/lib/intake-actions.ts @@ -80,7 +80,7 @@ export async function handleIntakeAction( if (actionId === "intake_use_candidate") { if (!Number.isInteger(value.transaction_id)) return note("That button carried no transaction id."); const txnId = value.transaction_id as number; - await replaceLineItems(txnId, ctx.items, { source: "intake", platform: ctx.platform, batchId: batch.id }); + await replaceLineItems(txnId, ctx.items, { source: "intake", platform: ctx.platform, batchId: batch.id, currency: ctx.currency }); await queryRaw( `UPDATE intake_batches SET status = 'enriched', kind = COALESCE(kind, 'food'), transaction_id = $2, resolved_at = now() WHERE id = $1::uuid`, [batch.id, txnId] diff --git a/src/lib/intake-router.ts b/src/lib/intake-router.ts index 5f33c0b..0cbcdca 100644 --- a/src/lib/intake-router.ts +++ b/src/lib/intake-router.ts @@ -1,6 +1,6 @@ import { randomUUID } from "node:crypto"; -import { queryRaw, queryRow } from "@/lib/db"; -import { EXCLUDE_RECONCILED_SOURCE } from "@/lib/analytics-sql"; +import { queryRaw, queryRow, withTransaction } from "@/lib/db"; +import { EXCLUDE_RECONCILED_SOURCE, NATIVE_CURRENCY } from "@/lib/analytics-sql"; import { DEFAULT_OWNER_ID } from "@/lib/order-ingestion"; import { classifyInput, @@ -215,6 +215,8 @@ type MatchRow = { amount: string; merchant: string | null; platform: string | null; + /** What `amount` is denominated in — a candidate row is not always dollars. */ + currency: string; /** The `expense_metadata` row the match resolved through — null when there is none. */ metadata_id: number | null; /** Does that row already carry a receipt? */ @@ -240,9 +242,10 @@ export async function matchTransaction(header: OrderHeader, platform: string | n `SELECT DISTINCT ON (t.id) t.id AS transaction_id, t.description, t.transaction_date::text, t.amount::text, COALESCE(em.merchant_normalized, t.merchant_normalized, t.merchant_name, t.description) AS merchant, - em.platform, em.id AS metadata_id, + em.platform, em.id AS metadata_id, ${NATIVE_CURRENCY} AS currency, COALESCE(jsonb_typeof(em.line_items) = 'array' AND jsonb_array_length(em.line_items) > 0, false) AS has_line_items FROM transactions t + LEFT JOIN statements s ON s.id = t.statement_id LEFT JOIN expense_metadata em ON em.transaction_id = t.id OR em.matched_transaction_id = t.id WHERE ${EXCLUDE_RECONCILED_SOURCE} AND t.transaction_date = $1::date @@ -271,8 +274,9 @@ export async function findCandidates(header: OrderHeader): Promise `SELECT DISTINCT ON (t.id) t.id AS transaction_id, t.description, t.transaction_date::text, t.amount::text, COALESCE(em.merchant_normalized, t.merchant_normalized, t.merchant_name, t.description) AS merchant, - em.platform, em.id AS metadata_id, false AS has_line_items + em.platform, em.id AS metadata_id, ${NATIVE_CURRENCY} AS currency, false AS has_line_items FROM transactions t + LEFT JOIN statements s ON s.id = t.statement_id LEFT JOIN expense_metadata em ON em.transaction_id = t.id OR em.matched_transaction_id = t.id WHERE ${EXCLUDE_RECONCILED_SOURCE} AND t.transaction_date BETWEEN $1::date - 3 AND $1::date + 3 @@ -288,7 +292,7 @@ export async function findCandidates(header: OrderHeader): Promise return Math.abs(Number(a.amount) - header.total) - Math.abs(Number(b.amount) - header.total); }) .slice(0, 5) - .map((r) => ({ transaction_id: r.transaction_id, description: r.description, transaction_date: r.transaction_date, amount: r.amount })); + .map((r) => ({ transaction_id: r.transaction_id, description: r.description, transaction_date: r.transaction_date, amount: r.amount, currency: r.currency })); } // ─── the two destinations ─────────────────────────────────────────────────── @@ -335,19 +339,30 @@ const finish = (ctx: RouteContext, kind: IntakeKind, status: IntakeStatus, card: */ const unverifiable = (items: StoredLineItem[]) => items.every((i) => i.amount === null); +/** + * The hold, shared by EVERY path that enriches a matched transaction. + * + * It lived only in `routeFood` and `createFromBatch`'s re-match walked straight + * past it — the same upload destroyed the same receipt depending on which + * button reached it. Returns null when there is nothing to protect. + */ +async function holdRatherThanReplace(ctx: RouteContext, match: MatchRow): Promise { + if (!unverifiable(ctx.items) || !match.has_line_items) return null; + await updateBatch(ctx.batchId, { status: "held", items: ctx.items }); + const card = heldCard( + "this upload carries no prices, so nothing checked it against the total — and that order already has its items", + "I left the existing items alone" + ); + return finish(ctx, "held", "held", card, { transaction_id: match.transaction_id }); +} + /** Food: enrich the matched order's line_items, or show candidates. NEVER auto-creates. */ export async function routeFood(ctx: RouteContext): Promise { const { match } = await matchTransaction(ctx.header, ctx.platform); if (match) { - if (unverifiable(ctx.items) && match.has_line_items) { - await updateBatch(ctx.batchId, { status: "held", items: ctx.items }); - const card = heldCard( - "this upload carries no prices, so nothing checked it against the total — and that order already has its items", - "I left the existing items alone" - ); - return finish(ctx, "held", "held", card, { transaction_id: match.transaction_id }); - } - await replaceLineItems(match.transaction_id, ctx.items, { source: "intake", platform: ctx.platform, batchId: ctx.batchId, metadataId: match.metadata_id }); + const held = await holdRatherThanReplace(ctx, match); + if (held) return held; + await replaceLineItems(match.transaction_id, ctx.items, { source: "intake", platform: ctx.platform, batchId: ctx.batchId, metadataId: match.metadata_id, currency: ctx.currency }); await updateBatch(ctx.batchId, { kind: "food", status: "enriched", transaction_id: match.transaction_id, items: ctx.items }); return finish(ctx, "food", "enriched", enrichedCard({ ...cardInput(ctx), transactionId: match.transaction_id }), { transaction_id: match.transaction_id }); } @@ -406,7 +421,9 @@ export async function createFromBatch(batch: BatchRow, kind: "grocery" | "food") // whole slice exists to avoid. const { match } = await matchTransaction(header, header.platform); if (match) { - await replaceLineItems(match.transaction_id, items, { source: "intake", platform: header.platform, batchId: batch.id, metadataId: match.metadata_id }); + const held = await holdRatherThanReplace(ctx, match); + if (held) return held; + await replaceLineItems(match.transaction_id, items, { source: "intake", platform: header.platform, batchId: batch.id, metadataId: match.metadata_id, currency: header.currency }); await updateBatch(batch.id, { kind, status: "enriched", transaction_id: match.transaction_id, items }); return finish(ctx, kind, "enriched", enrichedCard({ ...cardInput(ctx), transactionId: match.transaction_id }), { transaction_id: match.transaction_id }); } @@ -440,35 +457,56 @@ export async function createFromBatch(batch: BatchRow, kind: "grocery" | "food") // than asserting a conversion — the same shape order-ingestion.ts writes. const currency = header.currency ?? "AUD"; const isAud = currency === "AUD"; - // source/source_ref make a second press of the same card a no-op rather than a - // second row — and ON CONFLICT, not SELECT-then-INSERT, so two presses that - // arrive together cannot both find nothing and both insert. - const inserted = await queryRow<{ id: number }>( - `INSERT INTO transactions ( - transaction_date, description, amount, amount_aud, category, payment_method, - merchant_name, merchant_normalized, transaction_type, owner_id, source, source_ref, - foreign_currency_amount, foreign_currency_code - ) VALUES ($1::date, $2, $3::numeric, $4::numeric, $5, NULL, $6, $6, 'debit', $7, 'intake', $8, $9::numeric, $10) - ON CONFLICT (source, source_ref) WHERE source IS NOT NULL AND source_ref IS NOT NULL - DO NOTHING - RETURNING id`, - [ - header.order_date, - description, - header.total, - isAud ? header.total : null, - category, - merchant, - ownerId, - batch.id, - isAud ? null : header.total, - isAud ? null : currency, - ] - ); - const txnId = - inserted?.id ?? - (await queryRow<{ id: number }>(`SELECT id FROM transactions WHERE source = 'intake' AND source_ref = $1`, [batch.id]))!.id; - await replaceLineItems(txnId, items, { source: "intake", platform: header.platform, batchId: batch.id }); + // Check and insert under one lock, on one connection. + // + // `source_ref = batch.id` makes a second press of THIS card a no-op, and + // ON CONFLICT makes that safe against a double tap. It cannot help two + // DIFFERENT batches for one purchase — their keys differ — and both taps run + // their match and duplicate checks before either inserts, so both can find + // nothing. The lock is keyed on the purchase, not the batch, which is the + // only thing the two share. It has to be transaction-scoped and therefore + // inside withTransaction: taken by a statement of its own it would be + // released before the next one ran (see db.ts). + const purchaseKey = [header.platform ?? "", merchantKey(merchant), header.order_date, header.total.toFixed(2)].join("|"); + const outcome = await withTransaction(async (q) => { + await q.queryRaw(`SELECT pg_advisory_xact_lock(hashtextextended($1, 0))`, [purchaseKey]); + // Mine (a double tap) or another batch's (the race). The ORDER BY is what + // tells them apart when both exist. + const existing = await q.queryRow<{ id: number; source_ref: string }>( + `SELECT id, source_ref FROM transactions + WHERE source = 'intake' AND superseded_by_id IS NULL + AND (source_ref = $1 + OR (transaction_date = $2::date AND amount = $3::numeric + AND lower(merchant_normalized) = lower($4))) + ORDER BY (source_ref = $1) DESC NULLS LAST, id + LIMIT 1`, + [batch.id, header.order_date, header.total, merchant] + ); + if (existing) return { id: existing.id, mine: existing.source_ref === batch.id }; + const row = await q.queryRow<{ id: number }>( + `INSERT INTO transactions ( + transaction_date, description, amount, amount_aud, category, payment_method, + merchant_name, merchant_normalized, transaction_type, owner_id, source, source_ref, + foreign_currency_amount, foreign_currency_code + ) VALUES ($1::date, $2, $3::numeric, $4::numeric, $5, NULL, $6, $6, 'debit', $7, 'intake', $8, $9::numeric, $10) + ON CONFLICT (source, source_ref) WHERE source IS NOT NULL AND source_ref IS NOT NULL + DO NOTHING + RETURNING id`, + [header.order_date, description, header.total, isAud ? header.total : null, category, merchant, ownerId, batch.id, isAud ? null : header.total, isAud ? null : currency] + ); + return { id: row!.id, mine: true }; + }); + + // Another batch booked this purchase in the moment between the two taps. Say + // the same thing the sequential case says one second later, and write nothing. + if (!outcome.mine) { + await updateBatch(batch.id, { kind, status: "duplicate", transaction_id: outcome.id }); + const card = duplicateCard({ merchant: header.merchant, total: header.total, currency: header.currency, transactionId: outcome.id, captureId: null, batchId: null }); + return finish(ctx, kind, "duplicate", card, { transaction_id: outcome.id }); + } + + const txnId = outcome.id; + await replaceLineItems(txnId, items, { source: "intake", platform: header.platform, batchId: batch.id, currency }); await updateBatch(batch.id, { kind, status: "created", transaction_id: txnId }); return finish(ctx, kind, "created", createdCard({ ...cardInput(ctx), transactionId: txnId }), { transaction_id: txnId }); }