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

// Watching together in a room (WAI extensions/co-watch.md), in a page.
//
// A watch room's service serves the source and carries its viewers' events in a log
// (co-watch §10). This module is the page's part, in two halves:
//
//   - the rules, with nothing of the page: the policy read (§3.1), controls and presentation
//     statements written byte for byte (§5.1, §7.1), statements read (§4.2), the controlled
//     position (§5.4, §6), the checks a control meets before a player applies it (§5.2, §5.3),
//     the catch-up rule (§9), and the clock estimate (§7.3). Its tests (cowatch.test.mjs) run
//     them against the corpus wai/live-conformance/cowatch/;
//   - CoWatch, the player: it enters the room by proving its grant's key over the service's
//     challenge (live-rooms §10.5.3), follows the log, applies each control at its instant on
//     its estimate of the room clock, keeps a media element at the controlled position by the
//     catch-up rule, signs a presentation statement twice a slot under the page's key, and shows
//     the room's sync: its own error, and each viewer's verdict from the clock's last skew
//     statement. Taken for gone after a silence, it enters again and signs again, after its new
//     `join`, the controls the room had not carried.
//
// Everything here is the platform's: WebCrypto's SHA-256 and Ed25519, an HTMLMediaElement.

const subtle = globalThis.crypto.subtle;
const enc = new TextEncoder();

export const NORMAL = 1000000;
export const ACTION = { play: 1, pause: 2, seek: 3, rate: 4 };
export const ACTION_NAME = { 1: "play", 2: "pause", 3: "seek", 4: "rate" };
export const STATE = { playing: 1, paused: 2, stalled: 3, ended: 4 };
export const VERDICT = { 1: "in_sync", 2: "out_of_sync", 3: "unknown" };
const MAGIC = { policy: "WXP1", control: "WXC1", presentation: "WXS1", skew: "WXK1" };
const PRESENTATION_DOMAIN = enc.encode("wai:co-watch-presentation\x01");
const PROOF_DOMAIN = enc.encode("wai:live-room-proof\x01");
const META = 0x01;

export const hex = (b) => Array.from(b, (x) => x.toString(16).padStart(2, "0")).join("");
export const unhex = (h) => new Uint8Array((h.match(/../g) || []).map((x) => parseInt(x, 16)));

export function concat(...parts) {
  const n = parts.reduce((a, p) => a + p.length, 0);
  const out = new Uint8Array(n);
  let o = 0;
  for (const p of parts) { out.set(p, o); o += p.length; }
  return out;
}

export async function sha256(...parts) {
  return new Uint8Array(await subtle.digest("SHA-256", concat(...parts)));
}

const le = (n, bytes) => {
  const out = new Uint8Array(bytes);
  let v = BigInt(n);
  for (let i = 0; i < bytes; i++) { out[i] = Number(v & 0xffn); v >>= 8n; }
  return out;
};

class Rd {
  constructor(b, p = 0) { this.b = b; this.p = p; }
  take(n) { if (this.p + n > this.b.length) throw new Error("short"); const s = this.b.subarray(this.p, this.p + n); this.p += n; return s; }
  u8() { return this.take(1)[0]; }
  uint(n) { const s = this.take(n); let v = 0n; for (let i = n - 1; i >= 0; i--) v = (v << 8n) | BigInt(s[i]); return v; }
  u16() { return Number(this.uint(2)); }
  u32() { return Number(this.uint(4)); }
  u64() { return Number(this.uint(8)); } // every instant, position and rate here is below 2^53
  done() { return this.p === this.b.length; }
}

// ---------------------------------------------------------------- §3 the policy

