/** Mounted I/O belongs to a PROCESS, not a worker thread: threads share
 * libuv's pool. A stuck FUSE stat must not starve parent DNS or /tmp queues. */
import { spawn, type ChildProcess } from "node:child_process";
import { EventEmitter } from "node:events";
import { mkdir, readdir, readFile, stat, writeFile } from "node:fs/promises";
import { isAbsolute, join, relative, resolve } from "node:path";
import { fileURLToPath } from "node:url";

export type WorkspaceWorkerStatus = {
  state: "starting" | "ready" | "stalled" | "backoff" | "stopped";
  pid: number | null;
  lastProgressAt: number | null;
  restarts: number;
  error: string | null;
};
export type WorkspaceFile = { absPath: string; size: number; mtimeMs: number };
type Request = { id: number; method: string; args: unknown[] };
type Reply = {
  id?: number;
  value?: unknown;
  error?: { message: string; code?: string };
  event?: string;
  path?: string;
};
const IO_TIMEOUT_MS = 8_000;
const SCAN_TIMEOUT_MS = 60_000;
const MAX_PENDING = 32;

export function shouldIgnore(relPath: string): boolean {
  const parts = relPath.split("/").filter(Boolean);
  const blockedDirs = new Set([
    ".chrome-profile",
    ".dench",
    ".git",
    ".next",
    ".vercel",
    "build",
    "coverage",
    "dist",
    "node_modules",
  ]);
  if (parts.some((part) => blockedDirs.has(part))) return true;
  const basename = parts.at(-1)?.toLowerCase() ?? "";
  if (basename === ".ds_store" || basename === "thumbs.db") return true;
  if (basename.startsWith("._")) return true;
  if (
    basename === ".env" ||
    (basename.startsWith(".env.") &&
      ![".env.example", ".env.sample", ".env.template"].includes(basename))
  ) {
    return true;
  }
  if (basename.endsWith(".pem") || basename.endsWith(".key")) return true;
  return false;
}

/** Reject incomplete snapshots. Treating EIO/timeout as an empty subtree can
 * propagate false deletes to Convex. Symlinks are never traversed. */
export async function scanWorkspace(
  workspace: string,
): Promise<WorkspaceFile[]> {
  const result: WorkspaceFile[] = [];
  const queue = [workspace];
  for (let index = 0; index < queue.length; index++) {
    const dir = queue[index];
    const entries = await readdir(dir, { withFileTypes: true });
    for (const entry of entries) {
      const absPath = join(dir, entry.name);
      if (shouldIgnore(relative(workspace, absPath))) continue;
      if (entry.isDirectory()) queue.push(absPath);
      else if (entry.isFile()) {
        // Even ENOENT invalidates this snapshot: retry the whole scan before
        // making any deletion decisions from a changing tree.
        const info = await stat(absPath);
        result.push({ absPath, size: info.size, mtimeMs: info.mtimeMs });
      }
    }
  }
  return result;
}

export class WorkspaceIo extends EventEmitter {
  private child: ChildProcess | null = null;
  private sequence = 0;
  private closed = false;
  private ready = false;
  private restartTimes: number[] = [];
  private restartTimer?: ReturnType<typeof setTimeout>;
  private watching: boolean | null = null;
  private pending = new Map<
    number,
    {
      request: Request;
      resolve: (value: unknown) => void;
      reject: (error: Error) => void;
      timer: ReturnType<typeof setTimeout>;
      callerTimer?: ReturnType<typeof setTimeout>;
      sent: boolean;
    }
  >();
  private health: WorkspaceWorkerStatus = {
    state: "starting",
    pid: null,
    lastProgressAt: null,
    restarts: 0,
    error: null,
  };

  constructor(
    readonly workspace: string,
    private readonly options: {
      timeoutMs?: number;
      restartDelayMs?: number;
      spawnWorker?: () => ChildProcess;
    } = {},
  ) {
    super();
    this.launch();
  }

