open-standards/joule-code/joule-code-rs/crates/joule-moq-web/web/room/cowatch.test.mjs

// The browser client's co-watch module (cowatch.js), against the corpus
// wai/live-conformance/cowatch/ (`node --test web/room/cowatch.test.mjs`): every catch-up case
// its correction (co-watch §9); each accepted room's controlled position recomputed from its
// policies and controls (§5.4, §6), as the corpus states it at every change and at every
// statement; its controls written byte for byte (§5.1); its presentation statements
// written and signed byte for byte (§7.1), each at the rate the catch-up rule gives (§9); the
// player taken for gone by the room re-entering with its chain whole (§7.3, §10); and the
// player stating in every slot while its loop runs, recording each stretch it did not (§7.3).
import test from "node:test";
import assert from "node:assert/strict";
import { createHash } from "node:crypto";
import { readFileSync } from "node:fs";
import * as cw from "./cowatch.js";

const doc = JSON.parse(readFileSync(new URL("../../../../../../wai/live-conformance/cowatch/vectors.json", import.meta.url), "utf8"));
const vector = (name) => doc.vectors.find((v) => v.name === name);
const ACCEPTED = ["watch_party", "next_episode", "drift"];

/** The corpus's key `i`: SHA-256 of its label, a zero byte and `i` (its README). */
const key = async (i) => cw.keyFromSeed(await cw.sha256(new TextEncoder().encode("wai:co-watch/corpus/key\x00"), Uint8Array.of(i)));
const INDEX = { authority: 0, clock: 1, host: 2, guest: 3, late: 4, fourth: 5 };

/** Every record of a vector, by its digest. */
async function records(v) {
  const out = new Map();
  for (const chain of v.chains) {
    for (const h of chain) {
      const b = cw.unhex(h);
      const ev = cw.readEvent(b);
      out.set(cw.hex(await cw.sha256(ev.body)), ev);
    }
  }
  return out;
}

const timelineOf = (v) =>
  new cw.Timeline(v.policies.map((h) => cw.readPolicy(cw.unhex(h))), v.controls.map((c) => ({ at: c.at, t: c.t, digest: c.digest, action: cw.ACTION[c.action], arg: c.arg })));

test("every catch-up case gets the corpus's correction", () => {
  const { params, cases } = doc.catch_up;
  assert.ok(cases.length > 0);
  for (const c of cases) assert.deepEqual(cw.catchUp(params, c.playing, c.rate, c.controlled, c.duration, c.presented), c.correction, c.about);
});

test("the controlled position of each accepted room is the corpus's", () => {
  for (const name of ACCEPTED) {
    const v = vector(name);
    const tl = timelineOf(v);
    assert.deepEqual(tl.points.map(([x, s]) => ({ at: x, playing: s.playing, rate: s.rate, position: s.pos })), v.timeline, name);
    for (const c of v.controls) assert.equal(!tl.overridden.has(c.digest), c.applied, `${name}: ${c.digest.slice(0, 8)}`);
    for (const p of v.presentations) assert.equal(tl.positionAt(p.at), p.controlled, `${name}: the statement at ${p.at}`);
  }
});

test("each accepted room's controls are written byte for byte", async () => {
  for (const name of ACCEPTED) {
    const v = vector(name);
    const recs = await records(v);
    for (const c of v.controls) {
      const ev = recs.get(c.digest);
      assert.ok(ev, `${name}: a control's record`);
      assert.deepEqual(cw.controlPayload({ action: cw.ACTION[c.action], at: c.at, arg: c.arg }), ev.payload, `${name}: ${c.action} at ${c.at}`);
      assert.deepEqual(cw.readStatement(ev.payload), { kind: "control", action: cw.ACTION[c.action], at: c.at, arg: c.arg });
    }
  }
});

