TypeScript 61.9%
HTML 37.2%
SQL 0.7%
1import { and, eq, isNull, lt, or, sql } from 'drizzle-orm';2import { certificates } from '@rareindex/database';3import { createCrawlContext, listConnectorMeta, loadConnector, missingRequirements, type RareIndexConnector } from '@rareindex/connectors';4import { logger, type NormalizedRecord } from '@rareindex/shared';5import { db } from './lib/db.ts';6import { getRouter } from './lib/router.ts';7import { flushCosts } from './lib/costs.ts';89/**10 * Certificate verification job (SPEC §22): for cert numbers seen on marketplaces that have not been11 * verified yet, call the grader's public verification page through its cert_lookup connector12 * (convention: meta.capabilities includes "cert_lookup", meta.config.grader = grader slug, and the13 * connector module exports `certUrl(cert)`). Results are stored on the certificate row; the cert14 * connector's own normalised output (catalog_item / population_report) is NOT written here — that15 * happens through the regular crawl pipeline when the connector is scheduled.16 */17export async function verifyCertificates(opts: { limit?: number; grader?: string; recheckDays?: number } = {}): Promise<{ checked: number; verified: number; notFound: number; failed: number; skippedGraders: string[] }> {18 const log = logger.child({ component: 'certs-verify' });19 const limit = opts.limit ?? 100;20 const recheckDays = opts.recheckDays ?? 180;21 const lookups = new Map<string, { connector: RareIndexConnector; certUrl: (cert: string, grade?: string | null) => string }>();22 const skippedGraders: string[] = [];23 for (const meta of listConnectorMeta({ enabled: true })) {24 if (!meta.capabilities.includes('cert_lookup') && !(meta.config as { grader?: string }).grader) continue;25 const grader = (meta.config as { grader?: string }).grader;26 if (!grader || (opts.grader && grader !== opts.grader)) continue;27 if (missingRequirements(meta).length) {28 skippedGraders.push(grader);29 continue;30 }31 try {32 const connector = await loadConnector(meta.id);33 const mod = (await import(`${new URL('../connectors/', import.meta.url).pathname}${meta.module}/index.ts`)) as { certUrl?: (cert: string, grade?: string | null) => string };34 if (!connector.lookup || !mod.certUrl) {35 skippedGraders.push(grader);36 continue;37 }38 lookups.set(grader, { connector, certUrl: mod.certUrl });39 } catch (err) {40 log.warn({ err, connector: meta.id }, 'cert connector unavailable');41 skippedGraders.push(grader);42 }43 }44 const res = { checked: 0, verified: 0, notFound: 0, failed: 0, skippedGraders };45 if (!lookups.size) return res;46 const graders = [...lookups.keys()];47 const rows = await db()48 .select()49 .from(certificates)50 .where(and(sql`${certificates.grader} = any(${sql.raw(`'{${graders.join(',')}}'::text[]`)})`, or(isNull(certificates.verifiedAt), lt(certificates.verifiedAt, new Date(Date.now() - recheckDays * 86_400_000)))))51 .orderBy(sql`${certificates.verifiedAt} nulls first`, sql`${certificates.sightings} desc`)52 .limit(limit);53 for (const row of rows) {54 const l = lookups.get(row.grader)!;55 const ctx = createCrawlContext({ router: getRouter(l.connector.meta.id), meta: l.connector.meta, options: { mode: 'probe', limit: 1 } });56 res.checked++;57 try {58 // NGC/PMG encode the holder grade in the URL ("cert/grade"); other graders ignore the suffix.59 const certArg = (row.grader === 'ngc' || row.grader === 'pmg') && row.grade ? `${row.certNumber}/${row.grade.replace(/\s+/g, '')}` : row.certNumber;60 const raws = await l.connector.lookup!(l.certUrl(certArg, row.grade), ctx);61 const records: NormalizedRecord[] = [];62 for (const raw of raws) records.push(...(await l.connector.normalize({ ...raw, externalId: raw.externalId ?? null, fetchedAt: raw.fetchedAt ?? new Date() })));63 const item = records.find((r) => r.kind === 'catalog_item');64 const pop = records.find((r) => r.kind === 'population_report');65 const found = Boolean(item);66 const verification = found67 ? { status: 'verified', title: item && 'rawTitle' in item ? item.rawTitle : null, grade: item && 'grade' in item ? item.grade : null, attributes: item?.attributes ?? null, population: pop && pop.kind === 'population_report' ? { total: pop.total, byGrade: pop.byGrade, reportDate: pop.reportDate } : null, checkedAt: new Date().toISOString(), connector: l.connector.meta.id }68 : (row.grader === 'ngc' || row.grader === 'pmg') && !row.grade69 ? { status: 'needs_grade', note: 'NGC/PMG verification URLs require the holder grade; none was captured with this cert', checkedAt: new Date().toISOString(), connector: l.connector.meta.id }70 : { status: 'not_found', checkedAt: new Date().toISOString(), connector: l.connector.meta.id, anomalies: ctx.anomalies };71 await db()72 .update(certificates)73 .set({74 verifiedAt: new Date(),75 verification,76 grade: found && item && 'grade' in item && item.grade.grade ? sql`coalesce(${certificates.grade}, ${item.grade.grade})` : sql`${certificates.grade}`,77 qualifier: found && item && 'grade' in item && item.grade.qualifier ? sql`coalesce(${certificates.qualifier}, ${item.grade.qualifier})` : sql`${certificates.qualifier}`,78 })79 .where(eq(certificates.id, row.id));80 if (found) res.verified++;81 else res.notFound++;82 } catch (err) {83 res.failed++;84 log.warn({ err, cert: row.certNumber, grader: row.grader }, 'cert verification failed');85 await db().update(certificates).set({ verification: { status: 'error', error: err instanceof Error ? err.message : String(err), checkedAt: new Date().toISOString() } }).where(eq(certificates.id, row.id));86 }87 }88 await flushCosts();89 log.info(res, 'certificate verification done');90 return res;91}9293/**94 * One-off/idempotent backfill: register certificates + sightings from sales/listings/auction lots that already95 * carry a certification number (rows ingested before the certificates table existed). Safe to re-run.96 */97export async function backfillCertificates(): Promise<{ certificates: number; sightings: number }> {98 const norm = (col: string) => sql.raw(`upper(regexp_replace(${col}, '[^A-Za-z0-9]', '', 'g'))`);99 const where = (tbl: string) => sql.raw(`${tbl}.certification_number is not null and ${tbl}.grader is not null and length(regexp_replace(${tbl}.certification_number, '[^A-Za-z0-9]', '', 'g')) between 5 and 20`);100 const inserted = (await db().execute(sql`101 with src as (102 select lower(s.grader) as grader, ${norm('s.certification_number')} as cert, s.asset_id, s.variant_id, s.grade, s.source_id, s.source_url, s.price_usd, s.sale_date as observed_at103 from sales s where ${where('s')}104 union all105 select lower(l.grader), ${norm('l.certification_number')}, l.asset_id, l.variant_id, l.grade, l.source_id, l.source_url, l.price_usd, l.last_seen_at106 from listings l where ${where('l')}107 ), agg as (108 select grader, cert, min(asset_id) as asset_id, min(variant_id) as variant_id, min(grade) as grade, min(observed_at) as first_seen, max(observed_at) as last_seen, count(*)::int as n,109 array_agg(distinct source_id) as sources, (array_agg(source_url order by observed_at desc))[1] as last_url, (array_agg(price_usd order by observed_at desc))[1] as last_price110 from src group by grader, cert111 )112 insert into certificates (id, grader, cert_number, asset_id, variant_id, grade, first_seen_at, last_seen_at, sightings, source_ids, last_source_url, last_price_usd)113 select 'cert_' || substr(md5(grader || '|' || cert), 1, 20), grader, cert, asset_id, variant_id, grade, first_seen, last_seen, n, sources, last_url, last_price from agg114 on conflict (grader, cert_number) do update set sightings = greatest(certificates.sightings, excluded.sightings), first_seen_at = least(certificates.first_seen_at, excluded.first_seen_at), last_seen_at = greatest(certificates.last_seen_at, excluded.last_seen_at), asset_id = coalesce(certificates.asset_id, excluded.asset_id)115 returning id`)) as unknown as unknown[];116 const sightings = (await db().execute(sql`117 insert into certificate_sightings (id, certificate_id, kind, target_id, source_id, connector_id, source_url, price_usd, currency, price, observed_at)118 select 'evt_' || substr(md5('sale' || s.id), 1, 20), c.id, 'sale', s.id, s.source_id, s.connector_id, s.source_url, s.price_usd, s.currency, s.price, s.sale_date119 from sales s join certificates c on c.grader = lower(s.grader) and c.cert_number = ${norm('s.certification_number')} where ${where('s')}120 union all121 select 'evt_' || substr(md5('listing' || l.id), 1, 20), c.id, 'listing', l.id, l.source_id, l.connector_id, l.source_url, l.price_usd, l.currency, l.price, l.last_seen_at122 from listings l join certificates c on c.grader = lower(l.grader) and c.cert_number = ${norm('l.certification_number')} where ${where('l')}123 on conflict do nothing returning id`)) as unknown as unknown[];124 return { certificates: inserted.length, sightings: sightings.length };125}126