Something went wrong. Try again.
A self-hosted household ledger.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356import type { DatabaseSync } from "node:sqlite";import { fetchAccounts } from "../simplefin.ts";import { type NormalizedAccount, normalizePayload } from "./normalize.ts";import { listConnections } from "./connections.ts";import { appendCategorizationEvent } from "./categorization.ts";import { applyRulesToUncategorized } from "./rules.ts";
// The Bridge recommends ranges <= 45 days (observed: larger requests add a// "Requested date range exceeds recommended range of 45 days" advisory to the// payload's errors). Routine daily syncs stay under that; when a sync// discovers an account it has never seen, a one-off deep fetch pulls that// account's full available backfill (~90 days) via the idempotent upserts.const ROUTINE_LOOKBACK_DAYS = 44;const BACKFILL_LOOKBACK_DAYS = 89;/** Max gap between a pending transaction and the posted one replacing it. */const RECONCILE_WINDOW_DAYS = 5;
export interface SyncOutcome { connectionId: number; ok: boolean; error?: string; newTransactions: number; reconciled: number; removedPending: number; ruleCategorized: number; connectionErrors: string[];}
/** Sync every connection: archive verbatim, normalize, snapshot, reconcile, rule-fire. */export async function runSync( db: DatabaseSync, fetchFn: typeof fetch = fetch,): Promise<SyncOutcome[]> { const outcomes: SyncOutcome[] = []; for (const connection of listConnections(db)) { outcomes.push( await syncConnection(db, connection.id, connection.accessUrl, fetchFn), ); } return outcomes;}
async function syncConnection( db: DatabaseSync, connectionId: number, accessUrl: string, fetchFn: typeof fetch,): Promise<SyncOutcome> { const outcome: SyncOutcome = { connectionId, ok: true, newTransactions: 0, reconciled: 0, removedPending: 0, ruleCategorized: 0, connectionErrors: [], };
const routine = await syncPass( db, connectionId, accessUrl, ROUTINE_LOOKBACK_DAYS, outcome, fetchFn, );
// Never-before-seen accounts (including all of them, on a connection's very // first sync) get one deep fetch for their full available backfill. if (routine.newAccountIds.length > 0) { await syncPass( db, connectionId, accessUrl, BACKFILL_LOOKBACK_DAYS, outcome, fetchFn, ); }
if (outcome.ok) outcome.ruleCategorized = applyRulesToUncategorized(db); return outcome;}
/** One fetch → archive → normalize → ingest pass; mutates `outcome` counters. */async function syncPass( db: DatabaseSync, connectionId: number, accessUrl: string, lookbackDays: number, outcome: SyncOutcome, fetchFn: typeof fetch,): Promise<{ newAccountIds: string[] }> { const startDate = new Date(Date.now() - lookbackDays * 86400_000); const fetched = await fetchAccounts( accessUrl, { startDate, pending: true }, fetchFn, ); const fetchedAt = new Date().toISOString();
// Archive verbatim before any processing — success or failure. db.prepare( "INSERT INTO raw_syncs (connection_id, fetched_at, ok, payload, error) VALUES (?, ?, ?, ?, ?)", ).run( connectionId, fetchedAt, fetched.ok ? 1 : 0, fetched.body ?? null, fetched.error ?? null, );
if (!fetched.ok || !fetched.body) { outcome.ok = false; outcome.error = fetched.error; return { newAccountIds: [] }; }
let normalized; try { normalized = normalizePayload(fetched.body); } catch (err) { outcome.ok = false; outcome.error = `Normalization failed: ${ err instanceof Error ? err.message : err }`; db.prepare( "UPDATE raw_syncs SET ok = 0, error = ? WHERE connection_id = ? AND fetched_at = ?", ).run( outcome.error, connectionId, fetchedAt, ); return { newAccountIds: [] }; } outcome.connectionErrors = normalized.errors;
const newAccountIds: string[] = []; for (const account of normalized.accounts) { if (upsertAccount(db, connectionId, account, fetchedAt)) { newAccountIds.push(account.id); } const result = ingestTransactions(db, account); outcome.newTransactions += result.inserted; outcome.reconciled += result.reconciled; outcome.removedPending += result.removedPending; db.prepare( "INSERT INTO balance_snapshots (account_id, captured_at, balance_cents, available_balance_cents) VALUES (?, ?, ?, ?)", ).run( account.id, fetchedAt, account.balanceCents, account.availableBalanceCents, ); }
// Lifecycle: ACTIVE accounts of this connection absent from the feed go INACTIVE. const seenIds = normalized.accounts.map((a) => a.id); const placeholders = seenIds.map(() => "?").join(","); db.prepare( `UPDATE accounts SET state = 'INACTIVE' WHERE connection_id = ? AND state = 'ACTIVE' ${seenIds.length ? `AND id NOT IN (${placeholders})` : ""}`, ).run(connectionId, ...seenIds);
return { newAccountIds };}
/** Returns true when the account was newly discovered by this pass. */function upsertAccount( db: DatabaseSync, connectionId: number, account: NormalizedAccount, fetchedAt: string,): boolean { const existing = db.prepare("SELECT id, state FROM accounts WHERE id = ?") .get(account.id) as | { id: string; state: string } | undefined; if (!existing) { db.prepare( `INSERT INTO accounts (id, connection_id, org_name, org_domain, org_sfin_url, name, currency, state, last_successful_data_at, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, 'NEW', ?, ?)`, ).run( account.id, connectionId, account.org.name, account.org.domain, account.org.sfinUrl, account.name, account.currency, fetchedAt, fetchedAt, ); return true; } // Reappearing INACTIVE accounts return to ACTIVE if already classified, NEW otherwise. db.prepare( `UPDATE accounts SET org_name = ?, org_domain = ?, org_sfin_url = ?, name = ?, currency = ?, last_successful_data_at = ?, state = CASE WHEN state = 'INACTIVE' AND account_type IS NOT NULL THEN 'ACTIVE' WHEN state = 'INACTIVE' THEN 'NEW' ELSE state END WHERE id = ?`, ).run( account.org.name, account.org.domain, account.org.sfinUrl, account.name, account.currency, fetchedAt, account.id, ); return false;}
function ingestTransactions( db: DatabaseSync, account: NormalizedAccount,): { inserted: number; reconciled: number; removedPending: number } { const now = new Date().toISOString(); let inserted = 0; let reconciled = 0;
const feedSfinIds = new Set(account.transactions.map((t) => t.sfinId));
for (const txn of account.transactions) { const existing = db .prepare( "SELECT id, pending FROM transactions WHERE account_id = ? AND sfin_id = ?", ) .get(account.id, txn.sfinId) as | { id: number; pending: number } | undefined;
if (existing) { // Idempotent refresh; a known-pending transaction that posts under the // SAME id is just updated in place. db.prepare( `UPDATE transactions SET posted = ?, transacted_at = ?, amount_cents = ?, description = ?, payee = ?, memo = ?, pending = ?, extra = ?, removed_at = NULL WHERE id = ?`, ).run( txn.posted, txn.transactedAt, txn.amountCents, txn.description, txn.payee, txn.memo, txn.pending ? 1 : 0, txn.extra, existing.id, ); continue; }
if (!txn.pending) { // New posted transaction: does it replace a pending row that vanished // from the feed? Match: same account + exact amount + date proximity. const windowStart = (txn.posted ?? 0) - RECONCILE_WINDOW_DAYS * 86400; const windowEnd = (txn.posted ?? 0) + RECONCILE_WINDOW_DAYS * 86400; const match = db .prepare( `SELECT id, sfin_id FROM transactions WHERE account_id = ? AND pending = 1 AND removed_at IS NULL AND amount_cents = ? AND COALESCE(transacted_at, posted, ?) BETWEEN ? AND ? ORDER BY id LIMIT 1`, ) .get( account.id, txn.amountCents, txn.posted ?? 0, windowStart, windowEnd, ) as | { id: number; sfin_id: string } | undefined;
if (match && !feedSfinIds.has(match.sfin_id)) { // Replace in place: same row id keeps the event history attached; // a reconciliation event records the carry-forward. const category = db .prepare("SELECT category_id FROM transactions WHERE id = ?") .get(match.id) as { category_id: number | null }; db.prepare( `UPDATE transactions SET sfin_id = ?, posted = ?, transacted_at = ?, amount_cents = ?, description = ?, payee = ?, memo = ?, pending = 0, extra = ?, removed_at = NULL WHERE id = ?`, ).run( txn.sfinId, txn.posted, txn.transactedAt, txn.amountCents, txn.description, txn.payee, txn.memo, txn.extra, match.id, ); if (category.category_id != null) { appendCategorizationEvent(db, { transactionId: match.id, categoryId: category.category_id, source: "reconciliation", }); } reconciled++; continue; } }
db.prepare( `INSERT INTO transactions (account_id, sfin_id, posted, transacted_at, amount_cents, description, payee, memo, pending, extra, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, ).run( account.id, txn.sfinId, txn.posted, txn.transactedAt, txn.amountCents, txn.description, txn.payee, txn.memo, txn.pending ? 1 : 0, txn.extra, now, ); inserted++; }
// Stale pending rows: still pending in the DB, absent from the feed, and // not matched by any posted transaction above. Soft-remove (events are // never deleted). const stale = db .prepare( "SELECT id, sfin_id FROM transactions WHERE account_id = ? AND pending = 1 AND removed_at IS NULL", ) .all(account.id) as { id: number; sfin_id: string }[]; let removedPending = 0; for (const row of stale) { if (feedSfinIds.has(row.sfin_id)) continue; db.prepare("UPDATE transactions SET removed_at = ? WHERE id = ?").run( now, row.id, ); removedPending++; }
return { inserted, reconciled, removedPending };}