server.mjs

// Production HTTP entry.
//
// WHY THIS EXISTS. The built server bundle (`dist/server/server.js`) exports a
// Web Fetch handler and nothing else — no `listen`, no port. The only thing
// that bound a port was `vite preview`, which Vite documents as a way to look
// at a production build locally, not as a production server: it carries the
// dev toolchain into the runtime image and gives you no control over
// concurrency, timeouts or shutdown.
//
// The consequence that actually mattered was concurrency. Node runs JavaScript
// on one thread, so one `vite preview` process served roughly one core no
// matter how large the host was. A 64-core machine ran 1/64th of itself unless
// the operator hand-rolled many containers. This forks one worker per core by
// default, so a big machine is used by default rather than by accident.
//
// Everything here is deliberately boring: srvx (already a dependency, and what
// the framework itself uses) bridges Node HTTP to Fetch, `node:cluster` forks
// the workers, and the parent restarts a worker that dies.
import cluster from "node:cluster";
import { randomBytes } from "node:crypto";
import { readFileSync } from "node:fs";
import os from "node:os";
import { fileURLToPath } from "node:url";
import path from "node:path";
import { installServerGuards, quietClientGone, rememberClientAborts } from "./serverGuards.mjs";
import { compressStatic } from "./serverCompression.mjs";

// Before anything else, in the primary and in every worker: undici can throw
// from a microtask when it closes a response stream twice, which used to end
// the process (R325). See serverGuards.mjs.
installServerGuards();

const HERE = path.dirname(fileURLToPath(import.meta.url));
const PORT = Number(process.env.PORT || 8080);
const HOSTNAME = process.env.HOST || "0.0.0.0";

/**
 * The container's CPU quota in cores, or null when there is no quota to find
 * (bare metal, an unlimited cgroup, or a non-Linux host).
 *
 * WHY THIS IS NEEDED, MEASURED on Docker 28 / Node 22:
 *
 *   --cpus=4    availableParallelism() = 4    correct
 *   --cpus=2    availableParallelism() = 2    correct
 *   --cpus=1.5  availableParallelism() = 1    correct
 *   --cpus=0.5  availableParallelism() = 8    WRONG — the host's core count
 *
 * libuv reads the cgroup quota, but below one whole CPU it gives up and reports
 * the machine. That is the dangerous direction: `resources.limits.cpu: 500m` is
 * an ordinary Kubernetes request, and on a 64-core node it would fork 64
 * workers at ~0.5-1 GB RSS each — an OOMKill presented as a crash loop. Reading
 * the quota ourselves turns the one case libuv gets wrong into one worker.
 */
function cgroupCpuLimit() {
  // cgroup v2: "<quota> <period>", or "max <period>" when unlimited.
  try {
    const [quota, period] = readFileSync("/sys/fs/cgroup/cpu.max", "utf8").trim().split(/\s+/);
    if (quota === "max") return null;
    const cores = Number(quota) / Number(period || 100000);
    return Number.isFinite(cores) && cores > 0 ? cores : null;
  } catch {
    // Not cgroup v2, or not readable — try v1 below.
  }
  // cgroup v1 splits the same numbers across two files; quota -1 is unlimited.
  try {
    const quota = Number(readFileSync("/sys/fs/cgroup/cpu/cpu.cfs_quota_us", "utf8"));
    const period = Number(readFileSync("/sys/fs/cgroup/cpu/cpu.cfs_period_us", "utf8"));
    if (quota > 0 && period > 0) return quota / period;
  } catch {
    // No cgroup CPU controller — nothing is limiting us.
  }
  return null;
}

/**
 * Workers to fork. `WEB_CONCURRENCY` is the convention Heroku/Render/Fly and
 * most Node images already use; 0 or 1 means "run in this process", which is
 * what a container-per-core deployment wants.
 *
 * Otherwise: every core the process may actually use. On a bare host that is
 * the whole machine — the point of the exercise. In a container it is the CPU
 * quota, never more, so the default is safe to leave alone anywhere.
 */
