1#!/usr/bin/env tsx2/**3 * RareIndex operator CLI.4 * ri connectors | ri crawl <id> [--mode probe|incremental|backfill] [--limit N]5 * ri normalize [--connector id] [--limit N] | ri resolve [--limit N]6 * ri value [--asset id | --all] [--history] | ri premiums [--category slug]7 * ri index | ri snapshots | ri radar | ri health [--probe] | ri fx [--backfill] | ri benchmarks8 * ri expire | ri stats | ri run-all [--limit N] | ri worker9 */10import { loadDotenv } from './lib/env.ts';11loadDotenv();12process.env.RI_SERVICE ??= 'ri-cli';1314import { sql } from 'drizzle-orm';15import { listConnectorMeta } from '@rareindex/connectors';16import { logger } from '@rareindex/shared';17import { closeDb, db } from './lib/db.ts';18import { flushCosts } from './lib/costs.ts';19import { runCrawl } from './crawler/run.ts';20import { normalizeBatch } from './normalizer/index.ts';21import { pendingCount, resolveBatch } from './entity-resolution/index.ts';22import { assetsNeedingValuation, computePremiums, valueAsset, valueMany } from './valuation/run.ts';23import { runCategorySnapshots, runIndices, runRadar } from './indices/run.ts';24import { syncFx } from './fx.ts';25import { syncBenchmarks } from './benchmarks.ts';26import { computeHealth } from './health.ts';27import { expireListings } from './listings-expire.ts';28import { imageStats, processImages } from './image-processing/index.ts';2930const [cmd = 'help', ...rest] = process.argv.slice(2);31const flags: Record<string, string | boolean> = {};32const positional: string[] = [];33for (let i = 0; i < rest.length; i++) {34 const a = rest[i]!;35 if (a.startsWith('--')) {36 const k = a.slice(2);37 const next = rest[i + 1];38 if (next && !next.startsWith('--')) {39 flags[k] = next;40 i++;41 } else flags[k] = true;42 } else positional.push(a);43}44const num = (k: string, d?: number) => (flags[k] !== undefined ? Number(flags[k]) : d);45const str = (k: string) => (typeof flags[k] === 'string' ? (flags[k] as string) : undefined);46const print = (x: unknown) => console.log(typeof x === 'string' ? x : JSON.stringify(x, null, 2));4748async function drain(fn: () => Promise<{ processed: number }>, batch = 500): Promise<number> {49 let total = 0;50 for (let i = 0; i < 10_000; i++) {51 const r = await fn();52 total += r.processed;53 if (r.processed < batch) break;54 }55 return total;56}5758async function stats() {59 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'];60 const out: Record<string, number> = {};61 for (const t of tables) {62 const r = (await db().execute(sql.raw(`select count(*)::int as n from ${t}`))) as unknown as Array<{ n: number }>;63 out[t] = r[0]?.n ?? 0;64 }65 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 }>;66 out['assets_priced'] = priced[0]?.n ?? 0;67 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>>;68 return { counts: out, indices: idx };69}7071async function main() {72 switch (cmd) {73 case 'connectors': {74 const metas = listConnectorMeta();75 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 })));76 break;77 }78 case 'crawl': {79 const id = positional[0];80 if (!id) throw new Error('usage: ri crawl <connectorId> [--mode ...] [--limit N]');81 print(await runCrawl(id, { mode: (str('mode') as 'probe' | 'incremental' | 'backfill') ?? 'incremental', limit: num('limit'), trigger: 'manual' }));82 break;83 }84 case 'normalize':85 print({ processed: await drain(() => normalizeBatch({ connectorId: str('connector'), limit: num('limit', 500) }), num('limit', 500)) });86 break;87 case 'resolve': {88 const processed = await drain(() => resolveBatch({ limit: num('limit', 500), connectorId: str('connector') }), num('limit', 500));89 print({ processed, stillPending: await pendingCount() });90 break;91 }92 case 'premiums':93 print({ written: await computePremiums(str('category')) });94 break;95 case 'value': {96 const asset = str('asset');97 if (asset) print(await valueAsset(asset, { rebuildHistory: Boolean(flags.history) }));98 else {99 await computePremiums();100 const ids = await assetsNeedingValuation({ all: Boolean(flags.all), limit: num('limit', 200_000) });101 print(await valueMany(ids, { rebuildHistory: Boolean(flags.history) }));102 }103 break;104 }105 case 'index':106 print(await runIndices());107 break;108 case 'snapshots':109 print({ categories: await runCategorySnapshots() });110 break;111 case 'radar':112 print({ findings: await runRadar() });113 break;114 case 'health':115 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 })));116 break;117 case 'fx':118 print(await syncFx({ backfill: Boolean(flags.backfill) }));119 break;120 case 'benchmarks':121 print(await syncBenchmarks());122 break;123 case 'expire':124 print({ expired: await expireListings() });125 break;126 case 'stats':127 print(await stats());128 break;129 case 'images': {130 const r = await processImages({ limit: num('limit', 2000), recheck: Boolean(flags.recheck) });131 print({ ...r, totals: await imageStats() });132 break;133 }134 case 'run-all': {135 const limit = num('limit', 200);136 const [fx] = (await db().execute(sql`select count(*)::int as n from fx_rates`)) as unknown as Array<{ n: number }>;137 if (!fx || fx.n === 0) print({ fx: await syncFx({ backfill: true }) });138 const summary: Record<string, unknown> = {};139 for (const m of listConnectorMeta({ enabled: true })) {140 try {141 const r = await runCrawl(m.id, { mode: 'incremental', limit, trigger: 'manual' });142 summary[m.id] = { status: r.status, raw: r.recordsRaw, dupes: r.recordsDuplicate, error: r.error };143 } catch (err) {144 summary[m.id] = { status: 'failed', error: err instanceof Error ? err.message : String(err) };145 }146 }147 summary.normalized = await drain(() => normalizeBatch({ limit: 500 }));148 summary.resolved = await drain(() => resolveBatch({ limit: 500 }));149 summary.premiums = await computePremiums();150 const ids = await assetsNeedingValuation({ limit: 200_000 });151 summary.valuation = await valueMany(ids, { rebuildHistory: Boolean(flags.history) });152 summary.indices = await runIndices();153 summary.snapshots = await runCategorySnapshots();154 summary.radar = await runRadar();155 summary.health = (await computeHealth()).map((h) => `${h.connector}:${h.status}`);156 print(summary);157 break;158 }159 case 'worker': {160 const { startWorker } = await import('./main.ts');161 const stop = await startWorker();162 const shutdown = () => void stop().then(() => process.exit(0));163 process.on('SIGINT', shutdown);164 process.on('SIGTERM', shutdown);165 return; // keep running166 }167 default:168 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`);169 }170 await flushCosts();171 await closeDb();172}173174main().catch(async (err) => {175 logger.error({ err }, 'ri failed');176 console.error(err);177 await closeDb();178 process.exit(1);179});180