#!/usr/bin/env node
// The stream audit kit end to end (wai/extensions/stream-audit.md): a real browser plays an HLS
// stream another origin serves, through a web player, with the kit's player integration; the kit
// sends the session to the collector, which verifies and stores it; and the collector's report is
// recomputed from the bundles by the collector's own `check`. Everything listens on loopback.
//
//   node e2e/audit.e2e.mjs --collector <wai-audit> --web <web/> --page <e2e/page/> --player <player.js>
//                          --out <dir> [--browser <executable>] [--runs N] [--hold-playlist-ms N]
//
// The stream is made here with the host's media tool (H.264 and AAC in CMAF segments of 2 s), and
// its playlist given a program time and a marked ad (an EXT-X-DATERANGE with SCTE35-OUT) over its
// third and fourth segments. Each run plays it twice, once audited through the player's loader
// class and once through the network tap, and checks what §4 to §12 say of the result.
//
// Before the first run the browser plays the stream once, untimed, through the player alone
// (page/warm.html). A browser starts parts of itself the first time a page plays media: its audio
// service, whose start the first presentation waits for, and on macOS each helper program's first
// run after an install or update, which the system checks first. That is the browser's start, paid
// once per launch, not the session's: under load it reached 11 s in the first session of a run and
// 0.5 s in the second (a host's system log showed the audio service starting for 10.6 s of the
// 11.2). So every timed session's start time, from the session's start to its first presentation,
// is the kit's, the player's and the stream's, and keeps its bound of 10 s. `--hold-playlist-ms N`
// plants a slow start: the stream's origin holds its playlist N ms in every timed session (never
// in the warm-up), and the start-time check must fail with N at 10 000 or more.

import { spawn, spawnSync } from "node:child_process";
import { createHash } from "node:crypto";
import { createRequire } from "node:module";
import fs from "node:fs";
import http from "node:http";
import path from "node:path";

const args = Object.fromEntries(process.argv.slice(2).reduce((a, x, i, all) => (x.startsWith("--") ? [...a, [x.slice(2), all[i + 1]]] : a), []));
const require = createRequire(path.join(process.env.PLAYWRIGHT_HOST || path.join(path.dirname(new URL(import.meta.url).pathname), "node_modules"), "x"));
const { chromium } = require("playwright");

const OUT = path.resolve(args.out);
const RUNS = Number(args.runs ?? 1);
const HOLD_PLAYLIST_MS = Number(args["hold-playlist-ms"] ?? 0);
const CHANNEL = "e2e-channel";
let failures = 0;
const check = (ok, what) => {
  console.log(`${ok ? "ok  " : "FAIL"} ${what}`);
  if (!ok) failures += 1;
};
const sha = (b) => createHash("sha256").update(b).digest("hex");

function makeStream(dir) {
  fs.rmSync(dir, { recursive: true, force: true });
  fs.mkdirSync(dir, { recursive: true });
  const r = spawnSync("ffmpeg", [
    "-hide_banner", "-loglevel", "error",
    "-f", "lavfi", "-i", "testsrc2=size=640x360:rate=30", "-f", "lavfi", "-i", "sine=frequency=440:sample_rate=48000",
    "-t", "12", "-c:v", "libx264", "-preset", "veryfast", "-g", "60", "-keyint_min", "60", "-sc_threshold", "0", "-pix_fmt", "yuv420p",
    "-c:a", "aac", "-b:a", "96k", "-f", "hls", "-hls_time", "2", "-hls_playlist_type", "vod", "-hls_segment_type", "fmp4",
    "-hls_fmp4_init_filename", "init.mp4", "-hls_segment_filename", path.join(dir, "seg%d.m4s"), path.join(dir, "index.m3u8"),
  ], { stdio: "inherit" });
  if (r.status !== 0) throw new Error("the media tool could not make the stream");
  // A program time on the first segment, and a marked ad over the third and fourth.
  const lines = fs.readFileSync(path.join(dir, "index.m3u8"), "utf8").split("\n").filter((l) => !l.startsWith("#EXT-X-PROGRAM-DATE-TIME"));
  const out = [];
  let seg = 0;
  for (const l of lines) {
    if (l.startsWith("#EXTINF")) {
      if (seg === 0) out.push("#EXT-X-PROGRAM-DATE-TIME:2026-10-25T01:00:00.000Z");
      if (seg === 2) out.push('#EXT-X-DATERANGE:ID="e2e-ad",START-DATE="2026-10-25T01:00:04.000Z",DURATION=4.000,SCTE35-OUT=0xFC302000');
      seg += 1;
    }
    out.push(l);
  }
  fs.writeFileSync(path.join(dir, "index.m3u8"), out.join("\n"));
  return Object.fromEntries(fs.readdirSync(dir).map((f) => [f, sha(fs.readFileSync(path.join(dir, f)))]));
}