test("each accepted room's presentation statements are written and signed byte for byte, at the catch-up rule's rate", async () => {
  const keys = {};
  for (const [n, i] of Object.entries(INDEX)) keys[doc.principals[n]] = await key(i);
  for (const name of ACCEPTED) {
    const v = vector(name);
    const recs = await records(v);
    const policies = v.policies.map((h) => cw.readPolicy(cw.unhex(h)));
    const tl = timelineOf(v);
    for (const p of v.presentations) {
      const ev = recs.get(p.digest);
      const pol = policies[cw.policyAt(policies, p.at)];
      const st = tl.stateAt(p.at);
      const duration = tl.durationAt(p.at);
      // The player states its own rate: the controlled one, or the catch-up rule's (§9).
      let rate = st.rate;
      if (st.playing && p.controlled < duration) {
        const c = cw.catchUp(pol, true, st.rate, p.controlled, duration, p.position);
        if (c.action === "rate") rate = c.value;
      }
      const state = !st.playing ? cw.STATE.paused : p.controlled >= duration ? cw.STATE.ended : cw.STATE.playing;
      assert.equal(state, cw.STATE[p.state], `${name}: the state at ${p.at}`);
      const k = keys[p.player];
      assert.equal(cw.hex(k.publicKey), p.player);
      const mine = await cw.signPresentation({ room: v.room, player: p.player, source: pol.source, at: p.at, position: p.position, state, rate }, k.privateKey);
      assert.deepEqual(mine, ev.payload, `${name}: the statement of ${p.player.slice(0, 8)} at ${p.at}`);
    }
  }
});

test("skew statements read back with their verdicts", async () => {
  const v = vector("drift");
  const recs = await records(v);
  for (const s of v.skews) {
    const st = cw.readStatement(recs.get(s.digest).payload);
    assert.equal(st.kind, "skew");
    assert.deepEqual(st.viewers.map((x) => [x.principal, x.verdict, x.pq]), s.viewers.map((x) => [x.principal, x.verdict, x.pq]));
  }
});

test("a control the room would refuse is not applied", () => {
  const v = vector("watch_party");
  const policies = v.policies.map((h) => cw.readPolicy(cw.unhex(h)));
  const yes = () => true;
  assert.equal(cw.controlRefusal(policies, 5_000_000, { action: 2, at: 4_000_000, arg: 0 }, yes), "control_not_ahead");
  assert.equal(cw.controlRefusal(policies, 5_000_000, { action: 2, at: 5_100_000, arg: 0 }, yes), "control_not_ahead");
  assert.equal(cw.controlRefusal(policies, 5_000_000, { action: 2, at: 16_000_000, arg: 0 }, yes), "control_beyond_horizon");
  assert.equal(cw.controlRefusal(policies, 5_000_000, { action: 3, at: 6_000_000, arg: policies[0].duration_us + 1 }, yes), "control_out_of_range");
  assert.equal(cw.controlRefusal(policies, 5_000_000, { action: 4, at: 6_000_000, arg: 3_000_000 }, yes), "control_out_of_range");
  assert.equal(cw.controlRefusal(policies, 5_000_000, { action: 2, at: 6_000_000, arg: 0 }, () => false), "control_not_permitted");
  assert.equal(cw.controlRefusal(policies, 5_000_000, { action: 2, at: 6_000_000, arg: 0 }, yes), null);
});

test("the clock's offset is taken from the least round trip", () => {
  assert.equal(cw.clockOffset([{ sent: 1000, got: 1100, server: 5_000_000 }, { sent: 2000, got: 2010, server: 7_000_000 }]), 7_000_000 - 2_005_000);
});

// A record of the tests below: its fields as JSON, in place of a signed WEV1 record, and its
// digest the SHA-256 of those bytes.
const fakeWai = {
  eventSign: (seed, room, seq, prev, t, kind, payload) => new TextEncoder().encode(JSON.stringify({ seq: Number(seq), prev: cw.hex(prev), t: Number(t), payload: cw.hex(payload) })),
  eventRead: (rec) => JSON.stringify({ ok: true, digest: createHash("sha256").update(rec).digest("hex") }),
};