  snapshot(): WorkspaceWorkerStatus {
    return { ...this.health };
  }
  private changed() {
    this.emit("status", this.snapshot());
  }
  private launch() {
    if (this.closed || this.child) return;
    this.ready = false;
    this.health.state = "starting";
    this.health.lastProgressAt = null;
    const child =
      this.options.spawnWorker?.() ??
      spawn(
        process.execPath,
        [
          fileURLToPath(new URL("./fs-workspace-worker.mjs", import.meta.url)),
          resolve(this.workspace),
        ],
        {
          cwd: "/tmp",
          stdio: ["ignore", "inherit", "inherit", "ipc"],
          serialization: "advanced",
        },
      );
    this.child = child;
    this.health.pid = child.pid ?? null;
    this.changed();
    const bootTimer = setTimeout(
      () => this.stall("workspace_io_timeout: startup"),
      this.options.timeoutMs ?? IO_TIMEOUT_MS,
    );
    child.on("message", (message: Reply) => {
      if (this.child !== child || this.health.state === "stalled") return;
      if (message.event === "ready") {
        clearTimeout(bootTimer);
        this.ready = true;
        this.health.state = "ready";
        this.health.error = null;
        this.changed();
        for (const item of this.pending.values()) this.send(item.request);
        if (this.watching !== null)
          void this.call("watch", [this.watching]).catch(() => undefined);
        return;
      }
      if (message.event && message.path) {
        this.emit(message.event, message.path);
        return;
      }
      if (message.event === "watch_error") {
        this.stall("workspace_io_failed: watcher");
        return;
      }
      const item =
        message.id === undefined ? undefined : this.pending.get(message.id);
      if (!item) return;
      this.pending.delete(message.id!);
      clearTimeout(item.timer);
      clearTimeout(item.callerTimer);
      if (message.error)
        item.reject(
          Object.assign(new Error(message.error.message), {
            code: message.error.code,
          }),
        );
      else {
        this.health.lastProgressAt = Date.now();
        this.changed();
        item.resolve(message.value);
      }
    });
    const exited = () => {
      clearTimeout(bootTimer);
      if (this.child !== child) return;
      this.child = null;
      this.ready = false;
      this.health.pid = null;
      this.rejectPending(new Error("workspace_io_unavailable: worker exited"));
      if (this.closed) return;
      this.health.state = "backoff";
      const now = Date.now();
      this.restartTimes = this.restartTimes.filter((at) => now - at < 60_000);
      // At most two replacements per minute, and ONLY after observed exit.
      // SIGKILL is not proof of exit for an uninterruptible D-state process.
      const delay =
        this.restartTimes.length >= 2
          ? Math.max(1, 60_000 - (now - this.restartTimes[0]))
          : (this.options.restartDelayMs ?? 1_000);
      this.changed();
      this.restartTimer = setTimeout(() => {
        if (this.closed) return;
        this.restartTimes.push(Date.now());
        this.health.restarts++;
        this.launch();
      }, delay);
    };
    child.once("exit", exited);
    child.once("error", () => {
      if (this.child !== child) return;
      if (!child.pid) {
        this.health.error = "workspace_io_failed: spawn";
        exited();
      } else {
        // kill/send errors are NOT an exit notification. Retain ownership
        // until the OS confirms exit, including EPERM and D-state failures.
        this.stall("workspace_io_failed: child process");
      }
    });
  }
  private send(request: Request) {
    const child = this.child;
    const item = this.pending.get(request.id);
    if (!child || !item || item.sent) return;
    item.sent = true;
    child.send(request, (error) => {
      if (error && this.child === child) this.stall("workspace_io_failed: ipc");
    });
  }
  private rejectPending(error: Error) {
    for (const item of this.pending.values()) {
      clearTimeout(item.timer);
      clearTimeout(item.callerTimer);
      item.reject(error);
    }
    this.pending.clear();
  }
  private stall(message: string) {
    if (this.closed || this.health.state === "stalled") return;
    this.health.state = "stalled";
    this.health.error = message;
    this.rejectPending(new Error(message));
    this.changed();
    try {
      this.child?.kill("SIGKILL");
    } catch {
      /* Retain the unexited child. */
    }
    // No restart here! A stuck kernel syscall may ignore SIGKILL indefinitely.
  }
  call<T>(
    method: string,
    args: unknown[] = [],
    timeoutMs?: number,
    stallOnTimeout = timeoutMs === undefined,
  ): Promise<T> {
    if (this.closed || !this.child || this.health.state === "stalled")
      return Promise.reject(
        new Error("workspace_io_unavailable: " + this.health.state),
      );
    if (this.pending.size >= MAX_PENDING)
      return Promise.reject(new Error("workspace_io_busy: queue full"));
    const id = ++this.sequence;
    const request = { id, method, args };
    const usedTimeout = timeoutMs ?? this.options.timeoutMs ?? IO_TIMEOUT_MS;
    return new Promise<T>((resolveValue, reject) => {
      // Caller expiry does not cancel a syscall already executing in the
      // child. Keep its admission slot and independent liveness deadline.
      const livenessTimeout = stallOnTimeout
        ? usedTimeout
        : (this.options.timeoutMs ?? IO_TIMEOUT_MS);
      const timer = setTimeout(() => {
        if (this.pending.has(id))
          this.stall("workspace_io_timeout: " + method);
      }, Math.max(1, livenessTimeout));
      const callerTimer = stallOnTimeout
        ? undefined
        : setTimeout(() => {
            const item = this.pending.get(id);
            if (!item) return;
            if (!item.sent) {
              // Expired startup work must never be sent after ready.
              this.pending.delete(id);
              clearTimeout(item.timer);
            }
            item.reject(new Error("workspace_io_timeout: " + method));
          }, Math.max(1, usedTimeout));
      this.pending.set(id, {
        request,
        resolve: (value) => resolveValue(value as T),
        reject,
        timer,
        callerTimer,
        sent: false,
      });
      if (this.ready) this.send(request);
    });
  }
  async watch(isFuse: boolean) {
    this.watching = isFuse;
    if (this.ready) await this.call("watch", [isFuse]);
  }
  readFile(path: string) {
    return this.call<Buffer>("readFile", [path]);
  }
  writeFile(
    path: string,
    bytes: Uint8Array,
    options?: { flag?: string },
    timeoutMs?: number,
  ) {
    return this.call<void>("writeFile", [path, bytes, options], timeoutMs);
  }
  stat(path: string) {
    return this.call<{ isFile: boolean; size: number; mtimeMs: number }>(
      "stat",
      [path],
    );
  }
  mkdir(path: string, timeoutMs?: number) {
    return this.call<void>("mkdir", [path], timeoutMs);
  }
  scan() {
    return this.call<WorkspaceFile[]>(
      "scan",
      [],
      this.options.timeoutMs ?? SCAN_TIMEOUT_MS,
      true,
    );
  }
  close() {
    this.closed = true;
    clearTimeout(this.restartTimer);
    this.health.state = "stopped";
    this.rejectPending(new Error("workspace_io_unavailable: stopped"));
    try {
      this.child?.kill("SIGKILL");
    } catch {
      /* No replacement on shutdown. */
    }
    this.changed();
  }
}

