diff --git a/scripts/import-frollo.mts b/scripts/import-frollo.mts index cc95ce5..45f98c0 100644 --- a/scripts/import-frollo.mts +++ b/scripts/import-frollo.mts @@ -5,14 +5,18 @@ * node --experimental-strip-types scripts/import-frollo.mts --file --apply * * Dry run by default: it prints exactly what it would do and touches nothing. - * Rehearse first — every irreversible step in this repo that skipped that has - * cost something, and the failure mode here is doubling reported income. + * Rehearse first — the failure mode here is doubling reported income, and the + * de-duplication rule that shipped second only survived because a dry run was + * read against the salary rows. * * Export the file with PENDING TRANSACTIONS EXCLUDED. A pending row changes both * its id and its description when it settles, so importing one guarantees a - * duplicate on the next run; and every pending row observed so far has been on a - * credit card, which this importer does not cover anyway. See DECISIONS.md - * ING-11 in the smarthome repo. + * duplicate on the next run; every pending row observed has been on a credit + * card, which this importer does not cover anyway. See DECISIONS.md ING-11. + * + * All parsing, scoping, de-duplication and insertion live in + * `src/lib/frollo-ingest.ts`, shared with POST /api/frollo/ingest so the manual + * and automatic paths cannot drift. * * Needs DATABASE_URL. postgres-personal publishes no host port, so from the host: * export DATABASE_URL="postgresql://personal:@$(docker inspect postgres-personal \ @@ -21,239 +25,92 @@ */ import { readFileSync } from "node:fs"; import pg from "pg"; -import { parseCSVRows } from "../src/lib/csv-parser.ts"; -import { - accountFor, - accountReport, - dedupe, - readRows, - skipReason, - toLedgerRow, - unknownAccounts, - type FrolloRow, - type SkipReason, -} from "../src/lib/frollo-csv.ts"; - -const SOURCE = "frollo"; +import { ingestFrolloCsv, type SqlExecutor } from "../src/lib/frollo-ingest.ts"; function arg(name: string, fallback?: string): string | undefined { const i = process.argv.indexOf(`--${name}`); - if (i >= 0 && process.argv[i + 1] && !process.argv[i + 1].startsWith("--")) { - return process.argv[i + 1]; - } + if (i >= 0 && process.argv[i + 1] && !process.argv[i + 1].startsWith("--")) return process.argv[i + 1]; return fallback; } const has = (name: string) => process.argv.includes(`--${name}`); const file = arg("file"); -const apply = has("apply"); -const ownerId = Number(arg("owner", "1")); -const allowMissing = has("allow-missing-accounts"); - if (!file) { - console.error("usage: --file [--apply] [--owner ] [--allow-missing-accounts]"); + console.error("usage: --file [--apply] [--force] [--owner ]"); + process.exit(2); +} +if (!process.env.DATABASE_URL) { + console.error("DATABASE_URL is not set — cannot compare against the ledger."); process.exit(2); } const money = (n: number) => n.toLocaleString("en-AU", { minimumFractionDigits: 2, maximumFractionDigits: 2 }); -const rows = readRows(parseCSVRows(readFileSync(file, "utf8"))); -console.log(`\n${file}`); -console.log(` ${rows.length} rows in file\n`); - -// ---- 1. scope ------------------------------------------------------------ -const skipped = new Map(); -const inScope: FrolloRow[] = []; -for (const r of rows) { - const why = skipReason(r); - if (why) { - const list = skipped.get(why); - if (list) list.push(r); - else skipped.set(why, [r]); - } else inScope.push(r); -} -console.log("SCOPE"); -for (const [why, list] of [...skipped.entries()].sort((a, b) => b[1].length - a[1].length)) { - console.log(` skipped ${String(list.length).padStart(5)} ${why}`); -} -console.log(` in scope ${String(inScope.length).padStart(4)}\n`); - -if (skipped.has("pending")) { - console.log( - ` NOTE: ${skipped.get("pending")!.length} pending row(s) present — the export was taken\n` + - ` with pending INCLUDED. They are skipped, but re-export with pending\n` + - ` excluded so successive files are comparable.\n` - ); -} - -// ---- 2. accounts --------------------------------------------------------- -// A lapsed CDR consent removes an account from the export silently; the row -// count just drops. Assert the roster rather than trusting it. -console.log("ACCOUNTS"); -const report = accountReport(inScope); -for (const a of report) { - const mark = a.present ? " " : "!"; - console.log(` ${mark} ${a.spec.last4} ${a.spec.label.padEnd(24)} ${String(a.rows).padStart(5)} rows`); -} -const missing = report.filter((a) => !a.present); -const unknown = unknownAccounts(rows); -if (unknown.length > 0) { - console.log("\n accounts in the file this importer does not know (skipped):"); - for (const u of unknown) console.log(` ${u.accountNumber} ${u.accountName} ${u.rows} rows`); -} -if (missing.length > 0) { - console.log(`\n ${missing.length} configured account(s) contributed NO rows:`); - for (const m of missing) console.log(` ${m.spec.last4} ${m.spec.label} (${m.spec.provider})`); - console.log(" A CDR consent may have lapsed — check before treating this file as complete."); - if (!allowMissing) { - console.log(" Refusing to continue. Re-run with --allow-missing-accounts if this is expected.\n"); - process.exit(1); - } -} -console.log(); - -// ---- 3. dedupe ----------------------------------------------------------- -const { kept, dropped } = dedupe(inScope); -console.log("DE-DUPLICATION (re-ingest twins from CDR re-consent)"); -console.log(` ${dropped.length} dropped, ${kept.length} kept`); -if (dropped.length > 0) { - const byAccount = new Map(); - for (const d of dropped) { - const label = accountFor(d.row.accountNumber)?.label ?? "?"; - byAccount.set(label, (byAccount.get(label) ?? 0) + 1); - } - for (const [label, n] of [...byAccount.entries()].sort((a, b) => b[1] - a[1])) { - console.log(` ${String(n).padStart(4)} ${label}`); - } - console.log(" examples:"); - for (const d of dropped.slice(0, 3)) { - console.log( - ` ${d.row.transactionDate} ${money(d.row.amount).padStart(11)} ${d.row.description.slice(0, 34)}` + - ` id ${d.row.transactionId} duplicates ${d.duplicateOf.transactionId} (gap ${d.idGap})` - ); - } - // Ids this close look like a genuine same-day repeat rather than a twin. - // Nothing in scope has ever tripped it; if something does, look before - // trusting the collapse. - const suspect = dropped.filter((d) => d.suspectGenuineRepeat); - if (suspect.length > 0) { - console.log(`\n REVIEW: ${suspect.length} collapsed row(s) had near-consecutive ids and may be`); - console.log(" real repeats rather than duplicates:"); - for (const d of suspect.slice(0, 10)) { - console.log( - ` ${d.row.transactionDate} ${money(d.row.amount).padStart(11)} ${d.row.description.slice(0, 40)}` + - ` ids ${d.duplicateOf.transactionId}/${d.row.transactionId} gap ${d.idGap}` - ); - } - } -} -console.log(); - -// ---- 4. what would change ------------------------------------------------ -const ledger = kept.map(toLedgerRow); -if (!process.env.DATABASE_URL) { - console.error("DATABASE_URL is not set — cannot compare against the ledger."); - process.exit(2); -} const client = new pg.Client({ connectionString: process.env.DATABASE_URL }); await client.connect(); +const exec: SqlExecutor = async (sql: string, params: unknown[]) => + (await client.query(sql, params)).rows as T[]; -const existing = await client.query<{ source_ref: string }>( - "SELECT source_ref FROM transactions WHERE source = $1", - [SOURCE] -); -const seen = new Set(existing.rows.map((r) => r.source_ref)); -const fresh = ledger.filter((r) => !seen.has(r.sourceRef)); -const already = ledger.length - fresh.length; +let report; +try { + report = await ingestFrolloCsv(readFileSync(file, "utf8"), exec, { + ownerId: Number(arg("owner", "1")), + apply: has("apply"), + force: has("force"), + }); +} finally { + // Left open, the process hangs after an exception and looks like a slow query. + if (!has("keep-open")) await client.end().catch(() => {}); +} + +console.log(`\n${file}`); +console.log(` ${report.totalRows} rows in file\n`); + +console.log("SCOPE"); +for (const [why, n] of Object.entries(report.skipped).sort((a, b) => b[1]! - a[1]!)) { + console.log(` skipped ${String(n).padStart(5)} ${why}`); +} +console.log(` in scope ${String(report.inScope).padStart(4)}\n`); + +console.log("ACCOUNTS"); +for (const a of report.accounts) { + console.log( + ` ${a.present ? " " : "!"} ${a.spec.last4} ${a.spec.label.padEnd(24)} ${String(a.rows).padStart(5)} rows` + ); +} +if (report.unknownAccounts.length > 0) { + console.log("\n accounts in the file this importer does not know (skipped):"); + for (const u of report.unknownAccounts) { + console.log(` ${u.accountNumber} ${u.accountName} ${u.rows} rows`); + } +} +console.log(); + +console.log("DE-DUPLICATION (re-ingest twins from CDR re-consent)"); +console.log(` ${report.duplicatesDropped} dropped`); +if (report.suspectRepeats.length > 0) { + console.log(`\n REVIEW: ${report.suspectRepeats.length} collapsed row(s) had near-consecutive ids`); + console.log(" and may be real repeats rather than duplicates:"); + for (const s of report.suspectRepeats.slice(0, 10)) { + console.log(` ${s.transactionDate} ${money(s.amount).padStart(11)} ${s.description.slice(0, 40)} (gap ${s.idGap})`); + } +} +console.log(); console.log("LEDGER"); -console.log(` already imported : ${String(already).padStart(5)}`); -console.log(` new to insert : ${String(fresh.length).padStart(5)}`); -const debits = fresh.filter((r) => r.transactionType === "debit"); -const credits = fresh.filter((r) => r.transactionType === "credit"); -const aud = (rs: typeof fresh) => rs.filter((r) => !r.foreignCurrencyCode); -console.log( - ` of which AUD : ${aud(debits).length} debits ${money(aud(debits).reduce((s, r) => s + r.amount, 0))}` + - ` / ${aud(credits).length} credits ${money(aud(credits).reduce((s, r) => s + r.amount, 0))}` -); -const fx = fresh.filter((r) => r.foreignCurrencyCode); -if (fx.length > 0) { - const byCcy = new Map(); - for (const r of fx) byCcy.set(r.foreignCurrencyCode!, (byCcy.get(r.foreignCurrencyCode!) ?? 0) + 1); - console.log(` foreign : ${[...byCcy].map(([c, n]) => `${n} ${c}`).join(", ")} (amount_aud left NULL)`); -} -if (fresh.length > 0) { - console.log("\n first rows to insert:"); - for (const r of fresh.slice(0, 8)) { - console.log( - ` ${r.transactionDate} ${r.transactionType.padEnd(6)} ${money(r.amount).padStart(11)}` + - ` ${(r.foreignCurrencyCode ?? "AUD").padEnd(4)} ${r.description.slice(0, 40)}` - ); - } +console.log(` already imported : ${String(report.alreadyImported).padStart(5)}`); +console.log(` new to insert : ${String(report.toInsert).padStart(5)}`); + +if (report.anomalies.length > 0) { + console.log("\nANOMALIES"); + for (const a of report.anomalies) console.log(` ! ${a}`); } -// ---- 5. apply ------------------------------------------------------------ -if (!apply) { - console.log("\nDRY RUN — nothing written. Re-run with --apply to insert.\n"); - await client.end(); - process.exit(0); -} - -if (fresh.length === 0) { - console.log("\nNothing to insert.\n"); - await client.end(); - process.exit(0); -} - -// row_index is left NULL: it exists to order statement lines, and the -// uniqueness this import relies on is uq_transaction_source_ref, not -// uq_transaction_identity. -// -// category is not set, because Frollo's own categories are not trustworthy. Note -// what actually lands: trg_transactions_normalize_category rewrites NULL to -// 'other' on insert, so these rows arrive categorised 'other' rather than -// uncategorised. That is equivalent for display and for the rules engine, which -// writes transaction_overrides.category_override and is read ahead of -// t.category by EFFECTIVE_CATEGORY — so a rule still wins. -let inserted = 0; -try { - await client.query("BEGIN"); - for (const r of fresh) { - const res = await client.query( - `INSERT INTO transactions - (statement_id, owner_id, transaction_date, description, amount, transaction_type, - merchant_name, foreign_currency_amount, foreign_currency_code, - source, source_ref, source_account) - VALUES (NULL, $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) - ON CONFLICT (source, source_ref) WHERE source IS NOT NULL AND source_ref IS NOT NULL - DO NOTHING`, - [ - ownerId, - r.transactionDate, - r.description, - r.amount, - r.transactionType, - r.merchantName, - r.foreignCurrencyAmount, - r.foreignCurrencyCode, - SOURCE, - r.sourceRef, - r.sourceAccount, - ] - ); - inserted += res.rowCount ?? 0; - } - await client.query("COMMIT"); -} catch (e) { - await client.query("ROLLBACK"); - console.error("\nROLLED BACK:", e); - await client.end(); +if (report.applied) { + console.log(`\nInserted ${report.inserted} row(s).\n`); +} else if (report.anomalies.length > 0 && has("apply")) { + console.log("\nNOT APPLIED — anomalies above. Read them, then re-run with --force.\n"); process.exit(1); +} else { + console.log("\nDRY RUN — nothing written. Re-run with --apply to insert.\n"); } - -const after = await client.query<{ n: string }>( - "SELECT count(*) AS n FROM transactions WHERE source = $1", - [SOURCE] -); -console.log(`\nInserted ${inserted} row(s). ${after.rows[0].n} Frollo rows now in the ledger.\n`); -await client.end(); diff --git a/src/__tests__/unit/frollo-csv.test.ts b/src/__tests__/unit/frollo-csv.test.ts index bbd2aac..b15775b 100644 --- a/src/__tests__/unit/frollo-csv.test.ts +++ b/src/__tests__/unit/frollo-csv.test.ts @@ -2,6 +2,8 @@ import { describe, it, expect } from "vitest"; import { parseCSVRows } from "@/lib/csv-parser"; import { ACCOUNTS, + EXCLUDED_ACCOUNTS, + isKnownExcluded, SUSPECT_REPEAT_GAP, accountFor, accountReport, @@ -14,6 +16,7 @@ import { unknownAccounts, type FrolloRow, } from "@/lib/frollo-csv"; +import { LARGE_BATCH, findAnomalies } from "@/lib/frollo-ingest"; const HEADER = "transaction_id,description,user_description,amount,currency,transaction_date,posted_date," + @@ -307,3 +310,79 @@ describe("accountReport / unknownAccounts", () => { expect(unknown[0]).toMatchObject({ accountName: "New Card", rows: 1 }); }); }); + +describe("findAnomalies", () => { + const base = { missingAccounts: [], unknownAccounts: [], suspectRepeats: [], toInsert: 10, skipped: {} }; + + it("passes a boring run", () => { + expect(findAnomalies(base)).toEqual([]); + }); + + it("flags a configured account that contributed nothing — a lapsed consent", () => { + const out = findAnomalies({ ...base, missingAccounts: ["4830 Wise (income)"] }); + expect(out).toHaveLength(1); + expect(out[0]).toMatch(/consent may have lapsed/i); + }); + + it("flags an account that appeared in the file unannounced", () => { + const out = findAnomalies({ + ...base, + unknownAccounts: [{ accountNumber: "xxxx9999", accountName: "New Everyday", rows: 4 }], + }); + expect(out[0]).toMatch(/unknown account/i); + }); + + it("flags a de-duplication that might have collapsed a real repeat", () => { + const out = findAnomalies({ + ...base, + suspectRepeats: [{ description: "X", transactionDate: "2026-01-01", amount: -1, idGap: 2 }], + }); + expect(out[0]).toMatch(/genuine repeats/i); + }); + + it("flags a batch far larger than an ordinary month", () => { + expect(findAnomalies({ ...base, toInsert: LARGE_BATCH + 1 })).toHaveLength(1); + expect(findAnomalies({ ...base, toInsert: LARGE_BATCH })).toHaveLength(0); + }); + + it("flags an export taken with pending included", () => { + const out = findAnomalies({ ...base, skipped: { pending: 3 } }); + expect(out[0]).toMatch(/pending/i); + }); + + it("reports every anomaly at once rather than the first", () => { + expect( + findAnomalies({ ...base, missingAccounts: ["a"], toInsert: LARGE_BATCH + 1, skipped: { pending: 1 } }) + ).toHaveLength(3); + }); +}); + +describe("excluded vs unknown accounts", () => { + it("does not report a deliberately excluded credit card as unknown", () => { + // An alert that fires on every run is an alert nobody reads. + const rows = csv(line({ id: "1", amount: "-50.00", account: "xxxx xxxx xxxx 3893", accountName: "Ultimate Awards credit card" })); + expect(unknownAccounts(rows)).toEqual([]); + expect(isKnownExcluded("xxxx xxxx xxxx 3893")).toBe(true); + }); + + it("still reports a genuinely new account", () => { + const rows = csv(line({ id: "1", amount: "-50.00", account: "xxxx xxxx xxxx 9999", accountName: "Brand New" })); + expect(unknownAccounts(rows)).toHaveLength(1); + }); + + it("keeps the imported and excluded rosters disjoint", () => { + const a = new Set(ACCOUNTS.map((x) => x.last4)); + for (const e of EXCLUDED_ACCOUNTS) expect(a.has(e.last4)).toBe(false); + }); +}); + +describe("pending detection precedes account scoping", () => { + it("reports a pending credit-card row as pending, not out of scope", () => { + // Otherwise the wrong-export-option signal never fires, since every pending + // row observed so far has been on a card. + const row = csv( + line({ id: "1", amount: "-13.50", desc: "Pending - Breadworld Bakery Cafe", account: "xxxx xxxx xxxx 8032" }) + )[0]; + expect(skipReason(row)).toBe("pending"); + }); +}); diff --git a/src/app/api/frollo/ingest/route.ts b/src/app/api/frollo/ingest/route.ts new file mode 100644 index 0000000..535b577 --- /dev/null +++ b/src/app/api/frollo/ingest/route.ts @@ -0,0 +1,62 @@ +import { NextRequest, NextResponse } from "next/server"; +import { queryRaw } from "@/lib/db"; +import { ingestFrolloCsv } from "@/lib/frollo-ingest"; + +/** + * Machine ingest endpoint for a Frollo transaction export. + * + * Sibling to /api/orders/ingest and /api/receipts/ingest, shaped like them. Auth + * is a shared secret rather than the Traefik `x-forwarded-user` header, because + * n8n calls this app-to-app with no browser session to forward. + * + * Its own token so it can be rotated without touching the order or receipt + * flows, each of which runs on a schedule nobody is watching. + * + * The body is the CSV itself, as text. The caller does not parse anything — + * parsing, scoping, de-duplication and the insert all live in one place + * (`frollo-ingest.ts`), shared with `scripts/import-frollo.mts`. + */ +function authorised(req: NextRequest): boolean { + const expected = process.env.FROLLO_INGEST_TOKEN; + if (!expected) return false; // fail closed when unconfigured + const got = req.headers.get("x-ingest-token"); + return !!got && got === expected; +} + +export async function POST(req: NextRequest) { + if (!authorised(req)) return NextResponse.json({ error: "Unauthorized" }, { status: 401 }); + + let body: { csv?: string; dryRun?: boolean; force?: boolean }; + try { + body = await req.json(); + } catch { + return NextResponse.json({ error: "invalid JSON" }, { status: 400 }); + } + if (!body?.csv || typeof body.csv !== "string") { + return NextResponse.json({ error: "csv (string) is required" }, { status: 400 }); + } + + try { + const report = await ingestFrolloCsv(body.csv, queryRaw, { + apply: !body.dryRun, + force: body.force === true, + }); + + // An automatic caller must not be able to write through an anomaly by + // accident: the run reports 409 and inserts nothing, so a human decides. + // Today's session is the argument — the de-duplication rule passed 29 tests + // and a clean dry run while still doubling three salary payments. + if (!report.applied && report.anomalies.length > 0 && !body.dryRun) { + return NextResponse.json({ kind: "needs_review", ...report }, { status: 409 }); + } + return NextResponse.json({ kind: "frollo", ...report }); + } catch (e) { + // A missing or renamed column throws rather than importing blank rows, and + // that is worth surfacing as loudly as an unparseable receipt. + const message = e instanceof Error ? e.message : String(e); + if (/missing expected column|empty CSV/i.test(message)) { + return NextResponse.json({ kind: "rejected", reason: message }, { status: 422 }); + } + return NextResponse.json({ error: message }, { status: 500 }); + } +} diff --git a/src/lib/frollo-csv.ts b/src/lib/frollo-csv.ts index 24294dd..8eb1c74 100644 --- a/src/lib/frollo-csv.ts +++ b/src/lib/frollo-csv.ts @@ -97,6 +97,30 @@ export const ACCOUNTS: AccountSpec[] = [ { last4: "9807", label: "ANZ Online Saver", provider: "ANZ" }, ]; +/** + * Accounts knowingly left out, so "excluded" is distinguishable from "unknown". + * + * Without this the five credit cards read as unrecognised accounts on every + * single run, which would raise the new-account anomaly every time and make an + * automatic import permanently refuse. An alert that always fires is an alert + * nobody reads. + * + * Anything in neither list is genuinely new — a freshly connected account, or a + * re-masked number — and does deserve a look before its rows are ignored. + */ +export const EXCLUDED_ACCOUNTS: AccountSpec[] = [ + { last4: "3893", label: "CommBank Ultimate Awards (card)", provider: "CommBank" }, + { last4: "8032", label: "Westpac Altitude Qantas Black (card)", provider: "Westpac" }, + { last4: "1000", label: "Amex Platinum Business (card)", provider: "American Express" }, + { last4: "0351", label: "HSBC Star Alliance (card)", provider: "HSBC" }, + { last4: "6227", label: "MyCard Prestige (card)", provider: "Mycard" }, +]; + +export function isKnownExcluded(accountNumber: string): boolean { + const key = last4(accountNumber); + return EXCLUDED_ACCOUNTS.some((a) => a.last4 === key); +} + /** * Below this id gap, two rows sharing a natural key are *probably* a genuine * same-day repeat rather than a re-ingest twin — and are reported rather than @@ -196,8 +220,13 @@ export type SkipReason = export function skipReason(row: FrolloRow): SkipReason | null { if (!Number.isFinite(row.amount)) return "unparseable_amount"; if (row.amount === 0) return "zero_amount"; - if (!accountFor(row.accountNumber)) return "out_of_scope_account"; + // Pending is tested before account scope on purpose. Every pending row seen so + // far has been on a credit card, so scoping first would classify them all as + // out-of-scope and the "this export was taken with pending INCLUDED" signal + // would never fire — even though it is worth acting on wherever it appears, + // because the option makes successive exports incomparable. if (/^\s*pending\s*[-:]/i.test(row.description)) return "pending"; + if (!accountFor(row.accountNumber)) return "out_of_scope_account"; return null; } @@ -333,11 +362,17 @@ export function accountReport(rows: FrolloRow[]): AccountReport[] { })); } -/** Accounts in the file that this importer does not know about. */ +/** + * Accounts in the file that are neither imported nor knowingly excluded. + * + * Empty is the normal state. A name appearing here means a new account reached + * the export and its rows are being dropped on the floor — worth a look before + * that becomes the status quo. + */ export function unknownAccounts(rows: FrolloRow[]): { accountNumber: string; accountName: string; rows: number }[] { const seen = new Map(); for (const r of rows) { - if (accountFor(r.accountNumber)) continue; + if (accountFor(r.accountNumber) || isKnownExcluded(r.accountNumber)) continue; const k = last4(r.accountNumber); const e = seen.get(k); if (e) e.rows++; diff --git a/src/lib/frollo-ingest.ts b/src/lib/frollo-ingest.ts new file mode 100644 index 0000000..ef27865 --- /dev/null +++ b/src/lib/frollo-ingest.ts @@ -0,0 +1,201 @@ +import { + accountReport, + dedupe, + readRows, + skipReason, + toLedgerRow, + unknownAccounts, + type AccountReport, + type FrolloRow, + type SkipReason, +} from "./frollo-csv.ts"; +import { parseCSVRows } from "./csv-parser.ts"; + +/** + * Turns a Frollo CSV into ledger rows and (optionally) writes them. + * + * Shared by `scripts/import-frollo.mts` and `POST /api/frollo/ingest` so the + * hand-run and automatic paths cannot drift. They previously would have been two + * implementations of the same insert, which is exactly how the pantry healthcheck + * came to be fixed in one repo and left broken in the other. + * + * The database is reached through an injected executor rather than imported + * directly: the API route runs inside Next with Prisma, while the CLI runs under + * `node --experimental-strip-types` with a plain `pg` client and no path aliases. + */ +// The .ts extensions above are load-bearing: this module is imported both by +// Next (which resolves either form) and by scripts/import-frollo.mts running +// under `node --experimental-strip-types`, whose ESM resolver requires the +// extension written out. tsconfig sets allowImportingTsExtensions for this. +export type SqlExecutor = (sql: string, params: unknown[]) => Promise; + +export const SOURCE = "frollo"; + +/** + * Above this many new rows, an automatic run stops and asks. + * + * Steady state is ~50 rows a month. A batch several times that means something + * changed — a re-consent duplicating history, a longer export window, or a new + * account — and none of those should land unseen. The first (backfill) import + * was 550 and was run by hand with `force`. + */ +export const LARGE_BATCH = 200; + +export interface IngestOptions { + ownerId?: number; + /** Write. Without it, everything is computed and nothing is inserted. */ + apply?: boolean; + /** Proceed despite anomalies. For a human who has read them. */ + force?: boolean; + largeBatch?: number; +} + +export interface IngestReport { + totalRows: number; + inScope: number; + skipped: Partial>; + accounts: AccountReport[]; + missingAccounts: string[]; + unknownAccounts: { accountNumber: string; accountName: string; rows: number }[]; + duplicatesDropped: number; + suspectRepeats: { description: string; transactionDate: string; amount: number; idGap: number }[]; + alreadyImported: number; + toInsert: number; + inserted: number; + /** Non-empty means an automatic run should stop and a human should look. */ + anomalies: string[]; + applied: boolean; +} + +/** + * Reasons an automatic import should stop rather than write. + * + * Pure, so it is testable without a database. Every entry is a thing that has + * either happened or come one verification away from happening: a lapsed CDR + * consent silently shrinking the file, a new account appearing unnoticed, a + * de-duplication that might have collapsed a real repeat, and a batch far larger + * than the account activity can explain. + */ +export function findAnomalies( + report: Pick, + largeBatch = LARGE_BATCH +): string[] { + const out: string[] = []; + if (report.missingAccounts.length > 0) { + out.push( + `${report.missingAccounts.length} configured account(s) contributed no rows ` + + `(${report.missingAccounts.join(", ")}) — a CDR consent may have lapsed` + ); + } + if (report.unknownAccounts.length > 0) { + out.push( + `${report.unknownAccounts.length} unknown account(s) in the file, skipped ` + + `(${report.unknownAccounts.map((a) => a.accountName).join(", ")})` + ); + } + if (report.suspectRepeats.length > 0) { + out.push( + `${report.suspectRepeats.length} de-duplicated row(s) had near-consecutive ids ` + + `and may be genuine repeats rather than twins` + ); + } + if (report.toInsert > largeBatch) { + out.push(`${report.toInsert} new rows is more than the expected ${largeBatch}`); + } + if ((report.skipped.pending ?? 0) > 0) { + out.push( + `${report.skipped.pending} pending row(s) present — export was taken with pending ` + + `INCLUDED; they are skipped, but re-export with it excluded` + ); + } + return out; +} + +export async function ingestFrolloCsv( + csvText: string, + exec: SqlExecutor, + opts: IngestOptions = {} +): Promise { + const ownerId = opts.ownerId ?? 1; + const rows = readRows(parseCSVRows(csvText)); + + const skipped: Partial> = {}; + const inScope: FrolloRow[] = []; + for (const r of rows) { + const why = skipReason(r); + if (why) skipped[why] = (skipped[why] ?? 0) + 1; + else inScope.push(r); + } + + const accounts = accountReport(inScope); + const missingAccounts = accounts.filter((a) => !a.present).map((a) => `${a.spec.last4} ${a.spec.label}`); + const unknown = unknownAccounts(rows); + + const { kept, dropped } = dedupe(inScope); + const suspectRepeats = dropped + .filter((d) => d.suspectGenuineRepeat) + .map((d) => ({ + description: d.row.description, + transactionDate: d.row.transactionDate, + amount: d.row.amount, + idGap: d.idGap, + })); + + const ledger = kept.map(toLedgerRow); + const existing = await exec<{ source_ref: string }>( + "SELECT source_ref FROM transactions WHERE source = $1", + [SOURCE] + ); + const seen = new Set(existing.map((r) => r.source_ref)); + const fresh = ledger.filter((r) => !seen.has(r.sourceRef)); + + const base = { + totalRows: rows.length, + inScope: inScope.length, + skipped, + accounts, + missingAccounts, + unknownAccounts: unknown, + duplicatesDropped: dropped.length, + suspectRepeats, + alreadyImported: ledger.length - fresh.length, + toInsert: fresh.length, + }; + const anomalies = findAnomalies(base, opts.largeBatch); + + if (!opts.apply || (anomalies.length > 0 && !opts.force)) { + return { ...base, inserted: 0, anomalies, applied: false }; + } + + let inserted = 0; + for (const r of fresh) { + // ON CONFLICT against uq_transaction_source_ref: belt and braces over the + // source_ref check above, which is not transactional. + const res = await exec<{ id: number }>( + `INSERT INTO transactions + (statement_id, owner_id, transaction_date, description, amount, transaction_type, + merchant_name, foreign_currency_amount, foreign_currency_code, + source, source_ref, source_account) + VALUES (NULL, $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) + ON CONFLICT (source, source_ref) WHERE source IS NOT NULL AND source_ref IS NOT NULL + DO NOTHING + RETURNING id`, + [ + ownerId, + r.transactionDate, + r.description, + r.amount, + r.transactionType, + r.merchantName, + r.foreignCurrencyAmount, + r.foreignCurrencyCode, + SOURCE, + r.sourceRef, + r.sourceAccount, + ] + ); + inserted += res.length; + } + + return { ...base, inserted, anomalies, applied: true }; +}