/**
 * Market Atlas CLI:
 *   ma migrate                      apply SQL migrations
 *   ma seed                         reference data (countries, exchanges, holidays)
 *   ma connectors                   list connectors with rights/realtime status
 *   ma run <connector> [--file p]   run one poll (or normalize a fixture file offline) and print the batch
 *   ma replay <day> [connector]     re-run normalization on archived raw payloads of a day (no network)
 *   ma status                       quick DB/quotes/events status
 *   ma archive                      archive observation partitions older than the retention window
 *   ma backup                       pg_dump metadata to data/backups
 */
import { readFileSync, readdirSync, existsSync } from "node:fs";
import { join } from "node:path";
import { gunzipSync } from "node:zlib";
import { CONNECTORS } from "@market-atlas/connectors";
import { makeTestContext, normalizeFixture } from "@market-atlas/connector-sdk";
import { config } from "./config.js";
import { migrate } from "./db/migrate.js";
import { pool } from "./db/pool.js";
import { seed } from "./seed/index.js";
import { archiveOldPartitions, backupMetadata } from "./core/jobs.js";

const [cmd, ...args] = process.argv.slice(2);

async function main() {
  switch (cmd) {
    case "migrate": {
      const applied = await migrate();
      console.log(applied.length ? `applied: ${applied.join(", ")}` : "up to date");
      break;
    }
    case "seed":
      await migrate();
      await seed();
      console.log("seed ok");
      break;
    case "connectors":
      for (const c of CONNECTORS) {
        const m = c.metadata;
        console.log(`${m.id.padEnd(30)} ${m.sourceType.padEnd(12)} ${m.rightsStatus.padEnd(34)} ${m.realtimeStatus.padEnd(12)} ${m.assetClasses.join(",")}`);
      }
      break;
    case "run": {
      const id = args[0];
      const def = CONNECTORS.find((c) => c.metadata.id === id);
      if (!def) throw new Error(`unknown connector ${id}`);
      const fileIdx = args.indexOf("--file");
      if (fileIdx >= 0) {
        const path = args[fileIdx + 1]!;
        const text = readFileSync(path, "utf8");
        const payload = path.endsWith(".json") ? JSON.parse(text) : text;
        const kind = args[args.indexOf("--kind") + 1] ?? "poll";
        const batch = normalizeFixture(def, args.includes("--kind") ? kind : "poll", payload);
        console.log(JSON.stringify(batch, null, 2));
        break;
      }
      if (!def.poll) throw new Error("streaming connector: use the running service or --file");
      const ctx = makeTestContext(def, { fetchImpl: fetch, secrets: process.env as Record<string, string> });
      const raws = await def.poll(ctx);
      let total = 0;
      for (const r of raws) {
        const b = def.normalize(r);
        total += b.observations.length;
        console.log(`kind=${r.kind} observations=${b.observations.length} events=${b.events?.length ?? 0} instruments=${b.instruments?.length ?? 0} filings=${b.filings?.length ?? 0}`);
        for (const o of b.observations.slice(0, 8)) console.log(`  ${o.symbol.padEnd(12)} ${o.field.padEnd(15)} ${String(o.value).padEnd(14)} ${o.sourceTimestamp ? new Date(o.sourceTimestamp).toISOString() : "-"} ${o.realtimeStatus}`);
      }
      console.log(`total observations: ${total}`);
      break;
    }
    case "replay": {
      const day = args[0];
      if (!day) throw new Error("usage: ma replay YYYY-MM-DD [connector]");
      const dir = join(config.dataDir, "raw", day);
      if (!existsSync(dir)) throw new Error(`no raw archive for ${day}`);
      const connectors = args[1] ? [args[1]] : readdirSync(dir);
      for (const cid of connectors) {
        const def = CONNECTORS.find((c) => c.metadata.id === cid);
        if (!def) continue;
        let n = 0,
          obs = 0,
          failed = 0;
        for (const f of readdirSync(join(dir, cid))) {
          const rec = JSON.parse(gunzipSync(readFileSync(join(dir, cid, f))).toString("utf8"));
          try {
            const b = def.normalize({ connectorId: cid, sourceId: def.metadata.sourceId, kind: rec.kind, payload: rec.payload, receivedAt: Date.parse(rec.received_at) });
            obs += b.observations.length;
          } catch {
            failed++;
          }
          n++;
        }
        console.log(`${cid}: ${n} payloads → ${obs} observations, ${failed} normalization failures`);
      }
      break;
    }
    case "status": {
      const [i, q, e, o, c] = await Promise.all([
        pool.query<{ n: number }>("select count(*)::int as n from instruments"),
        pool.query<{ n: number }>("select count(*)::int as n from canonical_quotes where price is not null"),
        pool.query<{ n: number }>("select count(*)::int as n from market_events where ts > now() - interval '24 hours'"),
        pool.query<{ n: number }>("select coalesce(sum(observations),0)::bigint as n from daily_stats"),
        pool.query("select id, status, messages_total, reliability_score, last_message_at from connectors order by id"),
      ]);
      console.log(`instruments: ${i.rows[0]?.n}  quotes: ${q.rows[0]?.n}  events 24h: ${e.rows[0]?.n}  observations total: ${o.rows[0]?.n}`);
      for (const r of c.rows) console.log(`  ${String(r.id).padEnd(30)} ${String(r.status).padEnd(12)} msgs=${r.messages_total} score=${r.reliability_score ?? "-"} last=${r.last_message_at ?? "-"}`);
      break;
    }
    case "archive":
      console.log(await archiveOldPartitions());
      break;
    case "backup":
      console.log(await backupMetadata());
      break;
    default:
      console.log("usage: ma migrate|seed|connectors|run <id> [--file f --kind k]|replay <day> [connector]|status|archive|backup");
  }
  await pool.end();
}

main().catch((err) => {
  console.error(err instanceof Error ? err.message : err);
  process.exit(1);
});