1import type { ConnectorHealth } from '@rareindex/shared';2import type { HealthContext } from './types.js';34/**5 * Derive a ConnectorHealth snapshot (§105) from recent runs. Anomaly detectors that need data6 * distributions (price drift, duplicate explosion) run in the normalizer and are passed in as7 * run anomalies.8 */9export function healthFromRuns(ctx: HealthContext, schemaVersion: string): ConnectorHealth {10 const now = Date.now();11 const runs24h = ctx.recentRuns.filter((r) => now - r.startedAt.getTime() <= 24 * 3600_000);12 const attempted = runs24h.reduce((a, r) => a + r.pagesAttempted, 0);13 const success = runs24h.reduce((a, r) => a + r.pagesSuccess, 0);14 const rawRecords = runs24h.reduce((a, r) => a + r.recordsRaw, 0);15 const dupes = runs24h.reduce((a, r) => a + r.recordsDuplicate, 0);16 const fc = sumEngine(runs24h, 'firecrawl');17 const sf = sumEngine(runs24h, 'scrapfly');18 const lastOk = [...ctx.recentRuns].reverse().find((r) => r.status === 'success' || r.status === 'partial');19 const lastErr = [...ctx.recentRuns].reverse().find((r) => r.error)?.error ?? null;20 const anomalies = [...new Set(runs24h.flatMap((r) => r.anomalies))];21 const successRate = attempted > 0 ? success / attempted : null;22 const failing = ctx.recentRuns.slice(-3).length >= 3 && ctx.recentRuns.slice(-3).every((r) => r.status === 'failed');2324 let status: ConnectorHealth['status'] = 'unknown';25 if (ctx.recentRuns.length === 0) status = 'unknown';26 else if (failing) status = 'failing';27 else if (successRate !== null && successRate < 0.5) status = 'failing';28 else if ((successRate !== null && successRate < 0.85) || anomalies.length > 0) status = 'degraded';29 else status = 'healthy';3031 return {32 connector: ctx.meta.id,33 status,34 success_rate_24h: successRate === null ? null : round3(successRate),35 pages_attempted: attempted,36 pages_success: success,37 firecrawl_success_rate: fc.attempts ? round3(fc.success / fc.attempts) : null,38 scrapfly_fallback_rate: attempted ? round3(sf.attempts / attempted) : null,39 parse_failure_rate: rawRecords ? round3(runs24h.reduce((a, r) => a + (r.anomalies.filter((x) => x.startsWith('parse_failure')).length), 0) / rawRecords) : null,40 last_success: lastOk ? lastOk.startedAt.toISOString() : null,41 last_error: lastErr,42 schema_version: schemaVersion,43 records_24h: rawRecords,44 duplicates_24h: dupes,45 anomalies,46 };47}4849function sumEngine(runs: HealthContext['recentRuns'], engine: string) {50 return runs.reduce(51 (acc, r) => {52 const s = r.engineStats[engine];53 if (s) {54 acc.attempts += s.attempts;55 acc.success += s.success;56 }57 return acc;58 },59 { attempts: 0, success: 0 },60 );61}6263function round3(x: number): number {64 return Math.round(x * 1000) / 1000;65}6667/**68 * Detect abnormal price distributions between a baseline and a new batch (§105): median shift69 * beyond a factor or a collapse of dispersion suggests a parser/currency problem.70 */71export function priceDistributionAnomaly(baseline: number[], batch: number[], factor = 3): string | null {72 if (baseline.length < 20 || batch.length < 10) return null;73 const med = (xs: number[]) => {74 const s = [...xs].sort((a, b) => a - b);75 return s[Math.floor(s.length / 2)]!;76 };77 const b = med(baseline);78 const n = med(batch);79 if (b > 0 && (n / b > factor || b / n > factor)) return `price_distribution_shift: median ${b.toFixed(2)} → ${n.toFixed(2)}`;80 return null;81}8283/** Duplicate explosion: share of duplicate raw records in a run is far above baseline. */84export function duplicateExplosion(records: number, duplicates: number, threshold = 0.9): string | null {85 if (records < 50) return null;86 const share = duplicates / records;87 return share >= threshold ? `duplicate_explosion: ${(share * 100).toFixed(0)}% duplicates` : null;88}89