beb4679dc6
Merged last, as asked. Conflicts, all adjacent additions, both sides kept: - joule-moq-room: Cargo.toml's wai features (live_world, remote_desktop and live_messaging); lib.rs's modules (messaging, msgstore, remote), Config's remote, messaging and call, Service's remote, messaging and calls, the epoch through remote::epoch with the conversation and calls opened beside it, and settings()'s refusals (#482's world room, #493's conversation and call); the README's world, conversation and call sections. - joule-moq-web: app.js asks a remote room's grant as #489 does and takes a given grant as #493 does (a remote page's fallback to viewer only when it asked); run.sh runs room.e2e.mjs (now with --remote) and messaging.e2e.mjs, each its own --only scenarios; the usage names every scenario. - workflows: joule-moq-room.yml's and joule-moq-web.yml's accounts; wai-conformance.yml's WAI_FEATURES gains live_messaging (#477's union) and the first step's test build names live_messaging beside the train's targets. - wai-live-web's wai features (stream_termination, live_world, live_messaging), wai-rs lib.rs's modules, registry.rs's capability list, REGISTRY.md's rows; CHANGELOG, #493's entry on top. - capabilities.json and v2.ttl regenerated (registry export, then --check). Co-Authored-By: Claude Opus 5.5 <[email protected]>
2288 lines
116 KiB
JavaScript
2288 lines
116 KiB
JavaScript
// The room client: a participant publishes its camera and microphone through the relay
|
||
// and subscribes to everyone else; the room's clock (a room key) publishes the lifecycle.
|
||
//
|
||
// Two ways in:
|
||
// - a room service (joule-moq-room): the page makes its own Ed25519 key, asks the
|
||
// service for a grant for it (a publisher's or a viewer's), and presents the grant with
|
||
// a proof by that key (WAI live-rooms §10.5); the service's clock authors the lifecycle.
|
||
// The service's base URL is the `service` parameter, or `data-service` on #room-app;
|
||
// the page enters when `enter("publisher" | "viewer")` is called (the #join and #watch
|
||
// buttons). The relay is the WebSocket at `/moq` on the page's own origin unless `ws`
|
||
// names another, and WebTransport is not tried;
|
||
// - a test room (`fixture`): every key from the fixture, and a page with `role=clock`
|
||
// authors the lifecycle, driven by the test.
|
||
//
|
||
// Query parameters:
|
||
// service=<url> the room service, as above
|
||
// fixture=<url> the test room (descriptor, grants, composition descriptor, keys)
|
||
// who=<name> this participant, by name; `role=clock` for the room's clock
|
||
// wt=<url> the relay's WebTransport URL (https://127.0.0.1:<port>/moq)
|
||
// ws=<url> its WebSocket URL (ws://127.0.0.1:<port>/moq)
|
||
// hashes=<hex,...> serverCertificateHashes for a relay with a development certificate
|
||
// transport=<t> `websocket` forces the WebSocket binding; otherwise WebTransport is
|
||
// tried first and the binding is the fallback
|
||
// fetch=1 a late joiner: earlier groups of every media track by FETCH
|
||
// (the "Seek back" control, #seek, fetches a participant's last seconds from the relay's
|
||
// DVR window, verifies every object, plays the video from its first key frame in a view of
|
||
// its own, and returns to live; `roomTest.seekBack(seconds)` does the same)
|
||
// audioFrameMs=<n> the Opus frame duration, 20 by default
|
||
// issuer_directory=<url> where Privacy Pass issuers' directories are (RFC 9578 §4): an
|
||
// issuer's at <url>/<issuer name>/.well-known/private-token-issuer-directory;
|
||
// by default at https://<issuer name>/.well-known/…
|
||
// age_threshold=<n> the threshold of the token to obtain, in a room that admits by age; by
|
||
// default the least the room's challenges offer at or above its age
|
||
// rate=<n>, rate_scheme=<uri>, rate_value=<v> a publisher rates its own video from its
|
||
// first group: n years under the scheme, with that value (age-assurance §3.6)
|
||
// debug=binding the state of the frame each tile shows (WAI browser-sink §10): its
|
||
// binding, the receipts it verified under, its route and energy entry
|
||
// unseen_grace_ms=<n>, idle_ms=<n>, autoplay_grace_ms=<n> the termination policy of each
|
||
// stream this page receives (WAI stream-termination §4.3, ./termination.js):
|
||
// a video stream hidden or out of view this long is terminated with a signed
|
||
// record, its subscription ended and its audio kept; 10000, 0 (never) and 0
|
||
// by default
|
||
// autoplay=off the viewer's autoplay setting (§4.4), recorded in each stream's log: media
|
||
// is subscribed only after the viewer's first activation; `autoplay` in this
|
||
// page's storage keeps it across visits
|
||
//
|
||
// The endpoints among them (`service`, `ws`, `wt`, `fixture`, `issuer_directory`), and the URI
|
||
// a relay's GOAWAY moves the session to, are taken only on the page's own origin
|
||
// (`https://<host>`, or `wss://<host>` for a relay); a page served from the loopback interface
|
||
// (a fixture page, the end-to-end harness) takes loopback addresses, and `hashes` only there
|
||
// (./endpoint.js). A link that names another origin is refused and shown, and the page
|
||
// connects to nothing: otherwise a visitor who joined from it would send their camera and
|
||
// microphone there.
|
||
//
|
||
// A room that admits by age (WAI age-assurance §5.6; `GET /room`'s `age`): the service answers
|
||
// a viewer's grant request `401` with Privacy Pass challenges (RFC 9577); the page obtains a
|
||
// token from the issuer by blind issuance (RFC 9578 §6, ./privacy-pass.js) and asks again with
|
||
// it, and the grant states the threshold it was admitted at. A publisher carries the room's
|
||
// rating on every track it publishes (§5.7). A track the relay holds above the page's
|
||
// threshold is refused `below_threshold`, and the page says so (§5.5).
|
||
//
|
||
// An end-to-end encrypted room (WAI confidentiality): the room service names the room's
|
||
// confidentiality record (`GET /room`'s `e2ee`). The page checks it, loads the build of the
|
||
// codec with the room's MLS member (`joule_moq_web_e2ee`), leaves a KeyPackage, and follows
|
||
// the room's log (./e2ee.js). It seals each encoded frame before it is signed and published,
|
||
// and opens every object it receives through the receiving checks in their order before
|
||
// any decoder sees it; an object that does not open is counted with its reason and never
|
||
// decoded. Such a room has no seeking back: the relay keeps nothing of it.
|
||
//
|
||
// A room with co-performance (WAI co-performance): the room service names its policy and its
|
||
// audience's grants (`GET /room`'s `co`). The page shows the turn schedule and who holds the
|
||
// floor, each participant's intents as upcoming lines, the selection window open now with an
|
||
// input box for a viewer whose grant allows it, each selection with its credits, and the
|
||
// clock's fidelity statement (./co.js). It states its media's timestamps on the room clock, so
|
||
// that an object's due instant is its capture instant there, and sends media only while it
|
||
// holds the floor. A lead asks the room's clock for schedules, handovers and windows, its
|
||
// grant presented with a proof over the service's challenge (live-rooms §10.5.3).
|
||
// ai=scripted a demo AI participant: the page enters as a publisher whose grant marks
|
||
// it an AI participant, publishes a drawn picture and a tone instead of a
|
||
// camera and a microphone, and announces a line ahead of each it says, as
|
||
// a caption a machine made; `script=off` leaves the lines to a test
|
||
// A watch room (WAI co-watch): the room service names its policy and its source (`GET /room`'s
|
||
// `watch`). The page fetches the source, checks its bytes by the content hash the policy
|
||
// names, enters the room's co-watch log by a proof of its key over the service's challenge,
|
||
// applies each control at its instant, keeps the source at the controlled position by the
|
||
// catch-up rule, states what it presents twice a slot, and shows the room's sync (./cowatch.js).
|
||
// The page's events in such a room are its co-watch statements: it sends no caption or chat.
|
||
// A remote desktop room (WAI remote-desktop): the room service names its host and its rules
|
||
// (`GET /room`'s `remote`). The page publishes no media in it: #join asks for a controller's grant
|
||
// (the service's allow rule decides; refused, the page asks for a viewer's) and #watch for a
|
||
// viewer's, each for the key the page made. The page draws the host's screen, every frame
|
||
// verified against its receipt first, and sends its pointer, keys, composed and pasted text as
|
||
// input statements under its grant in effect, once the clock has recorded it joining (./remote.js);
|
||
// a grant the host exchanges for it, to give or take back control, is read again from the service.
|
||
//
|
||
// Everything a test reads is on `window.roomTest`.
|
||
|
||
import initWai, * as wai from "../pkg/wai_live_web.js";
|
||
import { E2ee } from "./e2ee.js";
|
||
import { CoWatch, ACTION } from "./cowatch.js";
|
||
import { RemoteViewer, keyFromSeed, readEvent } from "./remote.js";
|
||
import { connect } from "../moq/transport.js";
|
||
import { MoqSession, RequestRefused, withTimeout } from "../moq/session.js";
|
||
import * as media from "../moq/media.js";
|
||
import { Compositor } from "./compose.js";
|
||
import * as pp from "./privacy-pass.js";
|
||
import { Co, wireControls } from "./co.js";
|
||
import { SideTable, buildReport, askUserAgent, chainToken, stateOf, renderStates, metadataProbe, installFramePatches } from "./binding.js";
|
||
import { checkEndpoint, isLoopback, kindOf } from "./endpoint.js";
|
||
import * as term from "./termination.js";
|
||
|
||
// The codec's wasm: the plain build, or, for an end-to-end encrypted room, the build with
|
||
// the member of the room's MLS group, loaded for such a room alone.
|
||
let moq = null;
|
||
async function loadMoq(sealed) {
|
||
if (moq) return moq;
|
||
const m = await import(sealed ? "../pkg/joule_moq_web_e2ee.js" : "../pkg/joule_moq_web.js");
|
||
await m.default();
|
||
moq = m;
|
||
return m;
|
||
}
|
||
|
||
const q = new URLSearchParams(location.search);
|
||
|
||
// What each endpoint the query string may name is, for the page's words.
|
||
const ENDPOINT_NAMES = { service: "room service", ws: "relay", wt: "relay", fixture: "test room", issuer_directory: "Privacy Pass issuer directory" };
|
||
// The endpoint the query string names as `name`, held to ./endpoint.js; `fallback` when it
|
||
// names none. One it names and the page refuses is recorded in `state.refused` and shown, and
|
||
// gives null: the caller then connects to nothing.
|
||
function queryEndpoint(name, kind, fallback = null) {
|
||
const v = q.get(name);
|
||
if (v == null || v === "") return fallback;
|
||
const c = checkEndpoint(v, kind, location.href);
|
||
if (c.ok) return c.url;
|
||
state.refused.push({ name, value: v, reason: c.reason });
|
||
log(`refused the ${name} the link names: ${v} (${c.reason})`);
|
||
status(`This link names a ${ENDPOINT_NAMES[name] || name} that is not this page's own (${v}: ${c.reason}). The page does not use it, and connects to nothing.`);
|
||
return null;
|
||
}
|
||
|
||
// The browser sink (WAI browser-sink): each subscribed track is routed before it is decoded
|
||
// (§4), every frame a video decoder outputs is bound to the object it came from in a side
|
||
// table (§7, §9.2), and with `debug=binding` the page shows the state of the frame each
|
||
// tile last drew (§10).
|
||
const DEBUG_BINDING = q.get("debug") === "binding";
|
||
const bindings = new SideTable();
|
||
installFramePatches();
|
||
const $ = (id) => document.getElementById(id);
|
||
const hex = (b) => Array.from(b, (x) => x.toString(16).padStart(2, "0")).join("");
|
||
const unhex = (h) => new Uint8Array((h.match(/../g) || []).map((x) => parseInt(x, 16)));
|
||
const enc = new TextEncoder();
|
||
const dec = new TextDecoder();
|
||
|
||
const logLines = [];
|
||
function log(m) {
|
||
const line = `${(performance.now() / 1000).toFixed(3)} ${m}`;
|
||
logLines.push(line);
|
||
if (logLines.length > 2000) logLines.shift();
|
||
const el = $("log");
|
||
if (el) { el.textContent += line + "\n"; el.scrollTop = el.scrollHeight; }
|
||
console.log(line);
|
||
}
|
||
|
||
// Track Aliases this session's PUBLISHes give (§9.8): the publisher's for its own data.
|
||
const ALIAS = { catalog: 1, events: 2, audio: 10, audioReceipts: 11, video: 20, videoReceipts: 21 };
|
||
const RECEIPT_ALIAS = { audio: 1, video: 2 }; // the catalog's wai_receipt_alias
|
||
// Publisher priorities (MOQT §5.1.1, lower first): audio before video, the receipts
|
||
// tracks at their media track's, events before both, the catalog first.
|
||
const PRIORITY = { catalog: 0, events: 16, audio: 64, video: 96 };
|
||
// Subscriber priorities (§5.1.2) a participant gives its requests.
|
||
const SUB_PRIORITY = { catalog: 0, events: 8, audio: 32, video: 64, screen: 48 };
|
||
const VIDEO_TIMEOUT_MS = 2000; // OBJECT_DELIVERY_TIMEOUT on each video object
|
||
const AUDIO_TIMEOUT_MS = 1000;
|
||
const RECEIPT_WAIT_MS = 6000; // an object waits this long for its receipts
|
||
const RECOVER_AFTER_MS = 1000; // then, past this, its group of the receipts track is fetched
|
||
const MISSING_AFTER_MS = 4000; // receipts with no object after this: accounted by inclusion
|
||
|
||
const state = {
|
||
phase: "loading",
|
||
errors: [],
|
||
// Endpoints the query string named and the page refused (./endpoint.js).
|
||
refused: [],
|
||
transport: null,
|
||
published: { audio: 0, video: 0, events: 0, bytes: 0 },
|
||
publishCodec: {},
|
||
subs: {}, // principal -> per-track tallies
|
||
timeline: [],
|
||
captions: [],
|
||
goaways: [],
|
||
sessions: 0,
|
||
mode: null,
|
||
ended: null,
|
||
// Seeking back into the relay's DVR window: "live", "seeking" (fetching and verifying)
|
||
// or "playing", and each run's counts.
|
||
dvr: { mode: "live", runs: [] },
|
||
// An end-to-end encrypted room: the record's suite, and what this page sealed.
|
||
e2ee: null,
|
||
// A room that admits by age: its age, the threshold this page was admitted at, the token
|
||
// it presented (base64url; spent), how the service answered, and the tracks withheld.
|
||
age: null,
|
||
// The browser sink: each track's routing entry and the codec strings put to the user
|
||
// agent; what the user agent keeps of an unregistered metadata member.
|
||
binding: { routes: {}, asked: [], metadata: null },
|
||
// Stream termination (WAI stream-termination): the panel this page declared, whether its
|
||
// refresh rate was measured, the policy, and every stream log it signed.
|
||
termination: { panel: null, refresh_measured: false, policy: null, autoplay: "on", streams: [] },
|
||
};
|
||
window.roomTest = { state, logLines };
|
||
|
||
function fail(e) {
|
||
const m = (e && e.stack) || String(e);
|
||
state.errors.push(m);
|
||
log(`error: ${m}`);
|
||
status(`Something went wrong: ${(e && e.message) || e}`);
|
||
}
|
||
|
||
// What the page tells its reader, where it has a place for it.
|
||
function status(text) {
|
||
const el = $("status");
|
||
if (el) el.textContent = text;
|
||
}
|
||
window.addEventListener("error", (e) => fail(e.error || e.message));
|
||
window.addEventListener("unhandledrejection", (e) => fail(e.reason));
|
||
|
||
// -------------------------------------------------------------------------- the events track
|
||
|
||
// One principal's events track (live-events §8.1): an MSF event timeline whose groups
|
||
// are the room-clock minutes. Object 0 of a group is its join set (§8.2): here every
|
||
// earlier record of the chain (any other event MAY be repeated, and a late subscriber
|
||
// then holds the whole chain); each later object is an array of one record.
|
||
class EventsTrack {
|
||
constructor(seed, roomId, t0, clock) {
|
||
this.clock = clock;
|
||
this.seed = seed;
|
||
this.roomId = roomId;
|
||
this.t0 = t0;
|
||
this.seq = 0;
|
||
this.prev = new Uint8Array(32);
|
||
this.lastT = 0;
|
||
this.records = [];
|
||
this.group = null;
|
||
this.stream = null;
|
||
this.writer = null;
|
||
this.nextObject = 0;
|
||
this.session = null;
|
||
}
|
||
|
||
now() {
|
||
const t = Math.max(this.lastT, this.clock());
|
||
this.lastT = t;
|
||
return t;
|
||
}
|
||
|
||
// A record signed as this chain's next and sent, but not kept: the chain goes on as if it
|
||
// had never been signed. For a test that the relay withholds what it refuses.
|
||
async detached(kind, payload) {
|
||
const t = this.now();
|
||
const rec = wai.eventSign(this.seed, this.roomId, BigInt(this.seq), this.prev, BigInt(t), kind, payload);
|
||
await this.send(rec, t);
|
||
return JSON.parse(wai.eventRead(rec));
|
||
}
|
||
|
||
async attach(session) {
|
||
this.session = session;
|
||
this.group = null; // a new session opens a new stream for the current group
|
||
}
|
||
|
||
async emit(kind, payload) {
|
||
const t = this.now();
|
||
const rec = wai.eventSign(this.seed, this.roomId, BigInt(this.seq), this.prev, BigInt(t), kind, payload);
|
||
const v = JSON.parse(wai.eventRead(rec));
|
||
this.prev = unhex(v.digest);
|
||
this.seq++;
|
||
await this.send(rec, t);
|
||
this.records.push(rec);
|
||
return { record: rec, view: v };
|
||
}
|
||
|
||
async send(rec, t) {
|
||
const s = this.session;
|
||
if (!s) return;
|
||
const g = Number(wai.moqGroup(BigInt(t)));
|
||
const msf = (r) => JSON.parse(wai.msfRecord(r, BigInt(this.t0)));
|
||
if (this.group !== g) {
|
||
if (this.stream) await this.stream.close();
|
||
this.group = g;
|
||
this.writer = new moq.SubgroupWriter(BigInt(ALIAS.events), BigInt(g), PRIORITY.events, false);
|
||
this.stream = await s.openStream(0);
|
||
await this.stream.write(this.writer.header());
|
||
await this.stream.write(this.writer.object(0n, enc.encode(JSON.stringify(this.records.map(msf)))));
|
||
this.nextObject = 1;
|
||
}
|
||
await this.stream.write(this.writer.object(BigInt(this.nextObject++), enc.encode(JSON.stringify([msf(rec)]))));
|
||
state.published.events++;
|
||
}
|
||
}
|
||
|
||
// --------------------------------------------------------------------------------- room
|
||
|
||
let fx, me, roomId, t0, room, compositor, session, transportOpts, eventsTrack, service, e2ee, co, cowatch, remote;
|
||
// Where Privacy Pass issuers' directories are, when the link names it (`issuer_directory`).
|
||
let issuerDirectory = null;
|
||
// The service's clock less this page's, in µs, from `GET /room`'s `now_us`: room-clock instants
|
||
// this page states are the service's.
|
||
let skew = 0;
|
||
const subscribed = new Map(); // principal -> subscription state
|
||
const signers = {};
|
||
const published = []; // the PUBLISH requests held open
|
||
|
||
function roomNow() {
|
||
return Math.max(0, Math.round(Date.now() * 1000 + skew - t0));
|
||
}
|
||
|
||
// A principal as the page names it: the clock, this page, a fixture's participant, or a
|
||
// publisher by its grant's `order` attribute.
|
||
function nameOf(P) {
|
||
if (fx.clock && P === fx.clock.principal) return fx.participants.length ? "clock" : "room";
|
||
if (me && P === me.principal) return fx.participants.length ? me.name : "you";
|
||
const p = fx.participants.find((x) => x.principal === P);
|
||
if (p) return p.name;
|
||
const g = (state.grants || []).find((x) => x.principal === P);
|
||
const order = g && (g.attrs.find((a) => a[0] === "order") || [])[1];
|
||
if (order != null) return `${isAi(P) ? "AI participant" : "participant"} ${Number(order) + 1}`;
|
||
if (g && g.attrs.some((a) => a[0] === "audience_input")) return `audience member ${P.slice(0, 4)}`;
|
||
return P.slice(0, 8);
|
||
}
|
||
|
||
// Whether `P`'s grant marks it an AI participant (live-rooms §4.4, `ai`).
|
||
function isAi(P) {
|
||
const g = (state.grants || []).find((x) => x.principal === P);
|
||
return !!g && g.attrs.some((a) => a[0] === "ai");
|
||
}
|
||
|
||
// The room's grants as the page holds them, read.
|
||
function readGrants() {
|
||
const gv = JSON.parse(room.grants());
|
||
if (gv.outcome !== "read") throw new Error(`the room's grants: ${gv.reason}`);
|
||
state.grants = gv.grants;
|
||
}
|
||
|
||
async function boot() {
|
||
await loadMoq(false);
|
||
await initWai();
|
||
const fixture = queryEndpoint("fixture", "http");
|
||
const wt = queryEndpoint("wt", "http");
|
||
const ws = queryEndpoint("ws", "ws");
|
||
if (state.refused.length || !fixture) {
|
||
state.phase = "refused";
|
||
if (!fixture && !state.refused.length) status("This page needs a room service or a test room.");
|
||
return;
|
||
}
|
||
fx = await (await fetch(fixture)).json();
|
||
const role = q.get("role") || "participant";
|
||
me = role === "clock" ? { name: "clock", ...fx.clock } : fx.participants.find((p) => p.name === q.get("who"));
|
||
if (!me) throw new Error(`no participant "${q.get("who")}" in the fixture`);
|
||
state.mode = "fixture";
|
||
transportOpts = {
|
||
webtransport: wt,
|
||
websocket: ws,
|
||
// Certificate hashes let a session trust a development certificate: a loopback page's alone.
|
||
hashes: isLoopback(location.hostname) ? (q.get("hashes") || "").split(",").filter(Boolean).map(unhex) : [],
|
||
force: q.get("transport") || "",
|
||
};
|
||
window.roomTest.role = role;
|
||
await begin(role);
|
||
}
|
||
|
||
// The room service's answer, or why there is none. `headers` are added to a request.
|
||
async function api(path, body, headers = {}) {
|
||
const r = await fetch(`${service}${path}`, body ? { method: "POST", headers: { "content-type": "application/json", ...headers }, body: JSON.stringify(body), cache: "no-store" } : { cache: "no-store" });
|
||
let j = null;
|
||
try { j = await r.json(); } catch { /* not JSON */ }
|
||
return { status: r.status, body: j, headers: r.headers };
|
||
}
|
||
|
||
// A Privacy Pass token for one of a room's challenges (WAI age-assurance §5.6): the least
|
||
// threshold at or above the room's age the challenges offer (or `age_threshold`), each
|
||
// challenge checked as RFC 9577 §2.1.3 has it, the token obtained from its issuer by RFC 9578
|
||
// §6. The challenges are the `401`'s `WWW-Authenticate` fields; the keys' thresholds are the
|
||
// room's policy's.
|
||
async function obtainToken(r, age) {
|
||
const offered = pp.parseChallenges(r.headers.get("www-authenticate"));
|
||
const policy = JSON.parse(age.policy);
|
||
const thresholdOf = (key) => {
|
||
const h = Array.from(key, (x) => x.toString(16).padStart(2, "0")).join("");
|
||
for (const i of policy.issuers) for (const k of i.keys) if (k.key === h) return k.threshold;
|
||
return null;
|
||
};
|
||
const want = q.get("age_threshold") ? Number(q.get("age_threshold")) : null;
|
||
const usable = offered
|
||
.map((o) => ({ ...o, c: pp.readChallenge(o.challenge), threshold: o.tokenKey && thresholdOf(o.tokenKey) }))
|
||
.filter((o) => pp.acceptable(o.c, location.host) && o.threshold != null && (want != null ? o.threshold === want : o.threshold >= age.age))
|
||
.sort((a, b) => a.threshold - b.threshold);
|
||
if (!usable.length) throw new Error(`the room admits at ${age.age} or above, and none of its ${offered.length} challenges is one this page can answer`);
|
||
const o = usable[0];
|
||
status(`This room is rated ${age.age}. Asking ${o.c.issuerName} for a token of ${o.threshold} or over…`);
|
||
const token = await pp.issue({ challenge: o.challenge, tokenKey: o.tokenKey, directory: pp.directoryOf(o.c.issuerName, issuerDirectory) });
|
||
return { token, threshold: o.threshold, issuer: o.c.issuerName };
|
||
}
|
||
|
||
const ROLE_OF = { publisher: "participant", viewer: "viewer" };
|
||
|
||
// Enter the room the service keeps: as a publisher (camera and microphone) or a viewer.
|
||
// `given`: a grant issued already, with the key it is for (`{ seed, answer }`, the answer in the
|
||
// form of `POST /grant`'s), as a call page hands over the grant its call came with (./call.js).
|
||
async function enter(role, given = null) {
|
||
if (state.mode) throw new Error("this page has entered already");
|
||
// The relay and the issuer directory a link may name, held to the page's own origin before
|
||
// anything is asked of the camera, the room service or a relay.
|
||
const ws = queryEndpoint("ws", "ws", `${location.protocol === "https:" ? "wss" : "ws"}://${location.host}/moq`);
|
||
issuerDirectory = queryEndpoint("issuer_directory", "http");
|
||
if (state.refused.length) {
|
||
state.phase = "refused";
|
||
return state.phase;
|
||
}
|
||
state.mode = "service";
|
||
state.phase = "entering";
|
||
const asked = Date.now();
|
||
let r = await api("/room");
|
||
if (r.status !== 200) throw new Error(`the room service answered ${r.status}`);
|
||
// The service's clock less this page's, where it differs by more than the request's own
|
||
// uncertainty (half its round trip): a page on the service's clock keeps its own.
|
||
if (r.body.now_us) {
|
||
const back = Date.now();
|
||
const est = Math.round(r.body.now_us - ((asked + back) / 2) * 1000);
|
||
if (Math.abs(est) > ((back - asked) / 2) * 1000 + 20_000) skew = est;
|
||
}
|
||
const sealed = !!r.body.e2ee;
|
||
// A remote desktop room: a controller's grant or a viewer's, and no camera.
|
||
const remoteRoom = !!r.body.remote;
|
||
const ai = role === "publisher" && !remoteRoom && q.get("ai") === "scripted" ? await scriptedModel() : null;
|
||
await loadMoq(sealed);
|
||
await initWai();
|
||
let stream = null;
|
||
if (role === "publisher" && !remoteRoom) {
|
||
// The camera and microphone first: a room place is asked for only once they are on.
|
||
const d = JSON.parse(wai.roomDescriptor(unhex(r.body.descriptor)));
|
||
if (ai) {
|
||
stream = syntheticStream(d);
|
||
} else {
|
||
status("Asking for your camera and microphone…");
|
||
stream = await navigator.mediaDevices.getUserMedia(constraintsOf(d));
|
||
}
|
||
}
|
||
// The key this page signs with: made here, and never sent anywhere.
|
||
const seed = given ? unhex(given.seed) : crypto.getRandomValues(new Uint8Array(32));
|
||
const principal = hex(moq.publicKeyOf(seed));
|
||
status("Asking the room for a grant…");
|
||
const age = r.body.age;
|
||
if (age) state.age = { age: age.age, ratings: age.ratings.length, admitted: null, issuer: null, token: null, answer: null, withheld: [] };
|
||
// An AI participant asks for a grant that says so (live-rooms §4.4, `ai`). In a remote desktop
|
||
// room the page asks to control, or to watch. A grant given with the page (a call's) is taken
|
||
// as it is.
|
||
const request = ai ? { principal, role, ai } : remoteRoom ? { principal, role: role === "publisher" ? "controller" : "viewer" } : { principal, role };
|
||
r = given ? { status: 200, body: given.answer } : await api("/grant", request);
|
||
if (!given && remoteRoom && r.status === 403 && request.role === "controller") {
|
||
log(`the room does not let this page control (${r.body && r.body.error}): it watches`);
|
||
request.role = "viewer";
|
||
r = await api("/grant", request);
|
||
}
|
||
if (r.status === 401 && age) {
|
||
// The room admits by age: a token, then the request again with it (RFC 9577 §2.2.2).
|
||
const t = await obtainToken(r, age);
|
||
state.age.token = pp.b64url(t.token);
|
||
state.age.issuer = t.issuer;
|
||
r = await api("/grant", request, { authorization: pp.authorization(t.token) });
|
||
if (r.status === 200) state.age.admitted = t.threshold;
|
||
} else if (age && r.status === 200) {
|
||
state.age.admitted = 0;
|
||
}
|
||
if (age) state.age.answer = r.status === 200 ? "admitted" : (r.body && r.body.error) || String(r.status);
|
||
if ((r.status === 403 || r.status === 401) && age) {
|
||
if (stream) for (const t of stream.getTracks()) t.stop();
|
||
state.mode = null;
|
||
state.phase = "refused";
|
||
status(`The room is rated ${age.age}, and did not admit this page: ${r.body && r.body.error}.`);
|
||
return state.phase;
|
||
}
|
||
if (r.status === 409 && role === "publisher" && !remoteRoom) {
|
||
for (const t of stream.getTracks()) t.stop();
|
||
state.mode = null;
|
||
state.phase = "full";
|
||
status(r.body && r.body.error === "full" ? `The room has its ${r.body.max_publishers} publishers. You can watch, or try again later.` : "The room is busy. You can watch, or try again later.");
|
||
return state.phase;
|
||
}
|
||
if (r.status === 429) throw new Error(`too many grants from this address; try again in ${r.body.retry_after_s} s`);
|
||
if (r.status !== 200) throw new Error(`the room service answered ${r.status}${r.body && r.body.error ? ` (${r.body.error})` : ""}`);
|
||
const v = r.body.room;
|
||
const d = JSON.parse(wai.roomDescriptor(unhex(v.descriptor)));
|
||
if (sealed !== !!v.e2ee) throw new Error("the room the grant is for is not end-to-end encrypted as the room was");
|
||
// A room with co-performance lists its audience's grants beside its publishers': a page
|
||
// verifies their inputs with them.
|
||
fx = { room: v.room, t0_us: v.t0_us, descriptor: v.descriptor, compose: v.compose, grants: v.co ? v.grants.concat(v.co.audience) : v.grants, clock: { principal: d.clock }, participants: [], co: v.co, remote: v.remote };
|
||
me = { name: "you", seed: hex(seed), principal, grant: r.body.grant, grant_digest: r.body.digest, role: remoteRoom ? r.body.role : role };
|
||
// The room's rating, which every track this page publishes carries (age-assurance §5.7).
|
||
me.ratings = (v.age && v.age.ratings) || [];
|
||
state.grant = { role, exp: r.body.exp, digest: r.body.digest };
|
||
if (sealed) {
|
||
// The room's confidentiality record (confidentiality §3.3), checked against the room
|
||
// the grant is for: its id, its authority and its recording mode.
|
||
const c = JSON.parse(moq.e2eeCheckRecord(unhex(v.e2ee.record), unhex(v.room), unhex(d.authority), d.recording_mode));
|
||
if (!c.ok) throw new Error(`the room's confidentiality record is refused: ${c.reason}`);
|
||
state.e2ee = { suite: c.suite, record: v.e2ee.record.slice(0, 16), tampered: [] };
|
||
e2ee = new E2ee({
|
||
moq, wai, api, log, room: v.room, clock: d.clock, seed, suite: c.suite, role,
|
||
grantOf: (digest) => (state.grants || []).find((g) => g.digest === digest),
|
||
refreshGrants: () => refreshGrants(),
|
||
onRecord: (bytes) => { records = records.then(() => onRecord(bytes)).catch(fail); },
|
||
onChange: onKeys,
|
||
});
|
||
if ($("dvr")) $("dvr").hidden = true;
|
||
status("Leaving a key package with the room…");
|
||
if (role === "publisher") await e2ee.keyPackage();
|
||
}
|
||
transportOpts = { webtransport: null, websocket: ws, hashes: [], force: "websocket" };
|
||
window.roomTest.role = remoteRoom ? "viewer" : ROLE_OF[role];
|
||
// A watch room's page enters its co-watch log first: the clock's `join` for it is then on the
|
||
// clock's events track, which the page subscribes to next, and which a room of viewers alone
|
||
// has no other record on yet.
|
||
if (v.watch) await startCoWatch(v, r.body, seed, principal);
|
||
// A remote desktop room's page publishes nothing but its events: it enters as a viewer does.
|
||
await begin(remoteRoom ? "viewer" : ROLE_OF[role], stream);
|
||
pollRoom();
|
||
if (cowatch) runCoWatch();
|
||
if (e2ee) {
|
||
e2ee.follow(() => state.phase !== "left").catch(fail);
|
||
status(role === "publisher" ? "You are in the room. It is end-to-end encrypted: your media is sealed before it leaves this page, and sent once the room's group has keyed you." : "Watching. This room is end-to-end encrypted, and a viewer is not a member of its group: it opens nothing.");
|
||
renderE2ee();
|
||
}
|
||
return state.phase;
|
||
}
|
||
|
||
function constraintsOf(d) {
|
||
const vp = d.profiles.find((p) => p.media === 2);
|
||
const ap = d.profiles.find((p) => p.media === 1);
|
||
const vl = vp.layers[0];
|
||
return {
|
||
audio: { channelCount: ap.channels, sampleRate: ap.sample_rate, echoCancellation: (ap.processing & 1) !== 0, noiseSuppression: (ap.processing & 2) !== 0, autoGainControl: (ap.processing & 4) !== 0 },
|
||
video: { width: vl.width, height: vl.height, frameRate: vl.fps },
|
||
};
|
||
}
|
||
|
||
// The room, the composition and the session, once the page knows who it is.
|
||
async function begin(role, stream) {
|
||
roomId = unhex(fx.room);
|
||
t0 = fx.t0_us;
|
||
const who = $("who");
|
||
if (who) who.textContent = `${me.name} · ${me.principal.slice(0, 12)}`;
|
||
room = new wai.Room(unhex(fx.descriptor), JSON.stringify(fx.grants));
|
||
readGrants();
|
||
state.descriptor = JSON.parse(wai.roomDescriptor(unhex(fx.descriptor)));
|
||
eventsTrack = new EventsTrack(unhex(me.seed), roomId, t0, roomNow);
|
||
if (fx.co && fx.co.policy) startCo();
|
||
// A remote desktop room (WAI remote-desktop) has no composition: its host's screen is drawn
|
||
// on a canvas of its own.
|
||
if (role !== "clock" && fx.compose) {
|
||
compositor = new Compositor(wai, $("canvas"), unhex(fx.compose), { log, bindingOf: (vf) => SideTable.of(vf) });
|
||
if (DEBUG_BINDING) setInterval(() => { try { renderStates($("binding-debug"), bindingRows()); } catch (e) { fail(e); } }, 500);
|
||
await compositor.startAudio(new URL("./worklet.js", import.meta.url));
|
||
await startTermination();
|
||
setInterval(() => { try { compositor.draw(); } catch (e) { fail(e); } }, Math.round(1000 / compositor.desc.params.fps));
|
||
}
|
||
if (role !== "clock") setInterval(sweepMissing, 1000);
|
||
status("Connecting to the relay…");
|
||
await openSession(transportOpts);
|
||
if (fx.remote && state.mode === "service") await startRemote();
|
||
if (role === "clock") {
|
||
window.roomTest.clockEmit = clockEmit;
|
||
state.phase = "ready";
|
||
} else if (role === "viewer") {
|
||
await subscribeEvents(fx.clock.principal);
|
||
state.phase = "watching";
|
||
status("Watching. Everything you receive is verified before it is shown.");
|
||
} else {
|
||
await startPublishing(stream);
|
||
await subscribeEvents(fx.clock.principal);
|
||
state.phase = "publishing";
|
||
status("You are in the room. Everything you receive is verified before it is shown.");
|
||
}
|
||
log(`ready as ${role} ${me.name}`);
|
||
}
|
||
|
||
async function openSession(opts) {
|
||
const t = await connect({ ...opts, log });
|
||
// A participant presents its room grant with a proof by its key (live-rooms §10.5.2);
|
||
// the clock, a room key with no grant, presents its own token.
|
||
const id = me.grant ? { grant: { record: unhex(me.grant), seed: unhex(me.seed) } } : { token: unhex(me.token) };
|
||
const s = await MoqSession.connect(t, moq, { ...id, implementation: "joule-moq-web/room", log, properties: !!e2ee });
|
||
state.transport = { kind: t.kind, protocol: t.protocol };
|
||
state.sessions++;
|
||
s.onGoaway = (g) => onGoaway(s, g);
|
||
session = s;
|
||
window.roomTest.session = s;
|
||
s.closed.then((i) => onClosed(s, i));
|
||
// The events track is published on every session.
|
||
const ns = [fx.room, me.principal, "events"];
|
||
published.push(await s.publish(ns, "0", ALIAS.events, { publisher_priority: PRIORITY.events }));
|
||
await eventsTrack.attach(s);
|
||
return s;
|
||
}
|
||
|
||
// The session ended: by this page, by a move after GOAWAY, or by the relay (its session
|
||
// limit, the grant's expiry, a refusal).
|
||
function onClosed(s, info) {
|
||
if (s !== session || state.goaways.some((g) => !g.moved && !g.refused)) return;
|
||
if (state.goaways.some((g) => g.refused)) return; // ended by this page, and said so
|
||
state.ended = { code: info && info.code, reason: info && info.reason };
|
||
if (state.phase === "left") return;
|
||
state.phase = "ended";
|
||
log(`the session ended: ${JSON.stringify(state.ended)}`);
|
||
if (state.mode === "service") status(`The session ended${info && info.reason ? `: ${info.reason}` : ""}. Reload the page to enter again.`);
|
||
}
|
||
|
||
// Leave: the session closes, and the relay ends this page's tracks, which the room's clock
|
||
// records as the publisher leaving.
|
||
// ------------------------------------------------------------------------------- co-watch
|
||
|
||
// Whether `principal` may control `action` (co-watch §5.2): the direct power, or the action's
|
||
// bit of its grant's `watch_control`.
|
||
function mayControl(principal, action) {
|
||
const g = (state.grants || []).find((x) => x.principal === principal);
|
||
if (!g) return false;
|
||
if (g.powers & 2) return true;
|
||
const a = g.attrs.find((x) => x[0] === "watch_control");
|
||
return !!a && /^[1-9][0-9]?$/.test(a[1]) && Number(a[1]) <= 15 && ((Number(a[1]) >> (action - 1)) & 1) === 1;
|
||
}
|
||
|
||
async function startCoWatch(v, granted, seed, principal) {
|
||
status("Fetching what the room watches…");
|
||
const base = new URL(service, location.href);
|
||
cowatch = new CoWatch({
|
||
api, wai, room: v.room, t0: v.t0_us, seed, principal, grant: granted.grant,
|
||
media: $("cowatch-media"), policies: [v.watch.policy], objects: v.watch.source.objects,
|
||
sourceUrl: `${service}${v.watch.source.path}`,
|
||
mayControl: (p, a) => mayControl(p, a),
|
||
// The proof names what this page addresses: the service under its prefix (live-rooms §10.5.1).
|
||
target: `//${location.host}${base.pathname.replace(/\/$/, "")}/watch/join`,
|
||
log, onChange: renderCoWatch,
|
||
});
|
||
if ($("cowatch")) $("cowatch").hidden = false;
|
||
if ($("caption-form")) $("caption-form").hidden = true;
|
||
await cowatch.estimateClock();
|
||
await cowatch.loadSource();
|
||
await cowatch.join();
|
||
}
|
||
|
||
function runCoWatch() {
|
||
const principal = me.principal;
|
||
const alive = () => state.phase !== "left";
|
||
cowatch.follow(alive).catch(fail);
|
||
cowatch.run(alive).catch(fail);
|
||
for (const [id, a] of [["cw-play", ACTION.play], ["cw-pause", ACTION.pause], ["cw-back", ACTION.seek], ["cw-fwd", ACTION.seek], ["cw-rate", ACTION.rate]]) {
|
||
if ($(id)) $(id).disabled = !mayControl(principal, a);
|
||
}
|
||
status("Watching together. The room keeps every viewer at one position, and measures how close.");
|
||
}
|
||
|
||
function watchControl(name, arg) {
|
||
if (!cowatch) throw new Error("not a watch room");
|
||
return cowatch.control(ACTION[name], arg);
|
||
}
|
||
|
||
function renderCoWatch() {
|
||
const el = $("cw-sync");
|
||
if (!el || !cowatch) return;
|
||
const s = cowatch.snapshot();
|
||
const ms = (us) => `${us >= 0 ? "+" : "−"}${Math.abs(us / 1000).toFixed(0)} ms`;
|
||
const mine = s.error_us == null ? "" : `You: ${ms(s.error_us)} from the room`;
|
||
let room = "Room: no window measured yet";
|
||
if (s.last_window) {
|
||
const n = s.last_window.viewers.length;
|
||
const unknown = s.last_window.viewers.filter((x) => x.verdict === "unknown").length;
|
||
const out = s.last_window.viewers.filter((x) => x.verdict === "out_of_sync").length;
|
||
const bound = `${s.last_window.bound_us / 1000} ms`;
|
||
room = out ? `Room: ${out} of ${n} out of sync (bound ${bound})` : unknown ? `Room: ${n - unknown} of ${n} in sync within ${bound}, ${unknown} unknown` : `Room: all ${n} in sync within ${bound}`;
|
||
}
|
||
el.textContent = [mine, room].filter(Boolean).join(" · ");
|
||
}
|
||
|
||
async function leave() {
|
||
state.phase = "left";
|
||
for (const sub of subscribed.values()) for (const [m, st] of Object.entries(sub.tracks)) endStream(sub, m, st, "user", false);
|
||
if (state.capture && state.capture.stream) for (const t of state.capture.stream.getTracks()) t.stop();
|
||
if (session) session.close(0, "left");
|
||
status("You left the room. Reload the page to enter again.");
|
||
}
|
||
|
||
// A new room may start while this page watches an empty one: the page says so.
|
||
function pollRoom() {
|
||
setInterval(async () => {
|
||
if (state.phase !== "watching" && state.phase !== "publishing") return;
|
||
try {
|
||
const r = await api("/room");
|
||
if (r.status === 200 && r.body.room !== fx.room) {
|
||
state.roomMoved = r.body.room;
|
||
if (state.phase === "watching") status("A new room has started. Reload the page to watch it.");
|
||
}
|
||
} catch { /* the next poll */ }
|
||
}, 20000);
|
||
}
|
||
|
||
// ------------------------------------------------------------------------- co-performance
|
||
|
||
const b64u = (b) => btoa(String.fromCharCode(...b)).replace(/\+/g, "-").replace(/\//g, "_").replace(/=+$/, "");
|
||
const fromB64u = (s) => Uint8Array.from(atob(s.replace(/-/g, "+").replace(/_/g, "/")), (c) => c.charCodeAt(0));
|
||
let coChallenge = null;
|
||
|
||
// A request to the room service's clock (`POST <service>/co/...`), this page's grant presented
|
||
// with a proof over the service's last challenge (live-rooms §10.5.3); a first request, or one
|
||
// whose challenge was used, is answered 401 with a new one and sent again.
|
||
async function coApi(path, body) {
|
||
const u = new URL(`${service}${path}`, location.href);
|
||
for (let i = 0; ; i++) {
|
||
const headers = { "content-type": "application/json", authorization: `Bearer ${b64u(unhex(me.grant))}` };
|
||
if (coChallenge) {
|
||
headers["wai-room-proof"] = b64u(moq.grantProofHttp(unhex(me.grant), unhex(me.seed), coChallenge, "POST", `//${u.host}${u.pathname}`));
|
||
coChallenge = null;
|
||
}
|
||
const r = await fetch(u, { method: "POST", headers, body: JSON.stringify(body), cache: "no-store" });
|
||
const c = r.headers.get("wai-room-challenge");
|
||
if (c) coChallenge = fromB64u(c);
|
||
let j = null;
|
||
try { j = await r.json(); } catch { /* not JSON */ }
|
||
if (r.status !== 401 || i >= 1) return { status: r.status, body: j };
|
||
}
|
||
}
|
||
|
||
// The model digest this page's grant names, when it marks it an AI participant.
|
||
function myAi() {
|
||
const g = (state.grants || []).find((x) => x.principal === me.principal);
|
||
const a = g && g.attrs.find((x) => x[0] === "ai");
|
||
return a ? a[1] : null;
|
||
}
|
||
|
||
async function emitOwn(kind, payload) {
|
||
const r = await eventsTrack.emit(kind, payload);
|
||
await (records = records.then(() => onRecord(r.record)));
|
||
return r.view;
|
||
}
|
||
|
||
// A line this page says: a caption, a machine's with its model when this page is an AI
|
||
// participant (live-rooms §4.4), a person's otherwise.
|
||
function say(text) {
|
||
const m = myAi();
|
||
return emitOwn(0x0a, wai.captionPayload(Math.max(1000, text.length * 60), m ? 2 : 1, m || undefined, document.documentElement.lang || "en", text));
|
||
}
|
||
|
||
function startCo() {
|
||
co = new Co({
|
||
wai, log, $, nameOf, isAi, roomNow, say,
|
||
me: () => me,
|
||
grantOf: (P) => (state.grants || []).find((g) => g.principal === P),
|
||
isPublisher: (P) => (state.grants || []).some((g) => g.principal === P && g.publish),
|
||
present: () => ((state.room && state.room.present) || []).map((p) => p.principal),
|
||
emit: emitOwn,
|
||
emitDetached: (kind, payload) => eventsTrack.detached(kind, payload),
|
||
api: coApi,
|
||
policyBytes: unhex(fx.co.policy),
|
||
policyHex: fx.co.policy,
|
||
});
|
||
window.roomTest.co = co;
|
||
wireControls(co, $, fail);
|
||
co.render();
|
||
setInterval(() => { try { co.render(); } catch (e) { fail(e); } }, 250);
|
||
if (q.get("ai") === "scripted" && q.get("script") !== "off" && me.role === "publisher") scriptLines().catch(fail);
|
||
}
|
||
|
||
// The demo AI participant's lines: a scripted source. It announces each line two seconds
|
||
// ahead, then says it; every third it says otherwise, so that its fidelity is not all kept.
|
||
const LINES = ["the bridge comes next", "back to the top, softer", "a long note to end the phrase", "the call, then its answer"];
|
||
async function scriptLines() {
|
||
for (let n = 0; ; n++) {
|
||
await new Promise((r) => setTimeout(r, 7000));
|
||
if (state.phase !== "publishing") { if (state.phase === "left" || state.phase === "ended") return; continue; }
|
||
const text = LINES[n % LINES.length];
|
||
await co.announce({ text, inMs: 2000, say: n % 3 === 2 ? `${text}, or not` : text });
|
||
}
|
||
}
|
||
|
||
// The model a scripted AI participant names: SHA-256 of its name.
|
||
async function scriptedModel() {
|
||
const d = await crypto.subtle.digest("SHA-256", enc.encode("wai demo: a scripted participant"));
|
||
return hex(new Uint8Array(d));
|
||
}
|
||
|
||
// A drawn picture and a tone in place of a camera and a microphone, at the room's profiles.
|
||
function syntheticStream(d) {
|
||
const vl = d.profiles.find((p) => p.media === 2).layers[0];
|
||
const ap = d.profiles.find((p) => p.media === 1);
|
||
const canvas = document.createElement("canvas");
|
||
canvas.width = vl.width;
|
||
canvas.height = vl.height;
|
||
const g = canvas.getContext("2d");
|
||
let n = 0;
|
||
setInterval(() => {
|
||
n++;
|
||
g.fillStyle = "#1a2a3a";
|
||
g.fillRect(0, 0, canvas.width, canvas.height);
|
||
for (let i = 0; i < 8; i++) {
|
||
g.fillStyle = `hsl(${(i * 45 + n * 3) % 360} 60% 55%)`;
|
||
const h = (Math.sin((n + i * 7) / 6) * 0.4 + 0.5) * canvas.height;
|
||
g.fillRect(i * (canvas.width / 8) + 4, canvas.height - h, canvas.width / 8 - 8, h);
|
||
}
|
||
g.fillStyle = "#ffffff";
|
||
g.font = "16px system-ui, sans-serif";
|
||
g.fillText("AI participant (scripted)", 12, 24);
|
||
}, Math.round(1000 / vl.fps));
|
||
const vs = canvas.captureStream(vl.fps);
|
||
const ac = new AudioContext({ sampleRate: ap.sample_rate });
|
||
const osc = ac.createOscillator();
|
||
osc.frequency.value = 220;
|
||
const gain = ac.createGain();
|
||
gain.gain.value = 0.08;
|
||
const dest = ac.createMediaStreamDestination();
|
||
dest.channelCount = ap.channels;
|
||
dest.channelCountMode = "explicit";
|
||
osc.connect(gain).connect(dest);
|
||
osc.start();
|
||
return new MediaStream([vs.getVideoTracks()[0], dest.stream.getAudioTracks()[0]]);
|
||
}
|
||
|
||
// The clock's log of objects (co-performance §9.1), every page of it.
|
||
async function coObjects() {
|
||
const out = [];
|
||
for (let from = 0; ; ) {
|
||
const r = await api(`/co/objects?from=${from}`);
|
||
if (r.status !== 200) throw new Error(`the clock's log answered ${r.status}`);
|
||
out.push(...r.body.objects);
|
||
if (r.body.end) return out;
|
||
from = r.body.next;
|
||
}
|
||
}
|
||
|
||
// ---------------------------------------------------------------------- remote desktop
|
||
|
||
// A remote desktop room (WAI remote-desktop) the room service keeps: the host's screen on its own
|
||
// canvas, this page's pointer, keys, composed text and pasted text sent as input statements under
|
||
// its grant in effect, the floor asked for and given back, and a latency record of each input the
|
||
// host applied, presented at the refresh that showed its frame. The key is the one this page
|
||
// made, and the grant the service issued for it.
|
||
async function startRemote() {
|
||
// A room whose host has not come yet: the page waits for it.
|
||
for (let i = 0; !fx.remote.host; i++) {
|
||
if (i === 0) status("Waiting for the host to share their screen…");
|
||
await new Promise((ok) => setTimeout(ok, 1000));
|
||
const r = await api("/room");
|
||
if (r.status === 200 && r.body.room === fx.room && r.body.remote) fx.remote = r.body.remote;
|
||
else if (r.status === 200 && r.body.room !== fx.room) throw new Error("a new room has started; reload the page to enter it");
|
||
}
|
||
const key = await keyFromSeed(unhex(me.seed));
|
||
remote = new RemoteViewer({
|
||
room: fx.room, me: me.principal, host: fx.remote.host, clock: fx.clock.principal, grant: me.grant, key, roomNow, log,
|
||
unixNow: () => Math.floor((Date.now() * 1000 + skew) / 1e6),
|
||
grantOf: async (digest) => {
|
||
if (!state.grants.some((g) => g.digest === digest)) await refreshGrants();
|
||
const i = state.grants.findIndex((g) => g.digest === digest);
|
||
return i >= 0 ? fx.grants[i] : null;
|
||
},
|
||
emit: async (k, p) => (await eventsTrack.emit(k, p)).view, canvas: $("remote-canvas"), panel: $("remote-state"),
|
||
});
|
||
await remote.start();
|
||
window.roomTest.remote = remote;
|
||
$("remote").hidden = false;
|
||
const controls = () => {
|
||
$("remote-request").hidden = !remote.control;
|
||
$("remote-release").hidden = !remote.control;
|
||
};
|
||
controls();
|
||
setInterval(controls, 500);
|
||
$("remote-request").onclick = () => remote.requestFloor().catch(fail);
|
||
$("remote-release").onclick = () => remote.releaseFloor().catch(fail);
|
||
await subscribeEvents(fx.remote.host);
|
||
}
|
||
|
||
// ------------------------------------------------------------------------------- the clock
|
||
|
||
async function clockEmit(desc) {
|
||
const lp = wai.lifecyclePayload(JSON.stringify(desc));
|
||
const r = await eventsTrack.emit(lp[0], lp.subarray(1));
|
||
return r.view;
|
||
}
|
||
|
||
// ------------------------------------------------------------------------------ publishing
|
||
|
||
async function pickVideoCodec(profile) {
|
||
const l = profile.layers[0];
|
||
for (const codec of profile.codecs) {
|
||
const cfg = { codec, width: l.width, height: l.height, bitrate: l.max_kbps * 1000, framerate: l.fps, latencyMode: "realtime" };
|
||
try {
|
||
const s = await VideoEncoder.isConfigSupported(cfg);
|
||
if (s.supported) return cfg;
|
||
} catch { /* not this one */ }
|
||
}
|
||
throw new Error(`no video codec of the profile is supported: ${profile.codecs.join(", ")}`);
|
||
}
|
||
|
||
async function startPublishing(given) {
|
||
const d = state.descriptor;
|
||
const vp = d.profiles.find((p) => p.media === 2);
|
||
const ap = d.profiles.find((p) => p.media === 1);
|
||
const vcfg = await pickVideoCodec(vp);
|
||
const frameMs = Number(q.get("audioFrameMs") || 20);
|
||
const acfg = {
|
||
codec: "opus",
|
||
sampleRate: ap.sample_rate,
|
||
numberOfChannels: ap.channels,
|
||
bitrate: 32000,
|
||
opus: { frameDuration: frameMs * 1000, usedtx: (ap.processing & 8) !== 0, complexity: 5 },
|
||
};
|
||
const as = await AudioEncoder.isConfigSupported(acfg);
|
||
if (!as.supported) throw new Error("this browser's AudioEncoder does not take Opus at 48 kHz");
|
||
state.publishCodec = { video: vcfg.codec, audio: "opus", audioFrameMs: frameMs };
|
||
const vl = vp.layers[0];
|
||
const catalog = {
|
||
version: 1,
|
||
tracks: [
|
||
{
|
||
namespace: `${fx.room}-${me.principal}-video`, name: "0", packaging: "wai", isLive: true, role: "video",
|
||
codec: vcfg.codec, width: vl.width, height: vl.height, framerate: vl.fps, bitrate: vl.max_kbps * 1000, renderGroup: 1,
|
||
wai_capabilities: [media.CAPABILITY_OF(vcfg.codec)], wai_receipts: true, wai_receipt_alias: RECEIPT_ALIAS.video, wai_receipts_track: "0_receipts",
|
||
},
|
||
{ namespace: `${fx.room}-${me.principal}-video`, name: "0_receipts", packaging: "wai-receipts", isLive: true },
|
||
{
|
||
namespace: `${fx.room}-${me.principal}-audio`, name: "0", packaging: "wai", isLive: true, role: "audio",
|
||
codec: "opus", samplerate: ap.sample_rate, channelConfig: String(ap.channels), bitrate: 32000, renderGroup: 1,
|
||
wai_capabilities: ["wai.audio.opus"], wai_receipts: true, wai_receipt_alias: RECEIPT_ALIAS.audio, wai_receipts_track: "0_receipts",
|
||
},
|
||
{ namespace: `${fx.room}-${me.principal}-audio`, name: "0_receipts", packaging: "wai-receipts", isLive: true },
|
||
{ namespace: `${fx.room}-${me.principal}-events`, name: "0", packaging: "eventtimeline", eventType: "urn:wai:event:1", mimetype: "application/json", isLive: true },
|
||
],
|
||
};
|
||
state.catalog = catalog;
|
||
signers.video = new moq.TrackSigner(unhex(me.seed), BigInt(RECEIPT_ALIAS.video), BigInt(ALIAS.video), BigInt(ALIAS.videoReceipts), PRIORITY.video, me.name, BigInt(VIDEO_TIMEOUT_MS));
|
||
signers.audio = new moq.TrackSigner(unhex(me.seed), BigInt(RECEIPT_ALIAS.audio), BigInt(ALIAS.audio), BigInt(ALIAS.audioReceipts), PRIORITY.audio, me.name, BigInt(AUDIO_TIMEOUT_MS));
|
||
await publishTracks(session);
|
||
|
||
const stream = given || await navigator.mediaDevices.getUserMedia(constraintsOf(d));
|
||
state.capture = { video: stream.getVideoTracks()[0].getSettings(), audio: stream.getAudioTracks()[0].getSettings() };
|
||
Object.defineProperty(state.capture, "stream", { value: stream, enumerable: false });
|
||
let vgroup = 0, agroup = 0, vframes = 0;
|
||
// In a room with co-performance: each medium's offset from its capture clock to the room
|
||
// clock, so that a timestamp states the capture instant on the room clock (its due instant,
|
||
// co-performance §9.1); and whether the last frame was held back off the floor, so that the
|
||
// next one sent is a key frame.
|
||
// A video frame keeps its timestamp through the encoder, so its capture instant is its
|
||
// timestamp and the offset. An audio encoder numbers its output by the samples it has taken,
|
||
// so a stretch held back off the floor would be lost from its timestamps: each output's
|
||
// capture instant is read from the inputs it took, in order, instead (`aq`).
|
||
const offset = {};
|
||
const stamp = (m, ts) => (co && offset[m] != null ? ts + offset[m] : ts);
|
||
const aq = [];
|
||
let aqTaken = 0;
|
||
const audioDue = (samples, fallback) => {
|
||
if (!co || !aq.length) return fallback;
|
||
const due = aq[0].room + Math.round((aqTaken / ap.sample_rate) * 1e6);
|
||
aqTaken += samples;
|
||
while (aq.length && aqTaken >= aq[0].frames) { aqTaken -= aq[0].frames; aq.shift(); }
|
||
return due;
|
||
};
|
||
let heldBack = false;
|
||
const venc = new VideoEncoder({
|
||
output: (chunk) => {
|
||
// In an end-to-end encrypted room nothing is sent before the room's group has keyed
|
||
// this page, and no group is numbered for it: a track's groups run from 0 with no gap,
|
||
// so a late joiner's FETCH from group 0 asks for nothing the relay never had.
|
||
if (unkeyed("video")) return;
|
||
const g = vgroup++;
|
||
const b = new Uint8Array(chunk.byteLength);
|
||
chunk.copyTo(b);
|
||
const payload = vcfg.codec.startsWith("vp09") ? media.ivfOne(b, vl.width, vl.height, chunk.timestamp) : media.av1One(b, vl.width, vl.height, vl.fps, 1);
|
||
const env = wai.pack(media.manifestFor(media.CAPABILITY_OF(vcfg.codec), { width: vl.width, height: vl.height, timestamp_us: stamp("video", chunk.timestamp), key: chunk.type === "key" }), payload);
|
||
sendGroup("video", g, env);
|
||
},
|
||
error: fail,
|
||
});
|
||
venc.configure(vcfg);
|
||
const aenc = new AudioEncoder({
|
||
output: (chunk) => {
|
||
if (unkeyed("audio")) return;
|
||
const g = agroup++;
|
||
const b = new Uint8Array(chunk.byteLength);
|
||
chunk.copyTo(b);
|
||
const samples = Math.round(((chunk.duration || frameMs * 1000) / 1e6) * ap.sample_rate);
|
||
const late = co ? co.takeLate() : 0;
|
||
const due = audioDue(samples, chunk.timestamp);
|
||
const env = wai.pack(media.manifestFor("wai.audio.opus", { sample_rate: ap.sample_rate, channels: ap.channels, timestamp_us: Math.max(0, due - late), samples }), media.oggOpusOne(b, { channels: ap.channels, inputRate: ap.sample_rate, samples }));
|
||
sendGroup("audio", g, env);
|
||
},
|
||
error: fail,
|
||
});
|
||
aenc.configure(acfg);
|
||
// Frames at the profile's layer size: a capture that is another size is scaled first.
|
||
const scaler = new OffscreenCanvas(vl.width, vl.height);
|
||
const sctx = scaler.getContext("2d");
|
||
pump(new MediaStreamTrackProcessor({ track: stream.getVideoTracks()[0] }).readable, (frame) => {
|
||
if (co && offset.video == null) offset.video = roomNow() - frame.timestamp;
|
||
// Off the floor (co-performance §6.4): nothing is encoded, and the first frame back is a
|
||
// key frame.
|
||
if (co && !co.sends(roomNow())) { heldBack = true; frame.close(); return; }
|
||
// A key frame every two seconds, so a subscriber that joins can start within two.
|
||
const key = vframes % (vl.fps * 2) === 0 || heldBack;
|
||
heldBack = false;
|
||
if (venc.encodeQueueSize < 4) {
|
||
vframes++;
|
||
let f = frame;
|
||
if (frame.displayWidth !== vl.width || frame.displayHeight !== vl.height) {
|
||
sctx.drawImage(frame, 0, 0, vl.width, vl.height);
|
||
f = new VideoFrame(scaler, { timestamp: frame.timestamp });
|
||
}
|
||
venc.encode(f, { keyFrame: key });
|
||
// The participant's own tile shows what it sends (a self view, no audio of its own).
|
||
if (compositor) compositor.setVideo(me.principal, f.clone());
|
||
if (f !== frame) f.close();
|
||
}
|
||
frame.close();
|
||
});
|
||
pump(new MediaStreamTrackProcessor({ track: stream.getAudioTracks()[0] }).readable, (ad) => {
|
||
if (ad.sampleRate !== acfg.sampleRate || ad.numberOfChannels !== acfg.numberOfChannels) {
|
||
if (!state.audioMismatch) log(`the microphone gives ${ad.numberOfChannels} channel(s) at ${ad.sampleRate} Hz; the profile asks ${acfg.numberOfChannels} at ${acfg.sampleRate}`);
|
||
state.audioMismatch = { sampleRate: ad.sampleRate, channels: ad.numberOfChannels };
|
||
ad.close();
|
||
return;
|
||
}
|
||
if (co && !co.sends(roomNow())) { ad.close(); return; }
|
||
// Its first sample's instant on the room clock: it ends now.
|
||
if (co) aq.push({ room: roomNow() - Math.round((ad.numberOfFrames / ad.sampleRate) * 1e6), frames: ad.numberOfFrames });
|
||
aenc.encode(ad);
|
||
ad.close();
|
||
});
|
||
}
|
||
|
||
// The rating claims a track this page publishes carries (WAI age-assurance §5.7): the room's,
|
||
// and, for its video with `rate`, its own (§3.6: a principal rates its own tracks).
|
||
function ratingsFor(ns, name) {
|
||
const out = [...(me.ratings || [])];
|
||
if (q.get("rate") && ns[2] === "video" && name === "0") {
|
||
out.push(moq.ratingClaim(unhex(me.seed), JSON.stringify(ns), name, 0n, q.get("rate_scheme") || "", q.get("rate_value") || "", BigInt(q.get("rate")), new Date().toISOString().replace(/\.\d+Z$/, "Z")));
|
||
}
|
||
return out;
|
||
}
|
||
|
||
async function publishTracks(s) {
|
||
const P = me.principal;
|
||
const pub = async (ns, name, alias, props) => published.push(Object.assign(await s.publish(ns, name, alias, { ...props, ratings: ratingsFor(ns, name) }), { ns, name }));
|
||
await pub([fx.room, P], "catalog", ALIAS.catalog, { publisher_priority: PRIORITY.catalog });
|
||
for (const m of ["audio", "video"]) {
|
||
await pub([fx.room, P, m], "0_receipts", ALIAS[`${m}Receipts`], { publisher_priority: PRIORITY[m] });
|
||
await pub([fx.room, P, m], "0", ALIAS[m], { publisher_priority: PRIORITY[m], object_delivery_timeout: m === "video" ? VIDEO_TIMEOUT_MS : AUDIO_TIMEOUT_MS });
|
||
signers[m].setAliases(BigInt(ALIAS[m]), BigInt(ALIAS[`${m}Receipts`]));
|
||
}
|
||
// The catalog: group 0, one object (MSF §5); a session that moves sends it again.
|
||
const w = new moq.SubgroupWriter(BigInt(ALIAS.catalog), 0n, PRIORITY.catalog, true);
|
||
const b = enc.encode(JSON.stringify(state.catalog));
|
||
const obj = w.object(0n, b);
|
||
const h = w.header();
|
||
const all = new Uint8Array(h.length + obj.length);
|
||
all.set(h, 0);
|
||
all.set(obj, h.length);
|
||
await s.sendStream(all, 1000);
|
||
w.free();
|
||
}
|
||
|
||
async function pump(readable, each) {
|
||
const r = readable.getReader();
|
||
for (;;) {
|
||
const { value, done } = await r.read();
|
||
if (done) return;
|
||
try { each(value); } catch (e) { fail(e); }
|
||
}
|
||
}
|
||
|
||
// Whether an encoded frame of `m` goes unsent: in an end-to-end encrypted room, before the
|
||
// page holds an epoch (§4.9: it seals under its current epoch).
|
||
function unkeyed(m) {
|
||
if (!e2ee || e2ee.joined) return false;
|
||
e2ee.stats.unkeyed[m]++;
|
||
return true;
|
||
}
|
||
|
||
// Group g of a media track: its receipts first (no later than the group's first object,
|
||
// moq-streaming-format §4.1), then the object, each on a stream of its own.
|
||
async function sendGroup(m, g, envelope) {
|
||
const s = session;
|
||
if (state.stoppedPublishing) return;
|
||
try {
|
||
const now = new Date().toISOString();
|
||
let sg;
|
||
if (e2ee) {
|
||
// Sealed before it is signed (confidentiality §4.9, §8.1): the receipt is over the
|
||
// sealed bytes. Until the room's group has keyed this page, nothing is sent.
|
||
const sl = e2ee.seal([fx.room, me.principal, m], g, PRIORITY[m], envelope, m === "video", m === "video" ? VIDEO_TIMEOUT_MS : AUDIO_TIMEOUT_MS);
|
||
if (!sl) { e2ee.stats.unkeyed[m]++; return; }
|
||
if (m === "video" && state.e2ee.tamperNext) {
|
||
// A test plays a relay that changes a byte: signed as it travels, so the relay's
|
||
// door passes it, and every member refuses it bad_tag.
|
||
state.e2ee.tamperNext = false;
|
||
sl.payload[Math.min(5, sl.payload.length - 1)] ^= 1;
|
||
state.e2ee.tampered.push(g);
|
||
}
|
||
if (m === "video") {
|
||
// What this page sealed, by group, the last few hundred: what a test compares with
|
||
// what a subscriber outside the group received.
|
||
e2eeSent.set(g, { group: g, epoch: sl.epoch, plain_len: envelope.length, plain_head: hex(envelope.slice(0, 64)), sealed_len: sl.payload.length, sealed_head: hex(sl.payload.slice(0, 64)) });
|
||
if (e2eeSent.size > 400) e2eeSent.delete(e2eeSent.keys().next().value);
|
||
}
|
||
e2ee.stats.sealed[m]++;
|
||
sg = signers[m].signSealed(BigInt(g), sl.payload, sl.props, now, now);
|
||
} else {
|
||
sg = signers[m].sign(BigInt(g), envelope, now, now);
|
||
}
|
||
const r = sg.receiptsStream();
|
||
const md = sg.mediaStream();
|
||
sg.free();
|
||
// sendOrder: higher goes first; audio before video, receipts before their object.
|
||
const base = m === "audio" ? 2_000_000 : 1_000_000;
|
||
await s.sendStream(r, base + 1);
|
||
await s.sendStream(md, base);
|
||
state.published[m]++;
|
||
state.published.bytes += md.length + r.length;
|
||
} catch (e) {
|
||
if (s.goaway || s.closedInfo) return; // the session is moving; the next group goes on the new one
|
||
fail(e);
|
||
}
|
||
}
|
||
|
||
// ------------------------------------------------------------------------------ subscribing
|
||
|
||
function tally(principal) {
|
||
if (!state.subs[principal]) {
|
||
const t = () => ({ received: 0, verified: 0, refused: 0, no_receipt: 0, receipts: 0, receipts_refused: 0, decoded: 0, decode_errors: 0, reset: 0, dropped_accounted: 0, dropped_unaccounted: 0, fetched: 0, fetched_verified: 0, fetched_decoded: 0, first_live_group: null, first_fetched_group: null, last_fetched_group: null, chain: {} });
|
||
state.subs[principal] = { audio: t(), video: t(), screen: t(), catalog: false, events: 0 };
|
||
}
|
||
return state.subs[principal];
|
||
}
|
||
|
||
const eventsSubscribed = new Set();
|
||
|
||
async function subscribeEvents(principal) {
|
||
if (eventsSubscribed.has(principal)) return;
|
||
eventsSubscribed.add(principal);
|
||
const ns = [fx.room, principal, "events"];
|
||
const onRecords = (payload) => {
|
||
let arr;
|
||
try { arr = JSON.parse(dec.decode(payload)); } catch { return log(`events of ${principal.slice(0, 8)}: an object that is not JSON`); }
|
||
for (const rec of arr) {
|
||
let bytes;
|
||
try { bytes = wai.msfRecordRead(JSON.stringify(rec), BigInt(t0)); } catch (e) { log(`an event record refused: ${e.message || e}`); continue; }
|
||
records = records.then(() => onRecord(bytes)).catch(fail);
|
||
}
|
||
};
|
||
const handler = {
|
||
stream: () => ({ object: (j, p) => onRecords(p) }),
|
||
fill: () => ({ object: (j, p) => onRecords(p) }),
|
||
};
|
||
try {
|
||
await session.subscribe(ns, "0", {
|
||
subscriber_priority: SUB_PRIORITY.events, group_order: 1, forward: true, rendezvous_timeout: 10000,
|
||
location_filter: { type: "next" }, fill: { location_filter: { type: "relative", start_group: 1 } },
|
||
}, handler);
|
||
} catch (e) {
|
||
if (!withheld(e, principal, "events")) throw e;
|
||
}
|
||
}
|
||
|
||
// Records are taken one at a time, in the order they arrive: a join under a grant the page
|
||
// does not hold waits while the page asks the service for the room's grants again.
|
||
let records = Promise.resolve();
|
||
|
||
async function onRecord(bytes) {
|
||
const v = JSON.parse(room.offer(bytes));
|
||
if (!v.ok) return log(`an event refused: ${v.reason}`);
|
||
if (!v.new) return;
|
||
if (co) co.onRecord(v, bytes);
|
||
if (remote) remote.onRecord(bytes);
|
||
if (service && v.kind_name === "join" && v.payload && v.payload.detail && !state.grants.some((g) => g.digest === v.payload.detail.grant)) {
|
||
await refreshGrants();
|
||
}
|
||
// A grant exchanged (live-rooms §4.6): the room's grants read again when it names a new one.
|
||
if (service && v.kind_name === "grant_exchange") {
|
||
const exchanged = hex(readEvent(bytes).payload.subarray(64, 96));
|
||
if (!state.grants.some((g) => g.digest === exchanged)) await refreshGrants();
|
||
}
|
||
const who = nameOf(v.principal);
|
||
const entry = { t: v.t, who, kind: v.kind_name, payload: v.payload, digest: v.digest };
|
||
state.timeline.push(entry);
|
||
renderTimeline(entry);
|
||
if (v.kind_name === "caption") {
|
||
state.captions.push({ t: v.t, who, text: v.payload.text, origin: v.payload.origin_label, model: v.payload.model, lang: v.payload.lang });
|
||
renderCaption();
|
||
}
|
||
if (v.kind >= 0x10) onLifecycle(v.t);
|
||
}
|
||
|
||
async function refreshGrants() {
|
||
const r = await api("/room");
|
||
if (r.status !== 200 || r.body.room !== fx.room) return log("the room's grants could not be read again");
|
||
fx.grants = r.body.co ? r.body.grants.concat(r.body.co.audience) : r.body.grants;
|
||
if (r.body.co) fx.co = r.body.co;
|
||
room.setGrants(JSON.stringify(fx.grants));
|
||
readGrants();
|
||
log(`the room's grants read again: ${fx.grants.length}`);
|
||
}
|
||
|
||
// The room is laid out at the later of this page's instant and the lifecycle event's: an event
|
||
// the clock stamped a moment ahead of this page's clock still takes effect.
|
||
let lastLifecycleT = 0;
|
||
function onLifecycle(t = 0) {
|
||
lastLifecycleT = Math.max(lastLifecycleT, t);
|
||
const st = JSON.parse(room.state());
|
||
state.room = st;
|
||
if (compositor) compositor.relayout(room, Math.max(roomNow(), lastLifecycleT));
|
||
for (const own of st.own) {
|
||
if (own.principal === me.principal || subscribed.has(own.principal)) continue;
|
||
subscribeParticipant(own.principal).catch(fail);
|
||
}
|
||
// In a room with co-performance, every present principal's events: an audience member's
|
||
// inputs too, which a selection is made from. In a remote desktop room, every present
|
||
// principal's events too: a viewer shows who is typing, from statements whose keys and text are
|
||
// sealed to the host (remote-desktop §5.5).
|
||
if (co || (fx && fx.remote)) {
|
||
for (const p of st.present) {
|
||
if (p.principal === me.principal || eventsSubscribed.has(p.principal)) continue;
|
||
subscribeEvents(p.principal).catch(fail);
|
||
}
|
||
}
|
||
}
|
||
|
||
async function subscribeParticipant(P) {
|
||
if (subscribed.has(P)) return;
|
||
const sub = { principal: P, tracks: {}, catalog: null };
|
||
subscribed.set(P, sub);
|
||
tally(P);
|
||
await subscribeEvents(P);
|
||
let resolveCatalog;
|
||
const got = new Promise((r) => { resolveCatalog = r; });
|
||
const onCatalog = (p) => {
|
||
try { const c = JSON.parse(dec.decode(p)); if (!sub.catalog) { sub.catalog = c; tally(P).catalog = true; resolveCatalog(c); } } catch (e) { fail(e); }
|
||
};
|
||
try {
|
||
await session.subscribe([fx.room, P], "catalog", {
|
||
subscriber_priority: SUB_PRIORITY.catalog, group_order: 1, forward: true, rendezvous_timeout: 10000,
|
||
location_filter: { type: "next" }, fill: { location_filter: { type: "relative", start_group: 1 } },
|
||
}, { stream: () => ({ object: (j, p) => onCatalog(p) }), fill: () => ({ object: (j, p) => onCatalog(p) }) });
|
||
} catch (e) {
|
||
if (withheld(e, P, "catalog")) return;
|
||
throw e;
|
||
}
|
||
const catalog = await withTimeout(got, 10000, `the catalog of ${P.slice(0, 8)}`);
|
||
for (const tr of catalog.tracks.filter((x) => x.packaging === "wai")) {
|
||
await subscribeMedia(sub, tr);
|
||
}
|
||
}
|
||
|
||
// A request the relay refused `below_threshold` (WAI age-assurance §5.5): the track is rated
|
||
// above the threshold this page was admitted at. Recorded and shown, not an error.
|
||
function withheld(e, P, m) {
|
||
if (!(e instanceof RequestRefused) || !/below_threshold/.test(e.message)) return false;
|
||
const t = tally(P);
|
||
t.withheld = [...(t.withheld || []), m];
|
||
if (!state.age) state.age = { age: null, admitted: 0, withheld: [] };
|
||
state.age.withheld.push({ principal: P, track: m });
|
||
log(`${m} of ${P.slice(0, 8)}: rated above this page's threshold, withheld by the relay (below_threshold)`);
|
||
status(`Some of the room is rated above the age you were admitted at, and is withheld: ${m} of ${nameOf(P)}.`);
|
||
return true;
|
||
}
|
||
|
||
// A host's screen is a track of the room's `screen` kind (live-rooms §6.1), named by its namespace.
|
||
function mediaOf(tr) {
|
||
if (typeof tr.namespace === "string" && tr.namespace.endsWith("-screen")) return "screen";
|
||
return tr.role === "audio" ? "audio" : "video";
|
||
}
|
||
|
||
async function subscribeMedia(sub, tr) {
|
||
const P = sub.principal;
|
||
const m = mediaOf(tr);
|
||
await autoplayAllowed();
|
||
const T = tally(P)[m];
|
||
const ns = [fx.room, P, m];
|
||
const verifier = new moq.TrackVerifier(unhex(P), BigInt(tr.wai_receipt_alias));
|
||
const st = { verifier, pending: new Map(), seen: new Set(), receiptsAt: new Map(), decoder: null, waitingKey: true, tr, m, chainOf: new Map(), verifiedBy: new Map() };
|
||
sub.tracks[m] = st;
|
||
st.route = await routeTrack(P, tr, m);
|
||
st.decoder = makeDecoder(P, tr, T, st);
|
||
|
||
const onReceipts = (group, payload) => {
|
||
const r = JSON.parse(verifier.receipts(BigInt(group), payload, BigInt(Date.now())));
|
||
if (!r.ok) { T.receipts_refused++; log(`receipts of ${m} group ${group} of ${P.slice(0, 8)} refused: ${r.reason}`); return; }
|
||
T.receipts++;
|
||
T.chain[r.chain] = (T.chain[r.chain] || 0) + 1;
|
||
// Each group's admission to its chain, for the state of a frame decoded from it (§10).
|
||
st.chainOf.set(group, r.chain);
|
||
for (const e of r.released || []) st.chainOf.set(e.group, e.outcome);
|
||
if (st.chainOf.size > 2048) st.chainOf.delete(st.chainOf.keys().next().value);
|
||
if (!st.seen.has(group)) st.receiptsAt.set(group, performance.now());
|
||
const held = st.pending.get(group);
|
||
if (held) { st.pending.delete(group); onObject(group, held.payload, held.fetched, held.meta); }
|
||
};
|
||
// `meta`: the object's Object ID, Publisher Priority and property list, which an end-to-
|
||
// end encrypted room's objects are opened with.
|
||
const onObject = (group, payload, fetched, meta) => {
|
||
const r = JSON.parse(verifier.object(BigInt(group), 0n, payload));
|
||
if (!r.ok && r.reason === "no_receipt") {
|
||
if (!st.pending.has(group)) { st.pending.set(group, { payload, at: performance.now(), fetched, meta }); T.no_receipt++; }
|
||
return;
|
||
}
|
||
st.receiptsAt.delete(group);
|
||
if (!r.ok) { T.refused++; log(`${m} group ${group} of ${P.slice(0, 8)} refused: ${r.reason}`); return; }
|
||
if (fetched) T.fetched_verified++; else T.verified++;
|
||
// What a frame decoded from this object names (browser-sink §7.2): its content hash and
|
||
// its group receipt's link hash, as the verifier established them.
|
||
st.verifiedBy.set(r.content_hash, r.group_receipt);
|
||
if (st.verifiedBy.size > 1024) st.verifiedBy.delete(st.verifiedBy.keys().next().value);
|
||
deliver(P, st, T, group, payload, fetched, { ...(meta || {}), bind: { digest: r.content_hash, receipt: r.group_receipt, group } });
|
||
};
|
||
st.onObject = onObject;
|
||
st.onReceipts = onReceipts;
|
||
st.ns = ns;
|
||
|
||
const rh = {
|
||
stream: (h) => ({ object: (j, p) => onReceipts(h.group_id, p) }),
|
||
fill: () => ({ object: (j, p) => onReceipts(j.group_id, p) }),
|
||
};
|
||
const mh = {
|
||
stream: (h) => ({
|
||
object: (j, p, k, props) => {
|
||
if (st.seen.has(h.group_id)) return;
|
||
st.seen.add(h.group_id);
|
||
T.received++;
|
||
if (T.first_live_group == null) T.first_live_group = h.group_id;
|
||
if (T.last_live_group == null || h.group_id > T.last_live_group) T.last_live_group = h.group_id;
|
||
onObject(h.group_id, p, false, { object: j.object_id, priority: h.publisher_priority, props });
|
||
},
|
||
reset: () => { T.reset++; },
|
||
}),
|
||
fill: () => ({ object: () => {} }),
|
||
// The publisher ended the track: the stream ended (stream-termination §4.3, `ended`).
|
||
done: () => { if (st.sink && !st.sink.terminated) endStream(sub, m, st, "ended"); },
|
||
};
|
||
// The receipts track first and with no delivery timeout (moq-streaming-format §4.1),
|
||
// at the media track's subscriber priority.
|
||
st.subs = [await session.subscribe(ns, tr.wai_receipts_track, { subscriber_priority: SUB_PRIORITY[m], group_order: 1, forward: true, rendezvous_timeout: 10000, location_filter: { type: "next" } }, rh)];
|
||
try {
|
||
st.subs.push(await session.subscribe(ns, tr.name, { subscriber_priority: SUB_PRIORITY[m], group_order: 1, forward: true, rendezvous_timeout: 10000, location_filter: { type: "next" } }, mh));
|
||
} catch (e) {
|
||
if (withheld(e, P, m)) return;
|
||
throw e;
|
||
}
|
||
openSinkLog(sub, st, ns, tr, m);
|
||
log(`subscribed to ${m} of ${P.slice(0, 8)} (${tr.codec}, receipt alias ${tr.wai_receipt_alias})`);
|
||
|
||
if (q.get("fetch") === "1") lateJoin(sub, st, T, ns, tr).catch(fail);
|
||
}
|
||
|
||
// A late joiner (MOQT §3.2): every earlier group of the track and of its receipts track,
|
||
// by FETCH, from group 0 to the Largest Object; each object verified against its receipt
|
||
// like any other, and the video decoded from its first key frame.
|
||
async function lateJoin(sub, st, T, ns, tr) {
|
||
const params = { group_order: 1, subscriber_priority: SUB_PRIORITY[st.m] + 1, location_filter: { type: "absolute", start: [0, 0] } };
|
||
// A FETCH is refused before any object is sent, so one tried again counts nothing twice:
|
||
// a relay asked for groups still arriving may pass the request to the publisher, which
|
||
// takes no requests; a moment later its cache holds them.
|
||
const fetchAgain = async (name, onObject) => {
|
||
for (let i = 0; ; i++) {
|
||
try { return await session.fetch(ns, name, params, onObject); } catch (e) {
|
||
if (i >= 3 || !(e instanceof RequestRefused)) throw e;
|
||
log(`FETCH of ${st.m} ${name} of ${sub.principal.slice(0, 8)}: ${e.message}; again`);
|
||
T.fetch_retries = (T.fetch_retries || 0) + 1;
|
||
await new Promise((r) => setTimeout(r, 750));
|
||
}
|
||
}
|
||
};
|
||
const rec = [];
|
||
const r1 = await fetchAgain(tr.wai_receipts_track, (j, p) => rec.push([j.group_id, p]));
|
||
for (const [g, p] of rec) {
|
||
const r = JSON.parse(st.verifier.receipts(BigInt(g), p, BigInt(Date.now())));
|
||
if (r.ok) { T.receipts++; T.chain[r.chain] = (T.chain[r.chain] || 0) + 1; } else T.receipts_refused++;
|
||
}
|
||
const objs = [];
|
||
const r2 = await fetchAgain(tr.name, (j, p, props) => objs.push([j.group_id, p, { object: j.object_id, priority: j.publisher_priority, props }]));
|
||
for (const [g, p, meta] of objs) {
|
||
// In an end-to-end encrypted room a group already received live was opened once: its
|
||
// copy by FETCH is the same location, which the receiving checks refuse as a replay
|
||
// (confidentiality §6 step 8). It is not presented again.
|
||
if (e2ee && st.seen.has(g)) { T.fetched_live = (T.fetched_live || 0) + 1; continue; }
|
||
T.fetched++;
|
||
if (T.first_fetched_group == null) T.first_fetched_group = g;
|
||
T.last_fetched_group = g;
|
||
st.seen.add(g);
|
||
st.onObject(g, p, true, meta);
|
||
}
|
||
log(`FETCH of ${st.m} of ${sub.principal.slice(0, 8)}: ${rec.length} receipts (${r1.how}), ${objs.length} objects (${r2.how})`);
|
||
}
|
||
|
||
// An object verified against its receipt, on its way to a decoder. In an end-to-end
|
||
// encrypted room it is opened first, through the receiving checks in their order
|
||
// (confidentiality §6); one that does not open never reaches a decoder (§10). One sealed in
|
||
// an epoch this page has not reached, or sent before this page's Welcome, waits (§5.8).
|
||
const E2EE_HOLD_MS = 15000;
|
||
const e2eeSent = new Map();
|
||
const E2EE_HOLD_MAX = 600;
|
||
|
||
function deliver(P, st, T, group, payload, fetched, meta) {
|
||
if (!e2ee) return decodeEnvelope(P, st, T, group, payload, fetched, meta && meta.bind);
|
||
const E = T.e2ee || (T.e2ee = { opened: 0, refused: {}, held: 0, held_expired: 0, by_epoch: {}, fetched_opened: 0, fetched_refused: {}, fetched_epochs: {}, samples: [] });
|
||
if (st.intercept) {
|
||
const f = st.intercept;
|
||
st.intercept = null;
|
||
return f({ group, payload, meta, fetched });
|
||
}
|
||
const v = e2ee.open(st.ns, st.tr.name, group, meta.object || 0, meta.priority, meta.props, payload);
|
||
if (E.samples.length < 8 && st.m === "video" && !fetched && !v.ok) E.samples.push({ group, sealed_len: payload.length, sealed_head: hex(payload.slice(0, 64)), outcome: v.reason || `pending ${v.pending}` });
|
||
if (v.ok) {
|
||
if (fetched) E.fetched_opened++; else E.opened++;
|
||
E.by_epoch[v.epoch] = (E.by_epoch[v.epoch] || 0) + 1;
|
||
if (fetched) E.fetched_epochs[v.epoch] = (E.fetched_epochs[v.epoch] || 0) + 1;
|
||
return decodeEnvelope(P, st, T, group, v.payload, fetched, meta && meta.bind);
|
||
}
|
||
if (v.pending != null || (v.reason === "epoch_not_joined" && e2ee.awaitingWelcome)) {
|
||
st.sealedHeld = st.sealedHeld || [];
|
||
if (st.sealedHeld.length >= E2EE_HOLD_MAX) { E.held_expired++; return; }
|
||
st.sealedHeld.push({ group, payload, fetched, meta, at: performance.now() });
|
||
E.held++;
|
||
return;
|
||
}
|
||
const into = fetched ? E.fetched_refused : E.refused;
|
||
into[v.reason] = (into[v.reason] || 0) + 1;
|
||
if ((into[v.reason] || 0) <= 3) log(`${st.m} group ${group} of ${P.slice(0, 8)} does not open: ${v.reason}`);
|
||
}
|
||
|
||
// The room's group moved: what waited is opened again, in its order; the view says so.
|
||
function onKeys() {
|
||
for (const sub of subscribed.values()) {
|
||
for (const st of Object.values(sub.tracks)) {
|
||
if (!st.sealedHeld || !st.sealedHeld.length) continue;
|
||
const T = tally(sub.principal)[st.m];
|
||
const held = st.sealedHeld.sort((a, b) => a.group - b.group);
|
||
st.sealedHeld = [];
|
||
const now = performance.now();
|
||
for (const h of held) {
|
||
if (now - h.at > E2EE_HOLD_MS) { T.e2ee.held_expired++; continue; }
|
||
deliver(sub.principal, st, T, h.group, h.payload, h.fetched, h.meta);
|
||
}
|
||
}
|
||
}
|
||
renderE2ee();
|
||
}
|
||
|
||
function renderE2ee() {
|
||
const el = $("e2ee");
|
||
if (!el || !e2ee) return;
|
||
el.hidden = false;
|
||
const s = e2ee.state();
|
||
const roster = s.roster ? Object.entries(s.roster).map(([leaf, p]) => `${p === me.principal ? "you" : nameOf(p)} (leaf ${leaf})`) : [];
|
||
const where = e2ee.stats.removed != null ? `removed at epoch ${e2ee.stats.removed}` : s.epoch != null ? `epoch ${s.epoch}` : e2ee.role === "publisher" ? "waiting for the room's group" : "not a member";
|
||
$("e2ee-state").textContent = `${where}${roster.length ? ` · ${roster.length} member${roster.length === 1 ? "" : "s"}: ${roster.join(", ")}` : ""}`;
|
||
el.dataset.epoch = s.epoch != null ? String(s.epoch) : "";
|
||
}
|
||
|
||
// ------------------------------------------------------------------ stream termination
|
||
|
||
// WAI stream-termination: a stream log for each track this page receives (./termination.js),
|
||
// signed under a key the page makes for its session; a video stream nobody can see is ended.
|
||
const TERM_POLICY = {
|
||
unseen_grace_ms: Number(q.get("unseen_grace_ms") || 10000),
|
||
idle_ms: Number(q.get("idle_ms") || 0),
|
||
autoplay_grace_ms: Number(q.get("autoplay_grace_ms") || 0),
|
||
};
|
||
const UNSEEN_REASONS = ["hidden", "occluded", "display_off"];
|
||
let termSeed = null;
|
||
const attention = { occluded: false, activated: false, lastActivation: 0, waiters: [] };
|
||
const nowUs = () => Math.round((performance.timeOrigin + performance.now()) * 1000);
|
||
|
||
// The viewer's autoplay setting (§4.4): the `autoplay` parameter, else this page's storage.
|
||
function autoplaySetting() {
|
||
let v = q.get("autoplay");
|
||
if (!v) { try { v = localStorage.getItem("autoplay"); } catch { v = null; } }
|
||
return v === "off" ? "off" : "on";
|
||
}
|
||
|
||
// Under the setting off, media waits for the viewer's first activation: a stream is never
|
||
// started by autoplay against it (§4.4).
|
||
function autoplayAllowed() {
|
||
if (state.termination.autoplay !== "off" || attention.activated || (navigator.userActivation && navigator.userActivation.hasBeenActive)) return Promise.resolve();
|
||
status("Autoplay is off: press a key or click to start the room's media.");
|
||
return new Promise((r) => attention.waiters.push(r));
|
||
}
|
||
|
||
function eachSink(f) {
|
||
for (const sub of subscribed.values()) for (const st of Object.values(sub.tracks)) if (st.sink && !st.sink.terminated) f(st.sink);
|
||
}
|
||
|
||
// The session key, the panel (capability-constraints §4.1) and the observations (§4.2).
|
||
async function startTermination() {
|
||
termSeed = hex(crypto.getRandomValues(new Uint8Array(32)));
|
||
state.termination.policy = TERM_POLICY;
|
||
state.termination.autoplay = autoplaySetting();
|
||
const refresh = await term.measureRefresh(window);
|
||
state.termination.refresh_measured = refresh != null;
|
||
if (refresh == null) log("the refresh rate could not be measured (no animation frames): 60 Hz declared");
|
||
state.termination.panel = term.readDeclaration(window, refresh || 60000);
|
||
document.addEventListener("visibilitychange", () => {
|
||
const v = document.visibilityState === "hidden" ? term.UNSEEN : term.SEEN;
|
||
const t = nowUs();
|
||
eachSink((x) => x.observe(term.SIGNAL.visibility, v, t));
|
||
log(`visibility: ${document.visibilityState}`);
|
||
});
|
||
const canvas = $("canvas");
|
||
if (canvas && typeof IntersectionObserver === "function") {
|
||
new IntersectionObserver((entries) => {
|
||
attention.occluded = !entries[entries.length - 1].isIntersecting;
|
||
const t = nowUs();
|
||
eachSink((x) => x.observe(term.SIGNAL.occlusion, attention.occluded ? term.UNSEEN : term.SEEN, t));
|
||
}).observe(canvas);
|
||
}
|
||
for (const type of term.ACTIVATION_EVENTS) {
|
||
window.addEventListener(type, (e) => {
|
||
if (!term.isActivation(e)) return;
|
||
attention.activated = true;
|
||
for (const w of attention.waiters.splice(0)) w();
|
||
const t = nowUs();
|
||
attention.lastActivation = t;
|
||
eachSink((x) => x.observe(term.SIGNAL.interaction, 1, t));
|
||
}, { capture: true, passive: true });
|
||
}
|
||
// A panel that changes (a window moved to another display, its range switched) is declared
|
||
// again, and applies to every video stream from then (§3.3).
|
||
const redeclare = () => {
|
||
const d = term.readDeclaration(window, state.termination.panel.refresh_milli);
|
||
if (term.sameDeclaration(d, state.termination.panel)) return;
|
||
state.termination.panel = d;
|
||
const t = nowUs();
|
||
eachSink((x) => x.display(d, t));
|
||
log(`the panel changed: ${d.long}×${d.short}, ${d.bit_depth}-bit${d.hdr ? ", HDR" : ""}`);
|
||
};
|
||
window.addEventListener("resize", redeclare);
|
||
// A screen can change with no resize of the viewport: read again every second.
|
||
setInterval(() => { try { redeclare(); } catch (e) { fail(e); } }, 1000);
|
||
for (const mq of ["(video-dynamic-range: high)", "(dynamic-range: high)"]) {
|
||
try { matchMedia(mq).addEventListener("change", redeclare); } catch { /* no such feature */ }
|
||
}
|
||
setInterval(() => { try { terminationTick(); } catch (e) { fail(e); } }, 250);
|
||
}
|
||
|
||
function uaAutoplay(m) {
|
||
if (typeof navigator.getAutoplayPolicy !== "function") return null;
|
||
try { return navigator.getAutoplayPolicy(m === "audio" ? "audiocontext" : "mediaelement"); } catch { return null; }
|
||
}
|
||
|
||
// The open record of a stream just subscribed (§3.2), and the state at it (§4.2). The ladder is
|
||
// the participant's video tracks in its catalog; the rendition, this track's place among them.
|
||
function openSinkLog(sub, st, ns, tr, m) {
|
||
if (!termSeed) return;
|
||
const videoTracks = ((sub.catalog && sub.catalog.tracks) || []).filter((x) => x.packaging === "wai" && mediaOf(x) === "video");
|
||
const ua = uaAutoplay(m);
|
||
const open = {
|
||
stream: wai.trackDigest(JSON.stringify(ns), tr.name),
|
||
media: m,
|
||
start: attention.activated || (navigator.userActivation && navigator.userActivation.hasBeenActive) ? "user" : "autoplay",
|
||
autoplay: state.termination.autoplay,
|
||
ua_policy: ["allowed", "allowed-muted", "disallowed"].includes(ua) ? ua : null,
|
||
...TERM_POLICY,
|
||
display: m === "video" ? state.termination.panel : null,
|
||
ladder: m === "video" ? term.ladderOf(videoTracks) : null,
|
||
};
|
||
st.rendition = m === "video" ? Math.max(0, videoTracks.findIndex((x) => x.name === tr.name)) : null;
|
||
const t = nowUs();
|
||
st.sink = new term.SinkStream(wai, termSeed, open, t, { log });
|
||
if (document.visibilityState === "hidden") st.sink.observe(term.SIGNAL.visibility, term.UNSEEN, t);
|
||
if (attention.occluded) st.sink.observe(term.SIGNAL.occlusion, term.UNSEEN, t);
|
||
state.termination.streams.push({ principal: sub.principal, media: m, sink: st.sink });
|
||
}
|
||
|
||
// An object verified on its way to a decoder, gathered into its stream's segment (§4.5);
|
||
// false once the stream is terminated, and the object is then discarded (§4.3).
|
||
function receipted(st, T, bind, envelope, mediaUs, durUs) {
|
||
if (!st.sink || !bind) return true;
|
||
if (st.sink.received({ hash: bind.digest, bytes: envelope.length, mediaUs, durUs, rendition: st.rendition }, nowUs())) return true;
|
||
T.discarded_after_termination = (T.discarded_after_termination || 0) + 1;
|
||
return false;
|
||
}
|
||
|
||
// §4.3: a stream whose reason is due is terminated; a video stream ended as unseen is
|
||
// subscribed again, under a new log, once the page is seen, and one ended as idle or
|
||
// unattended once the viewer interacts.
|
||
function terminationTick() {
|
||
const t = nowUs();
|
||
const seen = document.visibilityState !== "hidden" && !attention.occluded;
|
||
for (const sub of subscribed.values()) {
|
||
for (const [m, st] of Object.entries(sub.tracks)) {
|
||
if (!st.sink) continue;
|
||
if (!st.sink.terminated) {
|
||
const reason = st.sink.due(t);
|
||
if (reason) endStream(sub, m, st, reason);
|
||
} else if (!st.reopening && state.phase !== "left" && ((seen && UNSEEN_REASONS.includes(st.sink.terminatedAs))
|
||
|| (["idle", "autoplay_unattended"].includes(st.sink.terminatedAs) && attention.lastActivation > st.sink.terminatedAt))) {
|
||
st.reopening = true;
|
||
log(`${m} of ${sub.principal.slice(0, 8)}: ${st.sink.terminatedAs === "idle" || st.sink.terminatedAs === "autoplay_unattended" ? "the viewer is back" : "seen again"}; subscribed under a new log`);
|
||
subscribeMedia(sub, st.tr).catch(fail);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// The termination signed, and the subscription ended (`unsubscribe` false where the session
|
||
// itself is going).
|
||
function endStream(sub, m, st, reason, unsubscribe = true) {
|
||
if (!st.sink || st.sink.terminated) return;
|
||
const r = st.sink.terminate(reason, nowUs());
|
||
const T = tally(sub.principal)[m];
|
||
T.terminated = (T.terminated || 0) + 1;
|
||
if (unsubscribe && session) for (const x of st.subs || []) session.unsubscribe(x);
|
||
log(`${m} of ${sub.principal.slice(0, 8)} terminated (${reason}) at media ${r.media} µs, after ${st.sink.segments} segment receipts`);
|
||
}
|
||
|
||
function terminationSnapshot() {
|
||
const tm = state.termination;
|
||
return {
|
||
panel: tm.panel, refresh_measured: tm.refresh_measured, policy: tm.policy, autoplay: tm.autoplay,
|
||
visibility: document.visibilityState, occluded: attention.occluded,
|
||
streams: tm.streams.map((x) => { const y = x.sink.summary(); delete y.records; return { principal: x.principal, ...y, records: x.sink.records.length }; }),
|
||
};
|
||
}
|
||
|
||
// Every stream log this page signed, with its records and the page's own verdict on it
|
||
// (stream-termination §6, through wai-live-web).
|
||
// The corpus live-conformance/termination at `url`, checked through this page's wasm build: the
|
||
// failures (none when every match and log gets its stated outcome).
|
||
window.roomTest.terminationCorpus = async (url) => JSON.parse(wai.terminationCorpusVerify(await (await fetch(url, { cache: "no-store" })).text()));
|
||
window.roomTest.terminationLogs = () => state.termination.streams.map((x) => ({ principal: x.principal, ...x.sink.summary(), verdict: x.sink.verdict() }));
|
||
|
||
// §4: a track's route, from the user agent's answers for its codec string. This page loads
|
||
// no wasm decoder for a room's tracks, so a track routed anywhere but the user agent's
|
||
// decoder is not decoded here.
|
||
async function routeTrack(P, tr, m) {
|
||
const capability = (tr.wai_capabilities && tr.wai_capabilities[0]) || media.CAPABILITY_OF(tr.codec);
|
||
const params = m === "audio"
|
||
? { "wai:codec": tr.codec, "wai:sample_rate": Number(tr.samplerate || 48000), "wai:channel_count": Number(tr.channelConfig || 1) }
|
||
: { "wai:codec": tr.codec, "wai:frame_width": Number(tr.width), "wai:frame_height": Number(tr.height) };
|
||
const object = { name: `${P.slice(0, 8)}/${m}`, capability, codec: tr.codec, need: tr.wai_determinism || "none", params };
|
||
const r = await buildReport(wai, [object], { ask: (o) => askAndCount(o) });
|
||
const entry = r.table[0] || { route: "none", reason: r.reason, tier: "none", energy: null };
|
||
state.binding.routes[object.name] = { capability, codec: tr.codec, ...entry };
|
||
if (entry.route !== "ua") log(`${m} of ${P.slice(0, 8)} (${capability}, ${tr.codec}) routed ${entry.route}: ${entry.reason || JSON.stringify(entry.legs)}`);
|
||
return entry;
|
||
}
|
||
|
||
function askAndCount(o) {
|
||
state.binding.asked.push(o.codec);
|
||
return askUserAgent(o, log);
|
||
}
|
||
|
||
// §10: each tile's frame, its binding as drawn, and its state.
|
||
function bindingRows() {
|
||
const rows = [];
|
||
if (!compositor) return rows;
|
||
for (const [P, d] of compositor.drawn) {
|
||
const sub = subscribed.get(P);
|
||
const st = sub && sub.tracks.video;
|
||
const chain = chainToken(st && d.group != null ? st.chainOf.get(d.group) : null);
|
||
const s = stateOf(wai, d.binding ? { binding: d.binding } : null, chain, "not_applicable");
|
||
const route = st && st.route;
|
||
rows.push({
|
||
// An unbound frame (this page's own camera, say) has no group, so no chain to show.
|
||
// A chain this page anchored, or filled below its anchor, says from which group it holds
|
||
// the chain (browser-sink §10).
|
||
principal: P, name: nameOf(P), ...s, provenance: "not_applicable", group: d.group,
|
||
chain: !d.binding ? "—" : chain === "anchor" ? `anchor at group ${d.group}` : chain === "history" ? `history, group ${d.group}, below the anchor` : chain,
|
||
route: route ? route.route : "none", energy: route ? route.energy : null,
|
||
// Whether the digest is one this page's verifier accepted, under that receipt.
|
||
receipt_verified: !!(st && s.digest && st.verifiedBy.get(s.digest) === s.receipt),
|
||
});
|
||
}
|
||
return rows;
|
||
}
|
||
|
||
function makeDecoder(P, tr, T, st) {
|
||
if (tr.role === "audio") {
|
||
const d = new AudioDecoder({
|
||
output: (ad) => { T.decoded++; if (compositor) compositor.addAudio(P, ad); else ad.close(); },
|
||
error: (e) => { T.decode_errors++; log(`audio decoder of ${P.slice(0, 8)}: ${e.message || e}`); },
|
||
});
|
||
d.configure({ codec: "opus", sampleRate: tr.samplerate, numberOfChannels: Number(tr.channelConfig || 1) });
|
||
return d;
|
||
}
|
||
const d = new VideoDecoder({
|
||
output: (vf) => {
|
||
T.decoded++;
|
||
// The binding entered for this frame's chunk, attached to the frame (§9.2 step 2).
|
||
if (bindings.output(d, vf)) T.bound = (T.bound || 0) + 1; else T.unbound = (T.unbound || 0) + 1;
|
||
// A host's screen is drawn by the remote desktop viewer (WAI remote-desktop), not tiled.
|
||
if (remote && st.m === "screen") remote.frame(vf);
|
||
else if (compositor) compositor.setVideo(P, vf); else vf.close();
|
||
},
|
||
error: (e) => { T.decode_errors++; bindings.reset(d); log(`video decoder of ${P.slice(0, 8)}: ${e.message || e}`); st.waitingKey = true; },
|
||
});
|
||
d.configure({ codec: tr.codec, codedWidth: tr.width, codedHeight: tr.height, optimizeForLatency: true });
|
||
return d;
|
||
}
|
||
|
||
function decodeEnvelope(P, st, T, group, envelope, fetched, bind) {
|
||
let e;
|
||
try { e = wai.unpack(envelope); } catch (err) { T.decode_errors++; T.envelope_errors = (T.envelope_errors || 0) + 1; return log(`an envelope that does not read: ${err.message || err}`); }
|
||
const manifest = JSON.parse(e.manifest);
|
||
const cap = e.capability;
|
||
const payload = e.payload();
|
||
e.free();
|
||
if (!st.tr.wai_capabilities.includes(cap)) {
|
||
// A capability the catalog did not declare is a protocol error (§4), not a fallback.
|
||
T.decode_errors++;
|
||
return log(`a capability the catalog does not declare: ${cap}`);
|
||
}
|
||
// Decoded only through its route (browser-sink §4.3): this page decodes on the user
|
||
// agent's decoders alone, and an object routed elsewhere is not decoded here.
|
||
if (!st.route || st.route.route !== "ua") { T.not_routed = (T.not_routed || 0) + 1; return; }
|
||
// A codec that met an error is closed for good (WebCodecs): a new one is made, and the
|
||
// video waits for its next key frame.
|
||
if (st.decoder.state === "closed") {
|
||
st.decoder = makeDecoder(P, st.tr, T, st);
|
||
st.waitingKey = true;
|
||
T.decoder_restarts = (T.decoder_restarts || 0) + 1;
|
||
log(`${st.m} decoder of ${P.slice(0, 8)}: made again after an error`);
|
||
}
|
||
const d = st.decoder;
|
||
if (d.state !== "configured") return;
|
||
try {
|
||
if (cap === "wai.audio.opus") {
|
||
const packets = media.oggOpusPackets(payload);
|
||
// Fetched audio is earlier than the live stream: decoded on a decoder of its own and
|
||
// counted, never mixed.
|
||
if (fetched) return decodeFetchedAudio(st, T, packets, manifest);
|
||
if (!receipted(st, T, bind, envelope, manifest.target.timestamp_us, packets.length * AUDIO_FRAME_MS * 1000)) return;
|
||
for (const pk of packets) {
|
||
d.decode(new EncodedAudioChunk({ type: "key", timestamp: manifest.target.timestamp_us, data: pk }));
|
||
}
|
||
return;
|
||
}
|
||
const units = cap === "wai.video.vp9" ? media.ivfFrames(payload).map((f) => f.data) : cap === "wai.video.avc" || cap === "wai.video.hevc" ? [payload] : media.av1Units(payload);
|
||
const key = !!manifest.target.key;
|
||
// A fetched group older than the live stream is decoded on its own decoder.
|
||
if (fetched) return decodeFetched(P, st, T, units, key, manifest);
|
||
if (st.waitingKey && !key) return;
|
||
// Each group rides a stream of its own and waits for its receipts, so under load a
|
||
// group can be released after a later one: decoded then, it would corrupt the
|
||
// decoder's references. It is dropped, and counted.
|
||
if (st.lastVideoGroup != null && group <= st.lastVideoGroup) {
|
||
T.late_dropped = (T.late_dropped || 0) + 1;
|
||
return;
|
||
}
|
||
if (!receipted(st, T, bind, envelope, manifest.target.timestamp_us, Math.round(1e6 / Number(st.tr.framerate || 30)))) return;
|
||
st.lastVideoGroup = group;
|
||
st.waitingKey = false;
|
||
// Every object this decoder is given was verified first, so each frame it outputs may
|
||
// be bound (§7.2); the tier depends on the frame's format, read when it is output.
|
||
const route = st.route;
|
||
const entry = bind ? { principal: P, group, binding: (vf) => wai.makeBinding(bind.digest, bind.receipt, wai.frameTier(route.route, route.tier, vf.format == null, false)) } : null;
|
||
if (remote && st.m === "screen" && bind) remote.chunk(group, manifest.target.timestamp_us, bind.digest);
|
||
for (const u of units) {
|
||
bindings.chunk(d, manifest.target.timestamp_us, entry);
|
||
d.decode(new EncodedVideoChunk({ type: key ? "key" : "delta", timestamp: manifest.target.timestamp_us, data: u }));
|
||
}
|
||
} catch (err) {
|
||
T.decode_errors++;
|
||
log(`decode of ${st.m} group ${group} of ${P.slice(0, 8)}: ${err.message || err}`);
|
||
}
|
||
}
|
||
|
||
function decodeFetchedAudio(st, T, packets, manifest) {
|
||
if (!st.fetchDecoder) {
|
||
st.fetchDecoder = new AudioDecoder({ output: (ad) => { T.fetched_decoded++; ad.close(); }, error: (e) => { T.decode_errors++; log(`fetched audio decoder: ${e.message || e}`); } });
|
||
st.fetchDecoder.configure({ codec: "opus", sampleRate: st.tr.samplerate, numberOfChannels: Number(st.tr.channelConfig || 1) });
|
||
}
|
||
for (const pk of packets) st.fetchDecoder.decode(new EncodedAudioChunk({ type: "key", timestamp: manifest.target.timestamp_us, data: pk }));
|
||
}
|
||
|
||
// The earlier groups a late joiner fetched, decoded in order on a decoder of their own from
|
||
// the first key frame, so the live decoder's state is not disturbed.
|
||
function decodeFetched(P, st, T, units, key, manifest) {
|
||
if (!st.fetchDecoder) {
|
||
if (!key) return;
|
||
st.fetchDecoder = new VideoDecoder({ output: (vf) => { T.fetched_decoded++; vf.close(); }, error: (e) => { T.decode_errors++; log(`fetched video decoder: ${e.message || e}`); } });
|
||
st.fetchDecoder.configure({ codec: st.tr.codec, codedWidth: st.tr.width, codedHeight: st.tr.height });
|
||
}
|
||
for (const u of units) st.fetchDecoder.decode(new EncodedVideoChunk({ type: key ? "key" : "delta", timestamp: manifest.target.timestamp_us, data: u }));
|
||
}
|
||
|
||
// Objects that waited past RECEIPT_WAIT_MS for their receipts, and receipts whose object
|
||
// never came: the latter accounted for by inclusion (jwp-receipts §4.4).
|
||
function sweepMissing() {
|
||
const now = performance.now();
|
||
for (const sub of subscribed.values()) {
|
||
for (const st of Object.values(sub.tracks)) {
|
||
const T = tally(sub.principal)[st.m];
|
||
for (const [g, p] of st.pending) {
|
||
if (now - p.at > RECEIPT_WAIT_MS) { st.pending.delete(g); T.refused++; log(`${st.m} group ${g}: its receipts did not arrive`); continue; }
|
||
// A receipts-track object the subscription did not deliver (one that reached the
|
||
// relay just before the subscription did) is recovered by FETCH of its group
|
||
// (moq-streaming-format §4.1).
|
||
if (now - p.at > RECOVER_AFTER_MS && !p.recovering) {
|
||
p.recovering = true;
|
||
T.receipt_fetches = (T.receipt_fetches || 0) + 1;
|
||
session.fetch(st.ns, st.tr.wai_receipts_track, { group_order: 1, location_filter: { type: "absolute_group_end", start: [g, 0], end_group_delta: 0 } }, (j, payload) => st.onReceipts(j.group_id, payload))
|
||
.then(() => { if (!st.pending.has(g)) T.receipts_recovered = (T.receipts_recovered || 0) + 1; })
|
||
.catch((e) => log(`${st.m} group ${g}: the FETCH of its receipts: ${e.message || e}`));
|
||
}
|
||
}
|
||
for (const [g, at] of st.receiptsAt) {
|
||
if (st.seen.has(g)) { st.receiptsAt.delete(g); continue; }
|
||
// Receipts of a group before the media subscription's first are of no object this
|
||
// subscription asked for (the receipts track was subscribed first): not a drop.
|
||
if (T.first_live_group == null || g < T.first_live_group) { if (now - at > MISSING_AFTER_MS) st.receiptsAt.delete(g); continue; }
|
||
if (now - at < MISSING_AFTER_MS) continue;
|
||
st.receiptsAt.delete(g);
|
||
const r = JSON.parse(st.verifier.inclusion(BigInt(g), 0n));
|
||
if (r.ok) T.dropped_accounted++; else T.dropped_unaccounted++;
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------- seeking back
|
||
|
||
// Seeking back (WAI live-recording §8, live-rooms §10.6): a participant's last seconds, by
|
||
// FETCH below the live edge of its tracks and of their receipts tracks, which the relay
|
||
// answers from its DVR window. Every object is verified against its receipt exactly as a
|
||
// live one is (the same verifier, which admits the earlier groups as history), the video
|
||
// is played from its first key frame in a view of its own at the pace it was captured, the
|
||
// audio decoded and counted, and the page then returns to live. What the relay no longer
|
||
// holds, or withholds, comes as an End of Unknown Range, counted and never played.
|
||
const AUDIO_FRAME_MS = Number(q.get("audioFrameMs") || 20);
|
||
let dvrStop = null;
|
||
|
||
function renderDvr() {
|
||
const d = state.dvr;
|
||
const run = d.runs[d.runs.length - 1];
|
||
const canvas = $("dvr-canvas");
|
||
if (canvas) canvas.hidden = d.mode !== "playing";
|
||
if ($("live")) $("live").hidden = d.mode === "live";
|
||
if ($("seek")) $("seek").disabled = d.mode !== "live" || !subscribedVideo();
|
||
const el = $("dvr-status");
|
||
if (!el) return;
|
||
if (!run) { el.textContent = ""; return; }
|
||
const v = run.video, a = run.audio;
|
||
const what = `${run.who}, ${run.seconds} s back: video ${v.verified} of ${v.fetched} verified, audio ${a.verified} of ${a.fetched}${v.unknown + a.unknown ? `; ${v.unknown + a.unknown} range(s) the relay no longer holds` : ""}`;
|
||
el.textContent = d.mode === "seeking" ? `Fetching ${what}` : d.mode === "playing" ? `Playing ${what}; ${v.played} frames shown` : `Back to live after ${what}; ${v.played} frames shown`;
|
||
}
|
||
|
||
function subscribedVideo() {
|
||
return [...subscribed.values()].find((s) => s.tracks.video && tally(s.principal).video.last_live_group != null);
|
||
}
|
||
|
||
// FETCH groups `from` to `to` of a track and of its receipts track, the receipts first,
|
||
// each verified; returns the verified objects in group order.
|
||
async function fetchVerified(st, m, from, to, c) {
|
||
const params = { group_order: 1, subscriber_priority: SUB_PRIORITY[m] + 2, location_filter: { type: "absolute_group_end", start: [from, 0], end_group_delta: to - from } };
|
||
const rec = [];
|
||
const r1 = await session.fetch(st.ns, st.tr.wai_receipts_track, params, (j, p) => rec.push([j.group_id, p]));
|
||
for (const [g, p] of rec) {
|
||
const r = JSON.parse(st.verifier.receipts(BigInt(g), p, BigInt(Date.now())));
|
||
if (!r.ok) c.receipts_refused++;
|
||
}
|
||
const objs = [];
|
||
const r2 = await session.fetch(st.ns, st.tr.name, params, (j, p) => objs.push([j.group_id, j.object_id, p]));
|
||
c.unknown += r1.ranges.length + r2.ranges.length;
|
||
const ok = [];
|
||
for (const [g, o, p] of objs) {
|
||
c.fetched++;
|
||
const r = JSON.parse(st.verifier.object(BigInt(g), BigInt(o), p));
|
||
if (r.ok) { c.verified++; ok.push([g, p]); } else { c.refused++; log(`seek back: ${m} group ${g} refused: ${r.reason}`); }
|
||
}
|
||
return ok;
|
||
}
|
||
|
||
async function seekBack(seconds = 10) {
|
||
if (e2ee) throw new Error("an end-to-end encrypted room has no seeking back: the relay keeps nothing of it");
|
||
if (state.dvr.mode !== "live") return state.dvr.runs[state.dvr.runs.length - 1];
|
||
const sub = subscribedVideo();
|
||
if (!sub) throw new Error("no participant's video to seek back into");
|
||
const counts = () => ({ from: null, to: null, fetched: 0, verified: 0, refused: 0, receipts_refused: 0, unknown: 0, decoded: 0, played: 0 });
|
||
const run = { principal: sub.principal, who: nameOf(sub.principal), seconds, video: counts(), audio: counts(), started_at: performance.now(), playing_at: null, live_at: null, error: null };
|
||
state.dvr.runs.push(run);
|
||
state.dvr.mode = "seeking";
|
||
renderDvr();
|
||
try {
|
||
const got = {};
|
||
for (const m of ["video", "audio"]) {
|
||
const st = sub.tracks[m];
|
||
const T = tally(sub.principal)[m];
|
||
if (!st || T.last_live_group == null) continue;
|
||
const perSecond = m === "video" ? Number(st.tr.framerate || 30) : 1000 / AUDIO_FRAME_MS;
|
||
const to = T.last_live_group;
|
||
const from = Math.max(0, to - Math.round(seconds * perSecond));
|
||
run[m].from = from;
|
||
run[m].to = to;
|
||
got[m] = await fetchVerified(st, m, from, to, run[m]);
|
||
}
|
||
log(`seek back ${seconds} s into ${run.who}: video groups ${run.video.from}..${run.video.to}, ${run.video.verified} of ${run.video.fetched} verified; audio ${run.audio.verified} of ${run.audio.fetched}; ${run.video.unknown + run.audio.unknown} unknown range(s)`);
|
||
state.dvr.mode = "playing";
|
||
run.playing_at = performance.now();
|
||
renderDvr();
|
||
await playBack(sub, got, run);
|
||
} catch (e) {
|
||
run.error = String(e.message || e);
|
||
log(`seek back: ${run.error}`);
|
||
} finally {
|
||
backToLive();
|
||
}
|
||
return run;
|
||
}
|
||
|
||
// The fetched video, decoded and drawn at the pace of its timestamps from its first key
|
||
// frame; the audio decoded alongside and counted.
|
||
async function playBack(sub, got, run) {
|
||
const canvas = $("dvr-canvas");
|
||
const ctx = canvas && canvas.getContext("2d");
|
||
const vst = sub.tracks.video;
|
||
const vdec = new VideoDecoder({
|
||
output: (vf) => {
|
||
run.video.decoded++;
|
||
if (ctx && state.dvr.mode === "playing") { ctx.drawImage(vf, 0, 0, canvas.width, canvas.height); run.video.played++; renderDvr(); }
|
||
vf.close();
|
||
},
|
||
error: (e) => log(`seek back: video decoder: ${e.message || e}`),
|
||
});
|
||
vdec.configure({ codec: vst.tr.codec, codedWidth: vst.tr.width, codedHeight: vst.tr.height });
|
||
const ast = sub.tracks.audio;
|
||
let adec = null;
|
||
if (ast && got.audio) {
|
||
adec = new AudioDecoder({ output: (ad) => { run.audio.decoded++; ad.close(); }, error: (e) => log(`seek back: audio decoder: ${e.message || e}`) });
|
||
adec.configure({ codec: "opus", sampleRate: ast.tr.samplerate, numberOfChannels: Number(ast.tr.channelConfig || 1) });
|
||
for (const [, env] of got.audio) {
|
||
try {
|
||
const e = wai.unpack(env);
|
||
const manifest = JSON.parse(e.manifest);
|
||
const payload = e.payload();
|
||
e.free();
|
||
for (const pk of media.oggOpusPackets(payload)) adec.decode(new EncodedAudioChunk({ type: "key", timestamp: manifest.target.timestamp_us, data: pk }));
|
||
} catch (err) { log(`seek back: audio: ${err.message || err}`); }
|
||
}
|
||
}
|
||
const frames = [];
|
||
for (const [, env] of got.video || []) {
|
||
try {
|
||
const e = wai.unpack(env);
|
||
const manifest = JSON.parse(e.manifest);
|
||
const cap = e.capability;
|
||
const payload = e.payload();
|
||
e.free();
|
||
const units = cap === "wai.video.vp9" ? media.ivfFrames(payload).map((f) => f.data) : cap === "wai.video.avc" || cap === "wai.video.hevc" ? [payload] : media.av1Units(payload);
|
||
frames.push({ key: !!manifest.target.key, ts: manifest.target.timestamp_us, units });
|
||
} catch (err) { log(`seek back: video: ${err.message || err}`); }
|
||
}
|
||
const first = frames.findIndex((f) => f.key);
|
||
let stopped = false;
|
||
dvrStop = () => { stopped = true; };
|
||
if (first >= 0) {
|
||
const t0 = performance.now();
|
||
const ts0 = frames[first].ts;
|
||
for (const f of frames.slice(first)) {
|
||
if (stopped) break;
|
||
const due = t0 + (f.ts - ts0) / 1000;
|
||
const wait = due - performance.now();
|
||
if (wait > 0) await new Promise((r) => setTimeout(r, Math.min(wait, 1000)));
|
||
for (const u of f.units) vdec.decode(new EncodedVideoChunk({ type: f.key ? "key" : "delta", timestamp: f.ts, data: u }));
|
||
}
|
||
}
|
||
try { await vdec.flush(); } catch { /* closed */ }
|
||
if (adec) { try { await adec.flush(); } catch { /* closed */ } adec.close(); }
|
||
vdec.close();
|
||
}
|
||
|
||
function backToLive() {
|
||
if (dvrStop) { dvrStop(); dvrStop = null; }
|
||
const run = state.dvr.runs[state.dvr.runs.length - 1];
|
||
if (run && run.live_at == null) run.live_at = performance.now();
|
||
state.dvr.mode = "live";
|
||
renderDvr();
|
||
}
|
||
|
||
window.roomTest.seekBack = (seconds) => seekBack(seconds);
|
||
if ($("seek")) {
|
||
$("seek").addEventListener("click", () => {
|
||
const s = Math.max(1, Math.min(900, Number(($("seek-s") || {}).value || 10)));
|
||
seekBack(s).catch(fail);
|
||
});
|
||
}
|
||
if ($("live")) $("live").addEventListener("click", () => { if (dvrStop) dvrStop(); });
|
||
setInterval(() => { if (state.dvr.mode === "live") renderDvr(); }, 1000);
|
||
|
||
// ------------------------------------------------------------------------------- GOAWAY
|
||
|
||
// GOAWAY (MOQT §9.2): no new request on the old session; a new session to the URI it
|
||
// names (or the same one), the tracks published again and every subscription made again
|
||
// there, then the old session closed by this end, before the relay's timeout.
|
||
async function onGoaway(old, g) {
|
||
state.goaways.push({ uri: g.uri, timeout_ms: g.timeout_ms, at: performance.now() });
|
||
try {
|
||
const opts = { ...transportOpts };
|
||
if (g.uri) {
|
||
// The session moves only within this page's own origin (or, on a loopback page, to a
|
||
// loopback address): a GOAWAY naming another is refused, and the session ends.
|
||
const c = checkEndpoint(g.uri, kindOf(g.uri), location.href);
|
||
if (!c.ok) {
|
||
state.goaways[state.goaways.length - 1].refused = c.reason;
|
||
state.ended = { code: 0, reason: "GOAWAY to another origin refused" };
|
||
state.phase = "ended";
|
||
log(`refused to move after GOAWAY to ${g.uri}: ${c.reason}`);
|
||
status(`The relay asked this page to move its session to ${g.uri}, which is not this page's own (${c.reason}). The page refused, and its session has ended. Reload the page to enter again.`);
|
||
old.close(0, "GOAWAY to another origin refused");
|
||
return;
|
||
}
|
||
if (c.url.startsWith("https://")) { opts.webtransport = c.url; opts.force = "webtransport"; }
|
||
else if (c.url.startsWith("ws")) { opts.websocket = c.url; opts.force = "websocket"; }
|
||
}
|
||
transportOpts = opts;
|
||
const s = await openSession(opts);
|
||
if (window.roomTest.role !== "clock") await publishTracks(s);
|
||
// This end's tracks now go to the new session, so the old one closes, by this end and
|
||
// before the relay's timeout; then every subscription is made again there.
|
||
old.close(0, "moved after GOAWAY");
|
||
state.goaways[state.goaways.length - 1].moved = { transport: s.t.kind, at: performance.now() };
|
||
log(`moved to ${g.uri || "the same relay"} after GOAWAY`);
|
||
if (window.roomTest.role !== "clock") {
|
||
const others = [...subscribed.keys()];
|
||
// Every stream of the old session ended with it; the new session's are new streams.
|
||
for (const sub of subscribed.values()) for (const [m, st] of Object.entries(sub.tracks)) endStream(sub, m, st, "ended", false);
|
||
subscribed.clear();
|
||
eventsSubscribed.clear();
|
||
await subscribeEvents(fx.clock.principal);
|
||
for (const P of others) await subscribeParticipant(P);
|
||
log(`subscriptions made again on the new session: ${others.length} participant(s)`);
|
||
}
|
||
} catch (e) {
|
||
fail(e);
|
||
}
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------- the views
|
||
|
||
function renderTimeline(e) {
|
||
const li = document.createElement("li");
|
||
li.dataset.kind = e.kind;
|
||
const p = e.payload || {};
|
||
const what = e.kind === "caption" ? `“${p.text}” (${p.origin_label})` : e.kind === "chat" ? p.text : p.detail ? `${(p.subject || "").slice(0, 8)} ${JSON.stringify(p.detail)}` : "";
|
||
li.textContent = `t=${(e.t / 1e6).toFixed(3)}s ${e.who} ${e.kind} ${what}`;
|
||
$("timeline").appendChild(li);
|
||
}
|
||
|
||
function renderCaption() {
|
||
const c = state.captions[state.captions.length - 1];
|
||
const el = $("caption");
|
||
el.innerHTML = "";
|
||
const label = document.createElement("span");
|
||
label.className = `origin origin-${c.origin}`;
|
||
label.textContent = c.origin === "human" ? "Human captioner" : c.origin === "machine" ? "Machine-generated" : "Machine-generated, reviewed";
|
||
const text = document.createElement("span");
|
||
text.className = "text";
|
||
text.textContent = c.text;
|
||
el.append(label, text);
|
||
el.dataset.origin = c.origin;
|
||
}
|
||
|
||
// A participant's own events: captions and chats, signed with its key.
|
||
window.roomTest.sendCaption = async ({ text, origin = 1, model = null, lang = "en", duration_ms = 3000 }) => {
|
||
if (cowatch) throw new Error("a watch room's page sends its co-watch statements alone");
|
||
const p = wai.captionPayload(duration_ms, origin, model || undefined, lang, text);
|
||
const r = await eventsTrack.emit(0x0a, p);
|
||
await (records = records.then(() => onRecord(r.record)));
|
||
return r.view;
|
||
};
|
||
window.roomTest.sendChat = async (text) => {
|
||
if (cowatch) throw new Error("a watch room's page sends its co-watch statements alone");
|
||
const r = await eventsTrack.emit(0x03, wai.chatPayload(text));
|
||
await (records = records.then(() => onRecord(r.record)));
|
||
return r.view;
|
||
};
|
||
window.roomTest.snapshot = () => ({
|
||
phase: state.phase,
|
||
refused: state.refused,
|
||
mode: state.mode,
|
||
me: me && { principal: me.principal, role: me.role || window.roomTest.role },
|
||
room: fx && fx.room,
|
||
grant: state.grant,
|
||
ended: state.ended,
|
||
errors: state.errors,
|
||
transport: state.transport,
|
||
sessions: state.sessions,
|
||
published: state.published,
|
||
codec: state.publishCodec,
|
||
capture: state.capture,
|
||
subs: state.subs,
|
||
dvr: state.dvr,
|
||
captions: state.captions,
|
||
timeline: state.timeline.map((e) => ({ t: e.t, who: e.who, kind: e.kind })),
|
||
goaways: state.goaways,
|
||
roomState: state.room,
|
||
frame: compositor && compositor.frame,
|
||
layouts: compositor && compositor.layouts,
|
||
drawn: compositor && Object.fromEntries(compositor.drawn),
|
||
draws: compositor && compositor.draws,
|
||
samples: compositor && compositor.samples(),
|
||
mixed: compositor && compositor.mixed,
|
||
worklet: compositor && compositor.workletStats,
|
||
audioState: compositor && compositor.ac && compositor.ac.state,
|
||
session: session && { counters: session.counters, transport: session.t.stats ? session.t.stats() : {} },
|
||
verifiers: Object.fromEntries([...subscribed.entries()].map(([P, s]) => [P, Object.fromEntries(Object.entries(s.tracks).map(([m, st]) => [m, JSON.parse(st.verifier.stats())]))])),
|
||
e2ee: e2ee && { ...state.e2ee, ...e2ee.stats, at: e2ee.at, stopped: e2ee.stopped, member: e2ee.state(), dom: $("e2ee") && !$("e2ee").hidden ? $("e2ee").textContent : null },
|
||
age: state.age,
|
||
co: co && co.snapshot(),
|
||
cowatch: cowatch && { ...cowatch.snapshot(), dom: $("cw-sync") ? $("cw-sync").textContent : null },
|
||
termination: terminationSnapshot(),
|
||
});
|
||
// A shared control of a watch room (co-watch §5): `play`, `pause`, `seek` (to, µs) or `rate`
|
||
// (millionths), at the policy's lead from now. Resolves with the instant it takes effect.
|
||
window.roomTest.watchControl = (name, arg = 0) => watchControl(name, arg);
|
||
// A participant's grant presented as a bare token, with no proof: the relay must refuse
|
||
// the session (live-rooms §10.5.2, UNAUTHORIZED). Resolves with how the session ended.
|
||
window.roomTest.tryBareGrant = async () => {
|
||
const t = await connect({ ...transportOpts, log: () => {} });
|
||
let setup = "exchanged";
|
||
try {
|
||
await MoqSession.connect(t, moq, { token: unhex(me.grant), implementation: "joule-moq-web/room-bare", log: () => {} });
|
||
} catch (e) {
|
||
setup = String(e.message || e);
|
||
}
|
||
const closed = await Promise.race([t.closed, new Promise((r) => setTimeout(() => r(null), 8000))]);
|
||
return { transport: t.kind, setup, closed };
|
||
};
|
||
|
||
// This page's grant presented with a proof made by another key: refused the session too
|
||
// (§10.5.4 step 4, pop_wrong_key).
|
||
window.roomTest.tryWrongKey = async () => {
|
||
const t = await connect({ ...transportOpts, log: () => {} });
|
||
let setup = "exchanged";
|
||
try {
|
||
await MoqSession.connect(t, moq, { grant: { record: unhex(me.grant), seed: unhex(me.seed), proofBy: crypto.getRandomValues(new Uint8Array(32)) }, implementation: "joule-moq-web/room-wrong-key", log: () => {} });
|
||
} catch (e) {
|
||
setup = String(e.message || e);
|
||
}
|
||
const closed = await Promise.race([t.closed, new Promise((r) => setTimeout(() => r(null), 8000))]);
|
||
return { transport: t.kind, setup, closed };
|
||
};
|
||
|
||
// End-to-end encryption, for the tests.
|
||
window.roomTest.e2eeSent = (groups) => groups.map((g) => e2eeSent.get(g)).filter(Boolean);
|
||
// The next video group is sealed and then has a byte changed, as a relay could change it.
|
||
window.roomTest.e2eeTamperNext = () => { state.e2ee.tamperNext = true; };
|
||
// An object sealed under this page's latest epoch on its own video track, not sent.
|
||
window.roomTest.e2eeSealAsMe = (group) => {
|
||
const ns = [fx.room, me.principal, "video"];
|
||
const sl = e2ee.seal(ns, group, PRIORITY.video, enc.encode("sealed by a member, kept back"), true, VIDEO_TIMEOUT_MS);
|
||
return sl && { ns, group, priority: PRIORITY.video, props: hex(sl.props), payload: hex(sl.payload), epoch: sl.epoch };
|
||
};
|
||
// An object as another page sealed it, opened here through the receiving checks.
|
||
window.roomTest.e2eeTryOpen = (o) => {
|
||
const v = e2ee.open(o.ns, "0", o.group, 0, o.priority, unhex(o.props), unhex(o.payload));
|
||
delete v.payload;
|
||
return v;
|
||
};
|
||
// The next video object received from `P`, before it is opened, presented again with each
|
||
// change a relay could make to it, then as it came, then once more: each outcome.
|
||
window.roomTest.e2eeTamperSuite = (P) => new Promise((resolve, reject) => {
|
||
const sub = subscribed.get(P);
|
||
const st = sub && sub.tracks.video;
|
||
if (!st) return reject(new Error(`no video of ${P.slice(0, 8)}`));
|
||
const t = setTimeout(() => { st.intercept = null; reject(new Error("no object arrived")); }, 20000);
|
||
st.intercept = (x) => { clearTimeout(t); resolve(tamperSuite(st, x)); };
|
||
});
|
||
// Stop publishing and stay: PUBLISH_DONE on each media track, the session and every
|
||
// subscription kept. The room's clock records the leave; the page keeps receiving.
|
||
window.roomTest.e2eeStopPublishing = async () => {
|
||
state.stoppedPublishing = true;
|
||
if (state.capture && state.capture.stream) for (const t of state.capture.stream.getTracks()) t.stop();
|
||
let n = 0;
|
||
for (const r of published) {
|
||
if (!r.ns || !["audio", "video"].includes(r.ns[2])) continue;
|
||
await r.writer.write(moq.encodeControl(JSON.stringify({ message: "PUBLISH_DONE", status: 2, reason: "stopped publishing" })));
|
||
n++;
|
||
}
|
||
return n;
|
||
};
|
||
|
||
function tamperSuite(st, x) {
|
||
const { group, payload, meta } = x;
|
||
const s = e2ee.state();
|
||
const open = (props, pl, o = {}) => {
|
||
const v = e2ee.open(st.ns, st.tr.name, o.group != null ? o.group : group, o.object != null ? o.object : meta.object || 0, o.priority != null ? o.priority : meta.priority, props, pl);
|
||
return v.ok ? "opened" : v.reason || `pending ${v.pending}`;
|
||
};
|
||
const alter = (kind, kid = 0n) => moq.e2eeAlteredProperties(meta.props, kind, kid);
|
||
const flipped = payload.slice();
|
||
flipped[flipped.length - 1] ^= 1;
|
||
const out = {
|
||
payload_byte_changed: open(meta.props, flipped),
|
||
priority_changed: open(meta.props, payload, { priority: (meta.priority + 1) & 255 }),
|
||
moved_to_another_group: open(meta.props, payload, { group: group + 1000000 }),
|
||
key_id_removed: open(alter("drop_key_id"), payload),
|
||
key_id_outside_too: open(alter("key_id_outside", 1n), payload),
|
||
object_id_past_32_bits: open(meta.props, payload, { object: 2 ** 32 }),
|
||
kid_of_an_empty_leaf: open(alter("key_id", moq.e2eeKid(BigInt(s.epoch), 1048575)), payload),
|
||
kid_of_this_member: open(alter("key_id", moq.e2eeKid(BigInt(s.epoch), s.leaf)), payload),
|
||
as_sent: open(meta.props, payload),
|
||
again: open(meta.props, payload),
|
||
};
|
||
return { group, epoch: s.epoch, outcomes: out };
|
||
}
|
||
|
||
window.roomTest.roomJson = () => room.roomJson();
|
||
// Co-performance, for the tests.
|
||
window.roomTest.coState = () => co && co.snapshot();
|
||
window.roomTest.coAnnounce = async (o) => { const r = await co.announce(o); return { at: r.at, intent: r.intent }; };
|
||
window.roomTest.coInput = (o) => co.input(o);
|
||
window.roomTest.coInputDetached = (o) => co.inputDetached(o);
|
||
window.roomTest.coPlayOutOfTurn = (ms) => co.playOutOfTurn(ms);
|
||
window.roomTest.coStampNextLate = (ms) => co.stampNextLate(ms);
|
||
window.roomTest.coRequest = (path, body) => co.ask(path, body);
|
||
window.roomTest.coObjects = () => coObjects();
|
||
// The room as this page holds it, verified as co-performance §9 verifies it, with the objects
|
||
// given (or the clock's whole log).
|
||
window.roomTest.coVerify = async (objects) => JSON.parse(room.coVerify(JSON.stringify({ policies: [fx.co.policy], objects: objects || await coObjects(), payloads: [] })));
|
||
window.roomTest.binding = () => ({
|
||
tiles: bindingRows(),
|
||
routes: state.binding.routes,
|
||
asked: state.binding.asked,
|
||
table: { refusals: bindings.refusals, struck: bindings.struck, bound: bindings.bound_frames, unbound: bindings.unbound_frames },
|
||
metadata: state.binding.metadata || (state.binding.metadata = metadataProbe()),
|
||
dom: $("binding-debug") && !$("binding-debug").hidden ? [...$("binding-debug").querySelectorAll("tbody tr")].map((tr) => ({ principal: tr.dataset.principal, state: tr.dataset.state, text: tr.textContent })) : null,
|
||
});
|
||
window.roomTest.compose = () => fx && fx.compose;
|
||
window.roomTest.frameAt = (t) => JSON.parse(room.frame(unhex(fx.compose), BigInt(t)));
|
||
window.roomTest.now = () => roomNow();
|
||
|
||
// A human caption, typed by this participant.
|
||
async function sendTypedCaption(e) {
|
||
e.preventDefault();
|
||
const el = $("caption-text");
|
||
const text = el.value.trim();
|
||
if (!text || state.phase !== "publishing") return;
|
||
el.value = "";
|
||
try { await window.roomTest.sendCaption({ text, origin: 1, lang: document.documentElement.lang || "en" }); } catch (err) { fail(err); }
|
||
}
|
||
|
||
// The room service: the link's, held to the page's own origin, or the page's own markup's.
|
||
service = queryEndpoint("service", "http") || (q.get("service") ? "" : ($("room-app") && $("room-app").dataset.service) || "");
|
||
if (state.refused.length) {
|
||
state.phase = "refused";
|
||
for (const id of ["join", "watch"]) if ($(id)) $(id).disabled = true;
|
||
} else if (service) {
|
||
window.roomTest.enter = (role, given) => enter(role, given);
|
||
window.roomTest.leave = leave;
|
||
const buttons = ["join", "watch"].map($).filter(Boolean);
|
||
const go = (role) => async () => {
|
||
for (const b of buttons) b.disabled = true;
|
||
try {
|
||
const phase = await enter(role);
|
||
if (phase === "refused") return;
|
||
if (phase === "full") for (const b of buttons) b.disabled = b.id === "join";
|
||
else if ($("leave")) $("leave").hidden = false;
|
||
if (phase === "publishing" && $("caption-form")) $("caption-form").hidden = false;
|
||
} catch (e) {
|
||
fail(e);
|
||
state.mode = state.mode === "service" && state.phase === "entering" ? null : state.mode;
|
||
for (const b of buttons) b.disabled = false;
|
||
}
|
||
};
|
||
if ($("join")) $("join").addEventListener("click", go("publisher"));
|
||
if ($("watch")) $("watch").addEventListener("click", go("viewer"));
|
||
if ($("leave")) $("leave").addEventListener("click", () => { $("leave").disabled = true; if ($("caption-form")) $("caption-form").hidden = true; leave(); });
|
||
if ($("caption-form")) $("caption-form").addEventListener("submit", sendTypedCaption);
|
||
// A watch room's shared controls, each at the room's lead (co-watch §5.3).
|
||
const cw = (fn) => () => { try { fn(); } catch (e) { fail(e); } };
|
||
const at = () => (cowatch ? cowatch.timeline.positionAt(cowatch.now()) : 0);
|
||
if ($("cw-play")) $("cw-play").addEventListener("click", cw(() => watchControl("play")));
|
||
if ($("cw-pause")) $("cw-pause").addEventListener("click", cw(() => watchControl("pause")));
|
||
if ($("cw-back")) $("cw-back").addEventListener("click", cw(() => watchControl("seek", Math.max(0, at() - 10e6))));
|
||
if ($("cw-fwd")) $("cw-fwd").addEventListener("click", cw(() => watchControl("seek", Math.min(cowatch.policy.duration_us, at() + 10e6))));
|
||
if ($("cw-rate")) $("cw-rate").addEventListener("change", cw(() => watchControl("rate", Number($("cw-rate").value))));
|
||
// Before entering: an end-to-end encrypted room says so, and offers no watching, since a
|
||
// viewer is not a member of its group and opens nothing, and no seeking back.
|
||
api("/room").then((r) => {
|
||
if (r.status !== 200 || !r.body || !r.body.e2ee) return;
|
||
if ($("e2ee")) { $("e2ee").hidden = false; $("e2ee-state").textContent = "only the room's members open its media; the relay and the room service hold no key"; }
|
||
if ($("watch")) $("watch").hidden = true;
|
||
if ($("dvr")) $("dvr").hidden = true;
|
||
}).catch(() => {});
|
||
state.phase = "idle";
|
||
} else {
|
||
boot().catch(fail);
|
||
}
|