function workerCount() {
  const raw = process.env.WEB_CONCURRENCY?.trim();
  if (raw) {
    const n = Number(raw);
    if (Number.isFinite(n) && n >= 0) return Math.trunc(n);
  }
  // An analytics node exists to run ONE big query engine, and the engine is
  // per process — forking would divide the memory limit it was sized for into
  // N independent copies, each able to claim the whole figure. One worker is
  // the only default that makes `LAKEHOUSE_MEMORY_LIMIT=64GB` mean what the
  // operator who typed it meant. WEB_CONCURRENCY above still overrides.
  // Kept in step with src/utils/appRole.ts by a test.
  if ((process.env.APP_ROLE ?? "").trim().toLowerCase() === "analytics") return 1;
  const reported = Math.max(1, os.availableParallelism?.() ?? os.cpus()?.length ?? 1);
  const quota = cgroupCpuLimit();
  // Floor, not round: half a core's worth of quota buys one worker, not two.
  return quota === null ? reported : Math.max(1, Math.min(reported, Math.floor(quota)));
}

/**
 * Content types the static handler gets wrong, by extension.
 *
 * FOUND FROM THE UI: the Data Catalog reported "Local tables: 0" on a workspace
 * with 33 datasets. The cause was two layers down — srvx's static handler
 * served `.wasm` as `application/octet-stream`, so
 * `WebAssembly.compileStreaming` refused the module ("Incorrect response MIME
 * type"), the browser DuckDB engine never started, and the catalog's loader —
 * which registers every dataset into that engine before returning their
 * metadata — threw and yielded an empty list. `vite preview` set this header,
 * so replacing it with this server is what introduced the bug.
 *
 * Only .wasm is wrong; js, css, woff2, svg and png were all checked and are
 * correct. It is a map rather than a special case so the next one is a line.
 */
const CONTENT_TYPE_FIXES = { ".wasm": "application/wasm" };

/** Wrap a static handler so those extensions get the type they need. */
function correctContentTypes(handler) {
  return async (request, next) => {
    const res = await handler(request, next);
    if (!res) return res;
    const ext = new URL(request.url).pathname.match(/\.[a-z0-9]+$/i)?.[0]?.toLowerCase();
    const want = ext && CONTENT_TYPE_FIXES[ext];
    if (!want || res.headers.get("content-type") === want) return res;
    const headers = new Headers(res.headers);
    headers.set("content-type", want);
    return new Response(res.body, { status: res.status, statusText: res.statusText, headers });
  };
}

