From a3e4e1102ac32c95b18e0ec9231b51021b95125a Mon Sep 17 00:00:00 2001 From: siddharthd Date: Sat, 3 Oct 2026 16:55:07 +1000 Subject: [PATCH] wallet: lock the ingest, pair on time and app, skip notices, ECB fallback Four defects found reviewing a month of overseas wallet rows: - Race: n8n posts a poll's items concurrently and both halves of a tap usually share a poll, so each ran findPairedRow before the other inserted. 11 taps became 22 rows (A$3,841.40), identical created_at on every pair. One advisory lock now covers re-check -> pair -> adopt -> insert, on the transaction's own connection (a second pooled one per request deadlocks a burst). Mutation-tested: without the lock the concurrent test fails 3/3. - Pair window: amount within +/-1 DAY folded the UNICEF $50 donation and an unrelated $50 Amex charge 17h later into one row and renamed it. A pair is now minutes apart AND from two different apps. Needs when/where a notification arrived: migration 0039 adds expense_metadata.notified_at / notif_app, backfilled from capture.wallet_events (144 rows). A merged half now records wallet_evt:, so a replay cannot merge it twice. - Not purchases: a CommBank overdraft notice became a $21.21 purchase named "To avoid fees and interest," (the Westpac fixed-width parser ran on a CommBank body). Account notices and transit-card balance drawdowns are skipped; the descriptor parser only runs on Westpac. - FX: with no ledger history (Hong Kong) every tap stored A$0.00. The ECB rate via Frankfurter is the fallback, flagged rate_ecb; the ledger median stays first because it carries the card's FX fee. Also: dates in Australia/Melbourne like the statements (UTC slicing filed 28 of 134 rows a day early), and test:setup now actually rebuilds personal_test -- replaying the dump over it never applied a migration. Co-Authored-By: Claude Opus 5.5 (1M context) --- CLAUDE.md | 9 +- .../migration.sql | 26 ++ prisma/schema.prisma | 3 + scripts/setup-test-db.sh | 19 +- .../integration/wallet-ingest.test.ts | 125 ++++++ src/lib/auto-reconcile.ts | 10 +- src/lib/db.ts | 9 +- src/lib/wallet-ingestion.ts | 395 ++++++++++++------ 8 files changed, 450 insertions(+), 146 deletions(-) create mode 100644 prisma/migrations/0039_wallet_notification_time/migration.sql diff --git a/CLAUDE.md b/CLAUDE.md index 69df4bd..bed5e82 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -65,7 +65,7 @@ docker logs finance -f npm test # unit — pure logic, no database npm run test:integration # database-backed; resolves the DSN itself (see below) npm run test:all # both -npm run test:setup # once, and after any migration: creates/refreshes personal_test +npm run test:setup # once, and after any migration: drops and rebuilds personal_test npm run lint # blocking in CI since #79 — 0 errors, warnings allowed npx tsc --noEmit # blocking in CI; nothing else compiles the test files ``` @@ -86,6 +86,13 @@ had pointed the suite at **postgres-pantry** — whose database it would have TRUNCATEd had the credentials happened to match. A loopback address cannot be reassigned to another container, which is why the published port wins. +**`test:setup` really rebuilds now (2026-10-03).** It used to replay the +schema dump over the existing database, where every `CREATE TABLE` failed with +"already exists" and the old shape survived — so no migration ever reached +`personal_test`, and new-column code failed every test with a misleading +`column does not exist`. It now `DROP DATABASE … WITH (FORCE)`s first, refusing +any name not ending `_test`, and loads with `ON_ERROR_STOP`. + **Two guards, and both matter.** `helpers.ts::assertTestDatabase` refuses to truncate anything whose `current_database()` is not `personal_test`, and the suite runs `fileParallelism: false` because every file shares that one database diff --git a/prisma/migrations/0039_wallet_notification_time/migration.sql b/prisma/migrations/0039_wallet_notification_time/migration.sql new file mode 100644 index 0000000..1f64f00 --- /dev/null +++ b/prisma/migrations/0039_wallet_notification_time/migration.sql @@ -0,0 +1,26 @@ +-- When, and from which app, a wallet notification arrived. +-- +-- WHY: one tap emits two notifications 2-8 seconds apart -- the issuer app +-- (card tail) and Apple Wallet (merchant, suburb) -- and wallet-ingestion.ts +-- merges them. It used to recognise the pair by amount alone, within +/-1 DAY, +-- which is wide enough to fold two different purchases together: on 2026-10-01 +-- an Amex "$50.00 at A MEDIUM" notification merged into the UNICEF $50 donation +-- from 17 hours earlier and renamed it. A real pair is seconds apart and comes +-- from two DIFFERENT apps; two notifications from one app are two charges. +-- Neither fact was stored anywhere on the ledger side, so the matcher could not +-- ask. These columns let it. +-- +-- Backfilled from capture.wallet_events through source_message_id ('evt:'), +-- which every wallet-written metadata row carries. + +ALTER TABLE expense_metadata + ADD COLUMN IF NOT EXISTS notified_at timestamptz, + ADD COLUMN IF NOT EXISTS notif_app text; + +UPDATE expense_metadata em + SET notified_at = e.received_at, + notif_app = e.notif_app + FROM capture.wallet_events e + WHERE em.source = 'wallet' + AND em.source_message_id = 'evt:' || e.id + AND em.notified_at IS NULL; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 234a5b0..a9aaed5 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -237,6 +237,9 @@ model expense_metadata { tender_raw String? receipt_sha256 String? receipt_group String? + // 0039 — when/where a wallet notification arrived; the pair merge needs both. + notified_at DateTime? @db.Timestamptz(6) + notif_app String? transaction transactions? @relation(fields: [transaction_id], references: [id], onDelete: Cascade) @@unique([source, order_reference], name: "uq_expense_source_order") diff --git a/scripts/setup-test-db.sh b/scripts/setup-test-db.sh index da47659..95a3260 100755 --- a/scripts/setup-test-db.sh +++ b/scripts/setup-test-db.sh @@ -19,14 +19,23 @@ if ! docker inspect "$PG_CONTAINER" >/dev/null 2>&1; then exit 1 fi -# Create test database (ignore error if it already exists) -docker exec "$PG_CONTAINER" psql -U "$PG_USER" -d postgres \ - -c "CREATE DATABASE $TEST_DB;" 2>/dev/null || true +# Drop and recreate, so the schema is rebuilt rather than replayed. Replaying +# the dump over an existing database was a silent no-op for every migration: +# each CREATE TABLE failed with "already exists" and the old shape survived, so +# a new column (0039's expense_metadata.notified_at, 2026-10-03) never reached +# the test database. The guard is because this now DROPs: it refuses anything +# not named *_test, whatever TEST_DB was set to. +case "$TEST_DB" in + *_test) ;; + *) echo "ERROR: refusing to drop '$TEST_DB' -- the test database must be named *_test"; exit 1 ;; +esac +docker exec "$PG_CONTAINER" psql -U "$PG_USER" -d postgres -q \ + -c "DROP DATABASE IF EXISTS $TEST_DB WITH (FORCE);" \ + -c "CREATE DATABASE $TEST_DB;" -# Wipe and rebuild schema from production (schema only, no data) echo "Copying schema from $PG_USER to $TEST_DB..." docker exec "$PG_CONTAINER" pg_dump -U "$PG_USER" --schema-only "$PG_USER" \ - | docker exec -i "$PG_CONTAINER" psql -U "$PG_USER" -d "$TEST_DB" -q + | docker exec -i "$PG_CONTAINER" psql -U "$PG_USER" -d "$TEST_DB" -q -v ON_ERROR_STOP=1 echo "Schema ready." diff --git a/src/__tests__/integration/wallet-ingest.test.ts b/src/__tests__/integration/wallet-ingest.test.ts index c745845..458b6b6 100644 --- a/src/__tests__/integration/wallet-ingest.test.ts +++ b/src/__tests__/integration/wallet-ingest.test.ts @@ -77,6 +77,10 @@ beforeAll(async () => { }); beforeEach(async () => { + // No network in tests. The ECB fallback treats any fetch failure as "no + // rate", which is what every pre-existing currency test assumes; the one test + // about the fallback stubs a rate of its own. + vi.stubGlobal("fetch", async () => { throw new Error("no network in tests"); }); await resetDB(pool); await pool.query("DELETE FROM expense_metadata"); ({ ownerId } = await seedParticipants(pool)); @@ -147,7 +151,128 @@ describe("the notification pair", () => { }); }); +describe("the notification pair is recognised by time and app, not amount alone", () => { + it("makes ONE row when both halves are posted at the same moment", async () => { + // n8n posts a poll's items concurrently, and both halves of a tap usually + // land in the same poll. Before the lock, each ran its pair lookup before + // the other inserted: 11 taps became 22 rows, identical created_at on each. + await Promise.all([ingest(req(breadworldBank)), ingest(req(breadworldWallet))]); + expect(await txns()).toHaveLength(1); + const [m] = await metas(); + expect(m.card_last4).toBe("8032"); + }); + + it("keeps two same-amount charges hours apart as two purchases", async () => { + // Event 275 ($50 at A MEDIUM) merged into the UNICEF $50 donation from 17 + // hours earlier and renamed it, because the old window was +/-1 day. + await ingest(req({ + ...amexRewardPay, event_id: 267, amount: 50, amount_text: "A$50.00", merchant: "UNICEF", + notif_body: "You have made a A$50.00 charge on your American Express® Card ending in 01000 at UNICEF.", + occurred_at: "2026-10-01T11:02:40+10:00", + })); + await ingest(req({ + ...amexRewardPay, event_id: 275, amount: 50, amount_text: "$50.00", merchant: "A MEDIUM", + notif_body: "You have made a $50.00 charge on your American Express® Card ending in 01000 at A MEDIUM.", + occurred_at: "2026-10-02T04:06:49+10:00", + })); + const rows = await txns(); + expect(rows).toHaveLength(2); + expect(rows.map((r) => r.merchant_normalized).sort()).toEqual(["A MEDIUM", "UNICEF"]); + }); + + it("never pairs two notifications from the SAME app, however close", async () => { + // A pair is the issuer half plus the Wallet half. Two from one app are two + // charges -- two $20 coffees bought together, say. + await ingest(req(breadworldBank)); + await ingest(req({ ...breadworldBank, event_id: 640, occurred_at: "2026-08-13T08:01:40+10:00" })); + expect(await txns()).toHaveLength(2); + }); + + it("records a merged half, so a replay cannot merge it again", async () => { + await ingest(req(breadworldBank)); + await ingest(req(breadworldWallet)); + const again = await (await ingest(req(breadworldWallet))).json(); + expect(again.skipped).toBe("already_ingested"); + const [m] = await metas(); + expect(m.flags).toContain("wallet_evt:65"); + }); +}); + +describe("notifications that name an amount but are not purchases", () => { + it("skips an overdraft notice", async () => { + // Event 235 became a $21.21 purchase named "To avoid fees and interest,". + const body = await (await ingest(req({ + event_id: 235, kind: "transaction", direction: "debit", amount: 21.21, + amount_text: "$21.21", currency: "$?", merchant: null, notif_app: "CommBank", + notif_body: "Your Smart Access 1910 is overdrawn by $21.21. To avoid fees and interest, bring your account to positive by midnight 21 Sep (Syd/Melb time).", + occurred_at: "2026-09-22T09:00:18+10:00", + }))).json(); + expect(body.skipped).toBe("account_notice"); + expect(await txns()).toHaveLength(0); + }); + + it("skips a transit-card fare, because the card's load is the spend", async () => { + // Event 200: the Octopus load is on the bank statement; each fare draws it + // down. 11 of 13 fares were already not_transaction; this makes the other 2 agree. + const body = await (await ingest(req({ + event_id: 200, kind: "transaction", direction: "debit", amount: 105, + amount_text: "HKD 105.00", currency: "HKD", merchant: "Octopus", notif_app: "Wallet", + notif_title: "Octopus", notif_subtitle: "HKD 105.00 for Train", + notif_body: "Your new balance is -HKD 12.60.", + occurred_at: "2026-09-14T02:00:37+10:00", + }))).json(); + expect(body.skipped).toBe("stored_value_drawdown"); + expect(await txns()).toHaveLength(0); + }); + + it("does not read a Westpac descriptor out of another bank's message", async () => { + await ingest(req({ + event_id: 2350, kind: "transaction", direction: "debit", amount: 9.5, + amount_text: "$9.50", currency: "$?", merchant: null, notif_app: "CommBank", + notif_body: "Payment of $9.50 made. Thanks for banking with us.", + occurred_at: "2026-09-22T09:00:18+10:00", + })); + const [t] = await txns(); + expect(t.merchant_normalized).toBe("Card purchase"); + }); +}); + +describe("dating", () => { + it("dates a tap in Melbourne local time, as the statements do", async () => { + // 21:30 UTC on 30 Sep is 07:30 on 1 Oct in Melbourne. Slicing the ISO + // string filed it a day early. + await ingest(req({ ...breadworldBank, occurred_at: "2026-09-30T21:30:00Z" })); + const { rows } = await pool.query( + `SELECT transaction_date::text AS d FROM transactions WHERE source = 'wallet'` + ); + expect(rows[0].d).toBe("2026-10-01"); + }); +}); + describe("currency", () => { + it("falls back to the ECB rate when the ledger has no history", async () => { + // Hong Kong: no HKD statement row existed, so every tap stored A$0.00. + const calls: string[] = []; + vi.stubGlobal("fetch", async (url: string) => { + calls.push(String(url)); + return { ok: true, json: async () => ({ rates: { AUD: 0.19 } }) }; + }); + await ingest(req({ + event_id: 167, kind: "transaction", direction: "debit", amount: 310.1, + amount_text: "HKD 310.10", currency: "HKD", merchant: "Wonder Taxi", notif_app: "Wallet", + notif_subtitle: "Wonder Taxi, Tsim Sha Tsui East, Kowloon", + occurred_at: "2026-09-10T08:54:00+10:00", + })); + const [t] = await txns(); + expect(Number(t.amount)).toBeCloseTo(58.92, 2); + expect(Number(t.foreign_currency_amount)).toBe(310.1); + expect(calls[0]).toContain("/2026-09-10?from=HKD&to=AUD"); + const [m] = await metas(); + // Flagged as the mid-rate, not the bank's: it lacks the card's FX fee. + expect(m.flags).toEqual(expect.arrayContaining(["aud_provisional", "rate_ecb"])); + expect(m.flags).not.toContain("no_rate_reference"); + }); + it("records AUD when Amex states it explicitly", async () => { await ingest(req(amexRewardPay)); const [t] = await txns(); diff --git a/src/lib/auto-reconcile.ts b/src/lib/auto-reconcile.ts index 8b04eea..8d4fb54 100644 --- a/src/lib/auto-reconcile.ts +++ b/src/lib/auto-reconcile.ts @@ -1,4 +1,4 @@ -import { queryRaw } from "@/lib/db"; +import { queryRaw, type Queryable } from "@/lib/db"; import { merchantMatches } from "@/lib/intake-rules"; /** @@ -168,12 +168,16 @@ export async function findWalletAdoptableTransaction( amount: number, date: string, merchant: string | null, - windowDays: number = AUTO_MATCH_WINDOW_DAYS + windowDays: number = AUTO_MATCH_WINDOW_DAYS, + // wallet-ingestion calls this while holding its advisory lock, and must pass + // that transaction's connection: borrowing a second pooled one per request + // is how a burst of concurrent posts exhausts the pool and deadlocks. + db: Pick = { queryRaw } ): Promise<{ id: number; metadata_id: number | null } | null> { // No usable merchant, no adoption. "Card purchase" is wallet-ingestion's own // fallback for a notification that named no merchant at all. if (!merchant || merchant === "Card purchase") return null; - const rows = await queryRaw<{ id: number; metadata_id: number | null; merchant: string | null }>( + const rows = await db.queryRaw<{ id: number; metadata_id: number | null; merchant: string | null }>( `SELECT t.id, em.id AS metadata_id, COALESCE(em.merchant_normalized, t.merchant_normalized, t.merchant_name, t.description) AS merchant FROM transactions t diff --git a/src/lib/db.ts b/src/lib/db.ts index 6fc3b3f..4068b38 100644 --- a/src/lib/db.ts +++ b/src/lib/db.ts @@ -38,7 +38,12 @@ export type Queryable = { * 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 { +export async function withTransaction( + fn: (q: Queryable) => Promise, + // Prisma's defaults (maxWait 2s, timeout 5s) count time spent WAITING on an + // advisory lock, so a caller that serialises a burst on one needs more. + opts?: { maxWait?: number; timeout?: number } +): Promise { return prisma.$transaction(async (tx) => { const q: Queryable = { queryRaw: (sql: string, params: unknown[] = []) => tx.$queryRawUnsafe(sql, ...params), @@ -48,5 +53,5 @@ export async function withTransaction(fn: (q: Queryable) => Promise): Prom }, }; return fn(q); - }); + }, opts); } diff --git a/src/lib/wallet-ingestion.ts b/src/lib/wallet-ingestion.ts index 4e5702f..bd1b13e 100644 --- a/src/lib/wallet-ingestion.ts +++ b/src/lib/wallet-ingestion.ts @@ -1,4 +1,4 @@ -import { queryRow, queryRaw } from "@/lib/db"; +import { queryRow, withTransaction, type Queryable } from "@/lib/db"; import { DEFAULT_OWNER_ID } from "@/lib/order-ingestion"; import { findWalletAdoptableTransaction } from "@/lib/auto-reconcile"; import { @@ -54,12 +54,40 @@ export type WalletEvent = { export type WalletIngestResult = { transactionId: number | null; metadataId: number | null; - skipped?: "not_a_transaction" | "already_ingested" | "no_amount" | "merged_pair" | "adopted"; + skipped?: + | "not_a_transaction" | "already_ingested" | "no_amount" | "merged_pair" | "adopted" + | "account_notice" | "stored_value_drawdown"; flags: string[]; }; -/** Days either side within which a second notification is the SAME tap. */ -const PAIR_WINDOW_DAYS = 1; +/** + * How far apart the two halves of ONE tap can arrive. Measured 2-8 seconds on + * every captured pair; ten minutes leaves room for a phone that was offline, + * and is still nowhere near the +/-1 DAY the matcher used to allow -- which + * folded the UNICEF $50 donation and an unrelated $50 bar charge 17 hours later + * into one row (event 275 into txn 8091, 2026-10-01). + */ +const PAIR_WINDOW_MINUTES = 10; + +/** The ledger is dated the way the bank statements are: Melbourne local. */ +const LEDGER_TZ = "Australia/Melbourne"; + +/** + * The calendar date of an instant, in LEDGER_TZ. + * + * Slicing the ISO string took whatever offset n8n serialised -- UTC in + * practice -- so a Melbourne-evening tap landed on the previous day (28 of the + * first 134 rows). Statements, which this row is reconciled against, date in + * Australian local time. + */ +export function ledgerDate(occurredAt: string | null | undefined, now: Date = new Date()): string { + const d = occurredAt ? new Date(occurredAt) : now; + const at = Number.isNaN(d.getTime()) ? now : d; + // en-CA formats as YYYY-MM-DD. + return new Intl.DateTimeFormat("en-CA", { + timeZone: LEDGER_TZ, year: "numeric", month: "2-digit", day: "2-digit", + }).format(at); +} /** * A ledger-derived AUD rate for a currency: the median of what the bank @@ -91,37 +119,70 @@ async function medianRateToAud(code: string): Promise { return rate && Number.isFinite(rate) && rate > 0 ? rate : null; } +/** + * The ECB reference rate for the day, via Frankfurter (free, no key; the same + * source the statement processor converts with). + * + * The fallback when the ledger has no history for a currency. Hong Kong was the + * first trip to need it: no HKD statement row existed, so every HKD tap stored + * A$0.00 and the trip read as free until the statement landed weeks later. The + * ledger median is still preferred -- it carries the card's ~3% foreign fee, + * which the ECB mid-rate does not -- so this one is flagged as such. + * + * Any failure is null, never a throw: no rate is a state this lane already + * handles (a flagged 0), and a slow FX API must not lose the purchase. + */ +async function ecbRateToAud(code: string, date: string): Promise { + try { + const res = await fetch( + `https://api.frankfurter.dev/v1/${date}?from=${encodeURIComponent(code)}&to=AUD`, + { signal: AbortSignal.timeout(5000) } + ); + if (!res.ok) return null; + const body = (await res.json()) as { rates?: { AUD?: number } }; + const rate = body?.rates?.AUD; + return typeof rate === "number" && Number.isFinite(rate) && rate > 0 ? rate : null; + } catch { + return null; + } +} + /** * The other half of a paired notification, if it has already been ingested. * * One tap emits TWO notifications ~2-8s apart: the issuer app (which has the * card tail) and Apple Wallet (which has the better merchant and the suburb). * Neither alone is complete, so they must be MERGED, never deduped -- dropping - * either loses a field nothing else supplies. Matched on amount and date rather - * than any id, because the notifications share no identifier. + * either loses a field nothing else supplies. The notifications share no + * identifier, so a pair is recognised by what a pair IS: the same amount, + * minutes apart, from two different apps. Two notifications from one app are + * two charges, however close together. */ async function findPairedRow( + db: Queryable, amount: number, - date: string, + notifiedAt: string, + notifApp: string | null, fx: { amount: number; code: string } | null = null ) { // A foreign row pairs on its native pair: its `amount` is a provisional - // conversion (or a 0 placeholder), so two unrelated foreign taps on one day - // could share it. - const rows = await queryRaw<{ id: number; meta_id: number | null }>( + // conversion (or a 0 placeholder), so two unrelated foreign taps could share it. + const rows = await db.queryRaw<{ id: number; meta_id: number }>( `SELECT t.id, em.id AS meta_id FROM transactions t - LEFT JOIN expense_metadata em ON em.transaction_id = t.id + JOIN expense_metadata em ON em.transaction_id = t.id WHERE t.source = 'wallet' AND t.statement_id IS NULL + AND t.superseded_by_id IS NULL AND CASE WHEN $4::text IS NULL THEN t.amount = $1::numeric AND t.foreign_currency_code IS NULL ELSE t.foreign_currency_code = $4::text AND t.foreign_currency_amount = $5::numeric END - AND t.transaction_date BETWEEN $2::date - $3::int AND $2::date + $3::int - AND t.superseded_by_id IS NULL + AND em.notified_at BETWEEN $2::timestamptz - make_interval(mins => $3::int) + AND $2::timestamptz + make_interval(mins => $3::int) + AND em.notif_app IS DISTINCT FROM $6::text LIMIT 2`, - [amount, date, PAIR_WINDOW_DAYS, fx?.code ?? null, fx?.amount ?? null] + [amount, notifiedAt, PAIR_WINDOW_MINUTES, fx?.code ?? null, fx?.amount ?? null, notifApp] ); return rows.length === 1 ? rows[0] : null; } @@ -144,13 +205,48 @@ async function findPairedRow( */ const STORED_VALUE = /gift\s*card|prepaid|travel\s*(?:money|card)|store\s*card/i; -/** How an adopted row records which wallet event folded into it. */ +/** + * A transit card reporting a drawdown of its own balance ("HKD 7.40 for Train" + * / "Your new balance is HKD 169.50."). + * + * Unlike a gift card, a transit card is loaded FROM a bank card -- that load is + * on the statement and is the spend. Each fare is that money being used, so + * booking it again counts it twice. The interpreter already called 11 of 13 + * Octopus fares not_transaction; the two it did not (events 191, 200) became + * rows, so the outcome depended on the model's mood. This makes it a rule. The + * body wording is the signal because it is what every such card says; the + * issuer name is not (Octopus, Opal, myki, Suica each name themselves). + */ +const BALANCE_DRAWDOWN = /\byour (?:new|current) balance is\b/i; + +/** + * Account notices that state an amount without money moving: "Your Smart Access + * 1910 is overdrawn by $21.21. To avoid fees and interest, ..." (event 235) + * became a $21.21 purchase named "To avoid fees and interest,". The model is + * right that the text names an amount; it is not a payment. + */ +const ACCOUNT_NOTICE = /\boverdrawn\b|\bto avoid fees\b|\bavailable balance\b|\blow balance\b/i; + +/** How an adopted or merged row records which wallet event folded into it. */ const adoptedFlag = (eventId: number) => `wallet_evt:${eventId}`; export function isStoredValueInstrument(ev: Pick): boolean { return [ev.notif_title, ev.issuer, ev.notif_app].some((v) => !!v && STORED_VALUE.test(v)); } +/** Already landed: a row this lane wrote, or one it adopted or merged into. */ +async function alreadyIngested(db: Pick, eventId: number) { + return db.queryRow<{ id: number }>( + `SELECT id FROM transactions WHERE source = 'wallet' AND source_ref = $1 + UNION ALL + SELECT em.transaction_id AS id FROM expense_metadata em + JOIN transactions t ON t.id = em.transaction_id + WHERE em.flags @> $2::jsonb AND t.superseded_by_id IS NULL + LIMIT 1`, + [`evt:${eventId}`, JSON.stringify([adoptedFlag(eventId)])] + ); +} + export async function processWalletIngestion(ev: WalletEvent): Promise { const flags: string[] = []; @@ -163,33 +259,36 @@ export async function processWalletIngestion(ev: WalletEvent): Promise( - `SELECT id FROM transactions WHERE source = 'wallet' AND source_ref = $1 - UNION ALL - SELECT em.transaction_id AS id FROM expense_metadata em - JOIN transactions t ON t.id = em.transaction_id - WHERE em.flags @> $2::jsonb AND t.superseded_by_id IS NULL - LIMIT 1`, - [sourceRef, JSON.stringify([adoptedFlag(ev.event_id)])] - ); - if (existing) { - return { transactionId: existing.id, metadataId: null, skipped: "already_ingested", flags }; + const text = [ev.notif_subtitle, ev.notif_body].filter(Boolean).join("\n"); + if (ACCOUNT_NOTICE.test(text)) { + return { transactionId: null, metadataId: null, skipped: "account_notice", flags }; + } + if (BALANCE_DRAWDOWN.test(ev.notif_body ?? "")) { + return { transactionId: null, metadataId: null, skipped: "stored_value_drawdown", flags }; } - const date = (ev.occurred_at ?? new Date().toISOString()).slice(0, 10); + // Two places this event may already have landed: a row this lane wrote (which + // carries the ref natively), or a row it ADOPTED or MERGED into, whose + // source_ref belongs to whoever created it -- marked in the metadata flags + // rather than by overwriting a field another lane owns. Both must be checked, + // because bumping the interpreter's prompt_version deliberately re-interprets + // and re-posts every event ever captured -- that is the replay mechanism, and + // it must not mint a second row for a purchase that has one. Checked here + // without the lock to skip the FX lookup on a replay, and again under it. + const seen = await alreadyIngested({ queryRow }, ev.event_id); + if (seen) return { transactionId: seen.id, metadataId: null, skipped: "already_ingested", flags }; + + const sourceRef = `evt:${ev.event_id}`; + const notifiedAt = ev.occurred_at ?? new Date().toISOString(); + const date = ledgerDate(ev.occurred_at); // Deterministic parsing of the positional fields. The model decides what the // amount and merchant ARE; it does not get to parse a fixed-width block. - const desc = parseWestpacDescriptor(ev.notif_body); + // Only Westpac writes that block: run over a CommBank overdraft notice it + // produced the merchant "To avoid fees and interest," (event 235). + const desc = /westpac/i.test(ev.notif_app ?? "") + ? parseWestpacDescriptor(ev.notif_body) + : { merchant: null, city: null, country: null }; const sub = parseWalletSubtitle(ev.notif_subtitle); const merchant = sub.merchant ?? ev.merchant ?? desc.merchant ?? "Card purchase"; // Two sources for a place, and they are not equal. The Apple Wallet subtitle @@ -219,19 +318,23 @@ export async function processWalletIngestion(ev: WalletEvent): Promise { + await db.queryRaw(`SELECT pg_advisory_xact_lock(hashtextextended('wallet-ingest', 0))::text`); - // The receipt got here first. Fold into its row instead of writing a second - // one for the same purchase (see findWalletAdoptableTransaction). The column - // list IS the ownership boundary, the mirror of receipt-ingestion.ts's: the - // receipt read the register tape, so merchant, description, category and the - // items are its business and are absent here; `location` is wallet's own -- - // nothing else writes a suburb -- and payment_method is filled only when the - // other lane left it unknown. - const adopted = await findWalletAdoptableTransaction(DEFAULT_OWNER_ID, amountAud, date, merchant); - if (adopted) { - await queryRaw( - `UPDATE transactions - SET location = COALESCE($2, location), - payment_method = COALESCE(payment_method, $3) - WHERE id = $1`, - [adopted.id, location, paymentMethod] + const existing = await alreadyIngested(db, ev.event_id); + if (existing) { + return { transactionId: existing.id, metadataId: null, skipped: "already_ingested" as const, flags }; + } + + // Merge the paired notification rather than creating a second row for one + // tap. The issuer half brings the card tail; the Wallet half brings the + // merchant and suburb. COALESCE so whichever arrives second only ever fills + // gaps -- it can never blank a field the first half supplied. + const pair = await findPairedRow( + db, amountAud, notifiedAt, ev.notif_app ?? null, + fxCode && fxAmount !== null ? { amount: fxAmount, code: fxCode } : null ); - let metadataId = adopted.metadata_id; - if (metadataId) { - await queryRaw( + if (pair) { + await db.queryRaw( + `UPDATE transactions + SET merchant_name = COALESCE(NULLIF($2,'Card purchase'), merchant_name), + merchant_normalized = COALESCE(NULLIF($2,'Card purchase'), merchant_normalized), + -- A subtitle-derived place overwrites; anything else only fills a + -- gap. Without the $4 branch the merged row depends on which + -- notification happened to arrive first, and the issuer half + -- usually wins by a few seconds -- keeping the worse string. + location = CASE WHEN $4::boolean THEN COALESCE($3, location) + ELSE COALESCE(location, $3) END + WHERE id = $1`, + [pair.id, merchant, location, subtitleLocation !== null] + ); + // The merged half records itself, or a replay finds nothing and merges + // it again (or, once the partner is superseded, mints a new row). + await db.queryRaw( `UPDATE expense_metadata SET card_last4 = COALESCE(card_last4, $2), payment_method_detail = COALESCE(payment_method_detail, $3), - currency = COALESCE(currency, $4), + merchant_normalized = COALESCE(NULLIF($4,'Card purchase'), merchant_normalized), flags = COALESCE(flags, '[]'::jsonb) || $5::jsonb WHERE id = $1`, - [metadataId, ev.card_tail ?? null, ev.issuer ?? null, stated, JSON.stringify([...flags, adoptedFlag(ev.event_id)])] + [pair.meta_id, ev.card_tail ?? null, ev.issuer ?? null, merchant, JSON.stringify([adoptedFlag(ev.event_id)])] ); - } else { - // expense_metadata.transaction_id is UNIQUE, so this branch only runs - // when the adopted row genuinely has none -- and it must produce one, or - // the card tail and the flag that makes this idempotent have nowhere to - // live. - const meta = await queryRow<{ id: number }>( - `INSERT INTO expense_metadata ( - transaction_id, source, source_message_id, merchant_normalized, amount, - transaction_date, card_last4, currency, payment_method, - payment_method_detail, line_items, flags, extraction_model - ) VALUES ($1,'wallet',$2,$3,$4,$5,$6,$7,$8,$9,'[]'::jsonb,$10::jsonb,'wallet-interpret') - RETURNING id`, - [ - adopted.id, sourceRef, merchant, ev.amount, date, ev.card_tail ?? null, stated, - paymentMethod, ev.issuer ?? null, JSON.stringify([...flags, adoptedFlag(ev.event_id)]), - ] - ); - metadataId = meta!.id; + return { transactionId: pair.id, metadataId: pair.meta_id, skipped: "merged_pair" as const, flags }; } - return { transactionId: adopted.id, metadataId, skipped: "adopted", flags }; - } - const txn = await queryRow<{ id: number }>( - `INSERT INTO transactions ( - transaction_date, description, amount, amount_aud, category, payment_method, - merchant_name, merchant_normalized, transaction_type, owner_id, location, - foreign_currency_amount, foreign_currency_code, source, source_ref, row_index - ) VALUES ($1,$2,$3,$3,'other',$10,$4,$4,'debit',$5,$6,$7,$8,'wallet',$9,0) - RETURNING id`, - [date, merchant, amountAud, merchant, DEFAULT_OWNER_ID, location, fxAmount, fxCode, sourceRef, paymentMethod] - ); + // The receipt got here first. Fold into its row instead of writing a second + // one for the same purchase (see findWalletAdoptableTransaction). The column + // list IS the ownership boundary, the mirror of receipt-ingestion.ts's: the + // receipt read the register tape, so merchant, description, category and the + // items are its business and are absent here; `location` is wallet's own -- + // nothing else writes a suburb -- and payment_method is filled only when the + // other lane left it unknown. + const adopted = await findWalletAdoptableTransaction(DEFAULT_OWNER_ID, amountAud, date, merchant, undefined, db); + if (adopted) { + await db.queryRaw( + `UPDATE transactions + SET location = COALESCE($2, location), + payment_method = COALESCE(payment_method, $3) + WHERE id = $1`, + [adopted.id, location, paymentMethod] + ); + let metadataId = adopted.metadata_id; + if (metadataId) { + await db.queryRaw( + `UPDATE expense_metadata + SET card_last4 = COALESCE(card_last4, $2), + payment_method_detail = COALESCE(payment_method_detail, $3), + currency = COALESCE(currency, $4), + flags = COALESCE(flags, '[]'::jsonb) || $5::jsonb + WHERE id = $1`, + [metadataId, ev.card_tail ?? null, ev.issuer ?? null, stated, JSON.stringify([...flags, adoptedFlag(ev.event_id)])] + ); + } else { + // expense_metadata.transaction_id is UNIQUE, so this branch only runs + // when the adopted row genuinely has none -- and it must produce one, or + // the card tail and the flag that makes this idempotent have nowhere to + // live. + const meta = await db.queryRow<{ id: number }>( + `INSERT INTO expense_metadata ( + transaction_id, source, source_message_id, merchant_normalized, amount, + transaction_date, card_last4, currency, payment_method, + payment_method_detail, line_items, flags, extraction_model, + notified_at, notif_app + ) VALUES ($1,'wallet',$2,$3,$4,$5,$6,$7,$8,$9,'[]'::jsonb,$10::jsonb,'wallet-interpret',$11,$12) + RETURNING id`, + [ + adopted.id, sourceRef, merchant, ev.amount, date, ev.card_tail ?? null, stated, + paymentMethod, ev.issuer ?? null, JSON.stringify([...flags, adoptedFlag(ev.event_id)]), + notifiedAt, ev.notif_app ?? null, + ] + ); + metadataId = meta!.id; + } + return { transactionId: adopted.id, metadataId, skipped: "adopted" as const, flags }; + } - const meta = await queryRow<{ id: number }>( - `INSERT INTO expense_metadata ( - transaction_id, source, source_message_id, merchant_normalized, amount, - transaction_date, card_last4, currency, payment_method, - payment_method_detail, line_items, flags, extraction_model - ) VALUES ($1,'wallet',$2,$3,$4,$5,$6,$7,$8,$9,'[]'::jsonb,$10::jsonb,'wallet-interpret') - RETURNING id`, - [ - txn!.id, sourceRef, merchant, ev.amount, date, - ev.card_tail ?? null, stated, paymentMethod, ev.issuer ?? null, JSON.stringify(flags), - ] - ); + const txn = await db.queryRow<{ id: number }>( + `INSERT INTO transactions ( + transaction_date, description, amount, amount_aud, category, payment_method, + merchant_name, merchant_normalized, transaction_type, owner_id, location, + foreign_currency_amount, foreign_currency_code, source, source_ref, row_index + ) VALUES ($1,$2,$3,$3,'other',$10,$4,$4,'debit',$5,$6,$7,$8,'wallet',$9,0) + RETURNING id`, + [date, merchant, amountAud, merchant, DEFAULT_OWNER_ID, location, fxAmount, fxCode, sourceRef, paymentMethod] + ); - return { transactionId: txn!.id, metadataId: meta!.id, flags }; + const meta = await db.queryRow<{ id: number }>( + `INSERT INTO expense_metadata ( + transaction_id, source, source_message_id, merchant_normalized, amount, + transaction_date, card_last4, currency, payment_method, + payment_method_detail, line_items, flags, extraction_model, + notified_at, notif_app + ) VALUES ($1,'wallet',$2,$3,$4,$5,$6,$7,$8,$9,'[]'::jsonb,$10::jsonb,'wallet-interpret',$11,$12) + RETURNING id`, + [ + txn!.id, sourceRef, merchant, ev.amount, date, + ev.card_tail ?? null, stated, paymentMethod, ev.issuer ?? null, JSON.stringify(flags), + notifiedAt, ev.notif_app ?? null, + ] + ); + + return { transactionId: txn!.id, metadataId: meta!.id, flags }; + }, { maxWait: 10_000, timeout: 30_000 }); }