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:<id>, 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) <noreply@anthropic.com>
2026-10-03 16:55:07 +10:00
parent 22fde30b26
commit a3e4e1102a
8 changed files with 450 additions and 146 deletions
@@ -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
@@ -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:<id>'),
+-- 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;
@@ -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")
@@ -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."

@@ -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();
@@ -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<Queryable, "queryRaw"> = { 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
@@ -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<T>(fn: (q: Queryable) => Promise<T>): Promise<T> {
+export async function withTransaction<T>(
+  fn: (q: Queryable) => Promise<T>,
+  // 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<T> {
  return prisma.$transaction(async (tx) => {
    const q: Queryable = {
      queryRaw: <R>(sql: string, params: unknown[] = []) => tx.$queryRawUnsafe<R[]>(sql, ...params),
@@ -48,5 +53,5 @@ export async function withTransaction<T>(fn: (q: Queryable) => Promise<T>): Prom
      },
    };
    return fn(q);
-  });
+  }, opts);
 }
@@ -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<number | null> {
  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<number | null> {
+  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<WalletEvent, "notif_title" | "issuer" | "notif_app">): 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<Queryable, "queryRow">, 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<WalletIngestResult> {
  const flags: string[] = [];

@@ -163,33 +259,36 @@ export async function processWalletIngestion(ev: WalletEvent): Promise<WalletIng
  if (ev.amount === null || !Number.isFinite(ev.amount) || ev.amount <= 0) {
    return { transactionId: null, metadataId: null, skipped: "no_amount", flags };
  }
-
-  const sourceRef = `evt:${ev.event_id}`;
-  // Two places this event may already have landed: a row this lane wrote (which
-  // carries the ref natively), or a row it ADOPTED, whose source_ref belongs to
-  // whoever created it. The adopted case is marked in the metadata flags rather
-  // than by overwriting a field another lane owns. Both must be checked here,
-  // 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.
-  const existing = await 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`,
-    [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<WalletIng

  // Foreign: the native figure is what both sides can agree on exactly, so it
  // is what gets stored. `amount` still needs a value (NOT NULL) and takes a
-  // provisional conversion, flagged, which the statement overwrites.
+  // provisional conversion, flagged, which the statement overwrites. Resolved
+  // BEFORE the lock: the ECB fallback is a network call, and the lock
+  // serialises every wallet post behind whatever runs inside it.
  let amountAud = ev.amount;
  let fxAmount: number | null = null;
  let fxCode: string | null = null;
  if (stated && stated !== "AUD") {
    fxAmount = ev.amount;
    fxCode = stated;
-    const rate = await medianRateToAud(stated);
+    const ledgerRate = await medianRateToAud(stated);
+    const rate = ledgerRate ?? (await ecbRateToAud(stated, date));
    if (rate) {
      amountAud = Number((ev.amount * rate).toFixed(2));
      flags.push("aud_provisional");
+      if (!ledgerRate) flags.push("rate_ecb");
    } else {
-      // No history to convert with. The native figure must NOT fall through as
+      // No rate from either source. The native figure must NOT fall through as
      // AUD -- 672006 VND once landed as A$672,006. Zero is a flagged
      // placeholder; the statement supplies the real AUD at reconcile time,
      // which matches on the native pair, not on this column.
@@ -246,15 +349,31 @@ export async function processWalletIngestion(ev: WalletEvent): Promise<WalletIng
  const paymentMethod = storedValue ? "credits" : "card";
  if (storedValue) flags.push("stored_value_instrument");

+  // Everything from the idempotency re-check to the INSERT holds one lock.
+  // n8n's HTTP node posts a poll's items CONCURRENTLY, and both halves of a tap
+  // usually land in the same 15-minute poll -- so each ran findPairedRow before
+  // the other had inserted, neither found its partner, and one tap became two
+  // rows: 11 pairs, A$3,841.40, every pair with an identical created_at. A
+  // single lock for the whole lane, not one per amount: wallet volume is a few
+  // dozen a day, and a coarse lock has no key to get wrong.
+  return withTransaction(async (db) => {
+    await db.queryRaw(`SELECT pg_advisory_xact_lock(hashtextextended('wallet-ingest', 0))::text`);
+
+    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(
-    amountAud, date, fxCode && fxAmount !== null ? { amount: fxAmount, code: fxCode } : null
+      db, amountAud, notifiedAt, ev.notif_app ?? null,
+      fxCode && fxAmount !== null ? { amount: fxAmount, code: fxCode } : null
    );
    if (pair) {
-    await queryRaw(
+      await db.queryRaw(
        `UPDATE transactions
            SET merchant_name       = COALESCE(NULLIF($2,'Card purchase'), merchant_name),
                merchant_normalized = COALESCE(NULLIF($2,'Card purchase'), merchant_normalized),
@@ -267,17 +386,18 @@ export async function processWalletIngestion(ev: WalletEvent): Promise<WalletIng
          WHERE id = $1`,
        [pair.id, merchant, location, subtitleLocation !== null]
      );
-    if (pair.meta_id) {
-      await queryRaw(
+      // 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),
-                merchant_normalized   = COALESCE(NULLIF($4,'Card purchase'), merchant_normalized)
+                merchant_normalized   = COALESCE(NULLIF($4,'Card purchase'), merchant_normalized),
+                flags                 = COALESCE(flags, '[]'::jsonb) || $5::jsonb
          WHERE id = $1`,
-        [pair.meta_id, ev.card_tail ?? null, ev.issuer ?? null, merchant]
+        [pair.meta_id, ev.card_tail ?? null, ev.issuer ?? null, merchant, JSON.stringify([adoptedFlag(ev.event_id)])]
      );
-    }
-    return { transactionId: pair.id, metadataId: pair.meta_id, skipped: "merged_pair", flags };
+      return { transactionId: pair.id, metadataId: pair.meta_id, skipped: "merged_pair" as const, flags };
    }

    // The receipt got here first. Fold into its row instead of writing a second
