frollo: don't re-import what the ledger already has, and stop hiding the queue

The first import wrote 550 rows on 2026-08-13. 422 of them (77%) were second
copies of transactions the ledger already held from statements — $1,023,824.63
of movement counted twice. The owner found it by opening the transactions view
and seeing one HDR Global salary listed twice, once as A$15,518.53 from the
statement and once as US$10,782.00 from the feed.

The currency was never the defect. toLedgerRow already left amount_aud NULL and
named the currency in foreign_currency_code, which is the documented contract for
a row whose AUD value is unknown, and the transactions page labels it. What made
it look wrong was the duplicate sitting beside it.

Two changes.

Upstream, ingestFrolloCsv now drops rows the ledger already holds, matching on
amount + direction within LEDGER_MATCH_DAYS (3). Three things had to be right and
the first two were not, each caught only by rehearsing against real data rather
than fixtures:

  - pg returns a DATE as a JS Date while Prisma and the CSV give strings.
    String(date).slice(0,10) is "Wed Mar 10", which parses to NaN, so the first
    dry run reported 550 to insert and zero duplicates. dayMs() takes both.
  - Direction has to be in the key. This ledger is full of internal transfers
    between the owner's own accounts and the feed carries both legs: 2026-05-18
    has +3076.04 into ANZ and -3076.04 out of AMP. Matching on amount alone let
    the credit leg consume the ledger's debit row, so the real duplicate was
    written — 20 rows got in that way.
  - 'refund' is money in. The feed calls a reversed account fee a credit and the
    statement importer types it 'refund'; classifying it as an outflow left every
    ANZ servicing-fee reversal behind.

Downstream, awaitsStatementLine() is removed. Its premise — feed rows never await
a statement line — was asserted, never tested, and false for almost every
account. Worse is how it got there: the reconcile queue jumped 8 -> 558 when the
feed landed, that jump was read as noise and filtered away, and filtering it
removed the only mechanism that would ever have collapsed the duplicates. The
queue was right. A feed row IS a row awaiting its statement line.

Re-imported: 375 dropped as already-on-ledger, 175 inserted. Residual duplicates
4 rows / $15.01, all sub-$5 account fees where several identical amounts fall in
overlapping windows and greedy consumption picks the wrong one; not chased
further at this scale.

The CLI prints the already-on-ledger count even when zero — a number you have to
go looking for is a number nobody looks at.

src/lib/frollo-ingest.ts

@@ -41,6 +41,46 @@ export const SOURCE = "frollo";
  */
 export const LARGE_BATCH = 200;
 