/** A co-watch policy's fields (§3.1); its signature is the room's to check. */
export function readPolicy(b) {
  if (new TextDecoder().decode(b.subarray(0, 4)) !== MAGIC.policy || b[4] !== 1) throw new Error("not a WXP1 policy");
  const r = new Rd(b, 5);
  const p = { room: hex(r.take(32)), prev: hex(r.take(32)), effective: r.u64(), source_type: r.u8(), source: hex(r.take(32)), duration_us: r.u64() };
  for (const k of ["lead_ms", "horizon_ms", "min_rate", "max_rate", "period_ms", "window_ms", "settle_ms", "bound_us"]) p[k] = r.u32();
  p.quantile = r.u8();
  for (const k of ["tolerance_us", "nudge", "seek_us"]) p[k] = r.u32();
  r.take(64);
  if (!r.done()) throw new Error("bytes after the policy");
  return p;
}

/** The policy in effect at `t` (§3.4): the last whose `effective` is at or below it. */
export function policyAt(policies, t) {
  let k = null;
  policies.forEach((p, i) => { if (p.effective <= t) k = i; });
  return k;
}

/** An object sequence's digest (§3.5). */
export async function sourceDigest(objects) {
  return sha256(enc.encode("wai:co-watch-source\x01"), le(objects.length, 4), ...objects.flatMap((o) => [unhex(o.content_hash), le(o.duration_ms, 4)]));
}

// ---------------------------------------------------------------- §5, §7 statements

/** A control (§5.1), 22 bytes. */
export function controlPayload(c) {
  return concat(enc.encode(MAGIC.control), Uint8Array.of(1, c.action), le(c.at, 8), le(c.arg, 8));
}

/** A presentation statement's record from `magic` through `rate` (§7.1). */
export function presentationBody(p) {
  return concat(enc.encode(MAGIC.presentation), Uint8Array.of(1), unhex(p.room), unhex(p.player), unhex(p.source), le(p.at, 8), le(p.position, 8), Uint8Array.of(p.state), le(p.rate, 4));
}

/** The statement, signed with Ed25519 under the player's key (§7.1): 186 bytes. */
export async function signPresentation(p, privateKey) {
  const body = presentationBody(p);
  const sig = new Uint8Array(await subtle.sign({ name: "Ed25519" }, privateKey, concat(PRESENTATION_DOMAIN, body)));
  return concat(body, sig);
}

/** WebCrypto's key for a 32-byte Ed25519 seed (RFC 8410's PKCS #8 form), and its public key. */
export async function keyFromSeed(seed) {
  const pkcs8 = concat(Uint8Array.of(0x30, 0x2e, 0x02, 0x01, 0x00, 0x30, 0x05, 0x06, 0x03, 0x2b, 0x65, 0x70, 0x04, 0x22, 0x04, 0x20), seed);
  const privateKey = await subtle.importKey("pkcs8", pkcs8, { name: "Ed25519" }, true, ["sign"]);
  const jwk = await subtle.exportKey("jwk", privateKey);
  const publicKey = Uint8Array.from(atob(jwk.x.replace(/-/g, "+").replace(/_/g, "/")), (c) => c.charCodeAt(0));
  return { privateKey, publicKey };
}

/** A statement of this extension (§4.1, §4.2) read, or null for another `meta` payload; a
 * statement that does not read is `{ kind: "malformed" }`. */
export function readStatement(payload) {
  const magic = new TextDecoder().decode(payload.subarray(0, 4));
  if (![MAGIC.control, MAGIC.presentation, MAGIC.skew].includes(magic)) return null;
  try {
    if (payload[4] !== 1) throw new Error("version");
    const r = new Rd(payload, 5);
    let s;
    if (magic === MAGIC.control) {
      s = { kind: "control", action: r.u8(), at: r.u64(), arg: r.u64() };
      const ok = s.action === 1 || s.action === 2 ? s.arg === 0 : s.action === 3 ? true : s.action === 4 ? s.arg >= 1 && s.arg <= 0xffffffff : false;
      if (!ok) throw new Error("arg");
    } else if (magic === MAGIC.presentation) {
      s = { kind: "presentation", room: hex(r.take(32)), player: hex(r.take(32)), source: hex(r.take(32)), at: r.u64(), position: r.u64(), state: r.u8(), rate: r.u32() };
      s.sig = r.take(64);
      if (!(s.state in { 1: 1, 2: 1, 3: 1, 4: 1 }) || s.rate === 0) throw new Error("state");
    } else {
      s = { kind: "skew", from: r.u64(), to: r.u64(), bound_us: r.u32(), quantile: r.u8(), viewers: [], pairs: [] };
      const nv = r.u8();
      for (let i = 0; i < nv; i++) {
        s.viewers.push({ principal: hex(r.take(32)), held: r.u64(), due: r.u16(), samples: r.u16(), counted: r.u16(), p50: r.u32(), pq: r.u32(), max: r.u32(), verdict: VERDICT[r.u8()] });
      }
      const np = r.u8();
      for (let i = 0; i < np; i++) s.pairs.push({ a: r.u8(), b: r.u8(), n: r.u16(), p50: r.u32(), pq: r.u32(), max: r.u32(), verdict: VERDICT[r.u8()] });
    }
    if (!r.done()) throw new Error("bytes after");
    return s;
  } catch {
    return { kind: "malformed" };
  }
}

