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 // Atomically claim a batch (status pending → processing) so several resolver workers can run27 // concurrently without processing the same rows; rows are released back to pending on crash.28 const connectorFilter = opts.connectorId ? sql` and connector_id = ${opts.connectorId}` : sql``;29 const rows = (await db().execute(sql`30 with claimed as (31 select id from normalized_records32 where status = 'pending'${connectorFilter}33 order by case 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, created_at34 limit ${limit}35 for update skip locked36 )37 update normalized_records n set status = 'processing'38 from claimed where n.id = claimed.id39 returning n.id, n.raw_record_id as "rawRecordId", n.connector_id as "connectorId", n.source_id as "sourceId", n.kind, n.payload, n.seq, n.asset_id as "assetId", n.variant_id as "variantId", n.match_method as "matchMethod", n.match_confidence as "matchConfidence", n.status, n.reject_reason as "rejectReason", n.target_id as "targetId", n.created_at as "createdAt", n.processed_at as "processedAt"40 `)) as unknown as Array<typeof normalizedRecords.$inferSelect>;41 const res: ResolveResult = { processed: 0, applied: 0, duplicates: 0, unmatched: 0, rejected: 0, fxMissing: 0, touchedAssets: [] };42 if (rows.length === 0) return res;43 const touched = new Set<string>();44 const events: Parameters<typeof emitMany>[0] = [];45 const missingFx = new Set<string>();46 const updates: Array<{ id: string; set: Partial<typeof normalizedRecords.$inferInsert> }> = [];4748 for (const row of rows) {49 res.processed++;50 const parsed = NormalizedRecordSchema.safeParse(row.payload);51 if (!parsed.success) {52 res.rejected++;53 updates.push({ id: row.id, set: { status: 'rejected', rejectReason: `schema: ${parsed.error.issues[0]?.message ?? 'invalid'}`, processedAt: new Date() } });54 continue;55 }56 try {57 const out = await applyRecord(parsed.data);58 if (out.assetId) touched.add(out.assetId);59 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 } });60 const status = out.status === 'applied' ? 'applied' : out.status === 'duplicate' ? 'applied' : out.status === 'unmatched' ? 'unmatched' : 'rejected';61 if (out.status === 'applied') res.applied++;62 else if (out.status === 'duplicate') res.duplicates++;63 else if (out.status === 'unmatched') res.unmatched++;64 else res.rejected++;65 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() } });66 } catch (err) {67 if (err instanceof FxMissingError) {68 res.fxMissing++;69 missingFx.add(err.currency);70 if (row.rejectReason?.startsWith('fx_missing')) {71 // second attempt after a sync: no rate exists for this currency/date → reject honestly72 res.rejected++;73 updates.push({ id: row.id, set: { status: 'rejected', rejectReason: `${err.message} (no rate after sync)`, processedAt: new Date() } });74 } else {75 updates.push({ id: row.id, set: { status: 'pending', rejectReason: err.message } }); // back to pending until fx sync76 }77 continue;78 }79 const msg = err instanceof Error ? err.message : String(err);80 log.warn({ err, record: row.id }, 'apply failed');81 res.rejected++;82 updates.push({ id: row.id, set: { status: 'rejected', rejectReason: msg.slice(0, 500), processedAt: new Date() } });83 }84 }85 for (const u of updates) await db().update(normalizedRecords).set(u.set).where(eq(normalizedRecords.id, u.id));86 await emitMany(events);87 if (missingFx.size) {88 log.warn({ currencies: [...missingFx] }, 'fx rates missing; syncing');89 try {90 await syncFx({ backfill: true });91 } catch (err) {92 log.error({ err }, 'fx sync failed');93 }94 }95 res.touchedAssets = [...touched];96 log.info({ ...res, touchedAssets: res.touchedAssets.length }, 'resolve batch done');97 return res;98}99100export async function pendingCount(): Promise<number> {101 const [row] = await db().select({ n: sql<number>`count(*)::int` }).from(normalizedRecords).where(eq(normalizedRecords.status, 'pending'));102 return row?.n ?? 0;103}104105export async function markRejected(ids: string[], reason: string): Promise<void> {106 if (!ids.length) return;107 await db().update(normalizedRecords).set({ status: 'rejected', rejectReason: reason, processedAt: new Date() }).where(inArray(normalizedRecords.id, ids));108}109