1import { and, eq, sql } from 'drizzle-orm';2import { auctionLots, auctions, images, listingEvents, listings, news, populationReports, priceObservations, sales, sources } from '@rareindex/database';3import { normalizeCondition } from '@rareindex/taxonomy';4import { newId, sha256, toDateOnly, type NormalizedAuctionLot, type NormalizedCatalogItem, type NormalizedListing, type NormalizedNewsItem, type NormalizedPopulationReport, type NormalizedPriceObservation, type NormalizedRecord, type NormalizedSale } from '@rareindex/shared';5import { db } from '../lib/db.ts';6import { toUsd } from '../lib/fx.ts';7import { audit } from '../lib/audit.ts';8import { enrichAttributesFromTitle, ensureVariant, resolveAsset, type Resolution } from './resolver.ts';9import { recordCertificate } from './certificates.ts';1011/** Record a slab certification sighting when a record carries one (SPEC §22); never throws into the writer. */12async function trackCert(rec: { grade: { grader: string | null; grade: string | null; qualifier: string | null; certificationNumber: string | null }; sourceId: string; connectorId: string; sourceUrl: string }, kind: 'sale' | 'listing' | 'auction_lot', targetId: string | null, assetId: string | null, variantId: string | null, priceUsd: number | null, price: number | null, currency: string | null, observedAt: Date): Promise<void> {13 if (!rec.grade.certificationNumber || !rec.grade.grader) return;14 try {15 await recordCertificate({ grader: rec.grade.grader, certNumber: rec.grade.certificationNumber, assetId, variantId, grade: rec.grade.grade, qualifier: rec.grade.qualifier, kind, targetId, sourceId: rec.sourceId, connectorId: rec.connectorId, sourceUrl: rec.sourceUrl, priceUsd, price, currency, observedAt });16 } catch {17 /* cert tracking is best-effort; the canonical write already succeeded */18 }19}2021export class FxMissingError extends Error {22 constructor(public currency: string, public date: Date) {23 super(`fx_missing:${currency}:${toDateOnly(date)}`);24 }25}2627export interface ApplyResult {28 assetId: string | null;29 variantId: string | null;30 targetId: string | null;31 method: Resolution['method'] | null;32 confidence: number | null;33 status: 'applied' | 'unmatched' | 'rejected' | 'duplicate';34 reason?: string;35 event?: 'sale_detected' | 'listing_created' | 'listing_updated' | 'entity_created';36}3738const trustCache = new Map<string, number>();39async function sourceTrust(sourceId: string): Promise<number> {40 const hit = trustCache.get(sourceId);41 if (hit !== undefined) return hit;42 const [row] = await db().select({ t: sources.trustScore }).from(sources).where(eq(sources.id, sourceId)).limit(1);43 const t = row ? Number(row.t) : 0.5;44 trustCache.set(sourceId, t);45 return t;46}4748/** Data quality 0–100 (§150): completeness × source trust × identification confidence. */49function dataQuality(fields: Record<string, unknown>, trust: number, confidence: number, extra = 0): number {50 const keys = Object.keys(fields);51 const present = keys.filter((k) => fields[k] !== null && fields[k] !== undefined && fields[k] !== '' && !(Array.isArray(fields[k]) && (fields[k] as unknown[]).length === 0)).length;52 const completeness = keys.length ? present / keys.length : 0;53 return Math.round(Math.max(0, Math.min(100, (0.4 * completeness + 0.3 * trust + 0.3 * confidence + extra) * 100)));54}5556function normCond(categorySlug: string, raw: string | null | undefined, already: string | null | undefined): string | null {57 return already ?? normalizeCondition(categorySlug, raw);58}5960async function storeImages(assetId: string, urls: string[], sourceId: string, ref: { listingId?: string; saleId?: string }): Promise<void> {61 if (!urls.length) return;62 const rows = urls.slice(0, 12).map((url, i) => ({ id: newId('image'), assetId, listingId: ref.listingId ?? null, saleId: ref.saleId ?? null, sourceId, url, role: i === 0 && !ref.listingId && !ref.saleId ? 'hero' : 'source' }));63 await db().insert(images).values(rows).onConflictDoNothing({ target: images.url });64}6566export async function applyRecord(rec: NormalizedRecord): Promise<ApplyResult> {67 switch (rec.kind) {68 case 'catalog_item':69 return applyCatalog(rec);70 case 'sale':71 return applySale(rec);72 case 'listing':73 return applyListing(rec);74 case 'price_observation':75 return applyObservation(rec);76 case 'auction_lot':77 return applyAuctionLot(rec);78 case 'population_report':79 return applyPopulation(rec);80 case 'news_item':81 return applyNews(rec);82 }83}8485async function applyCatalog(rec: NormalizedCatalogItem): Promise<ApplyResult> {86 const r = await resolveAsset(rec.attributes, { enrich: true, imageUrls: rec.imageUrls, description: rec.description });87 if (!r) return { assetId: null, variantId: null, targetId: null, method: null, confidence: null, status: 'unmatched' };88 if (rec.imageUrls.length) await storeImages(r.assetId, rec.imageUrls.slice(0, 1), rec.sourceId, {});89 return { assetId: r.assetId, variantId: null, targetId: r.assetId, method: r.method, confidence: r.confidence, status: 'applied', event: r.created ? 'entity_created' : undefined };90}9192async function applySale(rec: NormalizedSale): Promise<ApplyResult> {93 const r = await resolveAsset(enrichAttributesFromTitle(rec.attributes, rec.rawTitle), { imageUrls: rec.imageUrls });94 if (!r) return { assetId: null, variantId: null, targetId: null, method: null, confidence: null, status: 'unmatched' };95 const condition = normCond(rec.attributes.categorySlug, rec.condition.conditionRaw, rec.condition.condition);96 const variant = await ensureVariant(r.assetId, rec.grade, { condition, completeness: rec.condition.completeness }, rec.attributes.size);97 const fx = await toUsd(rec.price, rec.currency, rec.saleDate);98 if (!fx) throw new FxMissingError(rec.currency, rec.saleDate);99 const dedupeKey = sha256(['sale', rec.sourceId, rec.externalId ?? rec.sourceUrl, toDateOnly(rec.saleDate), rec.price.toFixed(2)].join('|'));100 const trust = await sourceTrust(rec.sourceId);101 const confidence = Math.min(rec.confidence, r.confidence);102 const flags: string[] = [];103 if (rec.isBundle || rec.quantity > 1) flags.push('bundle');104 if (rec.price <= 0) flags.push('zero_price');105 if (confidence < 0.6) flags.push('low_identification_confidence');106 const status = flags.includes('bundle') || flags.includes('zero_price') ? 'excluded' : 'valid';107 const id = newId('sale');108 const inserted = await db()109 .insert(sales)110 .values({111 id,112 assetId: r.assetId,113 variantId: variant.id,114 sourceId: rec.sourceId,115 connectorId: rec.connectorId,116 sourceUrl: rec.sourceUrl,117 externalId: rec.externalId,118 saleType: rec.saleType,119 saleDate: rec.saleDate,120 price: rec.price,121 currency: rec.currency,122 priceUsd: fx.usd,123 fxRate: fx.rate,124 fxDate: fx.fxDate,125 buyerPremiumIncluded: rec.buyerPremiumIncluded,126 quantity: rec.quantity,127 isBundle: rec.isBundle,128 condition,129 grader: rec.grade.grader,130 grade: rec.grade.grade,131 certificationNumber: rec.grade.certificationNumber,132 location: rec.location,133 auctionHouse: rec.auctionHouse,134 lotNumber: rec.lotNumber,135 imageUrls: rec.imageUrls,136 rawTitle: rec.rawTitle,137 confidence,138 dataQuality: dataQuality({ date: rec.saleDate, price: rec.price, currency: rec.currency, images: rec.imageUrls, grade: rec.grade.grade, condition, externalId: rec.externalId, type: rec.saleType === 'unknown' ? null : rec.saleType }, trust, confidence),139 status,140 flags,141 dedupeKey,142 })143 .onConflictDoNothing({ target: sales.dedupeKey })144 .returning({ id: sales.id });145 if (inserted.length === 0) return { assetId: r.assetId, variantId: variant.id, targetId: null, method: r.method, confidence, status: 'duplicate' };146 if (status === 'excluded') await audit({ entityType: 'sale', entityId: id, action: 'excluded', reason: flags.join(','), details: { rawTitle: rec.rawTitle, quantity: rec.quantity } });147 if (rec.imageUrls.length) await storeImages(r.assetId, rec.imageUrls.slice(0, 3), rec.sourceId, { saleId: id });148 await trackCert(rec, 'sale', id, r.assetId, variant.id, fx.usd, rec.price, rec.currency, rec.saleDate);149 return { assetId: r.assetId, variantId: variant.id, targetId: id, method: r.method, confidence, status: 'applied', event: 'sale_detected' };150}151152async function applyListing(rec: NormalizedListing): Promise<ApplyResult> {153 const r = await resolveAsset(enrichAttributesFromTitle(rec.attributes, rec.rawTitle), { imageUrls: rec.imageUrls });154 if (!r) return { assetId: null, variantId: null, targetId: null, method: null, confidence: null, status: 'unmatched' };155 const condition = normCond(rec.attributes.categorySlug, rec.condition.conditionRaw, rec.condition.condition);156 const variant = await ensureVariant(r.assetId, rec.grade, { condition, completeness: rec.condition.completeness }, rec.attributes.size);157 let priceUsd: number | null = null;158 if (rec.price !== null && rec.currency) {159 const fx = await toUsd(rec.price, rec.currency, rec.observedAt);160 if (!fx) throw new FxMissingError(rec.currency, rec.observedAt);161 priceUsd = fx.usd;162 }163 const externalId = rec.externalId ?? sha256(rec.sourceUrl);164 const trust = await sourceTrust(rec.sourceId);165 const confidence = Math.min(rec.confidence, r.confidence);166 const now = rec.observedAt;167 const [existing] = await db().select({ id: listings.id, price: listings.price, availability: listings.availability }).from(listings).where(and(eq(listings.sourceId, rec.sourceId), eq(listings.externalId, externalId))).limit(1);168 const quality = dataQuality({ price: rec.price, currency: rec.currency, images: rec.imageUrls, seller: rec.seller, condition, grade: rec.grade.grade, listedAt: rec.listedAt, description: rec.description }, trust, confidence);169 if (!existing) {170 const id = newId('listing');171 await db().insert(listings).values({172 id,173 assetId: r.assetId,174 variantId: variant.id,175 sourceId: rec.sourceId,176 connectorId: rec.connectorId,177 sourceUrl: rec.sourceUrl,178 externalId,179 listingType: rec.listingType,180 price: rec.price,181 currency: rec.currency,182 priceUsd,183 seller: rec.seller,184 sellerReputation: rec.sellerReputation,185 location: rec.location,186 shippingCost: rec.shippingCost,187 quantity: rec.quantity,188 condition,189 grader: rec.grade.grader,190 grade: rec.grade.grade,191 certificationNumber: rec.grade.certificationNumber,192 imageUrls: rec.imageUrls,193 rawTitle: rec.rawTitle,194 description: rec.description,195 listedAt: rec.listedAt,196 endsAt: rec.endsAt,197 availability: rec.availability,198 bidCount: rec.bidCount,199 firstSeenAt: now,200 lastSeenAt: now,201 confidence,202 dataQuality: quality,203 flags: confidence < 0.6 ? ['low_identification_confidence'] : [],204 });205 await db().insert(listingEvents).values({ id: newId('event'), listingId: id, eventType: 'new', newPrice: rec.price, currency: rec.currency, occurredAt: now });206 if (rec.imageUrls.length) await storeImages(r.assetId, rec.imageUrls.slice(0, 3), rec.sourceId, { listingId: id });207 await trackCert(rec, 'listing', id, r.assetId, variant.id, priceUsd, rec.price, rec.currency, now);208 return { assetId: r.assetId, variantId: variant.id, targetId: id, method: r.method, confidence, status: 'applied', event: 'listing_created' };209 }210 const priceChanged = rec.price !== null && existing.price !== null && Math.abs(Number(existing.price) - rec.price) > 0.009;211 const relisted = existing.availability !== 'available' && rec.availability === 'available';212 const ended = existing.availability === 'available' && rec.availability !== 'available';213 await db()214 .update(listings)215 .set({216 price: rec.price,217 currency: rec.currency,218 priceUsd,219 availability: rec.availability,220 bidCount: rec.bidCount,221 endsAt: rec.endsAt ?? sql`${listings.endsAt}`,222 lastSeenAt: now,223 priceChangedAt: priceChanged ? now : sql`${listings.priceChangedAt}`,224 quantity: rec.quantity,225 dataQuality: quality,226 updatedAt: new Date(),227 })228 .where(eq(listings.id, existing.id));229 const ev = priceChanged ? 'price_changed' : relisted ? 'relisted' : ended ? (rec.availability === 'sold' ? 'sold' : rec.availability === 'ended' ? 'auction_ended' : 'removed') : null;230 if (ev) await db().insert(listingEvents).values({ id: newId('event'), listingId: existing.id, eventType: ev, oldPrice: existing.price, newPrice: rec.price, currency: rec.currency, occurredAt: now });231 return { assetId: r.assetId, variantId: variant.id, targetId: existing.id, method: r.method, confidence, status: 'applied', event: ev ? 'listing_updated' : undefined };232}233234async function applyObservation(rec: NormalizedPriceObservation): Promise<ApplyResult> {235 const r = await resolveAsset(rec.attributes, { imageUrls: rec.imageUrls });236 if (!r) return { assetId: null, variantId: null, targetId: null, method: null, confidence: null, status: 'unmatched' };237 const condition = normCond(rec.attributes.categorySlug, rec.condition.conditionRaw, rec.condition.condition);238 const variant = await ensureVariant(r.assetId, rec.grade, { condition, completeness: rec.condition.completeness }, rec.attributes.size);239 const fx = await toUsd(rec.price, rec.currency, rec.observationDate);240 if (!fx) throw new FxMissingError(rec.currency, rec.observationDate);241 const dedupeKey = sha256(['obs', rec.sourceId, rec.externalId ?? rec.sourceUrl, rec.priceKind, variant.key, toDateOnly(rec.observationDate)].join('|'));242 const id = newId('valuation');243 const inserted = await db()244 .insert(priceObservations)245 .values({ id, assetId: r.assetId, variantId: variant.id, sourceId: rec.sourceId, connectorId: rec.connectorId, sourceUrl: rec.sourceUrl, priceKind: rec.priceKind, price: rec.price, currency: rec.currency, priceUsd: fx.usd, observationDate: toDateOnly(rec.observationDate), sampleSize: rec.sampleSize, dedupeKey })246 .onConflictDoNothing({ target: priceObservations.dedupeKey })247 .returning({ id: priceObservations.id });248 if (inserted.length === 0) return { assetId: r.assetId, variantId: variant.id, targetId: null, method: r.method, confidence: r.confidence, status: 'duplicate' };249 return { assetId: r.assetId, variantId: variant.id, targetId: id, method: r.method, confidence: r.confidence, status: 'applied' };250}251252async function applyAuctionLot(rec: NormalizedAuctionLot): Promise<ApplyResult> {253 const r = await resolveAsset(enrichAttributesFromTitle(rec.attributes, rec.rawTitle), { imageUrls: rec.imageUrls, allowCreate: rec.confidence >= 0.7 });254 const auctionUrl = (rec.attributes.metadata?.auctionUrl as string | undefined) ?? `${new URL(rec.sourceUrl).origin}#${rec.auctionHouse}:${rec.auctionName ?? 'auction'}`;255 const auctionId = sha256(auctionUrl).slice(0, 24);256 await db()257 .insert(auctions)258 .values({ id: `auc_${auctionId}`, sourceId: rec.sourceId, auctionHouse: rec.auctionHouse, name: rec.auctionName ?? rec.auctionHouse, url: auctionUrl, startsAt: rec.startsAt, endsAt: rec.endsAt, location: rec.location, categorySlugs: [rec.attributes.categorySlug], status: rec.status === 'unknown' ? 'upcoming' : rec.status, currency: rec.currency })259 .onConflictDoUpdate({ target: auctions.url, set: { endsAt: sql`coalesce(excluded.ends_at, ${auctions.endsAt})`, status: sql`excluded.status`, updatedAt: new Date() } });260 let variantId: string | null = null;261 if (r) variantId = (await ensureVariant(r.assetId, rec.grade, { condition: normCond(rec.attributes.categorySlug, rec.condition.conditionRaw, rec.condition.condition), completeness: rec.condition.completeness }, rec.attributes.size)).id;262 const id = `lot_${sha256(rec.sourceUrl).slice(0, 24)}`;263 await db()264 .insert(auctionLots)265 .values({ id, auctionId: `auc_${auctionId}`, assetId: r?.assetId ?? null, variantId, sourceId: rec.sourceId, lotNumber: rec.lotNumber, title: rec.rawTitle, url: rec.sourceUrl, estimateLow: rec.estimateLow, estimateHigh: rec.estimateHigh, currentBid: rec.currentBid, currency: rec.currency, startsAt: rec.startsAt, endsAt: rec.endsAt, status: rec.status === 'unknown' ? 'upcoming' : rec.status, imageUrls: rec.imageUrls, grader: rec.grade.grader, grade: rec.grade.grade })266 .onConflictDoUpdate({ target: auctionLots.url, set: { currentBid: sql`excluded.current_bid`, status: sql`excluded.status`, endsAt: sql`coalesce(excluded.ends_at, ${auctionLots.endsAt})`, assetId: sql`coalesce(${auctionLots.assetId}, excluded.asset_id)`, updatedAt: new Date() } });267 await trackCert(rec, 'auction_lot', id, r?.assetId ?? null, variantId, null, rec.currentBid, rec.currency, rec.observedAt);268 return { assetId: r?.assetId ?? null, variantId, targetId: id, method: r?.method ?? null, confidence: r?.confidence ?? null, status: r ? 'applied' : 'unmatched' };269}270271async function applyPopulation(rec: NormalizedPopulationReport): Promise<ApplyResult> {272 const r = await resolveAsset(rec.attributes, {});273 if (!r) return { assetId: null, variantId: null, targetId: null, method: null, confidence: null, status: 'unmatched' };274 const id = newId('event');275 const inserted = await db()276 .insert(populationReports)277 .values({ id, assetId: r.assetId, grader: rec.grader, sourceId: rec.sourceId, sourceUrl: rec.sourceUrl, reportDate: toDateOnly(rec.reportDate), total: rec.total, byGrade: rec.byGrade })278 .onConflictDoNothing()279 .returning({ id: populationReports.id });280 return { assetId: r.assetId, variantId: null, targetId: inserted[0]?.id ?? null, method: r.method, confidence: r.confidence, status: inserted.length ? 'applied' : 'duplicate' };281}282283async function applyNews(rec: NormalizedNewsItem): Promise<ApplyResult> {284 const id = newId('news');285 const inserted = await db()286 .insert(news)287 .values({ id, sourceId: rec.sourceId, url: rec.sourceUrl, title: rec.title, summary: rec.summary, publishedAt: rec.publishedAt, categorySlugs: rec.categorySlugs, imageUrl: rec.imageUrl, fetchedAt: new Date() })288 .onConflictDoNothing({ target: news.url })289 .returning({ id: news.id });290 return { assetId: null, variantId: null, targetId: inserted[0]?.id ?? null, method: null, confidence: null, status: inserted.length ? 'applied' : 'duplicate' };291}292