/** A WEV1 record's fields (live-events §3.1), read as far as this module needs them. */
export function readEvent(b) {
  const r = new Rd(b, 5);
  const ev = { room: hex(r.take(32)), principal: hex(r.take(32)), seq: r.u64(), prev: hex(r.take(32)), t: r.u64() };
  if (r.u8() === 1) r.take(48);
  ev.kind = r.u8();
  ev.payload = r.take(r.u16());
  ev.body = b.subarray(0, r.p);
  ev.sig = r.take(64);
  return ev;
}

// ---------------------------------------------------------------- §5.4, §6 the controlled position

function positionOf(st, x, duration) {
  if (!st.playing) return st.pos;
  const p = st.pos + Number((BigInt(Math.max(0, x - st.anchor)) * BigInt(st.rate)) / 1000000n);
  return Math.min(duration, p);
}

/** The controlled position over the room clock (§6), from the policies and the accepted
 * controls, each `{ at, t, digest (hex), action, arg }`. */
export class Timeline {
  constructor(policies, controls) {
    this.policies = policies;
    this.starts = [];
    policies.forEach((p, i) => {
      if (i === 0 || p.source_type !== policies[i - 1].source_type || p.source !== policies[i - 1].source) this.starts.push(p.effective);
    });
    const key = (c) => [c.at, c.t, c.digest];
    this.controls = [...controls].sort((a, b) => {
      const [x, y] = [key(a), key(b)];
      for (let i = 0; i < 3; i++) if (x[i] !== y[i]) return x[i] < y[i] ? -1 : 1;
      return 0;
    });
    const aspect = { 1: 0, 2: 0, 3: 1, 4: 2 };
    this.overridden = new Set();
    this.controls.forEach((c, i) => {
      for (const d of this.controls.slice(i + 1)) {
        if (d.at !== c.at) break;
        if (aspect[d.action] === aspect[c.action]) { this.overridden.add(c.digest); break; }
      }
    });
    const events = [...this.starts.map((s) => ({ x: s, start: true })), ...this.controls.map((c) => ({ x: c.at, c }))];
    events.sort((a, b) => a.x - b.x || (a.start ? 0 : 1) - (b.start ? 0 : 1));
    this.points = [];
    let st = null;
    for (const e of events) {
      if (e.start) st = { playing: false, rate: NORMAL, anchor: e.x, pos: 0 };
      else if (st) {
        st = { ...st, pos: positionOf(st, e.x, this.durationAt(e.x)), anchor: e.x };
        if (e.c.action === 1) st.playing = true;
        else if (e.c.action === 2) st.playing = false;
        else if (e.c.action === 3) st.pos = e.c.arg;
        else st.rate = e.c.arg;
      }
      if (st) this.points.push([e.x, st]);
    }
  }

  durationAt(x) {
    const k = policyAt(this.policies, x);
    return k == null ? 0 : this.policies[k].duration_us;
  }

  stateAt(x) {
    let st = null;
    for (const [at, s] of this.points) if (at <= x) st = s;
    return st;
  }