+/**
+ * How far a Frollo row may sit from a ledger row and still be the same event.
+ *
+ * The feed dates a transaction when the provider posted it; a statement dates it
+ * when the bank did. Those differ by a day or two around weekends. Three days is
+ * wide enough to catch that and narrow enough that two genuinely different
+ * charges for the identical amount in the same week are rare — and when they do
+ * collide, the cost is one missing row in a feed whose whole job is provisional
+ * visibility, against a permanent double-count the other way.
+ */
+export const LEDGER_MATCH_DAYS = 3;
+
+/**
+ * A date column, as midnight UTC, from whatever the driver handed back.
+ *
+ * `pg` maps a Postgres DATE to a JS `Date` in local time, while Prisma and the
+ * CSV both yield `"YYYY-MM-DD"` strings. Reading one as the other is silent:
+ * `String(new Date()).slice(0, 10)` is `"Wed Mar 10"`, which parses to NaN, and
+ * a guard built on it skips every row it was meant to compare — the first dry
+ * run of this code reported 550 rows to insert and zero duplicates against a
+ * ledger holding 422 of them.
+ */
+/**
+ * Ledger transaction types that mean money arriving.
+ *
+ * The whole set in use is debit | credit | payment | fee | interest | refund.
+ * `refund` is the one that bites: the feed calls a reversed account fee a
+ * `credit` while the statement importer types it `refund`, so classifying it as
+ * an outflow left five real duplicates behind — every ANZ servicing-fee reversal
+ * in the file. Small money, but it is the same shape of mismatch that would
+ * matter on a large refund.
+ */
+const INFLOW_TYPES = new Set(["credit", "refund"]);
+
+function dayMs(v: unknown): number {
+  if (v instanceof Date) return Date.UTC(v.getFullYear(), v.getMonth(), v.getDate());
+  const m = /^(\d{4})-(\d{2})-(\d{2})/.exec(String(v));
+  return m ? Date.UTC(Number(m[1]), Number(m[2]) - 1, Number(m[3])) : NaN;
+}
+
 export interface IngestOptions {
   ownerId?: number;
   /** Write. Without it, everything is computed and nothing is inserted. */
@@ -60,6 +100,12 @@ export interface IngestReport {
   duplicatesDropped: number;
   suspectRepeats: { description: string; transactionDate: string; amount: number; idGap: number }[];
   alreadyImported: number;
+  /**
+   * Rows dropped because the ledger already holds the same event from a
+   * statement. Counted and reported, never silent: a Frollo row vanishing can
+   * mean several things and one of them is worth following up.
+   */
+  ledgerDuplicates: number;
   toInsert: number;
   inserted: number;
   /** Non-empty means an automatic run should stop and a human should look. */
@@ -147,7 +193,79 @@ export async function ingestFrolloCsv(
     [SOURCE]
   );
   const seen = new Set(existing.map((r) => r.source_ref));
-  const fresh = ledger.filter((r) => !seen.has(r.sourceRef));
+  const notYetImported = ledger.filter((r) => !seen.has(r.sourceRef));
+
+  // Drop rows the ledger already holds from a statement.
+  //
+  // This guard is the whole lesson of the 2026-08-13 first import. The feed was
+  // scoped to "accounts that issue no monthly statement", but that was asserted,
+  // never tested against the ledger — and it was wrong for almost every account:
+  // 422 of the 550 rows imported (77%) had a statement twin within three days,
+  // $1,023,824.63 of movement counted twice. The HDR Global salary showed it
+  // most plainly, appearing as both A$15,518.53 (statement, converted) and
+  // US$10,782.00 (feed, native) for the same July payment.
+  //
+  // The check that would have caught it is one query and takes a second. The
+  // check that was run instead compared each row's date against the statement's
+  // min-max window, which for accounts whose statements span 182 to 460 days
+  // swallows a year and cannot distinguish covered from uncovered at all.
+  const dates = notYetImported.map((r) => dayMs(r.transactionDate)).filter(Number.isFinite);
+  let ledgerDuplicates = 0;
+  let fresh = notYetImported;
+  if (dates.length > 0) {
+    const pad = LEDGER_MATCH_DAYS * 86_400_000;
+    const from = new Date(Math.min(...dates) - pad).toISOString().slice(0, 10);
+    const to = new Date(Math.max(...dates) + pad).toISOString().slice(0, 10);
+
+    // Superseded rows are excluded: one has already been replaced by the row
+    // that supersedes it, so matching against both would hide a genuine gap.
+    const priorRows = await exec<{ transaction_date: string; amount: string; transaction_type: string }>(
+      `SELECT transaction_date, amount, transaction_type
+         FROM transactions
+        WHERE (source IS NULL OR source <> $1)
+          AND superseded_by_id IS NULL
+          AND transaction_date BETWEEN $2::date AND $3::date`,
+      [SOURCE, from, to]
+    );
+
+    // Keyed on amount AND direction.
+    //
+    // Amount is the field both sides agree on exactly — descriptions do not
+    // survive the trip (the feed writes "HDR Global Services (Bermuda)" where
+    // the statement writes "Received money from HDR Global Services (Bermuda)
+    // with reference ...") and dates drift by a day or two.
+    //
+    // Direction has to be in the key because this ledger is full of internal
+    // transfers between the owner's own accounts, and both legs are in the feed:
+    // 2026-05-18 carries a +3076.04 credit into ANZ and a -3076.04 debit out of
+    // AMP. On amount alone the credit leg consumed the ledger's debit row and
+    // the genuine duplicate was written — 20 rows got in this way on the first
+    // corrected run. A credit is never the duplicate of a debit.
+    const key = (amount: number, inflow: boolean) => `${Math.abs(amount).toFixed(2)}:${inflow ? "in" : "out"}`;
+    const byAmount = new Map<string, number[]>();
+    for (const p of priorRows) {
+      const k = key(Number(p.amount), INFLOW_TYPES.has(p.transaction_type));
+      const t = dayMs(p.transaction_date);
+      if (!Number.isFinite(t)) continue;
+      const list = byAmount.get(k);
+      if (list) list.push(t);
+      else byAmount.set(k, [t]);
+    }
+
+    fresh = notYetImported.filter((r) => {
+      const candidates = byAmount.get(key(r.amount, r.transactionType === "credit"));
+      if (!candidates) return true;
+      const t = dayMs(r.transactionDate);
+      if (!Number.isFinite(t)) return true;
+      const hit = candidates.findIndex((c) => Math.abs(c - t) <= pad);
+      if (hit === -1) return true;
+      // Consume the match so two feed rows cannot both claim one ledger row —
+      // a real pair of identical charges must not collapse into one.
+      candidates.splice(hit, 1);
+      ledgerDuplicates += 1;
+      return false;
+    });
+  }
 
   const base = {
     totalRows: rows.length,
@@ -158,7 +276,8 @@ export async function ingestFrolloCsv(
     unknownAccounts: unknown,
     duplicatesDropped: dropped.length,
     suspectRepeats,
-    alreadyImported: ledger.length - fresh.length,
+    alreadyImported: ledger.length - notYetImported.length,
+    ledgerDuplicates,
     toInsert: fresh.length,
   };
   const anomalies = findAnomalies(base, opts.largeBatch);