import { spawn } from "child_process";
import fs from "fs";
import path from "path";
import {
  DATA_DIR,
  Pipeline,
  Run,
  createRun,
  getConnection,
  getPipeline,
  getRun,
  updateRunStatus,
} from "./db";
import { SLING_HOME, buildReplicationYaml, connectionUrl } from "./sling";
import { buildApplyArgs, computeSchemaExcludes, prepareDevDatabase } from "./atlas";
import { resolveBinary } from "./binaries";

export interface JobEvent {
  line: string;
  ts: number;
}

export type LineListener = (ev: JobEvent) => void;
export type DoneListener = (status: string) => void;

// Mask credentials in echoed command lines (e.g. mysql://user:pass@host/db)
function maskSecret(arg: string): string {
  return arg.replace(/(\w[\w-]*:\/\/[^:/\s]+:)[^@\s]+(@)/, "$1****$2");
}

export class AlreadyRunningError extends Error {
  constructor(pipelineId: number) {
    super(`pipeline ${pipelineId} already has a running run`);
    this.name = "AlreadyRunningError";
  }
}

interface ActiveRun {
  runId: number;
  pipelineId: number;
  logPath: string;
  buffer: JobEvent[];
  tail: string[]; // last lines, used for the error summary
  listeners: Set<LineListener>;
  doneListeners: Set<DoneListener>;
}

const TAIL_LIMIT = 20;

class JobManager {
  private active = new Map<number, ActiveRun>(); // runId -> state
  private byPipeline = new Map<number, number>(); // pipelineId -> runId

  isRunning(pipelineId: number): boolean {
    return this.byPipeline.has(pipelineId);
  }

  startRun(pipelineId: number): Run {
    const pipeline = getPipeline(pipelineId);
    if (!pipeline) throw new Error(`pipeline ${pipelineId} not found`);
    if (this.isRunning(pipelineId)) throw new AlreadyRunningError(pipelineId);

    const run = createRun({
      pipeline_id: pipelineId,
      status: "running",
      started_at: new Date().toISOString(),
    });
    const runsDir = path.join(DATA_DIR, "runs");
    fs.mkdirSync(runsDir, { recursive: true });
    const logPath = path.join(runsDir, `${run.id}.log`);
    updateRunStatus(run.id, { log_path: logPath });

    const state: ActiveRun = {
      runId: run.id,
      pipelineId,
      logPath,
      buffer: [],
      tail: [],
      listeners: new Set(),
      doneListeners: new Set(),
    };
    this.active.set(run.id, state);
    this.byPipeline.set(pipelineId, run.id);

    void this.execute(pipeline, state);
    return getRun(run.id)!;
  }

  // Subscribe to live output. Replays the in-memory buffer first, then pushes
  // new lines. Returns an unsubscribe function, or null if the run is not
  // active in this process (e.g. after a dev HMR reload).
  subscribe(
    runId: number,
    onLine: LineListener,
    onDone?: DoneListener
  ): (() => void) | null {
    const state = this.active.get(runId);
    if (!state) return null;
    for (const ev of state.buffer) onLine(ev);
    state.listeners.add(onLine);
    if (onDone) state.doneListeners.add(onDone);
    return () => {
      state.listeners.delete(onLine);
      if (onDone) state.doneListeners.delete(onDone);
    };
  }

  private emit(state: ActiveRun, line: string) {
    const ev: JobEvent = { line, ts: Date.now() };
    state.buffer.push(ev);
    state.tail.push(line);
    if (state.tail.length > TAIL_LIMIT) state.tail.shift();
    try {
      fs.appendFileSync(state.logPath, line + "\n");
    } catch {
      // keep streaming to subscribers even if the log file fails
    }
    for (const l of state.listeners) l(ev);
  }

  private finish(state: ActiveRun, status: "success" | "failed", error: string | null) {
    updateRunStatus(state.runId, {
      status,
      error,
      finished_at: new Date().toISOString(),
    });
    this.active.delete(state.runId);
    this.byPipeline.delete(state.pipelineId);
    for (const cb of state.doneListeners) cb(status);
    state.listeners.clear();
    state.doneListeners.clear();
  }

  private async execute(pipeline: Pipeline, state: ActiveRun) {
    try {
      if (pipeline.schema_sync) {
        const src = getConnection(pipeline.source_conn_id);
        const tgt = getConnection(pipeline.target_conn_id);
        if (!src || !tgt) throw new Error("pipeline connections not found");
        const excludes = await computeSchemaExcludes(pipeline);
        const scope = pipeline.schema_scope || "selected";
        const tableExcludes = pipeline.include_fk
          ? excludes.length
          : excludes.filter((p) => p !== "*[type=foreign-key]").length;
        this.emit(
          state,
          `[schema] scope=${scope}, excluding ${tableExcludes} objects, fk=${pipeline.include_fk ? "on" : "off"}`
        );
        const devUrl = prepareDevDatabase(tgt);
        const code = await this.runProcess(
          state,
          resolveBinary("atlas") ?? "atlas",
          buildApplyArgs(connectionUrl(src), connectionUrl(tgt), devUrl, excludes),
          {}
        );
        if (code !== 0) {
          throw new Error(
            `atlas schema apply exited with code ${code}\n${state.tail.join("\n")}`
          );
        }
      }

      const yamlPath = buildReplicationYaml(pipeline, state.runId);
      const code = await this.runProcess(
        state,
        resolveBinary("sling") ?? "sling",
        ["run", "-r", yamlPath, "--home-dir", SLING_HOME],
        { SLING_LOGGING: "JSON", SLING_DISABLE_TELEMETRY: "1" }
      );
      if (code === 0) {
        this.finish(state, "success", null);
      } else {
        this.finish(
          state,
          "failed",
          `sling exited with code ${code}\n${state.tail.join("\n")}`
        );
      }
    } catch (err) {
      this.emit(state, `[error] ${(err as Error).message}`);
      this.finish(state, "failed", (err as Error).message);
    }
  }

  private runProcess(
    state: ActiveRun,
    bin: string,
    args: string[],
    extraEnv: Record<string, string>
  ): Promise<number> {
    return new Promise((resolve, reject) => {
      this.emit(state, `$ ${bin} ${args.map(maskSecret).join(" ")}`);
      const child = spawn(bin, args, {
        env: { ...process.env, ...extraEnv },
      });
      if (child.pid) {
        updateRunStatus(state.runId, { pid: child.pid });
      }
      child.on("error", reject);

      const hookStream = (stream: NodeJS.ReadableStream | null) => {
        if (!stream) return;
        let buf = "";
        stream.on("data", (chunk: Buffer) => {
          buf += chunk.toString();
          let idx: number;
          while ((idx = buf.indexOf("\n")) >= 0) {
            this.emit(state, buf.slice(0, idx));
            buf = buf.slice(idx + 1);
          }
        });
        stream.on("end", () => {
          if (buf) this.emit(state, buf);
        });
      };
      hookStream(child.stdout);
      hookStream(child.stderr);

      child.on("close", (code) => resolve(code ?? 1));
    });
  }
}

// Singleton surviving dev HMR reloads.
const g = globalThis as unknown as { __jobManager?: JobManager };
export const jobManager: JobManager =
  g.__jobManager ?? (g.__jobManager = new JobManager());