  positionAt(x) {
    const st = this.stateAt(x);
    return st ? positionOf(st, x, this.durationAt(x)) : 0;
  }
}

/** Why a control is not applied (§5.2, §5.3), or null: `mayControl(action)` says whether its
 * author may send it, under the policy in effect at its `t`. */
export function controlRefusal(policies, t, c, mayControl) {
  const k = policyAt(policies, t);
  if (k == null) return "cw_no_policy";
  const p = policies[k];
  if (!mayControl(c.action)) return "control_not_permitted";
  if (c.at < t + p.lead_ms * 1000) return "control_not_ahead";
  if (c.at > t + p.horizon_ms * 1000) return "control_beyond_horizon";
  const q = policies[policyAt(policies, c.at)];
  if ((c.action === 3 && c.arg > q.duration_us) || (c.action === 4 && (c.arg < q.min_rate || c.arg > q.max_rate))) return "control_out_of_range";
  return null;
}

// ---------------------------------------------------------------- §9 catch-up

/** The correction the catch-up rule gives (§9). */
export function catchUp(p, playing, rate, controlled, duration, presented) {
  const e = presented - controlled;
  const mag = Math.abs(e);
  if (!playing || controlled >= duration) return mag > p.tolerance_us ? { action: "seek", value: controlled } : { action: "hold" };
  if (mag <= p.tolerance_us) return { action: "rate", value: rate };
  if (mag > p.seek_us) return { action: "seek", value: controlled };
  const r = e > 0 ? rate - p.nudge : rate + p.nudge;
  return { action: "rate", value: Math.max(p.min_rate, Math.min(p.max_rate, r)) };
}

// ---------------------------------------------------------------- §7.3 the clock

/** The offset of the service's clock from this page's, in µs, from round trips each
 * `{ sent, got, server }` (this page's ms at sending and receiving, the server's µs): the
 * answer of least round trip, at its midpoint (§7.3). */
export function clockOffset(samples) {
  const best = samples.reduce((a, s) => (!a || s.got - s.sent < a.got - a.sent ? s : a), null);
  return Math.round(best.server - ((best.sent + best.got) / 2) * 1000);
}

// ---------------------------------------------------------------- the player

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

// A slot is stated twice (co-watch §7.3), so timer jitter leaves none empty.
const STATES_PER_SLOT = 2;
// Statements one post carries at most: a front may hold a request's body to a few KiB.
const POST_EVENTS = 4;
// Times a page taken for gone enters again with nothing of it carried since its last entry
// (taken for gone again at once, as by a second stall) before it stops.
const REENTRIES = 3;
// The player's wait between two passes of its loop, in ms, and the stalls of the loop a page
// keeps (`stalls`).
const PASS_MS = 50;
const STALLS_KEPT = 256;

export class CoWatch {
  // o: { api (path, body) -> {status, body}, wai, room, t0, seed (Uint8Array), principal (hex),
  //      grant (hex), grantDigest (hex), media (HTMLMediaElement), policies (hex), objects,
  //      sourceUrl, mayControl (principal, action) -> bool, target (the join's proof target,
  //      "//authority/path"), log, onChange }
  constructor(o) {
    Object.assign(this, o);
    this.policyList = o.policies.map((h) => readPolicy(unhex(h)));
    this.policy = this.policyList[0];
    this.offset = 0;
    this.at = 0; // the next entry of the log
    this.controls = [];
    this.timeline = new Timeline(this.policyList, []);
    this.seq = 0;
    this.prev = "00".repeat(32);
    // Where the room's copy of this page's chain ends: the next seq, and the last digest it carried.
    this.carried = { seq: 0, prev: "00".repeat(32) };
    this.lastT = 0;
    this.joinT = null;
    this.outbox = []; // each { rec, seq, digest, control }: signed, not yet carried
    this.reentered = 0; // entries again since the room last carried an event of this page
    this.stopped = null;
    this.skews = [];
    this.others = {}; // principal -> its last statement: at, position, error
    // `gone`: posts refused `not_present`, the room having taken the page for gone; `rejoins`:
    // its entries again; `resigned`: controls signed again after one; `dropped`: statements
    // and controls signed before one and not signed again. `stalls`: the stretches in which
    // the player's loop did not run (`run`), the last STALLS_KEPT of them.
    this.stats = { log: 0, controls: 0, applied: 0, refused: {}, statements: 0, posted: 0, post_refused: {}, seeks: 0, nudges: 0, verified_source: false, gone: 0, rejoins: 0, resigned: 0, dropped: 0, followed_at: null, stalls: [], errors: [] };
    this.onChange = o.onChange || (() => {});
  }

