1import { readFileSync } from 'node:fs';2import { and, desc, eq, inArray, isNull, sql } from 'drizzle-orm';3import { connectorRuns, normalizedRecords, rawRecords, sales } from '@rareindex/database';4import { loadConnector, priceDistributionAnomaly, type RareIndexConnector } from '@rareindex/connectors';5import { NormalizedRecordSchema, logger, newId, type NormalizedRecord } from '@rareindex/shared';6import { db } from '../lib/db.ts';78export interface NormalizeResult {9 processed: number;10 produced: number;11 failed: number;12 anomalies: string[];13}1415const connectorCache = new Map<string, Promise<RareIndexConnector>>();16function getConnector(id: string): Promise<RareIndexConnector> {17 let p = connectorCache.get(id);18 if (!p) {19 p = loadConnector(id);20 connectorCache.set(id, p);21 }22 return p;23}2425/**26 * Normalizer (§108): raw_records → normalized_records. Each raw row is processed exactly once;27 * failures are recorded on the raw row (process_error) and counted as anomalies — never swallowed.28 */29export async function normalizeBatch(opts: { connectorId?: string; rawIds?: string[]; limit?: number } = {}): Promise<NormalizeResult> {30 const log = logger.child({ component: 'normalizer', connector: opts.connectorId });31 const limit = opts.limit ?? 500;32 const where = opts.rawIds?.length ? inArray(rawRecords.id, opts.rawIds) : and(isNull(rawRecords.processedAt), opts.connectorId ? eq(rawRecords.connectorId, opts.connectorId) : undefined);33 const rows = await db().select().from(rawRecords).where(where).orderBy(rawRecords.fetchedAt).limit(limit);34 const result: NormalizeResult = { processed: 0, produced: 0, failed: 0, anomalies: [] };35 if (rows.length === 0) return result;3637 const byConnector = new Map<string, typeof rows>();38 for (const r of rows) byConnector.set(r.connectorId, [...(byConnector.get(r.connectorId) ?? []), r]);3940 for (const [connectorId, group] of byConnector) {41 let connector: RareIndexConnector;42 try {43 connector = await getConnector(connectorId);44 } catch (err) {45 const msg = `connector load failed: ${err instanceof Error ? err.message : String(err)}`;46 log.error({ err }, msg);47 await db().update(rawRecords).set({ processedAt: new Date(), processError: msg }).where(inArray(rawRecords.id, group.map((g) => g.id)));48 result.failed += group.length;49 result.anomalies.push(`parse_failure: ${msg}`);50 continue;51 }52 const inserts: Array<typeof normalizedRecords.$inferInsert> = [];53 const done: Array<{ id: string; error: string | null }> = [];54 const batchPrices: number[] = [];55 for (const raw of group) {56 try {57 let payload = raw.payload;58 if (raw.snapshotRef && payload && typeof payload === 'object' && !('snapshot' in (payload as object))) {59 try {60 payload = { ...(payload as object), snapshot: readFileSync(raw.snapshotRef, 'utf8') };61 } catch {62 /* snapshot missing: proceed with payload only */63 }64 }65 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' });66 let n = 0;67 for (const rec of out) {68 const parsed = NormalizedRecordSchema.safeParse(rec);69 if (!parsed.success) {70 result.anomalies.push(`parse_failure: schema ${parsed.error.issues[0]?.path.join('.')} ${parsed.error.issues[0]?.message}`);71 result.failed++;72 continue;73 }74 const p = parsed.data;75 if ((p.kind === 'sale' || p.kind === 'price_observation') && p.price > 0) batchPrices.push(p.price);76 inserts.push({ id: newId('raw'), rawRecordId: raw.id, connectorId, sourceId: raw.sourceId, kind: p.kind, seq: n, payload: p, status: 'pending' });77 n++;78 }79 done.push({ id: raw.id, error: n === 0 && out.length === 0 ? 'normalize produced no records' : null });80 result.produced += n;81 } catch (err) {82 const msg = err instanceof Error ? err.message : String(err);83 log.warn({ err, raw: raw.id }, 'normalize failed');84 done.push({ id: raw.id, error: `normalize: ${msg}`.slice(0, 500) });85 result.failed++;86 result.anomalies.push(`parse_failure: ${msg.slice(0, 120)}`);87 }88 result.processed++;89 }90 // normalized_records has a unique raw_record_id index; a raw record producing several records91 // is common (catalog + observations) so we key on (raw, kind, position) instead.92 for (let i = 0; i < inserts.length; i += 500) {93 await db().insert(normalizedRecords).values(inserts.slice(i, i + 500)).onConflictDoNothing();94 }95 const okIds = done.filter((d) => !d.error).map((d) => d.id);96 if (okIds.length) await db().update(rawRecords).set({ processedAt: new Date(), processError: null }).where(inArray(rawRecords.id, okIds));97 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));9899 // Price distribution anomaly vs the connector's recent accepted sales (§105)100 if (batchPrices.length >= 10) {101 const baseline = await db().select({ p: sales.priceUsd }).from(sales).where(eq(sales.connectorId, connectorId)).orderBy(desc(sales.saleDate)).limit(500);102 const a = priceDistributionAnomaly(baseline.map((b) => Number(b.p)), batchPrices);103 if (a) result.anomalies.push(a);104 }105 if (result.anomalies.length) {106 const uniq = [...new Set(result.anomalies)].slice(0, 20);107 const [lastRun] = await db().select({ id: connectorRuns.id }).from(connectorRuns).where(eq(connectorRuns.connectorId, connectorId)).orderBy(desc(connectorRuns.startedAt)).limit(1);108 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));109 }110 }111 log.info(result, 'normalize batch done');112 return result;113}114