/**
 * Route host subprocess entry point.
 *
 * A dedicated OS process that runs user-defined `/x/*` route handlers off the
 * daemon, receiving invocations from the daemon over a Unix domain socket at
 * `$VELLUM_WORKSPACE_DIR/procs/routes/routes.sock`. Because it's a separate
 * process, a handler that blocks synchronously pins only this process (the
 * daemon stays responsive), and a wedged handler can be reclaimed with a hard
 * `kill` — the guarantee a worker thread could not give on Bun.
 *
 * Lifecycle mirrors the resource-monitor worker (`monitoring/worker.ts`): bind
 * the socket, then write the PID file as the readiness signal, arm the
 * PID-file guard so a superseded instance self-exits, and clean up on exit.
 *
 * Handlers run with **no injected context**. Reaching daemon state (publishing
 * events, running conversation turns) is deferred to the plugin-api once it is
 * safe out-of-process; today the host serves pure request→response handlers
 * (CRUD, file persistence, external API proxying).
 */

import { existsSync, unlinkSync, writeFileSync } from "node:fs";
import { createServer, type Server, type Socket } from "node:net";

import type { IpcEnvelope } from "@vellumai/ipc-server-utils";
import { IpcFrameReader, writeMessage } from "@vellumai/ipc-server-utils";

import { runInPluginContext } from "../plugins/plugin-execution-context.js";
import { disableStreamSeqStamping } from "../runtime/assistant-stream-state.js";
import {
  evictRouteSourceTree,
  importRouteModule,
  sourceRootForHandler,
} from "../runtime/routes/user-route-import.js";
import { getLogger } from "../util/logger.js";
import {
  ensureProcDir,
  getProcPidPath,
  getProcSocketPath,
} from "../util/platform.js";
import {
  cleanupWorkerPidFile,
  startWorkerPidFileGuard,
} from "../util/worker-process.js";
import {
  ROUTE_HOST_PROC_NAME,
  ROUTE_INVOKE_METHOD,
  type RouteInvokeParams,
} from "./route-host-protocol.js";

const log = getLogger("route-host");

/** Last entry-file mtime imported in this process, keyed by handler path. */
const lastHandlerMtime = new Map<string, number>();

const socketPath = getProcSocketPath(ROUTE_HOST_PROC_NAME);
const pidPath = getProcPidPath(ROUTE_HOST_PROC_NAME);

// ---------------------------------------------------------------------------
// Invocation handling
// ---------------------------------------------------------------------------

/** Normalize framing's `Uint8Array` body into a `BodyInit`-safe `ArrayBuffer`. */
function toArrayBuffer(bytes: Uint8Array): ArrayBuffer {
  if (bytes.byteOffset === 0 && bytes.byteLength === bytes.buffer.byteLength) {
    return bytes.buffer as ArrayBuffer;
  }
  return bytes.slice().buffer as ArrayBuffer;
}

function reconstructRequest(
  params: RouteInvokeParams,
  body: Uint8Array | undefined,
): Request {
  const headers = new Headers();
  for (const [name, value] of params.headers) {
    headers.append(name, value);
  }
  const init: RequestInit = { method: params.method, headers };
  if (body && params.method !== "GET" && params.method !== "HEAD") {
    init.body = toArrayBuffer(body);
  }
  return new Request(params.url, init);
}

/** Send the handler's response back over the socket (body as a binary frame). */
function replyResult(
  socket: Socket,
  id: string,
  status: number,
  headers: [string, string][],
  body: Uint8Array | null,
): void {
  const envelope: IpcEnvelope = { id, result: { status, headers } };
  if (body && body.byteLength > 0) {
    envelope.headers = { "content-length": String(body.byteLength) };
    writeMessage(socket, envelope, body);
  } else {
    writeMessage(socket, envelope);
  }
}

function replyError(socket: Socket, id: string, message: string): void {
  writeMessage(socket, { id, error: message });
}

const HTTP_METHODS = [
  "GET",
  "POST",
  "PUT",
  "PATCH",
  "DELETE",
  "HEAD",
  "OPTIONS",
] as const;

