import { spawn, type ChildProcess } from "node:child_process";
import fs from "node:fs";
import path from "node:path";
import { fileURLToPath } from "node:url";
import { createLogger, isPlatform, newId, type AgentMode, type AppConfig } from "@src/shared";
import type { PostgresStore } from "@src/storage";
import { hasAdapter, SUPPORTED_PLATFORMS } from "@src/connectors";

const log = createLogger("jobs");
const here = path.dirname(fileURLToPath(import.meta.url));
const repoRoot = path.resolve(here, "..", "..", "..");

export interface JobRequest {
  platform: string;
  goal?: string;
  query?: string;
  mode?: AgentMode;
  seed?: string;
  minutes?: number;
  actions?: number;
  media?: number;
  headless?: boolean;
  account?: string;
}

/**
 * Job runner (§63 control plane, single-node version): spawns the worker CLI as a child process,
 * streams its log to data/jobs/<id>.log and tracks status in crawl_jobs. One browser per job.
 */
export class JobRunner {
  private running = new Map<string, ChildProcess>();
  readonly logsDir: string;
  constructor(private readonly store: PostgresStore, private readonly cfg: AppConfig, private readonly maxConcurrent = 2) {
    this.logsDir = path.join(cfg.dataDir, "jobs");
    fs.mkdirSync(this.logsDir, { recursive: true });
  }

  get activeCount() {
    return this.running.size;
  }

  async start(req: JobRequest): Promise<{ job_id: string }> {
    if (!isPlatform(req.platform)) throw new Error("unknown platform");
    if (!hasAdapter(req.platform)) throw new Error(`no adapter for ${req.platform} yet (available: ${SUPPORTED_PLATFORMS.join(", ")})`);
    if (this.running.size >= this.maxConcurrent) throw new Error(`max ${this.maxConcurrent} concurrent jobs`);
    const job_id = newId("job");
    const mode = req.mode ?? "research";
    const minutes = clamp(req.minutes ?? 10, 1, 120);
    const actions = clamp(req.actions ?? 60, 1, 1000);
    const media = clamp(req.media ?? 1, 0, 4);
    const goal = req.goal?.trim() || (req.query ? `Discover public content about “${req.query}”` : `Observe the ${req.platform} feed`);
    const args = ["node_modules/tsx/dist/cli.mjs", "apps/worker/src/cli.ts", mode === "learn" ? "learn" : "crawl", req.platform, "--job", job_id, "--goal", goal, "--mode", mode, "--minutes", String(minutes), "--actions", String(actions), "--media", String(media), "--account", req.account?.trim() || "research-01"];
    if (req.query) args.push("--query", req.query);
    if (req.seed) args.push("--seed", req.seed);
    if (req.headless !== false) args.push("--headless");

    await this.store.pool.query(`insert into crawl_jobs(job_id, platform, mode, goal, budget, status) values ($1,$2,$3,$4,$5,'queued')`, [job_id, req.platform, mode, goal, JSON.stringify({ minutes, actions, media, query: req.query ?? null, seed: req.seed ?? null, headless: req.headless !== false, account: req.account ?? "research-01" })]);

    const logFile = path.join(this.logsDir, `${job_id}.log`);
    const out = fs.openSync(logFile, "a");
    const child = spawn(process.execPath, args, { cwd: repoRoot, env: { ...process.env, SRC_LOG_PRETTY: "1", SRC_LOG_LEVEL: "info" }, stdio: ["ignore", out, out] });
    this.running.set(job_id, child);
    await this.store.pool.query(`update crawl_jobs set status='running' where job_id=$1`, [job_id]);
    log.info("job started", { job_id, platform: req.platform, mode, pid: child.pid });
    child.on("exit", async (code, signal) => {
      this.running.delete(job_id);
      fs.closeSync(out);
      const status = signal ? "stopped" : code === 0 ? "done" : "failed";
      let result: unknown = null;
      try {
        const sid = (await this.store.pool.query(`select session_id from crawl_jobs where job_id=$1`, [job_id])).rows[0]?.session_id as string | undefined;
        if (sid) {
          const f = path.join(this.cfg.sessionsDir, sid, "summary.json");
          if (fs.existsSync(f)) result = JSON.parse(fs.readFileSync(f, "utf8"));
        }
      } catch {
        /* ignore */
      }
      await this.store.pool.query(`update crawl_jobs set status=$2, finished_at=now(), result=$3 where job_id=$1`, [job_id, status, result ? JSON.stringify(result) : null]).catch(() => {});
      log.info("job finished", { job_id, status, code, signal });
    });
    return { job_id };
  }

  stop(job_id: string): boolean {
    const child = this.running.get(job_id);
    if (!child) return false;
    child.kill("SIGTERM");
    setTimeout(() => child.kill("SIGKILL"), 15_000).unref();
    return true;
  }

  logTail(job_id: string, lines = 200): string {
    const f = path.join(this.logsDir, `${job_id}.log`);
    if (!fs.existsSync(f)) return "";
    const all = fs.readFileSync(f, "utf8").split("\n");
    return all.slice(-lines).join("\n");
  }
}

function clamp(n: number, lo: number, hi: number) {
  return Math.max(lo, Math.min(hi, Number.isFinite(n) ? n : lo));
}