// A watch room's service as joule-moq-room's watch.rs carries one viewer's events (co-watch
// §10): while the viewer is present, each event the next of its chain, at or after its `join`
// and its last event, and a control at least the lead ahead of its `t`; the first that is not
// is refused with the service's reason, those before it carried. Its clock signs a `join`
// `late` µs after the page's estimate of now. `failJoins`: the next joins its clock cannot
// sign (503). `leaveAfterJoin`: the next joins whose presence a `leave` the room had asked
// before them, signed after them, ends at once (a service that ordered them so).
function watchService(lead_ms, late) {
  const s = { now: 0, present: false, joined: 0, lastT: 0, joins: 0, chain: [], controls: [], presentations: [], failJoins: 0, leaveAfterJoin: 0 };
  s.api = async (path, body) => {
    if (path === "/watch/challenge") return { status: 200, body: { challenge: "00".repeat(32) } };
    if (path === "/watch/join") {
      if (s.failJoins > 0) {
        s.failJoins--;
        return { status: 503, body: { error: "the clock has no session" } };
      }
      s.joined = s.now + late;
      s.lastT = Math.max(s.lastT, s.joined);
      s.present = true;
      s.joins++;
      if (s.leaveAfterJoin > 0) {
        s.leaveAfterJoin--;
        s.present = false;
      }
      return { status: 200, body: { ok: true, t: s.joined } };
    }
    assert.equal(path, "/watch/post");
    for (const [k, h] of body.events.entries()) {
      const ev = JSON.parse(new TextDecoder().decode(cw.unhex(h)));
      const st = cw.readStatement(cw.unhex(ev.payload));
      const why = !s.present ? "not_present"
        : ev.seq !== s.chain.length ? "delivery_gap"
        : ev.prev !== (s.chain.length ? s.chain[s.chain.length - 1] : "00".repeat(32)) ? "chain_broken"
        : ev.t < s.lastT || ev.t < s.joined ? "time_reversed"
        : st && st.kind === "control" && st.at < ev.t + lead_ms * 1000 ? "control_not_ahead"
        : null;
      if (why) return { status: 409, body: { error: why, carried: k } };
      s.chain.push(createHash("sha256").update(cw.unhex(h)).digest("hex"));
      s.lastT = ev.t;
      if (st && st.kind === "control") s.controls.push({ action: st.action, at: st.at, t: ev.t });
      if (st && st.kind === "presentation") s.presentations.push(st.at);
    }
    return { status: 200, body: { ok: true, carried: body.events.length } };
  };
  return s;
}

/** Until the page has posted all it holds, or stopped. */
async function settled(c) {
  const end = Date.now() + 5000;
  while ((c.posting || c.outbox.length) && !c.stopped && Date.now() < end) await new Promise((r) => setTimeout(r, 1));
}

