WIP main worktree changes before merge

lib/server/api/app.ts

@@ -18,6 +18,7 @@ import type {
 } from '@/lib/types';
 import { auth } from '@/lib/auth';
 import { requireAuthenticatedSession } from '@/lib/server/auth-session';
+import { getLatestFinancialIngestionSchemaStatus } from '@/lib/server/db/financial-ingestion-schema';
 import { asErrorMessage, jsonError } from '@/lib/server/http';
 import { buildPortfolioSummary } from '@/lib/server/portfolio';
 import {
@@ -391,16 +392,36 @@ export const app = new Elysia({ prefix: '/api' })
         getTaskQueueSnapshot(),
         checkWorkflowBackend()
       ]);
+      const ingestionSchema = getLatestFinancialIngestionSchemaStatus();
+      const ingestionSchemaPayload = ingestionSchema
+        ? {
+          ok: ingestionSchema.ok,
+          mode: ingestionSchema.mode,
+          missingIndexes: ingestionSchema.missingIndexes,
+          duplicateGroups: ingestionSchema.duplicateGroups,
+          lastCheckedAt: ingestionSchema.lastCheckedAt
+        }
+        : {
+          ok: false,
+          mode: 'failed' as const,
+          missingIndexes: [],
+          duplicateGroups: 0,
+          lastCheckedAt: new Date().toISOString()
+        };
+      const schemaHealthy = ingestionSchema?.ok ?? false;
 
-      if (!workflowBackend.ok) {
+      if (!workflowBackend.ok || !schemaHealthy) {
         return Response.json({
           status: 'degraded',
           version: '4.0.0',
           timestamp: new Date().toISOString(),
           queue,
+          database: {
+            ingestionSchema: ingestionSchemaPayload
+          },
           workflow: {
-            ok: false,
-            reason: workflowBackend.reason
+            ok: workflowBackend.ok,
+            ...(workflowBackend.ok ? {} : { reason: workflowBackend.reason })
           }
         }, { status: 503 });
       }
@@ -410,6 +431,9 @@ export const app = new Elysia({ prefix: '/api' })
         version: '4.0.0',
         timestamp: new Date().toISOString(),
         queue,
+        database: {
+          ingestionSchema: ingestionSchemaPayload
+        },
         workflow: {
           ok: true
         }
@@ -1366,12 +1390,13 @@ export const app = new Elysia({ prefix: '/api' })
       return jsonError('ticker is required');
     }
 
