From a03216ee55d1d96501ecb0d1d7e57056167cc097 Mon Sep 17 00:00:00 2001 From: null Date: Sun, 5 Jul 2026 18:54:34 -0500 Subject: [PATCH] perf(simplefin): batch sync writes in a transaction + align dedup key MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Wrap each source's account/transaction upserts + the status update in one better-sqlite3 db.transaction() — a single commit instead of one implicit commit per row (far fewer fsyncs on large syncs) and atomic (a mid-sync failure can't leave a half-written account). The write block is synchronous; the async SimpleFIN fetch already completed before it. - Key the dedup lookup on (user_id, provider_transaction_id) to match the UNIQUE index and the reconnect-stable provider id. A data_source_id-keyed lookup missed the prior row after a disconnect/reconnect, so pending→posted and amount-refresh were skipped; now they survive a reconnect. Co-Authored-By: Claude Opus 4.8 --- services/bankSyncService.js | 97 +++++++++++++++++++++---------------- 1 file changed, 55 insertions(+), 42 deletions(-) diff --git a/services/bankSyncService.js b/services/bankSyncService.js index 1b49534..9f65902 100644 --- a/services/bankSyncService.js +++ b/services/bankSyncService.js @@ -64,10 +64,14 @@ function upsertAccount(db, accountRow) { // a row the user has matched or ignored is never touched. // Returns: 'inserted' | 'posted' (pending→settled) | 'updated' (pending refreshed) | 'skipped'. function upsertTransaction(db, txRow) { + // Key on (user_id, provider_transaction_id) to match the UNIQUE dedup index — + // provider_transaction_id is deliberately stable across disconnect/reconnect, so + // this also lets the pending→posted / amount-refresh paths survive a reconnect + // (a data_source_id-keyed lookup would miss the prior row and skip the update). const existing = db.prepare(` SELECT id, pending, match_status FROM transactions - WHERE data_source_id = ? AND provider_transaction_id = ? - `).get(txRow.data_source_id, txRow.provider_transaction_id); + WHERE user_id = ? AND provider_transaction_id = ? + `).get(txRow.user_id, txRow.provider_transaction_id); if (!existing) { try { @@ -145,54 +149,63 @@ async function runSync(db, userId, dataSource, { days, debug = false } = {}) { AND provider_transaction_id NOT IN (SELECT value FROM json_each(?)) `); - for (const rawAccount of accounts) { - const accountRow = normalizeAccount(rawAccount, dataSource.id, userId); - const localAccount = upsertAccount(db, accountRow); - accountsUpserted += 1; - - const txList = rawAccount.transactions || []; - if (debug) console.log(`[bankSync:debug] Account "${rawAccount.name}" (monitored=${localAccount.monitored}): ${txList.length} transaction(s)`); - - if (localAccount.monitored === 0) continue; - - const seenTxIds = []; - for (const rawTx of txList) { - const txRow = normalizeTransaction( - rawTx, localAccount.id, dataSource.id, userId, rawAccount.id, rawAccount.currency, - ); - seenTxIds.push(txRow.provider_transaction_id); - const outcome = upsertTransaction(db, txRow); - if (outcome === 'inserted') { - transactionsNew += 1; - if (debug) console.log(`[bankSync:debug] tx ${txRow.provider_transaction_id}: inserted${txRow.pending ? ' (pending)' : ''} (${rawTx.description || rawTx.payee || '—'}, ${txRow.amount}¢)`); - } else if (outcome === 'posted') { - transactionsPosted += 1; - if (debug) console.log(`[bankSync:debug] tx ${txRow.provider_transaction_id}: pending → posted`); - } else if (outcome === 'updated') { - if (debug) console.log(`[bankSync:debug] tx ${txRow.provider_transaction_id}: pending amount refreshed`); - } else { - transactionsSkip += 1; - if (debug) console.log(`[bankSync:debug] tx ${txRow.provider_transaction_id}: duplicate — skipped`); - } - } - - const orphans = pruneOrphanPending.run(localAccount.id, userId, JSON.stringify(seenTxIds)); - if (orphans.changes > 0) { - pendingCleared += orphans.changes; - if (debug) console.log(`[bankSync:debug] Account "${rawAccount.name}": pruned ${orphans.changes} stale pending row(s)`); - } - } - // Store any errlist warnings alongside a successful sync so users can see them const partialError = raw._errlistSummary ? sanitizeErrorMessage(`Partial sync — some connections failed: ${raw._errlistSummary}`) : null; - db.prepare(` + const updateSource = db.prepare(` UPDATE data_sources SET last_sync_at = datetime('now'), last_error = ?, status = 'active', updated_at = datetime('now') WHERE id = ? AND user_id = ? - `).run(partialError, dataSource.id, userId); + `); + + // Batch every account + transaction write and the source-status update into ONE + // commit (better-sqlite3 db.transaction): far fewer fsyncs on large syncs, and + // atomic — a mid-sync failure can't leave a half-written account. The whole + // block is synchronous (the async fetch already happened above). + const ingest = db.transaction(() => { + for (const rawAccount of accounts) { + const accountRow = normalizeAccount(rawAccount, dataSource.id, userId); + const localAccount = upsertAccount(db, accountRow); + accountsUpserted += 1; + + const txList = rawAccount.transactions || []; + if (debug) console.log(`[bankSync:debug] Account "${rawAccount.name}" (monitored=${localAccount.monitored}): ${txList.length} transaction(s)`); + + if (localAccount.monitored === 0) continue; + + const seenTxIds = []; + for (const rawTx of txList) { + const txRow = normalizeTransaction( + rawTx, localAccount.id, dataSource.id, userId, rawAccount.id, rawAccount.currency, + ); + seenTxIds.push(txRow.provider_transaction_id); + const outcome = upsertTransaction(db, txRow); + if (outcome === 'inserted') { + transactionsNew += 1; + if (debug) console.log(`[bankSync:debug] tx ${txRow.provider_transaction_id}: inserted${txRow.pending ? ' (pending)' : ''} (${rawTx.description || rawTx.payee || '—'}, ${txRow.amount}¢)`); + } else if (outcome === 'posted') { + transactionsPosted += 1; + if (debug) console.log(`[bankSync:debug] tx ${txRow.provider_transaction_id}: pending → posted`); + } else if (outcome === 'updated') { + if (debug) console.log(`[bankSync:debug] tx ${txRow.provider_transaction_id}: pending amount refreshed`); + } else { + transactionsSkip += 1; + if (debug) console.log(`[bankSync:debug] tx ${txRow.provider_transaction_id}: duplicate — skipped`); + } + } + + const orphans = pruneOrphanPending.run(localAccount.id, userId, JSON.stringify(seenTxIds)); + if (orphans.changes > 0) { + pendingCleared += orphans.changes; + if (debug) console.log(`[bankSync:debug] Account "${rawAccount.name}": pruned ${orphans.changes} stale pending row(s)`); + } + } + + updateSource.run(partialError, dataSource.id, userId); + }); + ingest(); if (debug) console.log(`[bankSync:debug] Source #${dataSource.id}: applying merchant rules + auto-match`);