open-standards/joule-code/joule-code-rs/crates/joule-moq-web/e2e/lib.mjs

// What the end-to-end tests share: a static server for the pages, the relay on loopback
// from a test configuration, the host's composition (wai_live_web), and the browser.
//
// Every address is loopback, and every wait has a hard bound.

import http from "node:http";
import net from "node:net";
import fs from "node:fs";
import path from "node:path";
import os from "node:os";
import { spawn, execFileSync } from "node:child_process";
import crypto from "node:crypto";

export const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

export async function until(what, ms, f, every = 250) {
  const end = Date.now() + ms;
  let last;
  for (;;) {
    try {
      last = await f();
      if (last) return last;
    } catch (e) {
      last = e;
    }
    if (Date.now() > end) throw new Error(`${what}: not within ${ms} ms (last: ${last instanceof Error ? last.message : JSON.stringify(last)?.slice(0, 600)})`);
    await sleep(every);
  }
}

const MIME = { ".html": "text/html; charset=utf-8", ".js": "text/javascript; charset=utf-8", ".mjs": "text/javascript; charset=utf-8", ".wasm": "application/wasm", ".json": "application/json", ".css": "text/css" };

// A static server on 127.0.0.1: `root` at /, and each of `mounts` (a URL path to a file).
export function serve(root, mounts = {}) {
  const server = http.createServer((req, res) => {
    const u = new URL(req.url, "http://127.0.0.1");
    let file = mounts[u.pathname];
    if (!file) {
      const p = path.normalize(path.join(root, decodeURIComponent(u.pathname)));
      if (!p.startsWith(root)) { res.writeHead(403); return res.end(); }
      file = fs.existsSync(p) && fs.statSync(p).isDirectory() ? path.join(p, "index.html") : p;
    }
    fs.readFile(file, (err, b) => {
      if (err) { res.writeHead(404); return res.end("not found"); }
      res.writeHead(200, { "content-type": MIME[path.extname(file)] || "application/octet-stream", "cache-control": "no-store" });
      res.end(b);
    });
  });
  return new Promise((ok) => server.listen(0, "127.0.0.1", () => ok({ server, url: `http://127.0.0.1:${server.address().port}`, close: () => new Promise((r) => server.close(r)) })));
}