test("a page the room took for gone enters again, and its control signed before then is carried", async () => {
  const policy = vector("watch_party").policies[0];
  const lead = cw.readPolicy(cw.unhex(policy)).lead_ms;
  const svc = watchService(lead, 200_000);
  const seed = new Uint8Array(32).fill(7);
  const key = await cw.keyFromSeed(seed);
  const logs = [];
  const c = new cw.CoWatch({
    api: svc.api, wai: fakeWai, room: "11".repeat(32), t0: 0, seed, principal: cw.hex(key.publicKey), grant: "33".repeat(100), media: {},
    policies: [policy], objects: [], sourceUrl: "", mayControl: () => true, target: "//127.0.0.1/watch/join", log: (l) => logs.push(l),
  });
  c.now = () => svc.now;
  svc.now = 1_000_000;
  await c.join();
  svc.now = 1_500_000;
  const playAt = c.control(cw.ACTION.play);
  await settled(c);
  assert.deepEqual(svc.controls, [{ action: cw.ACTION.play, at: playAt, t: 1_500_000 }]);

  // The page falls silent past the room's limit (its thread held, on a loaded machine): the
  // room takes it for gone. Then it states what it presents, and a pause is pressed, both
  // signed before the room's next `join` for it, which its clock signs 200 ms later.
  svc.present = false;
  svc.now = 7_000_000;
  const source = cw.readPolicy(cw.unhex(policy)).source;
  const body = { room: c.room, player: c.principal, source, at: svc.now, position: 5_500_000, state: cw.STATE.playing, rate: cw.NORMAL };
  c.outbox.push(c.event(svc.now, await cw.signPresentation(body, key.privateKey)));
  const pauseAt = c.control(cw.ACTION.pause);
  await settled(c);

  assert.equal(c.stopped, null, logs.join("\n"));
  assert.deepEqual(c.stats.post_refused, {});
  assert.equal(svc.joins, 2);
  assert.deepEqual([c.stats.rejoins, c.stats.resigned, c.stats.dropped], [1, 1, 1]);
  // The room's copy of the chain is whole: the play, then the pause signed again after the
  // join, at the earliest instant its lead allows from there; the statement of an instant
  // before the page was present again is not carried.
  assert.equal(svc.chain.length, 2);
  const [, pause] = svc.controls;
  assert.equal(pause.action, cw.ACTION.pause);
  assert.ok(pause.t >= svc.joined, `the pause's t ${pause.t}, the join's ${svc.joined}`);
  assert.equal(pause.at, svc.joined + lead * 1000 + 50_000);
  assert.ok(pause.at > pauseAt);
  assert.deepEqual([c.seq, c.prev], [2, svc.chain[1]]);

  // And it goes on: the next event follows the re-signed one.
  svc.now = 8_000_000;
  c.control(cw.ACTION.play);
  await settled(c);
  assert.equal(c.stopped, null);
  assert.deepEqual([svc.chain.length, svc.controls.length], [3, 3]);
});

/** A page in a watch room kept by `watchService`, entered at room clock 1 s, its play carried
 * at 1.5 s. */
async function entered() {
  const policy = vector("watch_party").policies[0];
  const p = cw.readPolicy(cw.unhex(policy));
  const svc = watchService(p.lead_ms, 200_000);
  const seed = new Uint8Array(32).fill(7);
  const key = await cw.keyFromSeed(seed);
  const logs = [];
  const c = new cw.CoWatch({
    api: svc.api, wai: fakeWai, room: "11".repeat(32), t0: 0, seed, principal: cw.hex(key.publicKey), grant: "33".repeat(100), media: {},
    policies: [policy], objects: [], sourceUrl: "", mayControl: () => true, target: "//127.0.0.1/watch/join", log: (l) => logs.push(l),
  });
  c.now = () => svc.now;
  svc.now = 1_000_000;
  await c.join();
  svc.now = 1_500_000;
  c.control(cw.ACTION.play);
  await settled(c);
  assert.equal(svc.controls.length, 1);
  return { svc, c, logs, lead: p.lead_ms };
}

test("a page taken for gone again right after it entered again enters once more, its control carried", async () => {
  const { svc, c, logs, lead } = await entered();
  // Gone; and the `leave` the room asked for it is signed only after its next join, which it
  // ends (a service that signed them in that order).
  svc.present = false;
  svc.leaveAfterJoin = 1;
  svc.now = 7_000_000;
  c.control(cw.ACTION.pause);
  await settled(c);
  assert.equal(c.stopped, null, logs.join("\n"));
  assert.deepEqual(c.stats.post_refused, {});
  assert.equal(svc.joins, 3);
  assert.deepEqual([c.stats.gone, c.stats.rejoins, c.stats.resigned], [2, 2, 2]);
  assert.deepEqual(svc.controls.map((x) => x.action), [cw.ACTION.play, cw.ACTION.pause]);
  assert.ok(svc.controls[1].t >= svc.joined);
  assert.equal(svc.controls[1].at, svc.joined + lead * 1000 + 50_000);
});