@@ -287,9 +407,9 @@ export async function processWalletIngestion(ev: WalletEvent): Promise<WalletIng
    // 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);
+    const adopted = await findWalletAdoptableTransaction(DEFAULT_OWNER_ID, amountAud, date, merchant, undefined, db);
    if (adopted) {
-    await queryRaw(
+      await db.queryRaw(
        `UPDATE transactions
            SET location       = COALESCE($2, location),
                payment_method = COALESCE(payment_method, $3)
@@ -298,7 +418,7 @@ export async function processWalletIngestion(ev: WalletEvent): Promise<WalletIng
      );
      let metadataId = adopted.metadata_id;
      if (metadataId) {
-      await queryRaw(
+        await db.queryRaw(
          `UPDATE expense_metadata
              SET card_last4            = COALESCE(card_last4, $2),
                  payment_method_detail = COALESCE(payment_method_detail, $3),
@@ -312,24 +432,26 @@ export async function processWalletIngestion(ev: WalletEvent): Promise<WalletIng
        // 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 }>(
+        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
-         ) VALUES ($1,'wallet',$2,$3,$4,$5,$6,$7,$8,$9,'[]'::jsonb,$10::jsonb,'wallet-interpret')
+             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", flags };
+      return { transactionId: adopted.id, metadataId, skipped: "adopted" as const, flags };
    }

-  const txn = await queryRow<{ id: number }>(
+    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,
@@ -339,18 +461,21 @@ export async function processWalletIngestion(ev: WalletEvent): Promise<WalletIng
      [date, merchant, amountAud, merchant, DEFAULT_OWNER_ID, location, fxAmount, fxCode, sourceRef, paymentMethod]
    );

-  const meta = await queryRow<{ id: number }>(
+    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
-     ) VALUES ($1,'wallet',$2,$3,$4,$5,$6,$7,$8,$9,'[]'::jsonb,$10::jsonb,'wallet-interpret')
+         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 });
 }