/** Entry point invoked ONLY by the child launcher. No Convex, DNS, control
 * queue, or health-file operations belong here. */
export async function runWorkspaceWorker(workspace: string) {
  let watcher: import("chokidar").FSWatcher | undefined;
  const root = resolve(workspace);
  const checked = (input: unknown) => {
    if (typeof input !== "string")
      throw new Error("workspace_io_invalid: path");
    const path = resolve(input);
    const rel = relative(root, path);
    if (rel === ".." || rel.startsWith("../") || isAbsolute(rel))
      throw new Error("workspace_io_invalid: outside root");
    return path;
  };
  process.on("disconnect", () => process.exit(0));
  process.on("message", async (request: Request) => {
    try {
      const [path, bytes, options] = request.args;
      let value: unknown;
      switch (request.method) {
        case "readFile":
          value = await readFile(checked(path));
          break;
        case "writeFile":
          await writeFile(
            checked(path),
            bytes as Uint8Array,
            options as { flag?: string },
          );
          break;
        case "stat": {
          const info = await stat(checked(path));
          value = {
            isFile: info.isFile(),
            size: info.size,
            mtimeMs: info.mtimeMs,
          };
          break;
        }
        case "mkdir":
          await mkdir(checked(path), { recursive: true });
          break;
        case "scan":
          value = await scanWorkspace(root);
          break;
        case "watch": {
          if (watcher) break;
          await mkdir(root, { recursive: true });
          const { watch } = await import("chokidar");
          const isFuse = Boolean(path);
          watcher = watch(root, {
            ignoreInitial: true,
            persistent: true,
            usePolling: isFuse,
            interval: isFuse ? 2_500 : undefined,
            binaryInterval: isFuse ? 5_000 : undefined,
            awaitWriteFinish: isFuse
              ? false
              : { stabilityThreshold: 100, pollInterval: 50 },
            ignored: (path) => {
              const rel = relative(root, path);
              return rel.startsWith("..") || shouldIgnore(rel);
            },
          });
          for (const event of ["add", "change", "unlink"] as const)
            watcher.on(event, (path) => process.send?.({ event, path }));
          watcher.on("error", () => process.send?.({ event: "watch_error" }));
          break;
        }
        default:
          throw new Error("workspace_io_invalid: method");
      }
      process.send?.({ id: request.id, value });
    } catch (error) {
      const code = (error as NodeJS.ErrnoException).code;
      process.send?.({
        id: request.id,
        error: {
          message:
            "workspace_io_failed: " +
            request.method +
            (code ? " (" + code + ")" : ""),
          code,
        },
      });
    }
  });
  process.send?.({ event: "ready" });
}