  /** The room clock now, on this page's estimate (§7.3). */
  now() {
    return this.roomTime(Date.now());
  }

  /** The room clock at `ms` on this page's clock (`Date.now()`), on its estimate. */
  roomTime(ms) {
    return Math.max(0, Math.round(ms * 1000 + this.offset - this.t0));
  }

  async estimateClock() {
    const samples = [];
    for (let i = 0; i < 5; i++) {
      const sent = Date.now();
      const r = await this.api("/watch/time");
      const got = Date.now();
      if (r.status === 200) samples.push({ sent, got, server: r.body.now_us });
    }
    if (samples.length) this.offset = clockOffset(samples);
    this.stats.offset_us = this.offset;
  }

  /** The source, fetched whole and checked by its content hash (§3.5), as the media's own. */
  async loadSource() {
    const r = await fetch(this.sourceUrl, { cache: "no-store" });
    const bytes = new Uint8Array(await r.arrayBuffer());
    const h = hex(await sha256(bytes));
    const obj = this.objects[0];
    if (h !== obj.content_hash) throw new Error(`the source's bytes are not the object the policy names: ${h.slice(0, 16)}…`);
    if (hex(await sourceDigest(this.objects)) !== this.policy.source) throw new Error("the source's objects are not the policy's source");
    this.stats.verified_source = true;
    this.media.src = URL.createObjectURL(new Blob([bytes], { type: r.headers.get("content-type") || "application/octet-stream" }));
    this.media.preload = "auto";
    await new Promise((ok, bad) => {
      if (this.media.readyState >= 1) return ok();
      this.media.addEventListener("loadedmetadata", ok, { once: true });
      this.media.addEventListener("error", () => bad(new Error("the source does not play here")), { once: true });
    });
  }

  /** Enter: a proof of the grant's key over the service's challenge, for the request's own
   * method and target (live-rooms §10.5.1, §10.5.3); the clock then signs this page's `join`. */
  async join() {
    const c = await this.api("/watch/challenge");
    if (c.status !== 200) throw new Error(`no challenge: ${c.status}`);
    const grant = unhex(this.grant);
    const digest = await sha256(grant.subarray(0, grant.length - 64));
    const context = enc.encode(`POST ${this.target}`);
    const key = await keyFromSeed(this.seed);
    const body = concat(enc.encode("WRP1"), Uint8Array.of(1), digest, unhex(this.principal), unhex(c.body.challenge), le(context.length, 2), context);
    const sig = new Uint8Array(await subtle.sign({ name: "Ed25519" }, key.privateKey, concat(PROOF_DOMAIN, body)));
    const r = await this.api("/watch/join", { room: this.room, grant: this.grant, proof: hex(concat(body, sig)) });
    if (r.status !== 200) throw new Error(`the room refused the join: ${r.status} ${r.body && r.body.error}`);
    this.joinT = r.body.t;
    this.key = key;
    this.log(`co-watch: joined at room clock ${r.body.t}`);
  }

  /** A `meta` event of this page's chain, signed (live-events §3), at `t`: the outbox's entry
   * for it, with the control it carries, if it is one. */
  event(t, payload, control = null) {
    t = Math.max(t, this.lastT, this.joinT || 0);
    const seq = this.seq;
    const rec = this.wai.eventSign(this.seed, unhex(this.room), BigInt(seq), unhex(this.prev), BigInt(t), META, payload);
    const v = JSON.parse(this.wai.eventRead(rec));
    this.seq++;
    this.prev = v.digest;
    this.lastT = t;
    return { rec, seq, digest: v.digest, control };
  }

