local.js

#!/usr/bin/env node
// Local HTTP front for beamline.js — same handler as the Cloudflare Worker.

import { createServer } from "node:http";
import { gzipSync } from "node:zlib";
import { handle } from "./beamline.js";
import { readToken } from "./tok.js";

// Below this, gzip costs more bytes than it saves.
const GZIP_MIN_BYTES = 512;

const PORT = Number(process.env.PORT) || 8080;
const MAX_BYTES = Number(process.env.MAX_BYTES) || 16 * 1024 * 1024;
// Bodies are buffered whole, so MAX_INFLIGHT * MAX_BYTES is this process's
// memory ceiling — 512MB at the defaults. Raise them together, or not at all.
const MAX_INFLIGHT = Number(process.env.MAX_INFLIGHT) || 32;

// Every knob beamline.js reads from the environment, so a local run is tunable
// the same way a deployed Worker is. Anything absent falls back to the built-in
// default.
//
// Not here, and deliberately: BEAMLINE_KV, TOKENS_KV and BEAMLINE_AE are Worker
// bindings with no local equivalent, and CF_ZONE_ID / CF_PURGE_TOKEN purge a
// zone cache this process does not sit behind — /v1/flush reports that as
// `unconfigured` rather than pretending.
const TUNABLES = [
  "SCAN_URL",
  "BEAMLINE_TOKEN",
  "SCAN_TOKEN",
  "MAX_BYTES",
  "SCAN_TIMEOUT_MS",
  "SCAN_RETRIES",
  "SCAN_RETRY_BASE_MS",
  "SCAN_STREAM_IDLE_MS",
  "SCAN_STREAM_STALL_MS",
  "SCAN_STREAM_RESUMES",
  "SCAN_ORPHAN_MS",
  "VERDICT_MAX_AGE",
  "KV_MAX_AGE",
  "PYPI_URL",
  "NPM_REGISTRY_URL",
];

const env = Object.fromEntries(TUNABLES.map((name) => [name, process.env[name] || ""]));

// The client token is deliberately environment-only: if BEAMLINE_TOKEN is not
// passed, the API is open. SCAN_TOKEN still falls back to ~/.tok because it is
// a backend credential, not a policy switch for callers.
env.SCAN_TOKEN ||= readToken("scan");

// Workaround, not design. On Node 26 fetch does not dispatch a second
// concurrent request to a backend until the event loop wakes, costing a hedged
// lookup up to 200ms of dead time on an otherwise idle process. A 1ms tick
// while work is in flight keeps the loop awake. Cloudflare's fetch is
// unaffected, so this stays out of beamline.js.
//
// Delete this once fetch dispatches promptly. To check: park a slow request on
// a backend, fire a second one 20ms later, and compare its round trip against
// the same request over node:net. If they match, the tick is dead weight.
let inflight = 0;
let nudge = null;

/**
 * @param {number} delta - +1 as a request starts, -1 as it ends
 * @returns {void}
 */
function busy(delta) {
  inflight += delta;
  if (inflight > 0 && !nudge) nudge = setInterval(() => {}, 1).unref();
  if (inflight === 0 && nudge) {
    clearInterval(nudge);
    nudge = null;
  }
}

const server = createServer(async (req, res) => {
  // Shed rather than queue: a request we cannot start is cheaper to refuse now
  // than to accept and starve.
  if (inflight >= MAX_INFLIGHT) {
    res.statusCode = 429;
    res.setHeader("content-type", "application/json");
    res.setHeader("retry-after", "1");
    res.end(JSON.stringify({ error: "at capacity" }));
    return;
  }
  busy(1);
  try {
    const url = new URL(req.url || "/", `http://${req.headers.host || "127.0.0.1"}`);
    const buf = await readRequest(req, MAX_BYTES);
    const headers = new Headers();
    for (const [name, value] of Object.entries(req.headers)) {
      if (value == null) continue;
      headers.set(name, Array.isArray(value) ? value.join(", ") : value);
    }
    const request = new Request(url, {
      method: req.method,
      headers,
      body: req.method === "GET" || req.method === "HEAD" ? undefined : buf,
    });
    // A client that hangs up should not leave scan working.
    const ac = new AbortController();
    res.on("close", () => {
      if (!res.writableFinished) ac.abort();
    });
    const ctx = {
      signal: ac.signal,
      waitUntil(p) {
        Promise.resolve(p).catch(() => {});
      },
    };
    const out = await handle(request, env, ctx);
    res.statusCode = out.status;
    out.headers.forEach((value, name) => res.setHeader(name, value));
    // Nothing sits in front of this process, so compression is ours to do.
    // Behind Cloudflare the edge does it and beamline.js stays out of the way.
    let body = Buffer.from(await out.arrayBuffer());
    const accepts = String(req.headers["accept-encoding"] || "")
      .toLowerCase()
      .split(",")
      .some((coding) => coding.trim().startsWith("gzip"));
    if (accepts && body.length >= GZIP_MIN_BYTES && !res.getHeader("content-encoding")) {
      body = gzipSync(body);
      res.setHeader("content-encoding", "gzip");
      res.appendHeader("vary", "accept-encoding");
    }
    res.setHeader("content-length", String(body.length));
    res.end(body);
  } catch (err) {
    const tooLarge = err && err.code === 413;
    res.statusCode = tooLarge ? 413 : 500;
    res.setHeader("content-type", "application/json");
    res.end(JSON.stringify({ error: tooLarge ? "too large" : "internal" }));
  } finally {
    busy(-1);
  }
});

/**
 * @param {import("node:http").IncomingMessage} req
 * @param {number} maxBytes
 * @returns {Promise<Buffer>} rejects with code 413 past the limit
 */
function readRequest(req, maxBytes) {
  return new Promise((resolve, reject) => {
    const chunks = [];
    let size = 0;
    req.on("data", (chunk) => {
      size += chunk.length;
      if (size > maxBytes) {
        req.destroy();
        const err = new Error("too large");
        err.code = 413;
        reject(err);
        return;
      }
      chunks.push(chunk);
    });
    req.on("end", () => resolve(Buffer.concat(chunks)));
    req.on("error", reject);
  });
}

server.listen(PORT, "127.0.0.1", () => {
  process.stderr.write(`beamline listening on http://127.0.0.1:${PORT}\n`);
});