test("a join the service cannot make is tried again by the next post, and the page goes on", async () => {
  const { svc, c, logs } = await entered();
  svc.present = false;
  svc.failJoins = 1;
  svc.now = 7_000_000;
  c.control(cw.ACTION.pause);
  await settled(c);
  assert.equal(c.stopped, null, logs.join("\n"));
  assert.deepEqual(c.stats.post_refused, {});
  assert.equal(svc.joins, 2);
  assert.deepEqual([c.stats.gone, c.stats.rejoins, c.stats.resigned], [2, 1, 1]);
  assert.deepEqual(svc.controls.map((x) => x.action), [cw.ACTION.play, cw.ACTION.pause]);
  assert.ok(logs.some((l) => /entering again failed/.test(l)), logs.join("\n"));
});

test("a page the room takes for gone at every entry stops after a few", async () => {
  const { svc, c } = await entered();
  svc.present = false;
  svc.leaveAfterJoin = Infinity;
  svc.now = 7_000_000;
  c.control(cw.ACTION.pause);
  await settled(c);
  assert.match(c.stopped || "", /not_present/);
  assert.deepEqual(c.stats.post_refused, { not_present: 1 });
  assert.deepEqual([c.stats.gone, c.stats.rejoins, svc.joins], [3, 3, 4]);
  assert.equal(svc.controls.length, 1);
});

test("a control pressed while the room had taken the page for gone is dropped where signing it again would put it past another viewer's", async () => {
  const { svc, c, logs, lead } = await entered();
  const other = (at, action) => c.controls.push({ at, t: svc.now, digest: cw.hex(createHash("sha256").update(String(at)).digest()), action, arg: 0, author: "22".repeat(32) });
  // Gone; another viewer's pause is carried for 100 ms after this page's seek as pressed.
  // Signed again after the join, 200 ms later, the seek would apply after that pause: dropped.
  svc.present = false;
  svc.now = 7_000_000;
  other(svc.now + lead * 1000 + 50_000 + 100_000, cw.ACTION.pause);
  c.control(cw.ACTION.seek, 1_000_000);
  await settled(c);
  assert.equal(c.stopped, null, logs.join("\n"));
  assert.deepEqual([c.stats.rejoins, c.stats.resigned, c.stats.dropped], [1, 0, 1]);
  assert.equal(svc.controls.length, 1, "the seek is not carried");
  assert.ok(logs.some((l) => /is dropped: moved to/.test(l)), logs.join("\n"));
  // Gone again; other viewers' controls before its instant as pressed, and past where it
  // moves to: signed again, it keeps its place among them.
  svc.present = false;
  svc.now = 9_000_000;
  other(svc.now + 100_000, cw.ACTION.pause);
  other(svc.now + 5_000_000, cw.ACTION.play);
  const seekAt = c.control(cw.ACTION.seek, 2_000_000);
  await settled(c);
  assert.equal(c.stopped, null, logs.join("\n"));
  assert.deepEqual([c.stats.rejoins, c.stats.resigned], [2, 1]);
  assert.deepEqual(svc.controls.map((x) => x.action), [cw.ACTION.play, cw.ACTION.seek]);
  assert.ok(svc.controls[1].at > seekAt && svc.controls[1].at === svc.joined + lead * 1000 + 50_000);
});

/** The player's loop run for `passes` passes on a room clock the test moves, so the test, not
 * the host, sets each pass's instant: by `step(i)` µs in the loop's `i`th wait between two
 * passes, and by `work(i)` µs in the work of its `i`th pass (in `onChange`, which the loop
 * calls in each pass, as work of its own or an await it does not measure would be); then
 * until the page has posted all it holds. */
async function runFor(c, svc, passes, step, work = () => 0) {
  let i = 0;
  c.media = { currentTime: 0, paused: true, ended: false, readyState: 4, playbackRate: 1, play: () => Promise.resolve(), pause: () => {} };
  c.now = () => svc.now;
  c.onChange = () => { svc.now += work(i); };
  c.wait = async () => { svc.now += step(i++); await new Promise((r) => setTimeout(r, 0)); };
  await c.run(() => i < passes);
  await settled(c);
}

