import { spawn, ChildProcess } from "child_process";
import fs from "fs";
import path from "path";
import {
  DATA_DIR,
  Pipeline,
  Run,
  createRun,
  db,
  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>;
  child: ChildProcess | null; // current subprocess (atlas or sling)
  cancelled: boolean;
}

const TAIL_LIMIT = 20;

// True if a process with this pid exists (EPERM means it exists but is owned
// by another user).
export function isProcessAlive(pid: number): boolean {
  try {
    process.kill(pid, 0);
    return true;
  } catch (e) {
    return (e as NodeJS.ErrnoException).code === "EPERM";
  }
}

export interface StopResult {
  ok: boolean;
  message?: string;
  run?: Run | null;
}

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(),
      owner_pid: process.pid,
    });
    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(),
      child: null,
      cancelled: false,
    };
    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);
  }

  // Stop a run. If it is active in this process, flag it cancelled and
  // SIGTERM the current subprocess (SIGKILL after 5s if it ignores SIGTERM);
  // the execute loop observes `cancelled` and finishes as 'cancelled'.
  // If the run is not active here but the DB still says running/queued, it
  // belongs to a dead server process (HMR/restart): SIGTERM the recorded pid
  // if still alive, otherwise just repair the stale DB row.
  stopRun(runId: number): StopResult {
    const state = this.active.get(runId);
    if (state) {
      state.cancelled = true;
      const child = state.child;
      if (child && child.pid && child.exitCode === null) {
        this.emit(state, "[sling-ui] 收到停止请求，正在终止进程…");
        try {
          child.kill("SIGTERM");
        } catch {
          // already exiting
        }
        setTimeout(() => {
          if (state.child === child && child.exitCode === null) {
            try {
              child.kill("SIGKILL");
            } catch {
              // already gone
            }
          }
        }, 5000).unref();
      }
      return { ok: true, run: getRun(runId) };
    }

    const run = getRun(runId);
    if (!run) return { ok: false, message: "run not found" };
    if (run.status !== "running" && run.status !== "queued") {
      return { ok: false, message: `run ${runId} already ${run.status}`, run };
    }
    const now = new Date().toISOString();
    if (run.pid != null && isProcessAlive(run.pid)) {
      try {
        process.kill(run.pid, "SIGTERM");
      } catch {
        // died between the check and the kill
      }
      updateRunStatus(runId, {
        status: "cancelled",
        error: "用户手动停止",
        finished_at: now,
      });
    } else {
      updateRunStatus(runId, {
        status: "failed",
        error: "进程已退出（服务重启或异常）",
        finished_at: now,
      });
    }
    return { ok: true, run: getRun(runId) };
  }

  private finish(
    state: ActiveRun,
    status: "success" | "failed" | "cancelled",
    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) {
    const finishCancelled = () => {
      this.emit(state, "[sling-ui] 运行已被用户停止");
      this.finish(state, "cancelled", "用户手动停止");
    };
    try {
      if (state.cancelled) return finishCancelled();
      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 (state.cancelled) return finishCancelled();
        if (code !== 0) {
          throw new Error(
            `atlas schema apply exited with code ${code}\n${state.tail.join("\n")}`
          );
        }
      }

      if (state.cancelled) return finishCancelled();
      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 (state.cancelled) return finishCancelled();
      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) {
      if (state.cancelled) return finishCancelled();
      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 },
      });
      state.child = child;
      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) => {
        if (state.child === child) state.child = null;
        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());

// Reconcile runs left in running/queued by a previous server process.
// Each run records owner_pid (the Node process that started it). A run whose
// owner is still alive belongs to a live worker of the current server — skip
// it, otherwise a newly loaded worker would kill healthy runs mid-flight.
// Only runs with a dead (or legacy null) owner are reconciled: dead child pid
// -> failed; live child pid -> orphaned sling/atlas subprocess, SIGTERM it
// and mark cancelled.
function reconcileOrphanRuns() {
  const stale = db
    .prepare("SELECT * FROM runs WHERE status IN ('running', 'queued')")
    .all() as Run[];
  const now = new Date().toISOString();
  for (const run of stale) {
    if (run.owner_pid != null && isProcessAlive(run.owner_pid)) {
      continue;
    }
    const alive = run.pid != null && isProcessAlive(run.pid);
    if (alive) {
      try {
        process.kill(run.pid!, "SIGTERM");
      } catch {
        // died between the check and the kill
      }
      updateRunStatus(run.id, {
        status: "cancelled",
        error: "服务重启，运行已终止",
        finished_at: now,
      });
    } else {
      updateRunStatus(run.id, {
        status: "failed",
        error: "服务重启或进程异常退出",
        finished_at: now,
      });
    }
    if (run.log_path && fs.existsSync(run.log_path)) {
      try {
        fs.appendFileSync(run.log_path, "[sling-ui] 服务重启，运行终止\n");
      } catch {
        // log append is best-effort
      }
    }
    console.log(
      `[sling-ui] reconciled orphan run #${run.id} -> ${alive ? "cancelled" : "failed"} (pid ${run.pid ?? "none"})`
    );
  }
}

const gReconcile = globalThis as unknown as { __runsReconciled?: boolean };
if (!gReconcile.__runsReconciled) {
  gReconcile.__runsReconciled = true;
  try {
    reconcileOrphanRuns();
  } catch (e) {
    console.error("[sling-ui] orphan run reconcile failed", e);
  }
}