513 lines
23 KiB
JavaScript
513 lines
23 KiB
JavaScript
// 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.
|
|
//
|
|
// 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;
|
|
|
|
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);
|
|
this.lastT = 0;
|
|
this.joinT = null;
|
|
this.outbox = [];
|
|
this.stopped = null;
|
|
this.skews = [];
|
|
this.others = {}; // principal -> its last statement: at, position, error
|
|
this.stats = { log: 0, controls: 0, applied: 0, refused: {}, statements: 0, posted: 0, post_refused: {}, seeks: 0, nudges: 0, verified_source: false, errors: [] };
|
|
this.onChange = o.onChange || (() => {});
|
|
}
|
|
|
|
/** The room clock now, on this page's estimate (§7.3). */
|
|
now() {
|
|
return Math.max(0, Math.round(Date.now() * 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`. */
|
|
event(t, payload) {
|
|
t = Math.max(t, this.lastT, this.joinT || 0);
|
|
const rec = this.wai.eventSign(this.seed, unhex(this.room), BigInt(this.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;
|
|
}
|
|
|
|
/** Ask the room to `action` (ACTION) with `arg`, at the policy's lead from now (§5.3). */
|
|
control(action, arg = 0) {
|
|
const t = this.now();
|
|
const at = t + this.policy.lead_ms * 1000 + 50000;
|
|
this.outbox.push(this.event(t, controlPayload({ action, at, arg })));
|
|
this.flush();
|
|
return at;
|
|
}
|
|
|
|
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(hex) });
|
|
if (r.status === 200) {
|
|
this.outbox.splice(0, batch.length);
|
|
this.stats.posted += batch.length;
|
|
this.rejoined = false;
|
|
} else if (r.status === 409 && r.body && r.body.error === "not_present" && !this.rejoined) {
|
|
// The room took this page for gone (it was silent too long): it enters again, and
|
|
// its chain goes on where it was.
|
|
this.outbox.splice(0, r.body.carried || 0);
|
|
this.rejoined = true;
|
|
await this.join();
|
|
} 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.outbox.splice(0, (r.body.carried || 0) + 1);
|
|
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) {
|
|
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). */
|
|
async run(alive) {
|
|
const per = (this.policy.period_ms * 1000) / STATES_PER_SLOT;
|
|
let next = 0;
|
|
while (alive() && !this.stopped) {
|
|
const x = this.now();
|
|
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)) };
|
|
this.outbox.push(this.event(x, await signPresentation(body, this.key.privateKey)));
|
|
this.stats.statements++;
|
|
this.flush();
|
|
}
|
|
this.onChange();
|
|
await sleep(50);
|
|
}
|
|
}
|
|
|
|
/** 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) : [];
|
|
return {
|
|
joined: this.joinT, at: this.at, offset_us: this.offset, room_now: this.now(), controls: this.controls.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,
|
|
};
|
|
}
|
|
}
|