TypeScript 61.9%
HTML 37.2%
SQL 0.7%
1import { readFileSync } from 'node:fs';2import { and, desc, eq, inArray, isNull, sql } from 'drizzle-orm';3import { connectorFieldStats, connectorRuns, normalizedRecords, rawRecords, sales } from '@rareindex/database';4import { DRIFT_FIELDS, driftPresence, fieldNullDrift, loadConnector, priceDistributionAnomaly, type RareIndexConnector } from '@rareindex/connectors';5import { NormalizedRecordSchema, logger, newId, toDateOnly, type NormalizedRecord } from '@rareindex/shared';6import { db } from '../lib/db.ts';78/**9 * Schema-drift detection (SPEC §13): accumulate per-field presence counters for today, then compare10 * today's null rates with the trailing 7-day baseline. Returns anomaly strings (possibly empty).11 */12async function recordFieldStats(connectorId: string, records: NormalizedRecord[]): Promise<string[]> {13 if (!records.length) return [];14 const counts: Record<string, { total: number; nulls: number }> = {};15 for (const f of DRIFT_FIELDS) counts[f] = { total: 0, nulls: 0 };16 for (const r of records) {17 const p = driftPresence(r as unknown as Record<string, unknown>);18 for (const f of DRIFT_FIELDS) {19 counts[f]!.total++;20 if (!p[f]) counts[f]!.nulls++;21 }22 }23 const day = toDateOnly(new Date());24 for (const [field, c] of Object.entries(counts)) {25 await db()26 .insert(connectorFieldStats)27 .values({ connectorId, day, field, total: c.total, nulls: c.nulls })28 .onConflictDoUpdate({ target: [connectorFieldStats.connectorId, connectorFieldStats.day, connectorFieldStats.field], set: { total: sql`${connectorFieldStats.total} + ${c.total}`, nulls: sql`${connectorFieldStats.nulls} + ${c.nulls}` } });29 }30 const rows = (await db().execute(sql`select field, day::text as day, total, nulls from connector_field_stats where connector_id = ${connectorId} and day >= (current_date - interval '7 days')::date`)) as unknown as Array<{ field: string; day: string; total: number; nulls: number }>;31 const baseline: Record<string, { total: number; nulls: number }> = {};32 const today: Record<string, { total: number; nulls: number }> = {};33 for (const r of rows) {34 const target = r.day === day ? today : baseline;35 const t = (target[r.field] ??= { total: 0, nulls: 0 });36 t.total += Number(r.total);37 t.nulls += Number(r.nulls);38 }39 return fieldNullDrift(baseline, today);40}4142export interface NormalizeResult {43 processed: number;44 produced: number;45 failed: number;46 anomalies: string[];47}4849const connectorCache = new Map<string, Promise<RareIndexConnector>>();50function getConnector(id: string): Promise<RareIndexConnector> {51 let p = connectorCache.get(id);52 if (!p) {53 p = loadConnector(id);54 connectorCache.set(id, p);55 }56 return p;57}5859/**60 * Normalizer (§108): raw_records → normalized_records. Each raw row is processed exactly once;61 * failures are recorded on the raw row (process_error) and counted as anomalies — never swallowed.62 */63export async function normalizeBatch(opts: { connectorId?: string; rawIds?: string[]; limit?: number } = {}): Promise<NormalizeResult> {64 const log = logger.child({ component: 'normalizer', connector: opts.connectorId });65 const limit = opts.limit ?? 500;66 const where = opts.rawIds?.length ? inArray(rawRecords.id, opts.rawIds) : and(isNull(rawRecords.processedAt), opts.connectorId ? eq(rawRecords.connectorId, opts.connectorId) : undefined);67 const rows = await db().select().from(rawRecords).where(where).orderBy(rawRecords.fetchedAt).limit(limit);68 const result: NormalizeResult = { processed: 0, produced: 0, failed: 0, anomalies: [] };69 if (rows.length === 0) return result;7071 const byConnector = new Map<string, typeof rows>();72 for (const r of rows) byConnector.set(r.connectorId, [...(byConnector.get(r.connectorId) ?? []), r]);7374 for (const [connectorId, group] of byConnector) {75 let connector: RareIndexConnector;76 try {77 connector = await getConnector(connectorId);78 } catch (err) {79 const msg = `connector load failed: ${err instanceof Error ? err.message : String(err)}`;80 log.error({ err }, msg);81 await db().update(rawRecords).set({ processedAt: new Date(), processError: msg }).where(inArray(rawRecords.id, group.map((g) => g.id)));82 result.failed += group.length;83 result.anomalies.push(`parse_failure: ${msg}`);84 continue;85 }86 const inserts: Array<typeof normalizedRecords.$inferInsert> = [];87 const done: Array<{ id: string; error: string | null }> = [];88 const batchPrices: number[] = [];89 const produced: NormalizedRecord[] = [];90 for (const raw of group) {91 try {92 let payload = raw.payload;93 if (raw.snapshotRef && payload && typeof payload === 'object' && !('snapshot' in (payload as object))) {94 try {95 payload = { ...(payload as object), snapshot: readFileSync(raw.snapshotRef, 'utf8') };96 } catch {97 /* snapshot missing: proceed with payload only */98 }99 }100 const out = await connector.normalize({ id: raw.id, connectorId, sourceId: raw.sourceId, url: raw.url, externalId: raw.externalId, kind: raw.kind as NormalizedRecord['kind'], payload, fetchedAt: raw.fetchedAt, engine: raw.engine as 'api' });101 let n = 0;102 for (const rec of out) {103 const parsed = NormalizedRecordSchema.safeParse(rec);104 if (!parsed.success) {105 result.anomalies.push(`parse_failure: schema ${parsed.error.issues[0]?.path.join('.')} ${parsed.error.issues[0]?.message}`);106 result.failed++;107 continue;108 }109 const p = parsed.data;110 produced.push(p);111 if ((p.kind === 'sale' || p.kind === 'price_observation') && p.price > 0) batchPrices.push(p.price);112 inserts.push({ id: newId('raw'), rawRecordId: raw.id, connectorId, sourceId: raw.sourceId, kind: p.kind, seq: n, payload: p, status: 'pending' });113 n++;114 }115 done.push({ id: raw.id, error: n === 0 && out.length === 0 ? 'normalize produced no records' : null });116 result.produced += n;117 } catch (err) {118 const msg = err instanceof Error ? err.message : String(err);119 log.warn({ err, raw: raw.id }, 'normalize failed');120 done.push({ id: raw.id, error: `normalize: ${msg}`.slice(0, 500) });121 result.failed++;122 result.anomalies.push(`parse_failure: ${msg.slice(0, 120)}`);123 }124 result.processed++;125 }126 // normalized_records has a unique raw_record_id index; a raw record producing several records127 // is common (catalog + observations) so we key on (raw, kind, position) instead.128 for (let i = 0; i < inserts.length; i += 500) {129 await db().insert(normalizedRecords).values(inserts.slice(i, i + 500)).onConflictDoNothing();130 }131 const okIds = done.filter((d) => !d.error).map((d) => d.id);132 if (okIds.length) await db().update(rawRecords).set({ processedAt: new Date(), processError: null }).where(inArray(rawRecords.id, okIds));133 for (const d of done.filter((d) => d.error)) await db().update(rawRecords).set({ processedAt: new Date(), processError: d.error }).where(eq(rawRecords.id, d.id));134135 // Schema drift: field-null rates vs the trailing week (SPEC §13)136 try {137 result.anomalies.push(...(await recordFieldStats(connectorId, produced)));138 } catch (err) {139 log.warn({ err }, 'field stats failed');140 }141 // Price distribution anomaly vs the connector's recent accepted sales (§105)142 if (batchPrices.length >= 10) {143 const baseline = await db().select({ p: sales.priceUsd }).from(sales).where(eq(sales.connectorId, connectorId)).orderBy(desc(sales.saleDate)).limit(500);144 const a = priceDistributionAnomaly(baseline.map((b) => Number(b.p)), batchPrices);145 if (a) result.anomalies.push(a);146 }147 if (result.anomalies.length) {148 const uniq = [...new Set(result.anomalies)].slice(0, 20);149 const [lastRun] = await db().select({ id: connectorRuns.id }).from(connectorRuns).where(eq(connectorRuns.connectorId, connectorId)).orderBy(desc(connectorRuns.startedAt)).limit(1);150 if (lastRun) await db().update(connectorRuns).set({ anomalies: sql`(SELECT jsonb_agg(DISTINCT x) FROM jsonb_array_elements(${connectorRuns.anomalies} || ${JSON.stringify(uniq)}::jsonb) x)` }).where(eq(connectorRuns.id, lastRun.id));151 }152 }153 log.info(result, 'normalize batch done');154 return result;155}156