1import { and, eq, inArray, sql } from 'drizzle-orm';2import { normalizedRecords } from '@rareindex/database';3import { NormalizedRecordSchema, logger } from '@rareindex/shared';4import { db } from '../lib/db.ts';5import { emitMany } from '../lib/events.ts';6import { applyRecord, FxMissingError } from './writers.ts';7import { syncFx } from '../fx.ts';89export interface ResolveResult {10 processed: number;11 applied: number;12 duplicates: number;13 unmatched: number;14 rejected: number;15 fxMissing: number;16 touchedAssets: string[];17}1819/**20 * Entity resolution + canonical writes (§108 steps 5–9). Pending normalized records are applied in21 * kind order (catalog first so identifiers exist before sales/listings resolve against them).22 */23export async function resolveBatch(opts: { limit?: number; connectorId?: string } = {}): Promise<ResolveResult> {24 const log = logger.child({ component: 'resolver' });25 const limit = opts.limit ?? 500;26 const rows = await db()27 .select()28 .from(normalizedRecords)29 .where(and(eq(normalizedRecords.status, 'pending'), opts.connectorId ? eq(normalizedRecords.connectorId, opts.connectorId) : undefined))30 .orderBy(sql`case ${normalizedRecords.kind} when 'catalog_item' then 0 when 'population_report' then 1 when 'sale' then 2 when 'price_observation' then 3 when 'listing' then 4 else 5 end`, normalizedRecords.createdAt)31 .limit(limit);32 const res: ResolveResult = { processed: 0, applied: 0, duplicates: 0, unmatched: 0, rejected: 0, fxMissing: 0, touchedAssets: [] };33 if (rows.length === 0) return res;34 const touched = new Set<string>();35 const events: Parameters<typeof emitMany>[0] = [];36 const missingFx = new Set<string>();37 const updates: Array<{ id: string; set: Partial<typeof normalizedRecords.$inferInsert> }> = [];3839 for (const row of rows) {40 res.processed++;41 const parsed = NormalizedRecordSchema.safeParse(row.payload);42 if (!parsed.success) {43 res.rejected++;44 updates.push({ id: row.id, set: { status: 'rejected', rejectReason: `schema: ${parsed.error.issues[0]?.message ?? 'invalid'}`, processedAt: new Date() } });45 continue;46 }47 try {48 const out = await applyRecord(parsed.data);49 if (out.assetId) touched.add(out.assetId);50 if (out.event) events.push({ type: out.event, entityType: out.event === 'entity_created' ? 'asset' : out.event === 'sale_detected' ? 'sale' : 'listing', entityId: out.targetId ?? out.assetId ?? undefined, payload: { assetId: out.assetId, connectorId: row.connectorId } });51 const status = out.status === 'applied' ? 'applied' : out.status === 'duplicate' ? 'applied' : out.status === 'unmatched' ? 'unmatched' : 'rejected';52 if (out.status === 'applied') res.applied++;53 else if (out.status === 'duplicate') res.duplicates++;54 else if (out.status === 'unmatched') res.unmatched++;55 else res.rejected++;56 updates.push({ id: row.id, set: { status, assetId: out.assetId, variantId: out.variantId, targetId: out.targetId, matchMethod: out.method, matchConfidence: out.confidence, rejectReason: out.status === 'duplicate' ? 'duplicate' : (out.reason ?? null), processedAt: new Date() } });57 } catch (err) {58 if (err instanceof FxMissingError) {59 res.fxMissing++;60 missingFx.add(err.currency);61 if (row.rejectReason?.startsWith('fx_missing')) {62 // second attempt after a sync: no rate exists for this currency/date → reject honestly63 res.rejected++;64 updates.push({ id: row.id, set: { status: 'rejected', rejectReason: `${err.message} (no rate after sync)`, processedAt: new Date() } });65 } else {66 updates.push({ id: row.id, set: { rejectReason: err.message } }); // stays pending until fx sync67 }68 continue;69 }70 const msg = err instanceof Error ? err.message : String(err);71 log.warn({ err, record: row.id }, 'apply failed');72 res.rejected++;73 updates.push({ id: row.id, set: { status: 'rejected', rejectReason: msg.slice(0, 500), processedAt: new Date() } });74 }75 }76 for (const u of updates) await db().update(normalizedRecords).set(u.set).where(eq(normalizedRecords.id, u.id));77 await emitMany(events);78 if (missingFx.size) {79 log.warn({ currencies: [...missingFx] }, 'fx rates missing; syncing');80 try {81 await syncFx({ backfill: true });82 } catch (err) {83 log.error({ err }, 'fx sync failed');84 }85 }86 res.touchedAssets = [...touched];87 log.info({ ...res, touchedAssets: res.touchedAssets.length }, 'resolve batch done');88 return res;89}9091export async function pendingCount(): Promise<number> {92 const [row] = await db().select({ n: sql<number>`count(*)::int` }).from(normalizedRecords).where(eq(normalizedRecords.status, 'pending'));93 return row?.n ?? 0;94}9596export async function markRejected(ids: string[], reason: string): Promise<void> {97 if (!ids.length) return;98 await db().update(normalizedRecords).set({ status: 'rejected', rejectReason: reason, processedAt: new Date() }).where(inArray(normalizedRecords.id, ids));99}100