1import { desc, eq, gte } from 'drizzle-orm';2import { connectorHealth, connectorRuns, connectors as connectorsTable } from '@rareindex/database';3import { createHealthContext, listConnectorMeta, loadConnector } from '@rareindex/connectors';4import { logger, type ConnectorHealth } from '@rareindex/shared';5import { db } from './lib/db.ts';6import { getRouter } from './lib/router.ts';78/** Compute and persist ConnectorHealth for every registered connector (§105, §144). */9export async function computeHealth(opts: { connectorId?: string; probe?: boolean } = {}): Promise<ConnectorHealth[]> {10 const metas = listConnectorMeta().filter((m) => !opts.connectorId || m.id === opts.connectorId);11 const out: ConnectorHealth[] = [];12 const since = new Date(Date.now() - 7 * 86_400_000);13 for (const meta of metas) {14 const runs = await db().select().from(connectorRuns).where(eq(connectorRuns.connectorId, meta.id)).orderBy(desc(connectorRuns.startedAt)).limit(50);15 const recentRuns = runs16 .filter((r) => r.startedAt >= since)17 .reverse()18 .map((r) => ({ startedAt: r.startedAt, status: r.status, pagesAttempted: r.pagesAttempted, pagesSuccess: r.pagesSuccess, recordsRaw: r.recordsRaw, recordsDuplicate: r.recordsDuplicate, engineStats: r.engineStats as Record<string, { attempts: number; success: number; credits: number; ms: number }>, anomalies: (r.anomalies as string[]) ?? [], error: r.error }));19 let health: ConnectorHealth;20 try {21 const connector = await loadConnector(meta.id);22 const ctx = createHealthContext({ router: getRouter(meta.id), meta, recentRuns: opts.probe ? recentRuns : recentRuns.length ? recentRuns : [{ startedAt: new Date(0), status: 'unknown', pagesAttempted: 0, pagesSuccess: 0, recordsRaw: 0, recordsDuplicate: 0, engineStats: {}, anomalies: [], error: null }] });23 health = await connector.healthCheck(ctx);24 if (!recentRuns.length && !opts.probe) health.status = 'unknown';25 } catch (err) {26 health = { connector: meta.id, status: 'failing', success_rate_24h: null, pages_attempted: 0, pages_success: 0, firecrawl_success_rate: null, scrapfly_fallback_rate: null, parse_failure_rate: null, last_success: null, last_error: err instanceof Error ? err.message : String(err), schema_version: meta.schemaVersion, records_24h: 0, duplicates_24h: 0, anomalies: ['load_failure'] };27 }28 const [state] = await db().select({ status: connectorsTable.status }).from(connectorsTable).where(eq(connectorsTable.id, meta.id)).limit(1);29 if (state?.status === 'paused') health.status = 'paused';30 await db().insert(connectorHealth).values({ connectorId: meta.id, computedAt: new Date(), status: health.status, health }).onConflictDoUpdate({ target: connectorHealth.connectorId, set: { computedAt: new Date(), status: health.status, health } });31 out.push(health);32 }33 logger.info({ connectors: out.length }, 'health computed');34 return out;35}3637export { gte };38