perf(simplefin): batch sync writes in a transaction + align dedup key

- 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 <noreply@anthropic.com>

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 = ?
+    WHERE user_id = ? AND provider_transaction_id = ?
-  `).get(txRow.data_source_id, txRow.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`);