function listen(handler) {
  return new Promise((resolve) => {
    const s = http.createServer(handler);
    s.listen(0, "127.0.0.1", () => resolve({ server: s, origin: `http://127.0.0.1:${s.address().port}` }));
  });
}

const TYPES = { ".m3u8": "application/vnd.apple.mpegurl", ".m4s": "video/iso.segment", ".mp4": "video/mp4", ".html": "text/html", ".js": "text/javascript" };

function serveFile(res, file, headers = {}) {
  if (!fs.existsSync(file) || !fs.statSync(file).isFile()) {
    res.writeHead(404).end();
    return null;
  }
  const body = fs.readFileSync(file);
  res.writeHead(200, { "content-type": TYPES[path.extname(file)] ?? "application/octet-stream", "content-length": body.length, ...headers }).end(body);
  return body;
}

async function freePort() {
  const { server } = await listen(() => {});
  const port = server.address().port;
  await new Promise((r) => server.close(r));
  return port;
}

async function main() {
  fs.mkdirSync(OUT, { recursive: true });
  const streamDir = path.join(OUT, "stream");
  const hashes = makeStream(streamDir);
  const segments = Object.keys(hashes).filter((f) => f.endsWith(".m4s"));
  console.log(`stream: ${Object.keys(hashes).length} files, ${segments.length} segments`);

  // The stream's origin: another origin than the page's, CORS open, CMSD-Static exposed on segments.
  const served = [];
  let warming = false;
  const stream = await listen((req, res) => {
    const name = decodeURIComponent(new URL(req.url, "http://x").pathname).replace(/^\/+/, "");
    const cors = { "access-control-allow-origin": "*", "access-control-expose-headers": "CMSD-Static, CMSD-Dynamic" };
    if (req.method === "OPTIONS") return res.writeHead(204, { ...cors, "access-control-allow-headers": "*" }).end();
    const extra = name.endsWith(".m4s") ? { "cmsd-static": "ot=v,sf=h,st=v" } : {};
    const answer = () => {
      if (serveFile(res, path.join(streamDir, path.basename(name)), { ...cors, ...extra })) served.push(name);
    };
    // The planted slow start (--hold-playlist-ms): in a timed session only.
    if (HOLD_PLAYLIST_MS > 0 && !warming && name.endsWith(".m3u8")) return void setTimeout(answer, HOLD_PLAYLIST_MS);
    answer();
  });
  const page = await listen((req, res) => {
    const p = new URL(req.url, "http://x").pathname;
    if (p === "/" || p === "/index.html") return serveFile(res, path.join(args.page, "index.html"));
    if (p === "/warm.html") return serveFile(res, path.join(args.page, "warm.html"));
    if (p === "/vendor/player.js") return serveFile(res, args.player);
    if (p.startsWith("/web/")) return serveFile(res, path.join(args.web, path.basename(p)));
    res.writeHead(404).end();
  });

  // The collector.
  const port = await freePort();
  const collectorUrl = `http://127.0.0.1:${port}/`;
  const data = path.join(OUT, "collector");
  fs.rmSync(data, { recursive: true, force: true });
  const env = { ...process.env, WAI_AUDIT_CHANNELS: CHANNEL, WAI_AUDIT_ADDR: `127.0.0.1:${port}`, WAI_AUDIT_DATA: data, WAI_AUDIT_KEY: path.join(data, "signing.key"), WAI_AUDIT_ALLOW_ORIGINS: page.origin };
  fs.mkdirSync(data, { recursive: true });
  const collector = spawn(args.collector, ["serve"], { env, stdio: ["ignore", "inherit", "inherit"] });
  // However the run ends, the collector ends with it: left running it would hold the output of
  // whatever started this script open.
  process.on("exit", () => collector.kill());
  let owner = null;
  for (let i = 0; i < 100 && !owner; i++) {
    await new Promise((r) => setTimeout(r, 100));
    owner = await fetch(`${collectorUrl}v1/key`).then((r) => r.text()).catch(() => null);
  }
  check(/^[0-9a-f]{64}$/.test(owner ?? ""), "the collector is up and states its key");

  const opts = { headless: true, args: ["--autoplay-policy=no-user-gesture-required"] };
  if (args.browser) opts.executablePath = path.resolve(args.browser);
  const browser = await chromium.launch(opts);
  console.log(`browser: ${browser.version()}`);
  const src = `${stream.origin}/index.m3u8`;
  const bundles = [];
  try {
    // The browser's first playback, untimed (see the head of this file). Its own time is printed,
    // and checked against no bound: it is the browser's start, and the host's load.
    warming = true;
    const warmCtx = await browser.newContext({ viewport: { width: 1280, height: 720 } });
    const wp = await warmCtx.newPage();
    await wp.goto(`${page.origin}/warm.html?src=${encodeURIComponent(src)}`, { waitUntil: "domcontentloaded" });
    const warm = await wp.waitForFunction(() => window.warm, null, { timeout: 60_000 }).then((h) => h.jsonValue(), (e) => ({ error: e.message }));
    await warmCtx.close();
    warming = false;
    check(Number.isInteger(warm.firstFrameMs), `the browser played the stream once before the timed sessions, untimed (its first frame ${warm.firstFrameMs ?? "never"} ms after the page asked it to play${warm.error ? `: ${warm.error}` : ""})`);
    for (let run = 0; run < RUNS; run++) {
      for (const mode of ["loader", "tap"]) {
        served.length = 0;
        const ctx = await browser.newContext({ viewport: { width: 1280, height: 720 } });
        const p = await ctx.newPage();
        const logs = [];
        p.on("console", (m) => logs.push(`${m.type()}: ${m.text()}`));
        p.on("pageerror", (e) => logs.push(`pageerror: ${e.message}`));
        const url = `${page.origin}/?mode=${mode}&channel=${CHANNEL}&owner=${owner}&collector=${encodeURIComponent(collectorUrl)}&src=${encodeURIComponent(src)}`;
        await p.goto(url, { waitUntil: "domcontentloaded" });
        let result = null;
        try {
          await p.waitForFunction(() => window.auditResult, null, { timeout: 60_000 });
          result = await p.evaluate(() => window.auditResult);
        } catch (e) {
          console.log(logs.join("\n"));
          throw new Error(`${mode}: no bundle was sent: ${e.message}`);
        }
        await ctx.close();
        const o = result.answer.body?.outcome ?? {};
        const tag = `run ${run + 1}, ${mode}`;
        check(result.answer.status === 201 && result.answer.body?.stored === "new", `${tag}: the collector verified and stored the session (status ${result.answer.status}${o.reason ? `, ${o.reason}` : ""})`);
        check(o.outcome === "accept" && o.presentation === "on_demand" && o.ended === "ended", `${tag}: accepted, on demand, ended at its end`);
        check(o.presented_ms >= 10_500 && o.presented_ms <= 12_500, `${tag}: presented ${o.presented_ms} ms of a 12 s stream`);
        check(o.viewable_ms === o.presented_ms && o.audible_ms === 0, `${tag}: in view throughout, and muted (${o.viewable_ms} ms in view, ${o.audible_ms} ms audible)`);
        check(o.ad_ms >= 3_500 && o.ad_ms <= 4_500, `${tag}: ${o.ad_ms} ms within the marked ad`);
        const claim = o.claims?.[0] ?? {};
        check(o.claims?.length === 1 && claim.verdict === "presented" && claim.quartile === 5 && claim.viewable === true, `${tag}: the marked ad's claim is presented, every quartile, viewable (${JSON.stringify(o.claims)})`);
        // Every object the origin served is acknowledged by its SHA-256, and nothing else is.
        const acked = new Set();
        for (const a of result.bundle.acks) for (const list of a.items) for (let i = 0; i < list.length; i += 96) acked.add(list.slice(i, i + 64));
        const servedHashes = new Set(served.map((n) => hashes[n]));
        check(served.length > segments.length && [...servedHashes].every((h) => acked.has(h)) && [...acked].every((h) => servedHashes.has(h)), `${tag}: the ${acked.size} objects acknowledged are exactly the ${servedHashes.size} the origin served, by SHA-256`);
        const d = Object.fromEntries((o.delivery ?? []).map((x) => [x.class, x]));
        check(d.fetch?.origin === stream.origin && d.fetch.objects === segments.length + 1 && d.ancillary?.objects === 1, `${tag}: one origin; the init and ${segments.length} segments as fetch, the playlist as ancillary`);
        check(o.notes === segments.length, `${tag}: the CMSD-Static the origin exposed recorded for each of its ${segments.length} segments (${o.notes})`);
        check(JSON.stringify(o.energy) === JSON.stringify([{ role: "sink_decode", class: "unmetered", joules_micro: 0 }, { role: "display", class: "unmetered", joules_micro: 0 }]), `${tag}: a page's energy: sink decode and display unmetered, never zero joules`);
        check(Number.isInteger(o.startup_ms) && o.startup_ms < 10_000, `${tag}: start time ${o.startup_ms} ms in a browser that has played once (bound 10 000 ms), stalls ${o.stalls}`);
        bundles.push({ mode, bundle: result.bundle, sid: result.sid });
      }
    }
  } finally {
    await browser.close();
  }

  // The same bundle again is stored once; a byte changed is refused.
  const first = JSON.stringify(bundles[0].bundle);
  const again = await fetch(`${collectorUrl}v1/sessions`, { method: "POST", body: first });
  check(again.status === 200 && (await again.json()).stored === "again", "the same bundle again: stored once");
  const tampered = structuredClone(bundles[0].bundle);
  tampered.session = tampered.session.slice(0, -2) + (tampered.session.endsWith("00") ? "01" : "00");
  const bad = await fetch(`${collectorUrl}v1/sessions`, { method: "POST", body: JSON.stringify(tampered) });
  const badBody = await bad.json();
  check(bad.status === 422 && badBody.reason === "bad_signature", `a session record with its signature altered: refused ${badBody.reason}`);

  // Event-mode CMCD reached the collector, for the sessions' sid.
  const cmcd = fs.existsSync(path.join(data, "cmcd")) ? fs.readdirSync(path.join(data, "cmcd")).map((f) => fs.readFileSync(path.join(data, "cmcd", f), "utf8")).join("") : "";
  check(bundles.every((b) => cmcd.includes(`sid="${b.sid}"`) && cmcd.includes("e=ps") && cmcd.includes("v=2")), "Event-mode CMCD version 2 batches reached the collector for every session");

  // The channel's report: every session, every ad; recomputed from the bundles by `check`.
  const rep = await fetch(`${collectorUrl}v1/channels/${CHANNEL}/report?from=0&to=4102444800000000`).then((r) => r.json());
  const r = rep.report ?? {};
  check(r.sessions?.length === bundles.length && r.ads?.presented === bundles.length && r.ads?.viewable === bundles.length, `the report: ${r.sessions?.length} sessions, ${r.ads?.presented} ads presented, ${r.ads?.viewable} viewable`);
  check(r.rebuffering?.suppressed === true && r.rebuffering?.presented === null, "the report's rebuffering is suppressed below 50 sessions");
  check(r.presented?.presented_ms > 0 && r.unmetered?.sink_decode === bundles.length && r.energy?.length === 0, "the report: time presented; sink decode unmetered in every session, and no joules stated");
  fs.writeFileSync(path.join(OUT, "report.json"), JSON.stringify(rep));
  const files = bundles.map((b, i) => {
    const f = path.join(OUT, `bundle-${i}-${b.mode}.json`);
    fs.writeFileSync(f, JSON.stringify(b.bundle));
    return f;
  });
  const recompute = spawnSync(args.collector, ["check", path.join(OUT, "report.json"), CHANNEL, ...files], { env: { ...process.env, WAI_AUDIT_OWNER: owner }, encoding: "utf8" });
  check(recompute.status === 0, `the served report recomputes from the bundles: ${recompute.stdout.trim()}${recompute.stderr.trim()}`);
  const fewer = spawnSync(args.collector, ["check", path.join(OUT, "report.json"), CHANNEL, ...files.slice(1)], { env: { ...process.env, WAI_AUDIT_OWNER: owner }, encoding: "utf8" });
  check(fewer.status === 1 && fewer.stdout.includes("record_missing"), `and is refused with a bundle withheld: ${fewer.stdout.trim()}`);

  collector.kill();
  stream.server.close();
  page.server.close();
  console.log(failures === 0 ? "all checks ok" : `${failures} check(s) failed`);
  process.exit(failures === 0 ? 0 : 1);
}

main().catch((e) => {
  console.error(e.stack || e);
  process.exit(1);
});