finance-app/src/lib/frollo-ingest.ts

import {
  accountReport,
  dedupe,
  last4,
  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 = <T = unknown>(sql: string, params: unknown[]) => Promise<T[]>;

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;

/**
 * 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;

/**
 * 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"]);

/**
 * 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.
 */
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. */
  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<Record<SkipReason, number>>;
  accounts: AccountReport[];
  missingAccounts: string[];
  unknownAccounts: { accountNumber: string; accountName: string; rows: number }[];
  duplicatesDropped: number;
  suspectRepeats: { description: string; transactionDate: string; amount: number; idGap: number }[];
  alreadyImported: number;
  /**
   * Rows dropped because a statement already covers that account up to that
   * date. The primary guard — see `statementWatermarks`.
   */
  coveredByStatement: number;
  /** Per-account coverage boundary used above, for the report. */
  statementWatermarks: { last4: string; coveredTo: string }[];
  /**
   * Rows dropped because the ledger holds the same amount, same direction,
   * within a few days. The secondary guard, for accounts with no statement.
   *
   * Both counts are 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. */
  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<IngestReport, "missingAccounts" | "unknownAccounts" | "suspectRepeats" | "toInsert" | "skipped">,
  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<IngestReport> {
  const ownerId = opts.ownerId ?? 1;
  const rows = readRows(parseCSVRows(csvText));

  const skipped: Partial<Record<SkipReason, number>> = {};
  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 notYetImported = ledger.filter((r) => !seen.has(r.sourceRef));

  // GUARD 1 — statement coverage. Do not import what the statements already
  // have.
  //
  // A statement's `billing_end_date` is a hard watermark: everything on that
  // account up to that date is already in the ledger, itemised and converted.
  // Amount-matching cannot replace this, because the two sources decompose the
  // same event differently — Frollo bundles the Wise fee into the transfer
  // (10001.13) where the statement itemises it (10000.00 + 1.13), so 38 Wise USD
  // rows survived the amount guard as "new" while being the same money.
  //
  // This is the coverage test the first import needed and got wrong. It asked
  // whether a row's date fell inside a statement's min–max *window*, which for
  // periods spanning 182 to 460 days swallows a year and answers nothing. The
  // watermark asks a different and answerable question: up to what date is this
  // account complete?
  //
  // Matching is on last4 alone. Verified against the live statement set: the
  // eight in-scope accounts with statements each map to exactly one bank, no
  // cross-bank collision. The per-account watermarks are reported so a wrong one
  // is visible rather than inferred from a row count.
  const coverage = await exec<{ last4: string; covered_to: unknown }>(
    `SELECT right(regexp_replace(account_number, '[^0-9]', '', 'g'), 4) AS last4,
            MAX(billing_end_date) AS covered_to
       FROM statements
      WHERE account_number IS NOT NULL
      GROUP BY 1`,
    []
  );
  const watermark = new Map<string, number>();
  for (const c of coverage) {
    if (!c.last4) continue; // Wise files one statement as account_number 'N/A'
    const t = dayMs(c.covered_to);
    if (!Number.isFinite(t)) continue;
    watermark.set(c.last4, Math.max(watermark.get(c.last4) ?? -Infinity, t));
  }

  let coveredByStatement = 0;
  const uncovered = notYetImported.filter((r) => {
    const w = watermark.get(last4(r.sourceAccount));
    if (w === undefined) return true; // no statement for this account, ever
    if (dayMs(r.transactionDate) <= w) {
      coveredByStatement += 1;
      return false;
    }
    return true;
  });

  const statementWatermarks = [...watermark.entries()]
    .filter(([l4]) => notYetImported.some((r) => last4(r.sourceAccount) === l4))
    .map(([l4, t]) => ({ last4: l4, coveredTo: new Date(t).toISOString().slice(0, 10) }))
    .sort((a, b) => a.last4.localeCompare(b.last4));

  // GUARD 2 — 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 = uncovered.map((r) => dayMs(r.transactionDate)).filter(Number.isFinite);
  let ledgerDuplicates = 0;
  let fresh = uncovered;
  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 = uncovered.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,
    inScope: inScope.length,
    skipped,
    accounts,
    missingAccounts,
    unknownAccounts: unknown,
    duplicatesDropped: dropped.length,
    suspectRepeats,
    alreadyImported: ledger.length - notYetImported.length,
    coveredByStatement,
    statementWatermarks,
    ledgerDuplicates,
    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 };
}