import { spawn } from "node:child_process";
import { openSync, readFileSync, statSync, openSync as open, closeSync, readSync } from "node:fs";
import { Effect } from "effect";
import type { HandleReference, InstanceRecord, LaunchPlan } from "../contracts";
import { realProcessRunner } from "../../../core/command";
import { LOG_TAIL_BYTES, spawnFailed, type Launcher } from "./launcher";

/**
 * Detached-daemon launcher, exo-style: stdout+stderr straight into a log file, the whole
 * process *group* signalled on stop (vLLM forks an EngineCore that holds the VRAM — the
 * leader dying does not free the GPU), and ownership proven before any signal is sent.
 */

const STOP_POLL_MS = 250;

const pidAlive = (pid: number): boolean => {
  if (!Number.isInteger(pid) || pid <= 0) return false;
  try {
    process.kill(pid, 0);
    return true;
  } catch {
    return false;
  }
};

const signal = (pid: number, sig: NodeJS.Signals): void => {
  try {
    process.kill(pid, sig);
  } catch {
    /* already gone */
  }
};

/**
 * Linux: /proc/<pid>/stat field 22 is the process start time in clock ticks — a pid plus
 * its start token survives pid reuse across reboots. Elsewhere there is no equivalent
 * cheap token; ownership falls back to the command-line check alone.
 */
export const processStartToken = (pid: number): string | null => {
  try {
    const stat = readFileSync(`/proc/${pid}/stat`, "utf8");
    // Field 2 is the comm, which may contain spaces/parens; everything after the last
    // ')' is fixed-position.
    const afterComm = stat.slice(stat.lastIndexOf(")") + 2).split(" ");
    return afterComm[19] ?? null;
  } catch {
    return null;
  }
};

const pidCommandLine = (pid: number): string => {
  const result = realProcessRunner.runSync("ps", ["-o", "command=", "-p", String(pid)], {
    timeoutMs: 3_000,
  });
  return result.status === 0 ? result.stdout : "";
};

const readTailBytes = (path: string, bytes: number): string => {
  try {
    const size = statSync(path).size;
    const start = Math.max(0, size - bytes);
    const length = size - start;
    if (length <= 0) return "";
    const buffer = Buffer.alloc(length);
    const fd = open(path, "r");
    try {
      readSync(fd, buffer, 0, length, start);
    } finally {
      closeSync(fd);
    }
    return buffer.toString("utf8");
  } catch {
    return "";
  }
};

export const makeProcessLauncher = (logPathFor: (name: string) => string): Launcher => ({
  start: (plan: LaunchPlan, record: InstanceRecord) =>
    Effect.gen(function* () {
      const [binary, ...args] = plan.argv;
      if (!binary) return yield* spawnFailed("plan.argv is empty");
      const logFd = yield* Effect.try({
        try: () => openSync(logPathFor(record.name), "w"),
        catch: (error) => error,
      }).pipe(
        Effect.catch((error) =>
          spawnFailed(`cannot open log file for ${record.name}: ${String(error)}`),
        ),
      );
      const child = spawn(binary, args, {
        detached: true,
        stdio: ["ignore", logFd, logFd],
        env: { ...process.env, ...plan.env },
        ...(plan.workdir ? { cwd: plan.workdir } : {}),
      });
      const pid = yield* Effect.callback<number, never>((resume) => {
        child.on("error", () => resume(Effect.succeed(-1)));
        child.on("spawn", () => resume(Effect.succeed(child.pid ?? -1)));
      });
      closeSync(logFd);
      if (pid <= 0) return yield* spawnFailed(`failed to spawn ${binary}`);
      child.unref();
      return { kind: "process", pid, startToken: processStartToken(pid) } as const;
    }),

  alive: (reference: HandleReference) =>
    Effect.sync(() => (reference.kind === "process" ? pidAlive(reference.pid) : false)),

  owns: (reference: HandleReference, record: InstanceRecord) =>
    Effect.sync(() => {
      if (reference.kind !== "process") return false;
      if (!pidAlive(reference.pid)) return false;
      // Start token is decisive where the OS provides one (Linux).
      if (reference.startToken !== null) return processStartToken(reference.pid) === reference.startToken;
      // Elsewhere the pid's command line must still carry our unmistakable argument:
      // every plan passes `--port <port>`, and the port is unique per node. A recycled
      // pid belonging to something else will not be serving on our port.
      return pidCommandLine(reference.pid).includes(`--port ${record.port}`);
    }),

  stop: (reference: HandleReference, graceMs: number) =>
    Effect.gen(function* () {
      if (reference.kind !== "process") return;
      const { pid } = reference;
      // The group first: -pid reaches children even when the leader is already gone.
      signal(-pid, "SIGTERM");
      signal(pid, "SIGTERM");
      const deadline = Date.now() + graceMs;
      while (pidAlive(pid) && Date.now() < deadline) {
        yield* Effect.sleep(STOP_POLL_MS);
      }
      signal(-pid, "SIGKILL");
      signal(pid, "SIGKILL");
    }),

  logTail: (reference: HandleReference, record: InstanceRecord) =>
    Effect.sync(() => readTailBytes(logPathFor(record.name), LOG_TAIL_BYTES)),
});