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); and its presentation statements
// written and signed byte for byte (§7.1), each at the rate the catch-up rule gives (§9).
import test from "node:test";
import assert from "node:assert/strict";
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);
});