1739817397
By meaning: - Page: app.js keeps the age parameters and paragraph and adds debug=binding; imports, state and roomTest hooks of age, co and binding side by side. index.html carries the co section and the binding table, each with its CSS. - room.e2e.mjs carries scenarioBinding whole beside scenarioAge and scenarioCo; default list `...,age,co` with a room, `binding` always; run.sh's node unit tests are every page module's (capsule, media, presentation, privacy-pass, binding). - wai-live-web: the wai dependency takes co_performance and browser_sink; the crate doc lists both. - joule-moq-web README: one node-test bullet for capsule, media and binding, then privacy-pass and ad-presentation; scenarios age, co and binding. - REGISTRY prose names exploration and browser-sink together; the shared lists keep both sides; export regenerated. Co-Authored-By: Claude Opus 5.5 <[email protected]>
1862 lines
91 KiB
JavaScript
1862 lines
91 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
|
|
//
|
|
// 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
|
|
//
|
|
// Everything a test reads is on `window.roomTest`.
|
|
|
|
import initWai, * as wai from "../pkg/wai_live_web.js";
|
|
import { E2ee } from "./e2ee.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";
|
|
|
|
// 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);
|
|
|
|
// 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 };
|
|
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: [],
|
|
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 },
|
|
};
|
|
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;
|
|
// 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();
|
|
fx = await (await fetch(q.get("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: q.get("wt"),
|
|
websocket: q.get("ws"),
|
|
hashes: (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, q.get("issuer_directory")) });
|
|
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.
|
|
async function enter(role) {
|
|
if (state.mode) throw new Error("this page has entered already");
|
|
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;
|
|
const ai = role === "publisher" && q.get("ai") === "scripted" ? await scriptedModel() : null;
|
|
await loadMoq(sealed);
|
|
await initWai();
|
|
let stream = null;
|
|
if (role === "publisher") {
|
|
// 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 = 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`).
|
|
const request = ai ? { principal, role, ai } : { principal, role };
|
|
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") {
|
|
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 };
|
|
me = { name: "you", seed: hex(seed), principal, grant: r.body.grant, grant_digest: r.body.digest, 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();
|
|
}
|
|
const ws = q.get("ws") || `${location.protocol === "https:" ? "wss" : "ws"}://${location.host}/moq`;
|
|
transportOpts = { webtransport: null, websocket: ws, hashes: [], force: "websocket" };
|
|
window.roomTest.role = ROLE_OF[role];
|
|
await begin(ROLE_OF[role], stream);
|
|
pollRoom();
|
|
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();
|
|
if (role !== "clock") {
|
|
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));
|
|
setInterval(() => { try { compositor.draw(); } catch (e) { fail(e); } }, Math.round(1000 / compositor.desc.params.fps));
|
|
setInterval(sweepMissing, 1000);
|
|
}
|
|
status("Connecting to the relay…");
|
|
await openSession(transportOpts);
|
|
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)) return;
|
|
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.
|
|
async function leave() {
|
|
state.phase = "left";
|
|
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;
|
|
}
|
|
}
|
|
|
|
// ------------------------------------------------------------------------------- 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(), 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 (service && v.kind_name === "join" && v.payload && v.payload.detail && !state.grants.some((g) => g.digest === v.payload.detail.grant)) {
|
|
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.
|
|
if (co) {
|
|
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;
|
|
}
|
|
|
|
function mediaOf(tr) {
|
|
return tr.role === "audio" ? "audio" : "video";
|
|
}
|
|
|
|
async function subscribeMedia(sub, tr) {
|
|
const P = sub.principal;
|
|
const m = mediaOf(tr);
|
|
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 receipts track first and with no delivery timeout (moq-streaming-format §4.1),
|
|
// at the media track's subscriber priority.
|
|
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 {
|
|
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;
|
|
}
|
|
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) : "";
|
|
}
|
|
|
|
// §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;
|
|
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);
|
|
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) : 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;
|
|
}
|
|
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;
|
|
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) : 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) {
|
|
if (g.uri.startsWith("https://")) { opts.webtransport = g.uri; opts.force = "webtransport"; }
|
|
else if (g.uri.startsWith("ws")) { opts.websocket = g.uri; 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()];
|
|
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 }) => {
|
|
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) => {
|
|
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,
|
|
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(),
|
|
});
|
|
// 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); }
|
|
}
|
|
|
|
service = q.get("service") || ($("room-app") && $("room-app").dataset.service) || "";
|
|
if (service) {
|
|
window.roomTest.enter = (role) => enter(role);
|
|
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 === "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);
|
|
// 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);
|
|
}
|