intake: lock the purchase, not the batch, and hold on both write paths
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.
This commit is contained in:
@@ -20,3 +20,33 @@ export async function queryRow<T>(sql: string, params: unknown[] = []): Promise<
|
||||
const rows = await prisma.$queryRawUnsafe<T[]>(sql, ...params);
|
||||
return rows.length > 0 ? rows[0] : null;
|
||||
}
|
||||
|
||||
/** The same two calls, scoped to one connection inside one transaction. */
|
||||
export type Queryable = {
|
||||
queryRaw<T>(sql: string, params?: unknown[]): Promise<T[]>;
|
||||
queryRow<T>(sql: string, params?: unknown[]): Promise<T | null>;
|
||||
};
|
||||
|
||||
/**
|
||||
* 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<T>(fn: (q: Queryable) => Promise<T>): Promise<T> {
|
||||
return prisma.$transaction(async (tx) => {
|
||||
const q: Queryable = {
|
||||
queryRaw: <R>(sql: string, params: unknown[] = []) => tx.$queryRawUnsafe<R[]>(sql, ...params),
|
||||
queryRow: async <R>(sql: string, params: unknown[] = []) => {
|
||||
const rows = await tx.$queryRawUnsafe<R[]>(sql, ...params);
|
||||
return rows.length > 0 ? rows[0] : null;
|
||||
},
|
||||
};
|
||||
return fn(q);
|
||||
});
|
||||
}
|
||||
|
||||
@@ -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]
|
||||
|
||||
+71
-33
@@ -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<Candidate[]>
|
||||
`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<Candidate[]>
|
||||
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);
|
||||
|
||||
/** Food: enrich the matched order's line_items, or show candidates. NEVER auto-creates. */
|
||||
export async function routeFood(ctx: RouteContext): Promise<IntakeResult> {
|
||||
const { match } = await matchTransaction(ctx.header, ctx.platform);
|
||||
if (match) {
|
||||
if (unverifiable(ctx.items) && match.has_line_items) {
|
||||
/**
|
||||
* 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<IntakeResult | null> {
|
||||
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 });
|
||||
}
|
||||
await replaceLineItems(match.transaction_id, ctx.items, { source: "intake", platform: ctx.platform, batchId: ctx.batchId, metadataId: match.metadata_id });
|
||||
}
|
||||
|
||||
/** Food: enrich the matched order's line_items, or show candidates. NEVER auto-creates. */
|
||||
export async function routeFood(ctx: RouteContext): Promise<IntakeResult> {
|
||||
const { match } = await matchTransaction(ctx.header, ctx.platform);
|
||||
if (match) {
|
||||
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,10 +457,33 @@ 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 }>(
|
||||
// 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,
|
||||
@@ -452,23 +492,21 @@ export async function createFromBatch(batch: BatchRow, kind: "grocery" | "food")
|
||||
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,
|
||||
]
|
||||
[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 });
|
||||
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 });
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user