-    const [filings, holding, watchlistItem, liveQuote, priceHistory, journalPreview, memo, secProfile] = await Promise.all([
+    const [filings, holding, watchlistItem, liveQuote, priceHistory, benchmarkHistory, journalPreview, memo, secProfile] = await Promise.all([
       listFilingsRecords({ ticker, limit: 40 }),
       getHoldingByTicker(session.user.id, ticker),
       getWatchlistItemByTicker(session.user.id, ticker),
       getQuote(ticker),
       getPriceHistory(ticker),
+      getPriceHistory('^GSPC'),
       listResearchJournalEntries(session.user.id, ticker, 6),
       getResearchMemoByTicker(session.user.id, ticker),
       getSecCompanyProfile(ticker)
@@ -1478,6 +1503,7 @@ export const app = new Elysia({ prefix: '/api' })
         quote: liveQuote,
         position: holding,
         priceHistory,
+        benchmarkHistory,
         financials,
         filings: redactedFilings.slice(0, 20),
         aiReports,

lib/server/api/task-workflow-hybrid.e2e.test.ts

@@ -74,11 +74,44 @@ function resetDbSingletons() {
   const globalState = globalThis as typeof globalThis & {
     __fiscalSqliteClient?: { close?: () => void };
     __fiscalDrizzleDb?: unknown;
+    __financialIngestionSchemaStatus?: unknown;
   };
 
   globalState.__fiscalSqliteClient?.close?.();
   globalState.__fiscalSqliteClient = undefined;
   globalState.__fiscalDrizzleDb = undefined;
+  globalState.__financialIngestionSchemaStatus = undefined;
+}
+
+function setFinancialIngestionSchemaStatus(input: {
+  ok: boolean;
+  mode: 'healthy' | 'repaired' | 'drifted' | 'failed';
+  missingIndexes?: string[];
+  duplicateGroups?: number;
+}) {
+  const globalState = globalThis as typeof globalThis & {
+    __financialIngestionSchemaStatus?: {
+      ok: boolean;
+      mode: 'healthy' | 'repaired' | 'drifted' | 'failed';
+      requestedMode: 'auto' | 'check-only' | 'off';
+      missingIndexes: string[];
+      duplicateGroups: number;
+      lastCheckedAt: string;
+      repair: null;
+      error: string | null;
+    };
+  };
+
+  globalState.__financialIngestionSchemaStatus = {
+    ok: input.ok,
+    mode: input.mode,
+    requestedMode: 'auto',
+    missingIndexes: input.missingIndexes ?? [],
+    duplicateGroups: input.duplicateGroups ?? 0,
+    lastCheckedAt: new Date().toISOString(),
+    repair: null,
+    error: input.ok ? null : 'schema drift injected by test'
+  };
 }
 
 function applySqlMigrations(client: { exec: (query: string) => void }) {
@@ -250,6 +283,10 @@ if (process.env.RUN_TASK_WORKFLOW_E2E === '1') {
     runStatuses.clear();
     runCounter = 0;
     workflowBackendHealthy = true;
+    setFinancialIngestionSchemaStatus({
+      ok: true,
+      mode: 'healthy'
+    });
   });
 
   it('queues multiple analyze jobs and suppresses duplicate in-flight analyze jobs', async () => {
@@ -808,9 +845,100 @@ if (process.env.RUN_TASK_WORKFLOW_E2E === '1') {
 
     const healthy = await jsonRequest('GET', '/api/health');
     expect(healthy.response.status).toBe(200);
-    expect((healthy.json as { status: string; workflow: { ok: boolean } }).status).toBe('ok');
+    expect((healthy.json as {
+      status: string;
+      workflow: { ok: boolean };
+      database: {
+        ingestionSchema: {
+          ok: boolean;
+          mode: string;
+          missingIndexes: string[];
+          duplicateGroups: number;
+        };
+      };
+    }).status).toBe('ok');
     expect((healthy.json as { status: string; workflow: { ok: boolean } }).workflow.ok).toBe(true);
+    expect((healthy.json as {
+      database: {
+        ingestionSchema: {
+          ok: boolean;
+          mode: string;
+          missingIndexes: string[];
+          duplicateGroups: number;
+        };
+      };
+    }).database.ingestionSchema.ok).toBe(true);
+    expect((healthy.json as {
+      database: {
+        ingestionSchema: {
+          ok: boolean;
+          mode: string;
+        };
+      };
+    }).database.ingestionSchema.mode).toBe('healthy');
 
+    setFinancialIngestionSchemaStatus({
+      ok: false,
+      mode: 'drifted',
+      missingIndexes: ['company_financial_bundle_uidx'],
+      duplicateGroups: 1
+    });
+    const schemaDrifted = await jsonRequest('GET', '/api/health');
+    expect(schemaDrifted.response.status).toBe(503);
+    expect((schemaDrifted.json as {
+      status: string;
+      workflow: { ok: boolean };
+      database: {
+        ingestionSchema: {
+          ok: boolean;
+          mode: string;
+          missingIndexes: string[];
+          duplicateGroups: number;
+        };
+      };
+    }).status).toBe('degraded');
+    expect((schemaDrifted.json as {
+      workflow: { ok: boolean };
+    }).workflow.ok).toBe(true);
+    expect((schemaDrifted.json as {
+      database: {
+        ingestionSchema: {
+          ok: boolean;
+          mode: string;
+          missingIndexes: string[];
+          duplicateGroups: number;
+        };
+      };
+    }).database.ingestionSchema.ok).toBe(false);
+    expect((schemaDrifted.json as {
+      database: {
+        ingestionSchema: {
+          ok: boolean;
+          mode: string;
+          missingIndexes: string[];
+          duplicateGroups: number;
+        };
+      };
+    }).database.ingestionSchema.mode).toBe('drifted');
+    expect((schemaDrifted.json as {
+      database: {
+        ingestionSchema: {
+          missingIndexes: string[];
+        };
+      };
+    }).database.ingestionSchema.missingIndexes).toEqual(['company_financial_bundle_uidx']);
+    expect((schemaDrifted.json as {
+      database: {
+        ingestionSchema: {
+          duplicateGroups: number;
+        };
+      };
+    }).database.ingestionSchema.duplicateGroups).toBe(1);
+
+    setFinancialIngestionSchemaStatus({
+      ok: true,
+      mode: 'healthy'
+    });
     workflowBackendHealthy = false;
     const degraded = await jsonRequest('GET', '/api/health');
     expect(degraded.response.status).toBe(503);