import { spawn } from "node:child_process";
import { createWriteStream, mkdirSync } from "node:fs";
import { stat } from "node:fs/promises";
import { join } from "node:path";
import { createGzip } from "node:zlib";
import { pipeline as streamPipeline } from "node:stream/promises";
import { Readable } from "node:stream";
import { config } from "../config.js";
import { pool } from "../db/pool.js";
import { logger } from "../logger.js";
import { rollupBars } from "./bars.js";
import { bus } from "./bus.js";
import { calendar } from "./calendar.js";
import { eventEngine } from "./events.js";
import { health } from "./health.js";
import { instruments } from "./instruments.js";
import { pipeline } from "./pipeline.js";
import { quoteStore } from "./quotes.js";
import { telemetry } from "./telemetry.js";
import { connectorManager } from "./connector-manager.js";
import { lineage } from "./lineage.js";

/** Periodic maintenance: health snapshots, market open/close events, rollups, retention/archive, backups. */
export class Scheduler {
  private timers: NodeJS.Timeout[] = [];
  private lastState = new Map<string, string>();
  private lastDaily: string | null = null;

  start(): void {
    this.every(60_000, () => health.tick((id) => staleAfter(id)));
    this.every(5_000, () => this.marketStates());
    this.every(10_000, () => pipeline.sweep());
    this.every(5 * 60_000, () => rollupBars(3));
    this.every(5 * 60_000, () => lineage.recompute());
    this.every(60_000, () => this.gauges());
    this.every(15 * 60_000, () => this.daily());
    this.marketStates();
    this.gauges();
  }

  stop(): void {
    for (const t of this.timers) clearInterval(t);
    this.timers = [];
  }

  private every(ms: number, fn: () => unknown) {
    const t = setInterval(() => {
      Promise.resolve()
        .then(fn)
        .catch((err) => logger.error({ err }, "scheduled job failed"));
    }, ms);
    this.timers.push(t);
  }

  private gauges() {
    telemetry.gauge("quotes_cached", quoteStore.size());
    telemetry.gauge("instruments_total", instruments.count());
    telemetry.gauge("connectors_healthy", connectorManager.list().filter((r) => health.state(r.def.metadata.id) === "HEALTHY").length);
    telemetry.gauge("connectors_total", connectorManager.list().length);
  }

  /** Emit MARKET_OPEN / MARKET_CLOSE events and reset session stats at session start. */
  private marketStates() {
    const now = Date.now();
    for (const ex of calendar.list()) {
      if (ex.sessions.continuous) continue;
      const st = calendar.status(ex.id, now);
      const prev = this.lastState.get(ex.id);
      this.lastState.set(ex.id, st.state);
      if (prev === undefined || prev === st.state) continue;
      bus.publish("market.state", { exchangeId: ex.id, state: st.state, at: now });
      if (st.state === "OPEN") {
        eventEngine.system("MARKET_OPEN", `${ex.id}:${new Date(now).toISOString().slice(0, 10)}`, now, "NOTICE", `${ex.name} market open`, { exchangeId: ex.id, localTime: st.localTime }, ["market-atlas-calendar"]);
        for (const inst of instruments.all()) {
          if (inst.exchangeId === ex.id) {
            pipeline.consensus.resetSession(inst.id, now);
            eventEngine.resetSession(inst.id);
          }
        }
      } else if (st.state === "CLOSED" && (prev === "OPEN" || prev === "POST")) {
        eventEngine.system("MARKET_CLOSE", `${ex.id}:${new Date(now).toISOString().slice(0, 10)}`, now, "NOTICE", `${ex.name} market close`, { exchangeId: ex.id, localTime: st.localTime }, ["market-atlas-calendar"]);
      }
    }
  }

  /** Once per day (UTC 04:40-ish): archive old observation partitions, backup metadata. */
  private async daily() {
    const now = new Date();
    const key = now.toISOString().slice(0, 10);
    if (this.lastDaily === key) return;
    if (now.getUTCHours() < 8) return; // ~04:00 America/Toronto
    this.lastDaily = key;
    await archiveOldPartitions().catch((err) => logger.error({ err }, "archive failed"));
    await backupMetadata().catch((err) => logger.error({ err }, "backup failed"));
    await pool.query("delete from connector_health where ts < now() - interval '90 days'");
  }
}

function staleAfter(connectorId: string): number {
  const rt = connectorManager.get(connectorId);
  const m = rt?.def.metadata;
  if (!m) return 10 * 60_000;
  if (m.sourceType === "WEBSOCKET") return 5 * 60_000;
  const s = rt?.def.schedule;
  if (s && "intervalMs" in s) return s.intervalMs * 3 + 60_000;
  if (s) return s.weekendMs * 2 + 60_000;
  return 24 * 3_600_000;
}

/** Export observation partitions older than the retention window to gzip NDJSON, then drop them. */
export async function archiveOldPartitions(): Promise<string[]> {
  const cutoff = new Date(Date.now() - config.observationRetentionDays * 86_400_000).toISOString().slice(0, 10).replace(/-/g, "");
  const parts = await pool.query<{ relname: string }>(
    `select c.relname from pg_inherits i join pg_class c on c.oid = i.inhrelid join pg_class p on p.oid = i.inhparent where p.relname = 'observations' order by 1`,
  );
  const archived: string[] = [];
  for (const { relname } of parts.rows) {
    const day = relname.replace("observations_", "");
    if (day >= cutoff) continue;
    const dir = join(config.dataDir, "archive", day.slice(0, 4));
    mkdirSync(dir, { recursive: true });
    const path = join(dir, `${relname}.ndjson.gz`);
    const rows = await pool.query(`select * from ${relname}`);
    await streamPipeline(Readable.from(rows.rows.map((r) => JSON.stringify(r) + "\n")), createGzip({ level: 9 }), createWriteStream(path));
    const { size } = await stat(path);
    await pool.query(`insert into observation_archives (partition_name, day, rows_archived, path, bytes) values ($1,$2,$3,$4,$5) on conflict (partition_name) do update set rows_archived = excluded.rows_archived, bytes = excluded.bytes, archived_at = now()`, [
      relname,
      `${day.slice(0, 4)}-${day.slice(4, 6)}-${day.slice(6, 8)}`,
      rows.rowCount ?? 0,
      path,
      size,
    ]);
    await pool.query(`drop table ${relname}`);
    archived.push(relname);
    logger.info({ partition: relname, rows: rows.rowCount, bytes: size }, "partition archived");
  }
  return archived;
}

/** pg_dump of metadata tables (schema + data, excluding observations/bars) into data/backups. */
export function backupMetadata(): Promise<string> {
  return new Promise((resolve, reject) => {
    const dir = join(config.dataDir, "backups");
    mkdirSync(dir, { recursive: true });
    const file = join(dir, `market-atlas-meta-${new Date().toISOString().slice(0, 10)}.sql.gz`);
    const pgDump = process.env.PG_DUMP ?? "pg_dump";
    const child = spawn(pgDump, ["--no-owner", "--exclude-table-data=observations*", "--exclude-table-data=bars", "--exclude-table-data=connector_health", config.databaseUrl], { stdio: ["ignore", "pipe", "pipe"] });
    const out = createWriteStream(file);
    child.stdout.pipe(createGzip()).pipe(out);
    let stderr = "";
    child.stderr.on("data", (d) => (stderr += d));
    child.on("error", reject);
    child.on("close", (code) => (code === 0 ? resolve(file) : reject(new Error(`pg_dump exited ${code}: ${stderr.slice(0, 300)}`))));
  });
}

export const scheduler = new Scheduler();