TypeScript 61.9%
HTML 37.2%
SQL 0.7%
1import type { ConnectorHealth } from '@rareindex/shared';2import { missingRequirements, type EngineStats, type HealthContext } from './types.js';34/**5 * Derive a ConnectorHealth snapshot (§105, SPEC §12–13) from recent runs. Anomaly detectors that need6 * data distributions (price drift, duplicate explosion, field-null drift) run in the normalizer and7 * arrive here as run anomalies. Result-count collapse and HTTP/challenge statistics are computed here.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 runs7d = ctx.recentRuns.filter((r) => now - r.startedAt.getTime() <= 7 * 24 * 3600_000);13 const attempted = runs24h.reduce((a, r) => a + r.pagesAttempted, 0);14 const success = runs24h.reduce((a, r) => a + r.pagesSuccess, 0);15 const rawRecords = runs24h.reduce((a, r) => a + r.recordsRaw, 0);16 const dupes = runs24h.reduce((a, r) => a + r.recordsDuplicate, 0);17 const fc = sumEngine(runs24h, 'firecrawl');18 const sf = sumEngine(runs24h, 'scrapfly');19 const lastOk = [...ctx.recentRuns].reverse().find((r) => r.status === 'success' || r.status === 'partial');20 const lastErr = [...ctx.recentRuns].reverse().find((r) => r.error)?.error ?? null;21 const anomalies = [...new Set(runs24h.flatMap((r) => r.anomalies))];22 const successRate = attempted > 0 ? success / attempted : null;23 const failing = ctx.recentRuns.slice(-3).length >= 3 && ctx.recentRuns.slice(-3).every((r) => r.status === 'failed');2425 // HTTP / latency / challenge statistics across engines (SPEC §12)26 const httpErrors: Record<string, number> = {};27 let challenges = 0;28 let circuit = 0;29 let totalMs = 0;30 let totalAttempts = 0;31 for (const r of runs24h) {32 for (const s of Object.values(r.engineStats ?? {})) {33 totalMs += s.ms;34 totalAttempts += s.attempts;35 challenges += s.blocked ?? 0;36 circuit += s.circuitOpen ?? 0;37 for (const [code, n] of Object.entries(s.statuses ?? {})) if (Number(code) >= 400) httpErrors[code] = (httpErrors[code] ?? 0) + n;38 }39 }40 const rateLimited = httpErrors['429'] ?? 0;4142 // Result-count collapse (SPEC §13): today's volume far below the trailing week's daily average.43 const days7 = Math.max(1, Math.min(7, Math.ceil((now - (runs7d[0]?.startedAt.getTime() ?? now)) / 86_400_000) || 1));44 const raw7d = runs7d.reduce((a, r) => a + r.recordsRaw, 0);45 const dailyAvg = runs7d.length ? raw7d / days7 : null;46 const collapsed = dailyAvg !== null && dailyAvg >= 50 && runs24h.length > 0 && rawRecords < dailyAvg * 0.2 && !anomalies.some((a) => a.startsWith('time_budget'));47 if (collapsed) anomalies.push(`result_count_collapse: ${rawRecords} vs ${Math.round(dailyAvg)}/day`);48 const drift = anomalies.filter((a) => /^schema_drift|^selector_missing|^parse_failure_page|^pagination_failure|^price_parse_failure/.test(a));4950 const missing = missingRequirements(ctx.meta);51 let status: ConnectorHealth['status'] = 'unknown';52 if (missing.length) status = 'disabled';53 else if (ctx.recentRuns.length === 0) status = 'unknown';54 else if (failing) status = 'failing';55 else if (successRate !== null && successRate < 0.5) status = 'failing';56 else if (collapsed || drift.length) status = 'degraded';57 else if ((successRate !== null && successRate < 0.85) || anomalies.length > 0) status = 'degraded';58 else status = 'healthy';5960 return {61 connector: ctx.meta.id,62 status,63 success_rate_24h: successRate === null ? null : round3(successRate),64 pages_attempted: attempted,65 pages_success: success,66 firecrawl_success_rate: fc.attempts ? round3(fc.success / fc.attempts) : null,67 scrapfly_fallback_rate: attempted ? round3(sf.attempts / attempted) : null,68 parse_failure_rate: rawRecords ? round3(runs24h.reduce((a, r) => a + r.anomalies.filter((x) => x.startsWith('parse_failure')).length, 0) / rawRecords) : null,69 last_success: lastOk ? lastOk.startedAt.toISOString() : null,70 last_error: lastErr,71 schema_version: schemaVersion,72 records_24h: rawRecords,73 duplicates_24h: dupes,74 anomalies,75 requests_24h: totalAttempts,76 latency_ms_avg: totalAttempts ? Math.round(totalMs / totalAttempts) : null,77 http_errors: httpErrors,78 challenges_24h: challenges,79 circuit_refusals_24h: circuit,80 rate_limited_24h: rateLimited,81 records_7d_daily_avg: dailyAvg === null ? null : Math.round(dailyAvg),82 schema_drift: drift,83 missing_requirements: missing,84 };85}8687function sumEngine(runs: HealthContext['recentRuns'], engine: string) {88 return runs.reduce(89 (acc, r) => {90 const s: EngineStats | undefined = r.engineStats[engine];91 if (s) {92 acc.attempts += s.attempts;93 acc.success += s.success;94 }95 return acc;96 },97 { attempts: 0, success: 0 },98 );99}100101function round3(x: number): number {102 return Math.round(x * 1000) / 1000;103}104105/**106 * Detect abnormal price distributions between a baseline and a new batch (§105): median shift107 * beyond a factor or a collapse of dispersion suggests a parser/currency problem.108 */109export function priceDistributionAnomaly(baseline: number[], batch: number[], factor = 3): string | null {110 if (baseline.length < 20 || batch.length < 10) return null;111 const med = (xs: number[]) => {112 const s = [...xs].sort((a, b) => a - b);113 return s[Math.floor(s.length / 2)]!;114 };115 const b = med(baseline);116 const n = med(batch);117 if (b > 0 && (n / b > factor || b / n > factor)) return `price_distribution_shift: median ${b.toFixed(2)} → ${n.toFixed(2)}`;118 return null;119}120121/** Duplicate explosion: share of duplicate raw records in a run is far above baseline. */122export function duplicateExplosion(records: number, duplicates: number, threshold = 0.9): string | null {123 if (records < 50) return null;124 const share = duplicates / records;125 return share >= threshold ? `duplicate_explosion: ${(share * 100).toFixed(0)}% duplicates` : null;126}127128/** Fields whose null-rate is tracked per connector for schema-drift detection (SPEC §13). */129export const DRIFT_FIELDS = ['price', 'currency', 'date', 'images', 'title', 'identifiers', 'grade', 'number', 'set', 'year', 'description'] as const;130export type DriftField = (typeof DRIFT_FIELDS)[number];131132/** Presence vector of the drift fields for one normalised record (any kind). */133export function driftPresence(rec: Record<string, unknown>): Record<DriftField, boolean> {134 const attrs = (rec.attributes ?? {}) as Record<string, unknown>;135 const grade = (rec.grade ?? {}) as Record<string, unknown>;136 const has = (v: unknown) => v !== null && v !== undefined && v !== '' && !(Array.isArray(v) && v.length === 0) && !(typeof v === 'object' && !Array.isArray(v) && Object.keys(v as object).length === 0);137 return {138 price: has(rec.price) || has(rec.currentBid) || has(rec.estimateLow) || has(rec.total),139 currency: has(rec.currency) || rec.kind === 'catalog_item' || rec.kind === 'population_report' || rec.kind === 'news_item',140 date: has(rec.saleDate) || has(rec.observationDate) || has(rec.listedAt) || has(rec.endsAt) || has(rec.reportDate) || has(rec.releaseDate) || has(rec.publishedAt),141 images: has(rec.imageUrls) || has(rec.imageUrl),142 title: has(rec.rawTitle) || has(rec.title),143 identifiers: has(attrs.identifiers),144 grade: has(grade.grade) || has(grade.grader),145 number: has(attrs.number),146 set: has(attrs.set) || has(attrs.setCode),147 year: has(attrs.year),148 description: has(rec.description) || has(rec.summary),149 };150}151152/**153 * Compare today's null rates against the trailing baseline. A field whose null rate rose by more154 * than `delta` (absolute) on ≥ `minRecords` records is reported as schema drift.155 */156export function fieldNullDrift(baseline: Record<string, { total: number; nulls: number }>, today: Record<string, { total: number; nulls: number }>, opts: { delta?: number; minRecords?: number } = {}): string[] {157 const delta = opts.delta ?? 0.3;158 const minRecords = opts.minRecords ?? 50;159 const out: string[] = [];160 for (const [field, t] of Object.entries(today)) {161 const b = baseline[field];162 if (!b || t.total < minRecords || b.total < minRecords) continue;163 const bRate = b.nulls / b.total;164 const tRate = t.nulls / t.total;165 if (tRate - bRate >= delta && bRate < 0.5) out.push(`schema_drift:${field}: null rate ${(bRate * 100).toFixed(0)}% → ${(tRate * 100).toFixed(0)}%`);166 }167 return out;168}169