#!/usr/bin/env tsx
/**
 * RareIndex operator CLI.
 *   ri connectors | ri crawl <id> [--mode probe|incremental|backfill] [--limit N]
 *   ri backfill <id…> [--start-only] [--reset] [--pause] [--status] [--start YYYY-MM-DD] [--end YYYY-MM-DD]
 *   ri certs [--limit N] [--grader psa] | ri certs --backfill   (verify cert numbers via cert_lookup connectors / register certs from existing sales+listings)
 *   ri normalize [--connector id] [--limit N] | ri resolve [--limit N]
 *   ri value [--asset id | --all] [--history] | ri premiums [--category slug] | ri regrade [--kind sale|listing|both] [--dry-run]
 *   ri fees --backfill [--limit N] [--connector id] [--force] | ri auctions --assess [--all] [--limit N]
 *   ri index | ri snapshots | ri radar | ri health [--probe] | ri fx [--backfill] | ri benchmarks
 *   ri expire | ri stats | ri run-all [--limit N] | ri worker
 */
import { loadDotenv } from './lib/env.ts';
loadDotenv();
process.env.RI_SERVICE ??= 'ri-cli';

import { sql } from 'drizzle-orm';
import { listConnectorMeta } from '@rareindex/connectors';
import { logger } from '@rareindex/shared';
import { closeDb, db } from './lib/db.ts';
import { flushCosts } from './lib/costs.ts';
import { activeBackfill, pauseBackfill, runCrawl, startBackfill } from './crawler/run.ts';
import { normalizeBatch } from './normalizer/index.ts';
import { pendingCount, resolveBatch } from './entity-resolution/index.ts';
import { regradeRecords } from './entity-resolution/regrade.ts';
import { backfillFees } from './fees-backfill.ts';
import { assessLots } from './auctions/assess.ts';
import { assetsNeedingValuation, computePremiums, valueAsset, valueMany } from './valuation/run.ts';
import { runCategorySnapshots, runIndices, runRadar } from './indices/run.ts';
import { syncFx } from './fx.ts';
import { syncBenchmarks } from './benchmarks.ts';
import { computeHealth } from './health.ts';
import { expireListings } from './listings-expire.ts';
import { imageStats, processImages } from './image-processing/index.ts';
import { backfillCertificates, verifyCertificates } from './certs-verify.ts';

const [cmd = 'help', ...rest] = process.argv.slice(2);
const flags: Record<string, string | boolean> = {};
const positional: string[] = [];
for (let i = 0; i < rest.length; i++) {
  const a = rest[i]!;
  if (a.startsWith('--')) {
    const k = a.slice(2);
    const next = rest[i + 1];
    if (next && !next.startsWith('--')) {
      flags[k] = next;
      i++;
    } else flags[k] = true;
  } else positional.push(a);
}
const num = (k: string, d?: number) => (flags[k] !== undefined ? Number(flags[k]) : d);
const str = (k: string) => (typeof flags[k] === 'string' ? (flags[k] as string) : undefined);
const print = (x: unknown) => console.log(typeof x === 'string' ? x : JSON.stringify(x, null, 2));

async function drain(fn: () => Promise<{ processed: number }>, batch = 500): Promise<number> {
  let total = 0;
  for (let i = 0; i < 10_000; i++) {
    const r = await fn();
    total += r.processed;
    if (r.processed < batch) break;
  }
  return total;
}

async function stats() {
  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'];
  const out: Record<string, number> = {};
  for (const t of tables) {
    const r = (await db().execute(sql.raw(`select count(*)::int as n from ${t}`))) as unknown as Array<{ n: number }>;
    out[t] = r[0]?.n ?? 0;
  }
  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 }>;
  out['assets_priced'] = priced[0]?.n ?? 0;
  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>>;
  return { counts: out, indices: idx };
}

