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> [--reset] [--pause] [--status] [--start YYYY-MM-DD] [--end YYYY-MM-DD]6 * ri certs [--limit N] [--grader psa] (verify cert numbers via cert_lookup connectors)7 * ri normalize [--connector id] [--limit N] | ri resolve [--limit N]8 * ri value [--asset id | --all] [--history] | ri premiums [--category slug]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 { assetsNeedingValuation, computePremiums, valueAsset, valueMany } from './valuation/run.ts';25import { runCategorySnapshots, runIndices, runRadar } from './indices/run.ts';26import { syncFx } from './fx.ts';27import { syncBenchmarks } from './benchmarks.ts';28import { computeHealth } from './health.ts';29import { expireListings } from './listings-expire.ts';30import { imageStats, processImages } from './image-processing/index.ts';31import { verifyCertificates } from './certs-verify.ts';3233const [cmd = 'help', ...rest] = process.argv.slice(2);34const flags: Record<string, string | boolean> = {};35const positional: string[] = [];36for (let i = 0; i < rest.length; i++) {37 const a = rest[i]!;38 if (a.startsWith('--')) {39 const k = a.slice(2);40 const next = rest[i + 1];41 if (next && !next.startsWith('--')) {42 flags[k] = next;43 i++;44 } else flags[k] = true;45 } else positional.push(a);46}47const num = (k: string, d?: number) => (flags[k] !== undefined ? Number(flags[k]) : d);48const str = (k: string) => (typeof flags[k] === 'string' ? (flags[k] as string) : undefined);49const print = (x: unknown) => console.log(typeof x === 'string' ? x : JSON.stringify(x, null, 2));5051async function drain(fn: () => Promise<{ processed: number }>, batch = 500): Promise<number> {52 let total = 0;53 for (let i = 0; i < 10_000; i++) {54 const r = await fn();55 total += r.processed;56 if (r.processed < batch) break;57 }58 return total;59}6061async function stats() {62 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'];63 const out: Record<string, number> = {};64 for (const t of tables) {65 const r = (await db().execute(sql.raw(`select count(*)::int as n from ${t}`))) as unknown as Array<{ n: number }>;66 out[t] = r[0]?.n ?? 0;67 }68 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 }>;69 out['assets_priced'] = priced[0]?.n ?? 0;70 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>>;71 return { counts: out, indices: idx };72}7374async function main() {75 switch (cmd) {76 case 'connectors': {77 const metas = listConnectorMeta();78 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 })));79 break;80 }81 case 'crawl': {82 const id = positional[0];83 if (!id) throw new Error('usage: ri crawl <connectorId> [--mode ...] [--limit N]');84 print(await runCrawl(id, { mode: (str('mode') as 'probe' | 'incremental' | 'backfill') ?? 'incremental', limit: num('limit'), trigger: 'manual' }));85 break;86 }87 case 'backfill': {88 // ri backfill <id> [--reset] [--pause] [--status] [--start YYYY-MM-DD] [--end YYYY-MM-DD] [--limit N]89 const id = positional[0];90 if (!id) throw new Error('usage: ri backfill <connectorId> [--reset|--pause|--status]');91 if (flags.status) {92 print(await activeBackfill(id));93 break;94 }95 if (flags.pause) {96 print({ paused: await pauseBackfill(id) });97 break;98 }99 const campaign = await startBackfill(id, { reset: Boolean(flags.reset), startDate: str('start') ?? null, endDate: str('end') ?? null });100 print({ campaign: campaign.id, resumedFrom: campaign.lastCursor });101 print(await runCrawl(id, { mode: 'backfill', limit: num('limit'), trigger: 'manual' }));102 print(await activeBackfill(id));103 break;104 }105 case 'certs':106 print(await verifyCertificates({ limit: num('limit', 100), grader: str('grader') }));107 break;108 case 'normalize':109 print({ processed: await drain(() => normalizeBatch({ connectorId: str('connector'), limit: num('limit', 500) }), num('limit', 500)) });110 break;111 case 'resolve': {112 const processed = await drain(() => resolveBatch({ limit: num('limit', 500), connectorId: str('connector') }), num('limit', 500));113 print({ processed, stillPending: await pendingCount() });114 break;115 }116 case 'premiums':117 print({ written: await computePremiums(str('category')) });118 break;119 case 'value': {120 const asset = str('asset');121 if (asset) print(await valueAsset(asset, { rebuildHistory: Boolean(flags.history) }));122 else {123 await computePremiums();124 const ids = await assetsNeedingValuation({ all: Boolean(flags.all), limit: num('limit', 200_000) });125 print(await valueMany(ids, { rebuildHistory: Boolean(flags.history) }));126 }127 break;128 }129 case 'index':130 print(await runIndices());131 break;132 case 'snapshots':133 print({ categories: await runCategorySnapshots() });134 break;135 case 'radar':136 print({ findings: await runRadar() });137 break;138 case 'health':139 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 })));140 break;141 case 'fx':142 print(await syncFx({ backfill: Boolean(flags.backfill) }));143 break;144 case 'benchmarks':145 print(await syncBenchmarks());146 break;147 case 'expire':148 print({ expired: await expireListings() });149 break;150 case 'stats':151 print(await stats());152 break;153 case 'images': {154 const r = await processImages({ limit: num('limit', 2000), recheck: Boolean(flags.recheck) });155 print({ ...r, totals: await imageStats() });156 break;157 }158 case 'run-all': {159 const limit = num('limit', 200);160 const [fx] = (await db().execute(sql`select count(*)::int as n from fx_rates`)) as unknown as Array<{ n: number }>;161 if (!fx || fx.n === 0) print({ fx: await syncFx({ backfill: true }) });162 const summary: Record<string, unknown> = {};163 for (const m of listConnectorMeta({ enabled: true })) {164 try {165 const r = await runCrawl(m.id, { mode: 'incremental', limit, trigger: 'manual' });166 summary[m.id] = { status: r.status, raw: r.recordsRaw, dupes: r.recordsDuplicate, error: r.error };167 } catch (err) {168 summary[m.id] = { status: 'failed', error: err instanceof Error ? err.message : String(err) };169 }170 }171 summary.normalized = await drain(() => normalizeBatch({ limit: 500 }));172 summary.resolved = await drain(() => resolveBatch({ limit: 500 }));173 summary.premiums = await computePremiums();174 const ids = await assetsNeedingValuation({ limit: 200_000 });175 summary.valuation = await valueMany(ids, { rebuildHistory: Boolean(flags.history) });176 summary.indices = await runIndices();177 summary.snapshots = await runCategorySnapshots();178 summary.radar = await runRadar();179 summary.health = (await computeHealth()).map((h) => `${h.connector}:${h.status}`);180 print(summary);181 break;182 }183 case 'worker': {184 const { startWorker } = await import('./main.ts');185 const stop = await startWorker();186 const shutdown = () => void stop().then(() => process.exit(0));187 process.on('SIGINT', shutdown);188 process.on('SIGTERM', shutdown);189 return; // keep running190 }191 default:192 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 | index | snapshots | radar | health [--probe] | fx [--backfill] | benchmarks | expire\n ri stats | run-all [--limit N] [--history] | worker`);193 }194 await flushCosts();195 await closeDb();196}197198main().catch(async (err) => {199 logger.error({ err }, 'ri failed');200 console.error(err);201 await closeDb();202 process.exit(1);203});204