  /** When a control asked for `at` is signed now: its `t`, and its instant, `at` or, if the
   * policy's lead (§5.3) no longer allows that, the earliest instant it does. */
  controlTime(at) {
    const t = Math.max(this.now(), this.lastT, this.joinT || 0);
    return { t, at: Math.max(at, t + this.policy.lead_ms * 1000 + 50000) };
  }

  /** A control signed now, at the instant `when` (`controlTime`) gives. */
  signControl(c, when = this.controlTime(c.at)) {
    const { t, at } = when;
    return this.event(t, controlPayload({ ...c, at }), { ...c, at });
  }

  /** Ask the room to `action` (ACTION) with `arg`, at the policy's lead from now (§5.3). */
  control(action, arg = 0) {
    const e = this.signControl({ action, at: 0, arg });
    this.outbox.push(e);
    this.flush();
    return e.control.at;
  }

  /** The outbox's first `n` events carried: the room's copy of the chain ends with them. */
  carry(n) {
    if (!n) return;
    const last = this.outbox[n - 1];
    this.carried = { seq: last.seq + 1, prev: last.digest };
    this.outbox.splice(0, n);
    this.stats.posted += n;
  }

  /** After entering again. Every event the room has not carried was signed before its new
   * `join`, so the room would refuse it as reversed in time: the chain goes on from where the
   * room's copy ends, at or after the join. The room holds none of those events (it carries
   * no event it refuses), so signing their positions again leaves it one event at each
   * (co-watch §10). A control is signed again, at its instant if the policy's lead still
   * allows it, else at the earliest it does, unless that moves it to or past the instant of
   * another viewer's control the page holds at or after its own: controls apply in the order
   * of their instants (§5.4), and the earlier press would then override the later one, so it
   * is dropped, as a press the room never had. A presentation statement of an instant before
   * the page was present again is dropped, and the page states anew within half a slot. */
  resign() {
    const pending = this.outbox;
    this.outbox = [];
    this.seq = this.carried.seq;
    this.prev = this.carried.prev;
    let resigned = 0;
    for (const e of pending) {
      if (!e.control) continue;
      const c = e.control;
      const when = this.controlTime(c.at);
      const passed = this.controls.find((x) => x.author !== this.principal && x.at >= c.at && x.at <= when.at);
      if (passed) {
        this.log(`co-watch: a ${ACTION_NAME[c.action]} at ${c.at}, signed while the room had taken this page for gone, is dropped: moved to ${when.at}, it would apply after another viewer's ${ACTION_NAME[passed.action]} at ${passed.at}`);
        continue;
      }
      const again = this.signControl(c, when);
      this.outbox.push(again);
      if (again.control.at !== c.at) this.log(`co-watch: a ${ACTION_NAME[c.action]} moved from ${c.at} to ${again.control.at}, the earliest the lead allows`);
      resigned++;
    }
    this.stats.rejoins++;
    this.stats.resigned += resigned;
    this.stats.dropped += pending.length - resigned;
    this.log(`co-watch: entered again; ${resigned} control(s) signed again, ${pending.length - resigned} earlier event(s) dropped`);
  }

  /** Enter again, the room having taken this page for gone, and sign again what it had not
   * carried (`resign`). Whether it entered: a join the service could not make now (the clock
   * could not sign it) leaves the outbox as it is, for the next post to try again. */
  async enterAgain() {
    try {
      await this.join();
    } catch (e) {
      this.log(`co-watch: entering again failed (${e.message}); the next post tries again`);
      return false;
    }
    this.resign();
    return true;
  }