async function startWorker() {
  const [{ serve }, { serveStatic }, mod] = await Promise.all([
    import("srvx"),
    import("srvx/static"),
    import("./dist/server/server.js"),
  ]);
  const app = mod.default ?? mod.server;
  if (typeof app?.fetch !== "function") {
    throw new Error(
      "dist/server/server.js did not export a fetch handler — run `npm run build` first.",
    );
  }

  // A browser that leaves mid-request is not a server fault, and is not
  // printed as one (R362). See serverGuards.mjs.
  quietClientGone();
  const server = serve({
    port: PORT,
    hostname: HOSTNAME,
    // Every worker binds the same port; the OS balances accepts between them.
    // Without this the second worker dies with EADDRINUSE.
    reusePort: true,
    // Client assets are hashed and immutable; the SSR handler should never see
    // a request for one.
    // Order matters: the type fix runs first (innermost) so the compressor
    // sees the corrected Content-Type when deciding what is compressible.
    middleware: [
      compressStatic(correctContentTypes(serveStatic({ dir: path.join(HERE, "dist", "client") }))),
    ],
    fetch: rememberClientAborts(app.fetch),
  });

  await server.ready?.();
  const who = cluster.isWorker ? `worker ${process.pid}` : `single process ${process.pid}`;
  console.log(`[agentswarms] ${who} listening on http://${HOSTNAME}:${PORT}`);

  // With an external key provider, prove the data key unwraps BEFORE this
  // process takes traffic. Serving without it would surface as hundreds of
  // per-feature credential errors; failing once here, naming the provider,
  // is the honest shape. The env provider has nothing to reach, so it is
  // not checked — a missing PROVIDER_CREDS_SECRET fails at first use as before.
  const kmsProvider = (process.env.KMS_PROVIDER ?? "env").trim().toLowerCase();
  if (kmsProvider && kmsProvider !== "env") {
    try {
      const res = await app.fetch(new Request(`http://${HOSTNAME}:${PORT}/api/health/ready`));
      const body = await res.json().catch(() => ({}));
      if (body?.checks?.keys === false) {
        console.error(
          `[agentswarms] refusing to start: the credential keyring did not load under KMS_PROVIDER=${kmsProvider} — ${body.error ?? "no detail"}`,
        );
        process.exit(1);
      }
      console.log(`[agentswarms] ${who} credential keyring loaded from ${kmsProvider}`);
    } catch (e) {
      console.error(`[agentswarms] refusing to start: keyring check failed — ${e?.message ?? e}`);
      process.exit(1);
    }
  }

  // FOUND FROM THE SURVEY (R95). Start the scheduler now, as this worker comes
  // up. It used to wait for the notification bell: ensureScheduler() is only
  // reached through /api/bi/cron, and the bell calls that once, when a
  // signed-in page mounts. So after every restart — the documented update,
  // `docker compose up -d --build`, included — no BI refresh, alert, scheduled
  // report, swarm schedule, ETL run, crawl, retention purge or kernel reap ran
  // until somebody opened the app. MEASURED: see ADVERSARIAL_LOG R95.
  //
  // In-process, like the keyring check above, and not awaited: the pass it
  // triggers can take seconds, and the SIGTERM handler below must be in place
  // for all of them. Skipped when the operator has handed scheduling to an
  // external cron, exactly as ensureScheduler() itself would skip.
  if (!/^(1|true|yes)$/i.test(process.env.DISABLE_INPROCESS_SCHEDULER ?? "")) {
    void (async () => {
      try {
        const res = await app.fetch(
          new Request(`http://${HOSTNAME}:${PORT}/api/bi/cron`, {
            method: "POST",
            headers: { Authorization: `Bearer ${process.env.AGENTSWARMS_BOOT_TOKEN}` },
          }),
        );
        const body = await res.json().catch(() => ({}));
        if (res.ok) {
          console.log(
            `[agentswarms] ${who} scheduler started; catch-up pass ${body.skipped ? "left to another worker" : "ran"}`,
          );
        } else {
          console.warn(
            `[agentswarms] ${who} could not start the scheduler at boot (HTTP ${res.status}${body.error ? `: ${body.error}` : ""}); scheduled work will wait for the first signed-in page load`,
          );
        }
      } catch (e) {
        console.warn(
          `[agentswarms] ${who} could not start the scheduler at boot: ${e?.message ?? e}; scheduled work will wait for the first signed-in page load`,
        );
      }
    })();
  }

  // SIGTERM is how a container runtime asks for a clean stop. Without a handler
  // Node exits immediately and in-flight requests are cut mid-response.
  let closing = false;
  for (const signal of ["SIGTERM", "SIGINT"]) {
    process.on(signal, () => {
      if (closing) return;
      closing = true;
      console.log(`[agentswarms] ${who} draining on ${signal}`);
      Promise.resolve(server.close?.()).finally(() => process.exit(0));
    });
  }
}

const workers = workerCount();

// Publish the count to the app, which needs it to size PER-PROCESS resources —
// the lakehouse engine lives in every worker, so its memory limit is charged N
// times over. Passing it through the environment (forked workers inherit it)
// keeps ONE source of truth: the app never re-derives the number and so can
// never disagree with how many processes actually exist. Unset under `vite dev`,
// where the answer is correctly 1.
process.env.AGENTSWARMS_WORKERS = String(workers < 1 ? 1 : workers);

// The credential each worker uses to start its own scheduler at boot (below).
// Random per boot, set before any fork so every worker inherits it, and known
// to nothing outside this process group: it buys one ordinary pass through
// /api/bi/cron, which is what a signed-in user's token already buys (R95).
process.env.AGENTSWARMS_BOOT_TOKEN = randomBytes(32).toString("hex");

if (workers > 1 && cluster.isPrimary) {
  console.log(
    `[agentswarms] primary ${process.pid} forking ${workers} workers ` +
      `(WEB_CONCURRENCY to override; 1 = no cluster)`,
  );
  for (let i = 0; i < workers; i++) cluster.fork();

  // A worker that dies takes its share of traffic with it, so replace it —
  // except during shutdown, where refilling the pool would stop the container
  // from ever exiting.
  let shuttingDown = false;
  cluster.on("exit", (worker, code, signal) => {
    if (shuttingDown) return;
    console.warn(
      `[agentswarms] worker ${worker.process.pid} exited (${signal || code}) — restarting`,
    );
    cluster.fork();
  });
  for (const signal of ["SIGTERM", "SIGINT"]) {
    process.on(signal, () => {
      shuttingDown = true;
      for (const w of Object.values(cluster.workers ?? {})) w?.kill(signal);
    });
  }
} else {
  await startWorker();
}