feat(financials-v2): hydrate filing statements and aggregate history

lib/server/task-processors.ts

@@ -10,9 +10,14 @@ import { buildPortfolioSummary } from '@/lib/server/portfolio';
 import { getQuote } from '@/lib/server/prices';
 import {
   getFilingByAccession,
+  listFilingsRecords,
   saveFilingAnalysis,
   upsertFilingsRecords
 } from '@/lib/server/repos/filings';
+import {
+  getFilingStatementSnapshotByFilingId,
+  upsertFilingStatementSnapshot
+} from '@/lib/server/repos/filing-statements';
 import {
   applyRefreshedPrices,
   listHoldingsForPriceRefresh,
@@ -22,7 +27,8 @@ import { createPortfolioInsight } from '@/lib/server/repos/insights';
 import {
   fetchFilingMetricsForFilings,
   fetchPrimaryFilingText,
-  fetchRecentFilings
+  fetchRecentFilings,
+  hydrateFilingStatementSnapshot
 } from '@/lib/server/sec';
 
 const EXTRACTION_REQUIRED_KEYS = [
@@ -40,6 +46,8 @@ const EXTRACTION_REQUIRED_KEYS = [
 const EXTRACTION_MAX_ITEMS = 6;
 const EXTRACTION_ITEM_MAX_LENGTH = 280;
 const EXTRACTION_SUMMARY_MAX_LENGTH = 900;
+const STATEMENT_HYDRATION_DELAY_MS = 120;
+const STATEMENT_HYDRATION_MAX_FILINGS = 80;
 const SEGMENT_PATTERNS = [
   /\boperating segment\b/i,
   /\bsegment revenue\b/i,
@@ -556,11 +564,67 @@ async function processSyncFilings(task: Task) {
     }))
   );
 
+  let statementSnapshotsHydrated = 0;
+  let statementSnapshotsFailed = 0;
+  const hydrateCandidates = (await listFilingsRecords({
+    ticker,
+    limit: Math.min(Math.max(limit * 3, 40), STATEMENT_HYDRATION_MAX_FILINGS)
+  }))
+    .filter((filing): filing is Filing & { filing_type: '10-K' | '10-Q' } => {
+      return filing.filing_type === '10-K' || filing.filing_type === '10-Q';
+    });
+
+  for (const filing of hydrateCandidates) {
+    const existingSnapshot = await getFilingStatementSnapshotByFilingId(filing.id);
+    const shouldRefresh = !existingSnapshot
+      || Date.parse(existingSnapshot.updated_at) < Date.parse(filing.updated_at);
+
+    if (!shouldRefresh) {
+      continue;
+    }
+
+    try {
+      const snapshot = await hydrateFilingStatementSnapshot({
+        filingId: filing.id,
+        ticker: filing.ticker,
+        cik: filing.cik,
+        accessionNumber: filing.accession_number,
+        filingDate: filing.filing_date,
+        filingType: filing.filing_type,
+        filingUrl: filing.filing_url,
+        primaryDocument: filing.primary_document ?? null,
+        metrics: filing.metrics
+      });
+
+      await upsertFilingStatementSnapshot(snapshot);
+      statementSnapshotsHydrated += 1;
+    } catch (error) {
+      await upsertFilingStatementSnapshot({
+        filing_id: filing.id,
+        ticker: filing.ticker,
+        filing_date: filing.filing_date,
+        filing_type: filing.filing_type,
+        period_end: filing.filing_date,
+        statement_bundle: null,
+        standardized_bundle: null,
+        dimension_bundle: null,
+        parse_status: 'failed',
+        parse_error: error instanceof Error ? error.message : 'Statement hydration failed',
+        source: 'companyfacts_fallback'
+      });
+      statementSnapshotsFailed += 1;
+    }
+
+    await Bun.sleep(STATEMENT_HYDRATION_DELAY_MS);
+  }
+
   return {
     ticker,
     fetched: filings.length,
     inserted: saveResult.inserted,
-    updated: saveResult.updated
+    updated: saveResult.updated,
+    statementSnapshotsHydrated,
+    statementSnapshotsFailed
   };
 }