  async flush() {
    if (this.posting || !this.outbox.length) return;
    this.posting = true;
    try {
      while (this.outbox.length) {
        const batch = this.outbox.slice(0, POST_EVENTS);
        const r = await this.api("/watch/post", { room: this.room, events: batch.map((e) => hex(e.rec)) });
        if (r.status === 200) {
          this.carry(batch.length);
          this.reentered = 0;
        } else if (r.status === 409 && r.body && r.body.error === "not_present" && this.reentered < REENTRIES) {
          // The room took this page for gone (it was silent too long, as a page on a loaded
          // machine can be): it enters again, and its chain goes on from what the room carried.
          // Taken for gone again before anything of it was carried (a second stall, right
          // after its join), it enters again too, a few times at most.
          this.carry(r.body.carried || 0);
          this.stats.gone++;
          if (await this.enterAgain()) this.reentered++;
          else await sleep(500);
        } else if (r.status === 409) {
          // Those before it were carried; this one never will be, and the chain cannot go
          // on past it: the page stops stating rather than leave a gap it cannot fill.
          const why = r.body.error;
          this.stats.post_refused[why] = (this.stats.post_refused[why] || 0) + 1;
          this.carry(r.body.carried || 0);
          this.outbox.shift();
          this.stop(`the room refused an event: ${why}`);
          break;
        } else {
          await sleep(500);
        }
      }
    } finally {
      this.posting = false;
    }
  }

  stop(why) {
    this.stopped = why;
    this.stats.errors.push(why);
    this.log(`co-watch: ${why}`);
  }

  /** Follow the room's log, each record checked before anything of it is applied. */
  async follow(alive) {
    this.stats.followed_at = this.now();
    while (alive() && !this.stopped) {
      let r;
      try { r = await this.api(`/watch?room=${this.room}&from=${this.at}`); } catch { await sleep(1000); continue; }
      if (r.status === 410) { this.stop("the room was replaced"); break; }
      if (r.status !== 200 || !r.body) { await sleep(1000); continue; }
      let changed = false;
      for (const e of r.body.entries) {
        if (e.seq !== this.at) { this.stop(`the log skipped from ${this.at} to ${e.seq}`); return; }
        changed = (await this.onRecord(unhex(e.event))) || changed;
        this.at++;
        this.stats.log++;
      }
      if (changed) this.timeline = new Timeline(this.policyList, this.controls);
      this.onChange();
    }
  }

  refuse(why) {
    this.stats.refused[why] = (this.stats.refused[why] || 0) + 1;
  }

  async onRecord(bytes) {
    const v = JSON.parse(this.wai.eventRead(bytes));
    if (!v.ok) { this.refuse(v.reason); return false; }
    if (v.room !== this.room) { this.refuse("wrong_room"); return false; }
    if (v.kind !== META) return false;
    const s = readStatement(readEvent(bytes).payload);
    if (!s) return false;
    if (s.kind === "malformed") { this.refuse("cw_statement_malformed"); return false; }
    if (s.kind === "control") {
      this.stats.controls++;
      const why = controlRefusal(this.policyList, v.t, s, (a) => this.mayControl(v.principal, a));
      if (why) { this.refuse(why); return false; }
      this.controls.push({ at: s.at, t: v.t, digest: v.digest, action: s.action, arg: s.arg, author: v.principal });
      return true;
    }
    if (s.kind === "skew") {
      this.skews.push(s);
      if (this.skews.length > 64) this.skews.shift();
      return false;
    }
    if (s.player !== v.principal) { this.refuse("presentation_wrong_key"); return false; }
    this.others[v.principal] = { at: s.at, position: s.position, error: s.position - this.timeline.positionAt(s.at), state: s.state };
    return false;
  }