const strip = (s) => s.replace(/\x1b\[[0-9;]*m/g, "");

// The relay, from a configuration this writes, on loopback ports it reads back from its
// log: the UDP port of QUIC and WebTransport, the TCP port of the WebSocket binding, and
// its development certificate's SHA-256.
export async function startRelay(bin, cfg, dir, name) {
  const toml = path.join(dir, `${name}.toml`);
  fs.writeFileSync(toml, relayToml(cfg));
  const log = path.join(dir, `${name}.log`);
  const out = fs.openSync(log, "w");
  const proc = spawn(bin, ["--config", toml], { stdio: ["ignore", "pipe", "pipe"], env: { ...process.env, NO_COLOR: "1", JOULE_MOQ_RELAY_LOG: process.env.JOULE_MOQ_RELAY_LOG || "info" } });
  const lines = [];
  const take = (b) => {
    fs.writeSync(out, b);
    for (const l of strip(b.toString()).split("\n")) if (l) lines.push(l);
  };
  proc.stdout.on("data", take);
  proc.stderr.on("data", take);
  const exited = new Promise((r) => proc.on("exit", (code, signal) => r({ code, signal })));
  const r = await until(`relay ${name} listening`, 30000, () => {
    if (proc.exitCode != null) throw new Error(`relay ${name} exited: ${lines.slice(-5).join(" | ")}`);
    const q = lines.map((l) => l.match(/on udp 127\.0\.0\.1:(\d+)/)).find(Boolean);
    const w = lines.map((l) => l.match(/on tcp 127\.0\.0\.1:(\d+)/)).find(Boolean);
    const h = lines.map((l) => l.match(/certificate sha-256 \(hex\):\s+([0-9a-f]{64})/)).find(Boolean);
    if (cfg.no_quic) return w && lines.some((l) => l.includes("session summaries signed by")) && { quic: null, ws: Number(w[1]), hash: null };
    return q && w && h && { quic: Number(q[1]), ws: Number(w[1]), hash: h[1] };
  }, 100);
  const path_ = cfg.path || "/moq";
  return {
    ...r,
    name,
    proc,
    lines,
    log,
    exited,
    wt: `https://127.0.0.1:${r.quic}${path_}`,
    wsUrl: `ws://127.0.0.1:${r.ws}${path_}`,
    // `signal` null: wait for an exit already asked for.
    async stop(signal = "SIGINT", ms = 20000) {
      if (signal && proc.exitCode == null && proc.signalCode == null) proc.kill(signal);
      const e = await Promise.race([exited, sleep(ms).then(() => null)]);
      if (!e) { proc.kill("SIGKILL"); await exited; }
      if (!this.closedLog) { this.closedLog = true; fs.closeSync(out); }
      return e;
    },
  };
}

function relayToml(c) {
  const q = (s) => JSON.stringify(s);
  const arr = (a) => `[${a.map(q).join(", ")}]`;
  let t = `${c.no_quic ? "" : 'quic_listen = "127.0.0.1:0"\n'}websocket_listen = "127.0.0.1:0"\npath = ${q(c.path || "/moq")}\ngoaway_uri = ${q(c.goaway_uri || "")}\n\n`;
  t += `[tls]\ndev_names = ["localhost", "127.0.0.1"]\n\n[limits]\ngoaway_timeout_ms = ${c.goaway_timeout_ms || 8000}\n\n`;
  t += `[receipts]\nreceipts_track_suffix = "_receipts"\n\n`;
  for (const d of c.door || []) t += `[[receipts.door]]\nprefix = ${arr(d.prefix)}\ntrust = ${arr(d.trust || [])}\n${d.trust_field != null ? `trust_field = ${d.trust_field}\n` : ""}hold_ms = ${d.hold_ms}\n\n`;
  if (c.limits) for (const [k, v] of Object.entries(c.limits)) t = t.replace("[limits]\n", `[limits]\n${k} = ${v}\n`);
  if (c.cache_bytes) t += `[cache]\nmax_bytes = ${c.cache_bytes}\n\n`;
  // A DVR window of every track (WAI live-recording §8).
  if (c.dvr) t += `[dvr]\npath = ${q(c.dvr.path)}\nroom_keys = ${arr(c.dvr.room_keys || [])}\n\n[[dvr.window]]\nprefix = ["*"]\nmax_age_s = ${c.dvr.max_age_s}\nmax_bytes = ${c.dvr.max_bytes}\n\n`;
  t += `[summary]\nout_path = ${q(c.summary_out)}\n\n[auth]\nrequired = ${c.auth ? "true" : "false"}\n\n`;
  // Lines a room service prints for the relay that trusts it (`joule-moq-room relay-lines`).
  if (c.extra) t += c.extra + "\n";
  // A grant that may subscribe may also seek back below the live edge (live-rooms §10.6),
  // as a late joiner does, unless the test says otherwise.
  for (const g of c.grants || []) t += `[[auth.grants]]\nprefix = ${arr(g.prefix)}\ntoken_sha256 = ${q(g.token_sha256)}\npublish = ${g.publish}\nsubscribe = ${g.subscribe}\nproof = ${g.proof === true}\nseek = ${g.seek != null ? g.seek : g.subscribe}\n\n`;
  return t;
}

// The test room, from wai_live_web on the host.
export function makeFixture(waiBin, names, dir) {
  const seed = crypto.randomBytes(32).toString("hex");
  const t0 = BigInt(Date.now()) * 1000n;
  const out = execFileSync(waiBin, ["fixture", seed, t0.toString(), names.join(",")], { maxBuffer: 16 << 20 }).toString();
  const f = JSON.parse(out);
  const file = path.join(dir, "fixture.json");
  fs.writeFileSync(file, JSON.stringify(f, null, 1));
  return { fixture: f, file };
}

// The composition the host computes at `t` for the room as a page holds it.
export function hostFrame(waiBin, roomJson, compose, t, dir) {
  const file = path.join(dir, `room-${t}.json`);
  fs.writeFileSync(file, roomJson);
  return JSON.parse(execFileSync(waiBin, ["frame", file, compose, String(t)], { maxBuffer: 16 << 20 }).toString());
}

// A room verified on the host as co-performance §9 verifies it (`wai_live_web co-verify`), from
// the room as a page holds it and `{policies, objects, payloads}`.
export function hostCoVerify(waiBin, roomJson, coDoc, dir, name = "co") {
  const r = path.join(dir, `${name}-room.json`);
  const c = path.join(dir, `${name}-co.json`);
  fs.writeFileSync(r, roomJson);
  fs.writeFileSync(c, JSON.stringify(coDoc));
  return JSON.parse(execFileSync(waiBin, ["co-verify", r, c], { maxBuffer: 64 << 20 }).toString());
}

// The browser, headless, with a fake camera and microphone, through memguard where the host
// has it. Every process this run starts carries `--moq-web-run=<id>` so that strays can be
// found and ended afterwards.
export const RUN_ID = crypto.randomBytes(6).toString("hex");

export function browserArgs() {
  return [
    "--use-fake-device-for-media-stream",
    "--use-fake-ui-for-media-stream",
    "--autoplay-policy=no-user-gesture-required",
    "--disable-background-timer-throttling",
    "--disable-renderer-backgrounding",
    "--disable-backgrounding-occluded-windows",
    `--moq-web-run=${RUN_ID}`,
  ];
}

// A wrapper that runs the browser under memguard with its own bound (MOQ_WEB_BROWSER_GB,
// 8 by default), so that one runaway browser is killed, not the host.
export function memguardedBrowser(bin, dir) {
  const mg = process.env.RB_ROOT ? path.join(process.env.RB_ROOT, "bin", "memguard") : process.env.MEMGUARD || "";
  if (!mg || !fs.existsSync(mg)) return { executable: bin, guarded: false };
  const w = path.join(dir, "browser-memguard.sh");
  fs.writeFileSync(w, `#!/bin/bash\nexec ${JSON.stringify(mg)} --max-gb ${process.env.MOQ_WEB_BROWSER_GB || 8} -- ${JSON.stringify(bin)} "$@"\n`, { mode: 0o755 });
  return { executable: w, guarded: true };
}

export function killStrays() {
  // The pattern without its leading dashes: pkill would read them as an option.
  try { execFileSync("pkill", ["-f", `moq-web-run=${RUN_ID}`]); } catch { /* none left */ }
}

export function tmpdir(label) {
  return fs.mkdtempSync(path.join(process.env.TMPDIR || os.tmpdir(), `moq-web-${label}-`));
}

// A room service (joule-moq-room) for a relay: its keys made in `dir` (mode 0600), the
// relay's lines that trust them, then, once the relay listens, the service itself.
export function roomKeys(bin, dir) {
  const keys = path.join(dir, "room-keys.toml");
  const lines = execFileSync(bin, ["keygen", keys], { stdio: ["ignore", "pipe", "pipe"] }).toString();
  return { keys, lines, authority: lines.match(/^# authority ([0-9a-f]{64})/m)[1], clock: lines.match(/^# clock\s+([0-9a-f]{64})/m)[1] };
}

export async function startRoomService(bin, { keys, relayWs, room = {}, rate = {}, co = null, top = {}, extra = "" }, dir) {
  const port = await freePort();
  const toml = path.join(dir, "room.toml");
  let t = `listen = "127.0.0.1:${port}"\nkeys = ${JSON.stringify(keys)}\nrelay_url = ${JSON.stringify(relayWs)}\n`;
  // Top-level fields (`public_prefix`), each value already in TOML's form.
  for (const [k, v] of Object.entries(top)) t += `${k} = ${v}\n`;
  t += "\n[room]\n";
  for (const [k, v] of Object.entries(room)) t += `${k} = ${v}\n`;
  t += "\n[rate]\n";
  for (const [k, v] of Object.entries(rate)) t += `${k} = ${v}\n`;
  // Co-performance (WAI co-performance): off unless given.
  if (co) {
    t += "\n[co]\n";
    for (const [k, v] of Object.entries(co)) t += `${k} = ${v}\n`;
  }
  // Tables of their own after it: `[age]` and its issuers and ratings.
  t += extra ? `\n${extra}\n` : "";
  fs.writeFileSync(toml, t);
  const log = path.join(dir, "room-service.log");
  const out = fs.openSync(log, "w");
  const proc = spawn(bin, ["--config", toml], { stdio: ["ignore", out, out], env: { ...process.env, NO_COLOR: "1" } });
  const exited = new Promise((r) => proc.on("exit", (code, signal) => r({ code, signal })));
  const url = `http://127.0.0.1:${port}`;
  await until("the room service's clock on the relay", 30000, async () => {
    if (proc.exitCode != null) throw new Error(`the room service exited: ${fs.readFileSync(log, "utf8").slice(-400)}`);
    try { const r = await fetch(`${url}/health`); return r.ok && (await r.json()).relay === true; } catch { return false; }
  }, 200);
  return {
    url, proc, log, exited,
    lines: () => fs.readFileSync(log, "utf8").split("\n"),
    async stop() {
      if (proc.exitCode == null) proc.kill("SIGTERM");
      const e = await Promise.race([exited, sleep(10000).then(() => null)]);
      if (!e) { proc.kill("SIGKILL"); await exited; }
      fs.closeSync(out);
      return e;
    },
  };
}

function freePort() {
  return new Promise((ok) => { const s = net.createServer(); s.listen(0, "127.0.0.1", () => { const p = s.address().port; s.close(() => ok(p)); }); });
}

// The pages, the room service and the relay on one origin, as the production front serves
// them: `root` at /, `/room-api/*` to the service with the prefix stripped, and the
// WebSocket upgrade of `/moq` to the relay, its request (Host and all) passed as it came.
export function serveSameOrigin(root, { service, relayPort }) {
  const files = serve(root);
  return files.then(({ server: fileServer, close: closeFiles }) => {
    const server = http.createServer((req, res) => {
      if (req.url.startsWith("/room-api/")) {
        const u = new URL(service);
        const up = http.request({ host: u.hostname, port: u.port, method: req.method, path: req.url.slice("/room-api".length), headers: { ...req.headers, "cf-connecting-ip": "127.0.0.1" } }, (r) => {
          res.writeHead(r.statusCode, r.headers);
          r.pipe(res);
        });
        up.on("error", () => { res.writeHead(502); res.end(); });
        req.pipe(up);
        return;
      }
      fileServer.emit("request", req, res);
    });
    // Every socket, the upgraded ones and their relay ends too, is ended when the server
    // closes: an upgraded socket otherwise holds `close` open for as long as it lives.
    const sockets = new Set();
    const keep = (x) => { sockets.add(x); x.on("close", () => sockets.delete(x)); };
    server.on("connection", keep);
    server.on("upgrade", (req, socket, head) => {
      if (!req.url.startsWith("/moq")) return socket.destroy();
      const up = net.connect(relayPort, "127.0.0.1", () => {
        let raw = `${req.method} ${req.url} HTTP/1.1\r\n`;
        for (let i = 0; i < req.rawHeaders.length; i += 2) raw += `${req.rawHeaders[i]}: ${req.rawHeaders[i + 1]}\r\n`;
        up.write(raw + "\r\n");
        if (head && head.length) up.write(head);
        up.pipe(socket);
        socket.pipe(up);
      });
      keep(up);
      up.on("error", () => socket.destroy());
      socket.on("error", () => up.destroy());
      up.on("close", () => socket.destroy());
      socket.on("close", () => up.destroy());
    });
    return new Promise((ok) => server.listen(0, "127.0.0.1", () => ok({
      server, url: `http://127.0.0.1:${server.address().port}`,
      close: () => {
        for (const x of sockets) x.destroy();
        return Promise.race([Promise.all([new Promise((r) => server.close(r)), closeFiles()]), sleep(5000)]);
      },
    })));
  });
}