1#!/usr/bin/env tsx2/**3 * RareIndex operator CLI.4 * ri connectors | ri crawl <id> [--mode probe|incremental|backfill] [--limit N]5 * ri backfill <id…> [--start-only] [--reset] [--pause] [--status] [--start YYYY-MM-DD] [--end YYYY-MM-DD]6 * ri certs [--limit N] [--grader psa] | ri certs --backfill (verify cert numbers via cert_lookup connectors / register certs from existing sales+listings)7 * ri normalize [--connector id] [--limit N] | ri resolve [--limit N]8 * ri value [--asset id | --all] [--history] | ri premiums [--category slug] | ri regrade [--kind sale|listing|both] [--dry-run]9 * ri index | ri snapshots | ri radar | ri health [--probe] | ri fx [--backfill] | ri benchmarks10 * ri expire | ri stats | ri run-all [--limit N] | ri worker11 */12import { loadDotenv } from './lib/env.ts';13loadDotenv();14process.env.RI_SERVICE ??= 'ri-cli';1516import { sql } from 'drizzle-orm';17import { listConnectorMeta } from '@rareindex/connectors';18import { logger } from '@rareindex/shared';19import { closeDb, db } from './lib/db.ts';20import { flushCosts } from './lib/costs.ts';21import { activeBackfill, pauseBackfill, runCrawl, startBackfill } from './crawler/run.ts';22import { normalizeBatch } from './normalizer/index.ts';23import { pendingCount, resolveBatch } from './entity-resolution/index.ts';24import { regradeRecords } from './entity-resolution/regrade.ts';25import { assetsNeedingValuation, computePremiums, valueAsset, valueMany } from './valuation/run.ts';26import { runCategorySnapshots, runIndices, runRadar } from './indices/run.ts';27import { syncFx } from './fx.ts';28import { syncBenchmarks } from './benchmarks.ts';29import { computeHealth } from './health.ts';30import { expireListings } from './listings-expire.ts';31import { imageStats, processImages } from './image-processing/index.ts';32import { backfillCertificates, verifyCertificates } from './certs-verify.ts';3334const [cmd = 'help', ...rest] = process.argv.slice(2);35const flags: Record<string, string | boolean> = {};36const positional: string[] = [];37for (let i = 0; i < rest.length; i++) {38 const a = rest[i]!;39 if (a.startsWith('--')) {40 const k = a.slice(2);41 const next = rest[i + 1];42 if (next && !next.startsWith('--')) {43 flags[k] = next;44 i++;45 } else flags[k] = true;46 } else positional.push(a);47}48const num = (k: string, d?: number) => (flags[k] !== undefined ? Number(flags[k]) : d);49const str = (k: string) => (typeof flags[k] === 'string' ? (flags[k] as string) : undefined);50const print = (x: unknown) => console.log(typeof x === 'string' ? x : JSON.stringify(x, null, 2));5152async function drain(fn: () => Promise<{ processed: number }>, batch = 500): Promise<number> {53 let total = 0;54 for (let i = 0; i < 10_000; i++) {55 const r = await fn();56 total += r.processed;57 if (r.processed < batch) break;58 }59 return total;60}6162async function stats() {63 const tables = ['sources', 'connectors', 'connector_runs', 'raw_records', 'normalized_records', 'assets', 'asset_variants', 'sets', 'brands', 'sales', 'listings', 'price_observations', 'auction_lots', 'population_reports', 'valuations', 'asset_stats', 'price_snapshots', 'index_values', 'category_snapshots', 'radar_findings', 'fx_rates', 'benchmarks', 'costs', 'events', 'audit_log'];64 const out: Record<string, number> = {};65 for (const t of tables) {66 const r = (await db().execute(sql.raw(`select count(*)::int as n from ${t}`))) as unknown as Array<{ n: number }>;67 out[t] = r[0]?.n ?? 0;68 }69 const priced = (await db().execute(sql`select count(*)::int as n from asset_stats where riv_usd is not null`)) as unknown as Array<{ n: number }>;70 out['assets_priced'] = priced[0]?.n ?? 0;71 const idx = (await db().execute(sql`select i.ticker, v.date, v.value, v.constituents_count from index_values v join indices i on i.id = v.index_id where v.date = (select max(date) from index_values v2 where v2.index_id = v.index_id) order by i.ticker`)) as unknown as Array<Record<string, unknown>>;72 return { counts: out, indices: idx };73}7475async function main() {76 switch (cmd) {77 case 'connectors': {78 const metas = listConnectorMeta();79 print(metas.map((m) => ({ id: m.id, source: m.sourceName, type: m.sourceType, engines: m.enginePriority.join('>'), categories: m.categories.join(','), enabled: m.enabled, refreshMin: m.refreshFrequencyMinutes })));80 break;81 }82 case 'crawl': {83 const id = positional[0];84 if (!id) throw new Error('usage: ri crawl <connectorId> [--mode ...] [--limit N]');85 print(await runCrawl(id, { mode: (str('mode') as 'probe' | 'incremental' | 'backfill') ?? 'incremental', limit: num('limit'), trigger: 'manual' }));86 break;87 }88 case 'backfill': {89 // ri backfill <id> [--reset] [--pause] [--status] [--start YYYY-MM-DD] [--end YYYY-MM-DD] [--limit N]90 const id = positional[0];91 if (!id) throw new Error('usage: ri backfill <connectorId> [--reset|--pause|--status]');92 if (flags.status) {93 print(await activeBackfill(id));94 break;95 }96 if (flags.pause) {97 print({ paused: await pauseBackfill(id) });98 break;99 }100 const ids = positional.length > 1 ? positional : [id];101 for (const cid of ids) {102 const campaign = await startBackfill(cid, { reset: Boolean(flags.reset), startDate: str('start') ?? null, endDate: str('end') ?? null });103 print({ connector: cid, campaign: campaign.id, status: campaign.status, resumedFrom: campaign.lastCursor });104 // --start-only: register the campaign and let the worker scheduler run the slices (recommended for many sources)105 if (!flags['start-only']) {106 print(await runCrawl(cid, { mode: 'backfill', limit: num('limit'), trigger: 'manual' }));107 print(await activeBackfill(cid));108 }109 }110 break;111 }112 case 'certs':113 if (flags.backfill) print(await backfillCertificates());114 else print(await verifyCertificates({ limit: num('limit', 100), grader: str('grader') }));115 break;116 case 'normalize':117 print({ processed: await drain(() => normalizeBatch({ connectorId: str('connector'), limit: num('limit', 500) }), num('limit', 500)) });118 break;119 case 'resolve': {120 const processed = await drain(() => resolveBatch({ limit: num('limit', 500), connectorId: str('connector') }), num('limit', 500));121 print({ processed, stillPending: await pendingCount() });122 break;123 }124 case 'premiums':125 print({ written: await computePremiums(str('category')) });126 break;127 case 'regrade': {128 // ri regrade [--kind sale|listing|both] [--connector id] [--limit N] [--dry-run]129 const kind = (str('kind') ?? 'both') as 'sale' | 'listing' | 'both';130 print(await regradeRecords({ kind, connectorId: str('connector'), limit: num('limit', 50_000), dryRun: Boolean(flags['dry-run']), recheck: Boolean(flags.recheck) }));131 break;132 }133 case 'value': {134 const asset = str('asset');135 if (asset) print(await valueAsset(asset, { rebuildHistory: Boolean(flags.history) }));136 else {137 await computePremiums();138 const ids = await assetsNeedingValuation({ all: Boolean(flags.all), limit: num('limit', 200_000) });139 print(await valueMany(ids, { rebuildHistory: Boolean(flags.history) }));140 }141 break;142 }143 case 'index':144 print(await runIndices());145 break;146 case 'snapshots':147 print({ categories: await runCategorySnapshots() });148 break;149 case 'radar':150 print({ findings: await runRadar() });151 break;152 case 'health':153 print((await computeHealth({ connectorId: positional[0], probe: Boolean(flags.probe) })).map((h) => ({ connector: h.connector, status: h.status, success24h: h.success_rate_24h, records24h: h.records_24h, lastError: h.last_error, anomalies: h.anomalies })));154 break;155 case 'fx':156 print(await syncFx({ backfill: Boolean(flags.backfill) }));157 break;158 case 'benchmarks':159 print(await syncBenchmarks());160 break;161 case 'expire':162 print({ expired: await expireListings() });163 break;164 case 'stats':165 print(await stats());166 break;167 case 'images': {168 const r = await processImages({ limit: num('limit', 2000), recheck: Boolean(flags.recheck) });169 print({ ...r, totals: await imageStats() });170 break;171 }172 case 'run-all': {173 const limit = num('limit', 200);174 const [fx] = (await db().execute(sql`select count(*)::int as n from fx_rates`)) as unknown as Array<{ n: number }>;175 if (!fx || fx.n === 0) print({ fx: await syncFx({ backfill: true }) });176 const summary: Record<string, unknown> = {};177 for (const m of listConnectorMeta({ enabled: true })) {178 try {179 const r = await runCrawl(m.id, { mode: 'incremental', limit, trigger: 'manual' });180 summary[m.id] = { status: r.status, raw: r.recordsRaw, dupes: r.recordsDuplicate, error: r.error };181 } catch (err) {182 summary[m.id] = { status: 'failed', error: err instanceof Error ? err.message : String(err) };183 }184 }185 summary.normalized = await drain(() => normalizeBatch({ limit: 500 }));186 summary.resolved = await drain(() => resolveBatch({ limit: 500 }));187 summary.premiums = await computePremiums();188 const ids = await assetsNeedingValuation({ limit: 200_000 });189 summary.valuation = await valueMany(ids, { rebuildHistory: Boolean(flags.history) });190 summary.indices = await runIndices();191 summary.snapshots = await runCategorySnapshots();192 summary.radar = await runRadar();193 summary.health = (await computeHealth()).map((h) => `${h.connector}:${h.status}`);194 print(summary);195 break;196 }197 case 'worker': {198 const { startWorker } = await import('./main.ts');199 const stop = await startWorker();200 const shutdown = () => void stop().then(() => process.exit(0));201 process.on('SIGINT', shutdown);202 process.on('SIGTERM', shutdown);203 return; // keep running204 }205 default:206 print(`RareIndex CLI\n ri connectors | crawl <id> [--mode probe|incremental|backfill] [--limit N] | normalize [--connector id] | resolve\n ri value [--asset id|--all] [--history] | premiums | regrade [--dry-run] | index | snapshots | radar | health [--probe] | fx [--backfill] | benchmarks | expire\n ri stats | run-all [--limit N] [--history] | worker`);207 }208 await flushCosts();209 await closeDb();210}211212main().catch(async (err) => {213 logger.error({ err }, 'ri failed');214 console.error(err);215 await closeDb();216 process.exit(1);217});218