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';2829const [cmd = 'help', ...rest] = process.argv.slice(2);30const flags: Record<string, string | boolean> = {};31const positional: string[] = [];32for (let i = 0; i < rest.length; i++) {33 const a = rest[i]!;34 if (a.startsWith('--')) {35 const k = a.slice(2);36 const next = rest[i + 1];37 if (next && !next.startsWith('--')) {38 flags[k] = next;39 i++;40 } else flags[k] = true;41 } else positional.push(a);42}43const num = (k: string, d?: number) => (flags[k] !== undefined ? Number(flags[k]) : d);44const str = (k: string) => (typeof flags[k] === 'string' ? (flags[k] as string) : undefined);45const print = (x: unknown) => console.log(typeof x === 'string' ? x : JSON.stringify(x, null, 2));4647async function drain(fn: () => Promise<{ processed: number }>, batch = 500): Promise<number> {48 let total = 0;49 for (let i = 0; i < 10_000; i++) {50 const r = await fn();51 total += r.processed;52 if (r.processed < batch) break;53 }54 return total;55}5657async function stats() {58 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'];59 const out: Record<string, number> = {};60 for (const t of tables) {61 const r = (await db().execute(sql.raw(`select count(*)::int as n from ${t}`))) as unknown as Array<{ n: number }>;62 out[t] = r[0]?.n ?? 0;63 }64 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 }>;65 out['assets_priced'] = priced[0]?.n ?? 0;66 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>>;67 return { counts: out, indices: idx };68}6970async function main() {71 switch (cmd) {72 case 'connectors': {73 const metas = listConnectorMeta();74 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 })));75 break;76 }77 case 'crawl': {78 const id = positional[0];79 if (!id) throw new Error('usage: ri crawl <connectorId> [--mode ...] [--limit N]');80 print(await runCrawl(id, { mode: (str('mode') as 'probe' | 'incremental' | 'backfill') ?? 'incremental', limit: num('limit'), trigger: 'manual' }));81 break;82 }83 case 'normalize':84 print({ processed: await drain(() => normalizeBatch({ connectorId: str('connector'), limit: num('limit', 500) }), num('limit', 500)) });85 break;86 case 'resolve': {87 const processed = await drain(() => resolveBatch({ limit: num('limit', 500), connectorId: str('connector') }), num('limit', 500));88 print({ processed, stillPending: await pendingCount() });89 break;90 }91 case 'premiums':92 print({ written: await computePremiums(str('category')) });93 break;94 case 'value': {95 const asset = str('asset');96 if (asset) print(await valueAsset(asset, { rebuildHistory: Boolean(flags.history) }));97 else {98 await computePremiums();99 const ids = await assetsNeedingValuation({ all: Boolean(flags.all), limit: num('limit', 200_000) });100 print(await valueMany(ids, { rebuildHistory: Boolean(flags.history) }));101 }102 break;103 }104 case 'index':105 print(await runIndices());106 break;107 case 'snapshots':108 print({ categories: await runCategorySnapshots() });109 break;110 case 'radar':111 print({ findings: await runRadar() });112 break;113 case 'health':114 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 })));115 break;116 case 'fx':117 print(await syncFx({ backfill: Boolean(flags.backfill) }));118 break;119 case 'benchmarks':120 print(await syncBenchmarks());121 break;122 case 'expire':123 print({ expired: await expireListings() });124 break;125 case 'stats':126 print(await stats());127 break;128 case 'run-all': {129 const limit = num('limit', 200);130 const [fx] = (await db().execute(sql`select count(*)::int as n from fx_rates`)) as unknown as Array<{ n: number }>;131 if (!fx || fx.n === 0) print({ fx: await syncFx({ backfill: true }) });132 const summary: Record<string, unknown> = {};133 for (const m of listConnectorMeta({ enabled: true })) {134 try {135 const r = await runCrawl(m.id, { mode: 'incremental', limit, trigger: 'manual' });136 summary[m.id] = { status: r.status, raw: r.recordsRaw, dupes: r.recordsDuplicate, error: r.error };137 } catch (err) {138 summary[m.id] = { status: 'failed', error: err instanceof Error ? err.message : String(err) };139 }140 }141 summary.normalized = await drain(() => normalizeBatch({ limit: 500 }));142 summary.resolved = await drain(() => resolveBatch({ limit: 500 }));143 summary.premiums = await computePremiums();144 const ids = await assetsNeedingValuation({ limit: 200_000 });145 summary.valuation = await valueMany(ids, { rebuildHistory: Boolean(flags.history) });146 summary.indices = await runIndices();147 summary.snapshots = await runCategorySnapshots();148 summary.radar = await runRadar();149 summary.health = (await computeHealth()).map((h) => `${h.connector}:${h.status}`);150 print(summary);151 break;152 }153 case 'worker': {154 const { startWorker } = await import('./main.ts');155 const stop = await startWorker();156 const shutdown = () => void stop().then(() => process.exit(0));157 process.on('SIGINT', shutdown);158 process.on('SIGTERM', shutdown);159 return; // keep running160 }161 default:162 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`);163 }164 await flushCosts();165 await closeDb();166}167168main().catch(async (err) => {169 logger.error({ err }, 'ri failed');170 console.error(err);171 await closeDb();172 process.exit(1);173});174