async function main() {
  switch (cmd) {
    case 'connectors': {
      const metas = listConnectorMeta();
      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 })));
      break;
    }
    case 'crawl': {
      const id = positional[0];
      if (!id) throw new Error('usage: ri crawl <connectorId> [--mode ...] [--limit N]');
      print(await runCrawl(id, { mode: (str('mode') as 'probe' | 'incremental' | 'backfill') ?? 'incremental', limit: num('limit'), trigger: 'manual' }));
      break;
    }
    case 'backfill': {
      // ri backfill <id> [--reset] [--pause] [--status] [--start YYYY-MM-DD] [--end YYYY-MM-DD] [--limit N]
      const id = positional[0];
      if (!id) throw new Error('usage: ri backfill <connectorId> [--reset|--pause|--status]');
      if (flags.status) {
        print(await activeBackfill(id));
        break;
      }
      if (flags.pause) {
        print({ paused: await pauseBackfill(id) });
        break;
      }
      const ids = positional.length > 1 ? positional : [id];
      for (const cid of ids) {
        const campaign = await startBackfill(cid, { reset: Boolean(flags.reset), startDate: str('start') ?? null, endDate: str('end') ?? null });
        print({ connector: cid, campaign: campaign.id, status: campaign.status, resumedFrom: campaign.lastCursor });
        // --start-only: register the campaign and let the worker scheduler run the slices (recommended for many sources)
        if (!flags['start-only']) {
          print(await runCrawl(cid, { mode: 'backfill', limit: num('limit'), trigger: 'manual' }));
          print(await activeBackfill(cid));
        }
      }
      break;
    }
    case 'certs':
      if (flags.backfill) print(await backfillCertificates());
      else print(await verifyCertificates({ limit: num('limit', 100), grader: str('grader') }));
      break;
    case 'normalize':
      print({ processed: await drain(() => normalizeBatch({ connectorId: str('connector'), limit: num('limit', 500) }), num('limit', 500)) });
      break;
    case 'resolve': {
      const processed = await drain(() => resolveBatch({ limit: num('limit', 500), connectorId: str('connector') }), num('limit', 500));
      print({ processed, stillPending: await pendingCount() });
      break;
    }
    case 'premiums':
      print({ written: await computePremiums(str('category')) });
      break;
    case 'regrade': {
      // ri regrade [--kind sale|listing|both] [--connector id] [--limit N] [--dry-run]
      const kind = (str('kind') ?? 'both') as 'sale' | 'listing' | 'both';
      print(await regradeRecords({ kind, connectorId: str('connector'), limit: num('limit', 50_000), dryRun: Boolean(flags['dry-run']), recheck: Boolean(flags.recheck) }));
      break;
    }
    case 'fees': {
      // ri fees --backfill [--limit N] [--connector id] [--force]   (buyer-pays price on existing sales, §35)
      if (!flags.backfill) {
        print('usage: ri fees --backfill [--limit N] [--connector id] [--force]');
        break;
      }
      print(await backfillFees({ limit: num('limit'), connectorId: str('connector'), force: Boolean(flags.force) }));
      break;
    }
    case 'auctions': {
      // ri auctions --assess [--all] [--limit N]   (all-in bid / estimate vs RIV on live lots, §33–§35)
      if (!flags.assess) {
        print('usage: ri auctions --assess [--all] [--limit N]');
        break;
      }
      print(await assessLots({ limit: num('limit', 50_000), all: Boolean(flags.all) }));
      break;
    }
    case 'value': {
      const asset = str('asset');
      if (asset) print(await valueAsset(asset, { rebuildHistory: Boolean(flags.history) }));
      else {
        await computePremiums();
        const ids = await assetsNeedingValuation({ all: Boolean(flags.all), limit: num('limit', 200_000) });
        print(await valueMany(ids, { rebuildHistory: Boolean(flags.history) }));
      }
      break;
    }
    case 'index':
      print(await runIndices());
      break;
    case 'snapshots':
      print({ categories: await runCategorySnapshots() });
      break;
    case 'radar':
      print({ findings: await runRadar() });
      break;
    case 'health':
      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 })));
      break;
    case 'fx':
      print(await syncFx({ backfill: Boolean(flags.backfill) }));
      break;
    case 'benchmarks':
      print(await syncBenchmarks());
      break;
    case 'expire':
      print({ expired: await expireListings() });
      break;
    case 'stats':
      print(await stats());
      break;
    case 'images': {
      const r = await processImages({ limit: num('limit', 2000), recheck: Boolean(flags.recheck) });
      print({ ...r, totals: await imageStats() });
      break;
    }
    case 'run-all': {
      const limit = num('limit', 200);
      const [fx] = (await db().execute(sql`select count(*)::int as n from fx_rates`)) as unknown as Array<{ n: number }>;
      if (!fx || fx.n === 0) print({ fx: await syncFx({ backfill: true }) });
      const summary: Record<string, unknown> = {};
      for (const m of listConnectorMeta({ enabled: true })) {
        try {
          const r = await runCrawl(m.id, { mode: 'incremental', limit, trigger: 'manual' });
          summary[m.id] = { status: r.status, raw: r.recordsRaw, dupes: r.recordsDuplicate, error: r.error };
        } catch (err) {
          summary[m.id] = { status: 'failed', error: err instanceof Error ? err.message : String(err) };
        }
      }
      summary.normalized = await drain(() => normalizeBatch({ limit: 500 }));
      summary.resolved = await drain(() => resolveBatch({ limit: 500 }));
      summary.premiums = await computePremiums();
      const ids = await assetsNeedingValuation({ limit: 200_000 });
      summary.valuation = await valueMany(ids, { rebuildHistory: Boolean(flags.history) });
      summary.indices = await runIndices();
      summary.snapshots = await runCategorySnapshots();
      summary.radar = await runRadar();
      summary.health = (await computeHealth()).map((h) => `${h.connector}:${h.status}`);
      print(summary);
      break;
    }
    case 'worker': {
      const { startWorker } = await import('./main.ts');
      const stop = await startWorker();
      const shutdown = () => void stop().then(() => process.exit(0));
      process.on('SIGINT', shutdown);
      process.on('SIGTERM', shutdown);
      return; // keep running
    }
    default:
      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] | fees --backfill | auctions --assess | index | snapshots | radar | health [--probe] | fx [--backfill] | benchmarks | expire\n  ri stats | run-all [--limit N] [--history] | worker`);
  }
  await flushCosts();
  await closeDb();
}

main().catch(async (err) => {
  logger.error({ err }, 'ri failed');
  console.error(err);
  await closeDb();
  process.exit(1);
});