  /** Keep the media at the controlled position (§9) and state what it presents (§7.3).
   *
   * Each pass of the loop states, if half a slot has passed since its last statement, then
   * waits PASS_MS. Two statements are so at most half a slot and the gap between two passes
   * apart, and while no such gap is over half a slot every slot holds one: a slot is left empty
   * only if no pass began in its second half. The page records each gap between two passes
   * over half a slot (`stalls`): `{ from, to }`, and how long the wait and the signing in it
   * took (`timer_us`, `signing_us`), all on its estimate of the room clock. The rest of the gap
   * is the loop's own work. A wait or a signing that took long is the page's thread held, by
   * the page's own work or by its machine not running it; the record does not tell which. */
  async run(alive) {
    const per = (this.policy.period_ms * 1000) / STATES_PER_SLOT;
    let next = 0;
    let last = null; // the room clock at the previous pass
    let timer = 0; // how long its wait and its signing took, on the room clock (µs)
    let signing = 0;
    while (alive() && !this.stopped) {
      const x = this.now();
      if (last != null && x - last > per) {
        this.stats.stalls.push({ from: last, to: x, timer_us: timer, signing_us: signing });
        if (this.stats.stalls.length > STALLS_KEPT) this.stats.stalls.shift();
      }
      last = x;
      signing = 0;
      const st = this.timeline.stateAt(x);
      const duration = this.timeline.durationAt(x);
      const controlled = this.timeline.positionAt(x);
      const presented = Math.round(this.media.currentTime * 1e6);
      if (st) {
        if (st.playing && controlled < duration && this.media.paused) this.media.play().catch(() => {});
        if ((!st.playing || controlled >= duration) && !this.media.paused) this.media.pause();
        const c = catchUp(this.policy, st.playing, st.rate, controlled, duration, presented);
        if (c.action === "seek") { this.media.currentTime = c.value / 1e6; this.stats.seeks++; }
        else if (c.action === "rate") {
          if (c.value !== st.rate) this.stats.nudges++;
          if (Math.abs(this.media.playbackRate - c.value / 1e6) > 1e-6) this.media.playbackRate = c.value / 1e6;
        }
        this.error = presented - controlled;
        this.controlled = controlled;
      }
      if (x >= next && this.key) {
        next = x + per;
        const ended = this.media.ended || (st && controlled >= duration);
        const state = this.media.readyState < 3 && !this.media.paused ? STATE.stalled : ended ? STATE.ended : this.media.paused ? STATE.paused : STATE.playing;
        const pos = Math.round(this.media.currentTime * 1e6);
        const body = { room: this.room, player: this.principal, source: this.policy.source, at: x, position: pos, state, rate: Math.max(1, Math.round(this.media.playbackRate * 1e6)) };
        const entries = this.stats.rejoins;
        const s0 = this.now();
        const signed = await signPresentation(body, this.key.privateKey);
        signing = this.now() - s0;
        if (this.stats.rejoins !== entries) {
          // The page entered again while it signed (`flush`): `x` is an instant from before its
          // present join, which it states nothing of. It states anew at once.
          this.stats.dropped++;
          next = 0;
        } else {
          this.outbox.push(this.event(x, signed));
          this.stats.statements++;
          this.flush();
        }
      }
      this.onChange();
      const w = this.now();
      await this.wait(PASS_MS);
      timer = this.now() - w;
    }
  }

  /** The loop's wait between two passes. */
  wait(ms) {
    return sleep(ms);
  }

  /** What a test reads, and what the page shows. */
  snapshot() {
    const last = this.skews.length ? this.skews[this.skews.length - 1] : null;
    const lastWindow = last ? this.skews.filter((s) => s.from === last.from) : [];
    const wall = Date.now();
    return {
      principal: this.principal, joined: this.joinT, at: this.at, offset_us: this.offset, room_now: this.roomTime(wall), wall_ms: wall, pass_ms: PASS_MS, controls: this.controls.length, outbox: this.outbox.length,
      overridden: this.controls.filter((c) => this.timeline.overridden.has(c.digest)).length,
      controlled: this.controlled ?? null, error_us: this.error ?? null,
      playing: !this.media.paused, rate: this.media.playbackRate, current_time: this.media.currentTime,
      skews: this.skews.length, last_window: lastWindow.length ? { from: lastWindow[0].from, to: lastWindow[0].to, bound_us: lastWindow[0].bound_us, viewers: lastWindow.flatMap((s) => s.viewers), pairs: lastWindow.flatMap((s) => s.pairs) } : null,
      others: this.others, stopped: this.stopped, ...this.stats,
    };
  }
}