spb/rareindex

1import { desc, eq, gte, sql } 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    else if (state?.status === 'maintenance') health.status = 'maintenance';31    else if (state?.status === 'disabled') health.status = 'disabled';32    // data freshness: newest source-side date ingested by this connector (SPEC §12)33    const [fresh] = (await db().execute(sql`select greatest((select max(sale_date) from sales where connector_id = ${meta.id}), (select max(last_seen_at) from listings where connector_id = ${meta.id}), (select max(observation_date)::timestamptz from price_observations where connector_id = ${meta.id})) as ts`)) as unknown as Array<{ ts: Date | string | null }>;34    health.data_freshness = fresh?.ts ? new Date(fresh.ts).toISOString() : null;35    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 } });36    out.push(health);37  }38  logger.info({ connectors: out.length }, 'health computed');39  return out;40}4142export { gte };43