async function handleInvoke(
  socket: Socket,
  id: string,
  params: RouteInvokeParams,
  body: Uint8Array | undefined,
): Promise<void> {
  // Each worker has its own module cache. When the entry file's mtime
  // moves, evict the whole source tree so a new handler cannot bind to a
  // stale helper, then re-import.
  const lastMtime = lastHandlerMtime.get(params.filePath);
  if (lastMtime !== params.mtimeMs) {
    evictRouteSourceTree(sourceRootForHandler(params.filePath));
    lastHandlerMtime.set(params.filePath, params.mtimeMs);
  }

  // A plugin's own routes execute as that plugin, matching the daemon's
  // in-thread path, so plugin-scoped host APIs a handler reaches
  // (`resolveCredential`, `indexDocument`) scope to the owning plugin rather
  // than falling through to their unscoped branch. The context covers the
  // import as well as the call: a route module can reach a scoped API at
  // evaluation time, and it is imported once per mtime.
  const serve = async (): Promise<
    { ok: true; response: Response } | { ok: false; allowed: string[] }
  > => {
    const mod = await importRouteModule(params.filePath);
    const handler = mod[params.method];
    if (typeof handler !== "function") {
      return {
        ok: false,
        allowed: HTTP_METHODS.filter((m) => typeof mod[m] === "function"),
      };
    }
    const request = reconstructRequest(params, body);
    const response = (await (handler as (req: Request) => unknown)(
      request,
    )) as Response;
    return { ok: true, response };
  };
  const outcome = await (params.pluginName
    ? runInPluginContext(params.pluginName, serve)
    : serve());

  if (!outcome.ok) {
    replyResult(
      socket,
      id,
      405,
      outcome.allowed.length ? [["allow", outcome.allowed.join(", ")]] : [],
      null,
    );
    return;
  }

  const { response } = outcome;
  const buffer = new Uint8Array(await response.arrayBuffer());
  const headers: [string, string][] = [];
  response.headers.forEach((value, name) => {
    headers.push([name, value]);
  });
  replyResult(
    socket,
    id,
    response.status,
    headers,
    buffer.byteLength > 0 ? buffer : null,
  );
}

function onConnection(socket: Socket): void {
  const reader = new IpcFrameReader(
    (envelope, binary) => {
      if (envelope.method !== ROUTE_INVOKE_METHOD || !envelope.id) {
        return;
      }
      const id = envelope.id;
      const params = envelope.params as unknown as RouteInvokeParams;
      handleInvoke(socket, id, params, binary).catch((err: unknown) => {
        const message = err instanceof Error ? err.message : String(err);
        log.error({ err, id }, "Route handler invocation failed");
        replyError(socket, id, message);
      });
    },
    (err) => log.warn({ err }, "Route host framing error"),
  );

  socket.on("data", (chunk: Buffer) => reader.push(chunk));
  socket.on("error", (err) => log.warn({ err }, "Route host socket error"));
}

// ---------------------------------------------------------------------------
// Lifecycle
// ---------------------------------------------------------------------------

let activeServer: Server | undefined;
let disposePidGuard: (() => void) | undefined;
let shuttingDown = false;

function shutdown(reason: string): void {
  if (shuttingDown) {
    return;
  }
  shuttingDown = true;
  log.info({ reason }, "Route host shutting down");
  disposePidGuard?.();
  activeServer?.close();
  cleanupWorkerPidFile(pidPath);
  process.exit(0);
}

function start(): void {
  // Only the daemon stamps SSE seqs and writes the shared reservation file; a
  // worker that stamped would issue overlapping seqs and race the daemon.
  disableStreamSeqStamping();

  ensureProcDir(ROUTE_HOST_PROC_NAME);

  // Clear a stale socket from a crashed predecessor (double-spawn is already
  // prevented by the PID-file check in spawnWorkerProcess).
  if (existsSync(socketPath)) {
    try {
      unlinkSync(socketPath);
    } catch {
      // best-effort — bind will surface a real conflict
    }
  }

  const server = createServer();
  activeServer = server;
  server.on("error", (err) => {
    log.error({ err }, "Route host server error — exiting");
    process.exit(1);
  });

  process.on("SIGTERM", () => shutdown("SIGTERM"));
  process.on("SIGINT", () => shutdown("SIGINT"));

  server.listen(socketPath, () => {
    // Publish the PID (readiness signal), then arm the identity guard BEFORE we
    // begin serving. The guard's on-arm check runs synchronously, so a worker
    // superseded during startup begins shutting down here instead of serving.
    writeFileSync(pidPath, String(process.pid), { mode: 0o600 });
    disposePidGuard = startWorkerPidFileGuard(pidPath, {
      onEvicted: (reason) => shutdown(`pid-file evicted: ${reason}`),
    });
    if (shuttingDown) {
      return;
    }
    // Work-start: begin accepting route invocations only once the guard is armed.
    server.on("connection", onConnection);
    log.info({ socketPath, pid: process.pid }, "Route host ready");
  });
}

start();