/** The slots from the page's first statement to its last: each held by a statement, else
 * empty, and of those whose second half a stall of the page spans. */
function slotsOf(ats, stalls, P) {
  const out = { held: 0, empty: [], stalled: [] };
  for (let s = Math.ceil(ats[0] / P) * P; s + P <= ats[ats.length - 1]; s += P) {
    if (ats.some((x) => x >= s && x < s + P)) out.held++;
    else (stalls.some((x) => x.from < s + P / 2 && x.to >= s + P) ? out.stalled : out.empty).push(s);
  }
  return out;
}

test("while its loop runs, the player states in every slot and records no stall", async () => {
  const { svc, c, logs } = await entered();
  const P = c.policy.period_ms * 1000;
  await runFor(c, svc, 40, () => 50_000);
  assert.equal(c.stopped, null, logs.join("\n"));
  assert.deepEqual(c.stats.stalls, []);
  const s = slotsOf(svc.presentations, c.stats.stalls, P);
  assert.ok(s.held >= 1, JSON.stringify(s));
  assert.deepEqual([s.empty, s.stalled], [[], []]);
  for (let k = 1; k < svc.presentations.length; k++) assert.ok(svc.presentations[k] - svc.presentations[k - 1] <= P / 2 + 50_000);
});

test("the player records the stretch its thread was held, and only then leaves a slot empty", async () => {
  const { svc, c, logs } = await entered();
  const P = c.policy.period_ms * 1000;
  // Its passes every 50 ms, but one: the page's thread held for three and a half slots while
  // the loop waited, so its wait took that long.
  const HELD = 15;
  await runFor(c, svc, 50, (i) => (i === HELD ? 3.5 * P : 50_000));
  assert.equal(c.stopped, null, logs.join("\n"));
  assert.equal(c.stats.stalls.length, 1, JSON.stringify(c.stats.stalls));
  const [st] = c.stats.stalls;
  assert.equal(st.to - st.from, 3.5 * P);
  // The whole stall is the wait: the loop's own work and its signing took no time on the clock.
  assert.deepEqual([st.timer_us, st.signing_us], [3.5 * P, 0]);
  // The slots the held thread left empty are those the stall spans; every other holds a
  // statement, and none is empty outside it.
  const s = slotsOf(svc.presentations, c.stats.stalls, P);
  assert.ok(s.stalled.length >= 2, JSON.stringify(s));
  assert.deepEqual(s.empty, []);
  for (const x of s.stalled) assert.ok(st.from < x + P / 2 && st.to >= x + P);
  // A pass more than half a slot after the one before is a stall: the loop states every half
  // slot, so only such a stretch can leave a slot empty.
  const { svc: svc2, c: c2 } = await entered();
  await runFor(c2, svc2, 30, (i) => (i === HELD ? P / 2 + 1 : i === HELD + 5 ? P / 2 : 50_000));
  assert.deepEqual(c2.stats.stalls.map((x) => x.to - x.from), [P / 2 + 1]);
});

test("a stall in the loop's own work is recorded with a wait that does not cover it", async () => {
  const { svc, c, logs } = await entered();
  const P = c.policy.period_ms * 1000;
  // One pass's own work takes two slots (or an await the loop does not measure would): its
  // wait after it took 50 ms, and the stall is the rest.
  const HELD = 15;
  await runFor(c, svc, 50, () => 50_000, (i) => (i === HELD ? 2 * P : 0));
  assert.equal(c.stopped, null, logs.join("\n"));
  assert.equal(c.stats.stalls.length, 1, JSON.stringify(c.stats.stalls));
  const [st] = c.stats.stalls;
  assert.equal(st.to - st.from, 2 * P + 50_000);
  assert.deepEqual([st.timer_us, st.signing_us], [50_000, 0]);
  assert.equal(st.to - st.from - st.timer_us - st.signing_us, 2 * P);
});