| 89 | log.info(res, 'certificate verification done'); |
89 | log.info(res, 'certificate verification done'); |
| 90 | return res; |
90 | return res; |
| 91 | } |
91 | } |
| |
92 | + |
| |
93 | +/** |
| |
94 | + * One-off/idempotent backfill: register certificates + sightings from sales/listings/auction lots that already |
| |
95 | + * carry a certification number (rows ingested before the certificates table existed). Safe to re-run. |
| |
96 | + */ |
| |
97 | +export 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_at |
| |
103 | + from sales s where ${where('s')} |
| |
104 | + union all |
| |
105 | + 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_at |
| |
106 | + 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_price |
| |
110 | + from src group by grader, cert |
| |
111 | + ) |
| |
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 agg |
| |
114 | + 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_date |
| |
119 | + 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 all |
| |
121 | + 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_at |
| |
122 | + 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 | +} |