TypeScript 61.9%
HTML 37.2%
SQL 0.7%
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 // Kind priority (catalog first so identifiers exist before sales/listings resolve against them).30 // Claim per kind so the query uses the (status, kind, created_at) index instead of sorting every pending row.31 const KIND_ORDER = ['catalog_item', 'population_report', 'sale', 'price_observation', 'listing', 'auction_lot', 'news_item'];32 let rows: Array<typeof normalizedRecords.$inferSelect> = [];33 for (const kind of KIND_ORDER) {34 rows = (await db().execute(sql`35 with claimed as (36 select id from normalized_records37 where status = 'pending' and kind = ${kind}${connectorFilter}38 order by created_at39 limit ${limit}40 for update skip locked41 )42 update normalized_records n set status = 'processing'43 from claimed where n.id = claimed.id44 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"45 `)) as unknown as Array<typeof normalizedRecords.$inferSelect>;46 if (rows.length) break;47 }48 const res: ResolveResult = { processed: 0, applied: 0, duplicates: 0, unmatched: 0, rejected: 0, fxMissing: 0, touchedAssets: [] };49 if (rows.length === 0) return res;50 const touched = new Set<string>();51 const events: Parameters<typeof emitMany>[0] = [];52 const missingFx = new Set<string>();53 const updates: Array<{ id: string; set: Partial<typeof normalizedRecords.$inferInsert> }> = [];5455 for (const row of rows) {56 res.processed++;57 const parsed = NormalizedRecordSchema.safeParse(row.payload);58 if (!parsed.success) {59 res.rejected++;60 updates.push({ id: row.id, set: { status: 'rejected', rejectReason: `schema: ${parsed.error.issues[0]?.message ?? 'invalid'}`, processedAt: new Date() } });61 continue;62 }63 try {64 const out = await applyRecord(parsed.data);65 if (out.assetId) touched.add(out.assetId);66 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 } });67 const status = out.status === 'applied' ? 'applied' : out.status === 'duplicate' ? 'applied' : out.status === 'unmatched' ? 'unmatched' : 'rejected';68 if (out.status === 'applied') res.applied++;69 else if (out.status === 'duplicate') res.duplicates++;70 else if (out.status === 'unmatched') res.unmatched++;71 else res.rejected++;72 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() } });73 } catch (err) {74 if (err instanceof FxMissingError) {75 res.fxMissing++;76 missingFx.add(err.currency);77 if (row.rejectReason?.startsWith('fx_missing')) {78 // second attempt after a sync: no rate exists for this currency/date → reject honestly79 res.rejected++;80 updates.push({ id: row.id, set: { status: 'rejected', rejectReason: `${err.message} (no rate after sync)`, processedAt: new Date() } });81 } else {82 updates.push({ id: row.id, set: { status: 'pending', rejectReason: err.message } }); // back to pending until fx sync83 }84 continue;85 }86 const msg = err instanceof Error ? err.message : String(err);87 log.warn({ err, record: row.id }, 'apply failed');88 res.rejected++;89 updates.push({ id: row.id, set: { status: 'rejected', rejectReason: msg.slice(0, 500), processedAt: new Date() } });90 }91 }92 for (const u of updates) await db().update(normalizedRecords).set(u.set).where(eq(normalizedRecords.id, u.id));93 await emitMany(events);94 if (missingFx.size) {95 log.warn({ currencies: [...missingFx] }, 'fx rates missing; syncing');96 try {97 await syncFx({ backfill: true });98 } catch (err) {99 log.error({ err }, 'fx sync failed');100 }101 }102 res.touchedAssets = [...touched];103 log.info({ ...res, touchedAssets: res.touchedAssets.length }, 'resolve batch done');104 return res;105}106107export async function pendingCount(): Promise<number> {108 const [row] = await db().select({ n: sql<number>`count(*)::int` }).from(normalizedRecords).where(eq(normalizedRecords.status, 'pending'));109 return row?.n ?? 0;110}111112export async function markRejected(ids: string[], reason: string): Promise<void> {113 if (!ids.length) return;114 await db().update(normalizedRecords).set({ status: 'rejected', rejectReason: reason, processedAt: new Date() }).where(inArray(normalizedRecords.id, ids));115}116