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

// A participant in a room's shared world (WAI extensions/live-worlds.md), in a page.
//
// The room service keeps the room (joule-moq-room, `[world]`): `GET /room` gives its
// descriptor, its grants and its world statement (`WLW1`), which the page checks against the
// descriptor before it enters. The page makes an Ed25519 key with the browser's random
// generator, asks for an actor's grant for it (`POST /grant`), and opens a MOQT session over
// the relay's WebSocket binding, the grant presented with a proof by its key (live-rooms
// §10.5.2). Then:
//
// - It publishes its world track, `[room, principal, "world"]`, and the track's receipts track
//   (§4.1): each group one op batch, an action of `wai.world.model` signed by the page's key,
//   the group signed into the page's receipt chain under receipt alias 4 (§4.3) by
//   joule-moq-web's signer, with the unmetered marker as its figure: this page measures
//   nothing, and says so.
// - It follows the room clock's events track (live-events §8.1): every lifecycle record checked
//   by wai-live-web, and each actor the clock records joining, present or since gone (§6), has
//   its world track and receipts track subscribed, and, in a world that is not private, fetched
//   from group 0 (§7). Every object is verified against its receipt by joule-moq-web's verifier,
//   the relay door's code, then offered in group order to the page's copy of the world
//   (wai-live-web's `World`, which is wai-rs's `live_world`): each op's signature, owner, kind,
//   Lamport rule and equivocation (§9.2). A track whose fetch does not reach group 0 to the live
//   edge leaves the world incomplete (§7).
// - In an end-to-end encrypted room (a private world, §8) it is a member of the room's group,
//   keying through the service's log (../room/e2ee.js, confidentiality §15): every object it
//   publishes is sealed under its current epoch before its receipt is signed, nothing is sent
//   before it holds an epoch, every object it receives opens before it is offered, and in each
//   epoch it reaches it first publishes a carry of every op it holds (§8.2).
//
// The page reads `snapshot()` and `sessionHash()`; `act(dx, dy)` signs and publishes a move.

import initWai, * as wai from "../pkg/wai_live_web.js";
import { E2ee } from "../room/e2ee.js";
import { connect } from "../moq/transport.js";
import { MoqSession, RequestRefused } from "../moq/session.js";

// The world track's receipt alias (live-worlds §4.3).
export const RECEIPT_ALIAS = 4;
// Track Aliases this session's PUBLISHes give (MOQT §9.8).
const ALIAS = { world: 30, worldReceipts: 31 };
// Publisher and subscriber priorities (MOQT §5.1): the world before media, the clock's events
// before the world.
const PRIORITY = 48;
const SUB = { events: 8, world: 24 };
// What a page holds while it waits: objects for their receipts, sealed objects for an epoch.
const HOLD_MAX = 512;
// In a private world a track's first object opened may be at any group (§9.2 step 2): the
// page takes the lowest it holds once that one has waited this long, so that a group whose
// stream arrived a moment after the next one's is not passed over.
const FIRST_SETTLE_MS = 300;

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 sleep = (ms) => new Promise((r) => setTimeout(r, ms));
const dec = new TextDecoder();

let codec = null;
async function loadMoq(sealed) {
  if (codec) return codec;
  const m = await import(sealed ? "../pkg/joule_moq_web_e2ee.js" : "../pkg/joule_moq_web.js");
  await m.default();
  codec = m;
  return m;
}

let waiReady = null;
const initLive = () => waiReady || (waiReady = initWai());

const bump = (o, k) => { o[k] = (o[k] || 0) + 1; };

export class WorldClient {
  // o: { service: the room service's base URL, ws: the relay's WebSocket URL, log, onChange,
  //      role: "publisher" (an actor) or "viewer" }
  constructor(o) {
    this.service = String(o.service || "").replace(/\/$/, "");
    this.ws = o.ws;
    this.log = o.log || (() => {});
    this.onChange = o.onChange || (() => {});
    this.role = o.role || "publisher";
    this.phase = "idle";
    this.errors = [];
    this.tracks = new Map(); // owner -> the track's state
    this.ops = []; // own ops waiting to be published
    this.lamport = 0;
    this.nextOwn = 0;
    this.sending = Promise.resolve();
    this.carriedEpoch = null;
    this.records = Promise.resolve();
    this.joins = new Set();
    this.following = new Set(); // the world-track subscriptions under way
    this.stats = {
      published: 0, carried: 0,
      // In a private world: the actors whose live world track this page followed before it
      // published its own (null before it publishes, or in a world that is not private).
      followed_before_publish: null,
      receipts: 0, receipts_refused: {}, chain: {},
      received: 0, fetched: 0, verified: 0, refused: {}, waiting_receipts: 0,
      admitted: 0, redelivered: 0, offer_refused: {},
      opened: 0, open_refused: {}, open_skipped: 0, open_held: 0,
      fetches: [], incomplete: [],
    };
  }

  fail(e) {
    const m = (e && e.stack) || String(e);
    this.errors.push(m);
    this.log(`error: ${m}`);
    this.onChange();
  }

  async api(path, body) {
    const r = await fetch(`${this.service}${path}`, body ? { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify(body), cache: "no-store" } : { cache: "no-store" });
    let b = null;
    try { b = await r.json(); } catch { b = null; }
    return { status: r.status, body: b };
  }

  // Enter the room: its world checked, a grant for a key made here, the session, the world
  // track published, the clock's events followed.
  async enter() {
    this.phase = "entering";
    let r = await this.api("/room");
    if (r.status !== 200) throw new Error(`the room service answered ${r.status}`);
    if (!r.body.world) throw new Error("this room carries no shared world");
    const sealed = !!r.body.e2ee;
    this.private = sealed;
    this.moq = await loadMoq(sealed);
    await initLive();
    const seed = crypto.getRandomValues(new Uint8Array(32));
    this.seed = seed;
    this.me = hex(this.moq.publicKeyOf(seed));
    r = await this.api("/grant", { principal: this.me, role: this.role });
    if (r.status === 409) { this.phase = "full"; this.onChange(); return this.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;
    if (!v.world || sealed !== !!v.e2ee) throw new Error("the room the grant is for is not the room the page read");
    const descriptor = unhex(v.descriptor);
    const d = JSON.parse(wai.roomDescriptor(descriptor));
    const st = JSON.parse(wai.worldStatement(descriptor, unhex(v.world.statement)));
    if (!st.ok) throw new Error(`the room's world statement is refused: ${st.reason}`);
    this.statement = st;
    this.roomId = v.room;
    this.t0 = v.t0_us;
    this.clock = d.clock;
    this.grants = v.grants;
    this.grant = { record: r.body.grant, digest: r.body.digest, exp: r.body.exp, role: r.body.role };
    this.world = new wai.World(descriptor, JSON.stringify(v.grants), unhex(v.world.statement), sealed);
    this.room = new wai.Room(descriptor, JSON.stringify(v.grants));
    this.descriptor = descriptor;
    if (sealed) {
      const c = JSON.parse(this.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}`);
      this.e2ee = new E2ee({
        moq: this.moq, wai, api: (p, b) => this.api(p, b), log: this.log, room: v.room, clock: d.clock, seed, suite: c.suite, role: this.role,
        grantOf: (digest) => this.grantView().find((g) => g.digest === digest),
        refreshGrants: () => this.refreshGrants(),
        onRecord: (bytes) => { this.records = this.records.then(() => this.onRecord(bytes)).catch((e) => this.fail(e)); },
        onChange: () => this.onKeys(),
      });
      if (this.role === "publisher") await this.e2ee.keyPackage();
    }
    const dial = () => connect({ webtransport: null, websocket: this.ws, hashes: [], force: "websocket", log: this.log });
    // The grant's proof is over a challenge the relay's SETUP gives: a second session, where
    // the first held none (session.js).
    const s = await MoqSession.connect(await dial(), this.moq, { grant: { record: unhex(this.grant.record), seed }, redial: dial, implementation: "joule-moq-web/world", log: this.log, properties: sealed });
    this.session = s;
    s.closed.then((i) => { if (this.phase !== "left") { this.phase = "closed"; this.closedInfo = i; this.onChange(); } });
    // In a private world a publishing page follows the actors already in the room before it
    // publishes: its world track is what makes the clock record its join, which brings the
    // commit that adds it to the group and, in that epoch, each member's carry (§8.2). A carry
    // is sent once; one sent before this page followed its sender's track never reaches it.
    if (this.e2ee && this.role === "publisher") await this.followActors();
    if (this.role === "publisher") {
      const ns = [this.roomId, this.me, "world"];
      this.published = [await s.publish(ns, "0_receipts", ALIAS.worldReceipts, { publisher_priority: PRIORITY }), await s.publish(ns, "0", ALIAS.world, { publisher_priority: PRIORITY })];
      this.signer = new this.moq.TrackSigner(seed, BigInt(RECEIPT_ALIAS), BigInt(ALIAS.world), BigInt(ALIAS.worldReceipts), PRIORITY, "world", 0n);
      this.signer.setEnvelope(false);
    }
    await this.followClock();
    if (this.e2ee) this.e2ee.follow(() => this.phase !== "left").catch((e) => this.fail(e));
    this.phase = this.role === "publisher" ? "acting" : "watching";
    this.log(`in the room ${this.roomId.slice(0, 12)}…, ${sealed ? "a private world" : "a public world"}`);
    this.onChange();
    return this.phase;
  }

  // The room's grants, each read (its digest, its rights): what the E2EE member keys by.
  grantView() {
    const gv = JSON.parse(this.room.grants());
    return gv.outcome === "read" ? gv.grants : [];
  }

  async refreshGrants() {
    const r = await this.api("/room");
    if (r.status !== 200 || r.body.room !== this.roomId) return this.log("the room's grants could not be read again");
    this.grants = r.body.grants;
    this.world.setGrants(JSON.stringify(this.grants));
    this.room.setGrants(JSON.stringify(this.grants));
  }

  // The clock's events track: its records, each verified, in the order they arrive.
  async followClock() {
    const ns = [this.roomId, this.clock, "events"];
    const onRecords = (payload) => {
      let arr;
      try { arr = JSON.parse(dec.decode(payload)); } catch { return this.log("an events object that is not JSON"); }
      for (const rec of arr) {
        let bytes;
        try { bytes = wai.msfRecordRead(JSON.stringify(rec), BigInt(this.t0)); } catch (e) { this.log(`an event record refused: ${e.message || e}`); continue; }
        this.records = this.records.then(() => this.onRecord(bytes)).catch((e) => this.fail(e));
      }
    };
    const h = { stream: () => ({ object: (j, p) => onRecords(p) }), fill: () => ({ object: (j, p) => onRecords(p) }) };
    await this.session.subscribe(ns, "0", {
      subscriber_priority: SUB.events, group_order: 1, forward: true, rendezvous_timeout: 10000,
      location_filter: { type: "next" }, fill: { location_filter: { type: "relative", start_group: 1 } },
    }, h);
  }

  // Every actor the room's log records now, its world track followed: the log read to its
  // present end, each record taken, and each subscription it started answered.
  async followActors() {
    await this.e2ee.readToEnd();
    await this.records;
    while (this.following.size) await Promise.all([...this.following]);
    this.stats.followed_before_publish = [...this.tracks.values()].filter((t) => t.live).length;
    this.log(`following ${this.stats.followed_before_publish} actor(s) live before publishing`);
  }

  // A lifecycle record: each actor the clock records joining has its world track followed.
  async onRecord(bytes) {
    const v = JSON.parse(this.room.offer(bytes));
    if (!v.ok) return this.log(`an event refused: ${v.reason}`);
    if (!v.new || v.kind_name !== "join") return;
    const P = v.payload && v.payload.subject;
    if (!P || this.joins.has(P)) return;
    this.joins.add(P);
    if (!this.world.isActor(unhex(P))) await this.refreshGrants();
    if (!this.world.isActor(unhex(P))) return this.log(`${P.slice(0, 8)} joined under a grant that does not act`);
    if (P === this.me) return;
    const f = this.subscribeWorld(P).catch((e) => this.fail(e));
    this.following.add(f);
    f.finally(() => this.following.delete(f));
  }

  async subscribeWorld(P) {
    if (this.tracks.has(P)) return;
    const t = { P, ns: [this.roomId, P, "world"], verifier: new this.moq.TrackVerifier(unhex(P), BigInt(RECEIPT_ALIAS)), waiting: new Map(), ready: new Map(), seen: new Set(), broken: null, live: 0, fetched: 0 };
    this.tracks.set(P, t);
    const rh = { stream: (h) => ({ object: (j, p) => this.onReceipts(t, h.group_id, p) }), fill: () => ({ object: (j, p) => this.onReceipts(t, j.group_id, p) }) };
    const wh = {
      stream: (h) => ({ object: (j, p, k, props) => this.onObject(t, h.group_id, p, { priority: h.publisher_priority, props, object: j.object_id }, false) }),
      fill: () => ({ object: () => {} }),
    };
    const params = { subscriber_priority: SUB.world, group_order: 1, forward: true, rendezvous_timeout: 10000, location_filter: { type: "next" } };
    // An actor that has left has no live track: its subscription is refused (no publisher
    // within the rendezvous), and its ops are fetched all the same (§7).
    try {
      await this.session.subscribe(t.ns, "0_receipts", params, rh);
      await this.session.subscribe(t.ns, "0", params, wh);
      t.live = true;
      this.log(`following the world track of ${P.slice(0, 8)}`);
    } catch (e) {
      if (!(e instanceof RequestRefused)) throw e;
      t.live = false;
      this.log(`no live world track of ${P.slice(0, 8)}: ${e.message}`);
    }
    // A private world fetches nothing: what was sealed before this member joined does not open
    // here, and members carry it instead (§8.2).
    if (!this.private) await this.catchUp(t);
  }

  // §7: the track and its receipts track from group 0, each object verified, admitted in
  // group order before the live ones.
  async catchUp(t) {
    const params = { group_order: 1, subscriber_priority: SUB.world + 1, location_filter: { type: "absolute", start: [0, 0] } };
    // A FETCH refused before any object is sent counts nothing twice when tried again: a
    // relay asked for groups still arriving may pass it to the publisher, which takes no
    // requests, and a track with no group yet has nothing to fetch. The last refusal is kept.
    const refusals = [];
    const again = async (name, each) => {
      for (let i = 0; ; i++) {
        try { return await this.session.fetch(t.ns, name, params, each); } catch (e) {
          if (!(e instanceof RequestRefused)) throw e;
          if (i >= 4) { refusals.push(`${name}: ${e.message}`); return null; }
          await sleep(500);
        }
      }
    };
    const rec = [];
    const objs = [];
    const r1 = await again("0_receipts", (j, p) => rec.push([j.group_id, p]));
    const r2 = await again("0", (j, p, props) => objs.push([j.group_id, p, { object: j.object_id, priority: j.publisher_priority, props }]));
    for (const [g, p] of rec) this.onReceipts(t, g, p);
    for (const [g, p, meta] of objs) this.onObject(t, g, p, meta, true);
    const groups = objs.map((o) => o[0]);
    const f = { owner: t.P, receipts: rec.length, objects: objs.length, first: groups.length ? Math.min(...groups) : null, last: groups.length ? Math.max(...groups) : null, how: [r1 && r1.how, r2 && r2.how], refused: refusals };
    this.stats.fetches.push(f);
    // Whole when the fetch holds every group from 0 to its last (§7).
    const want = f.last == null ? 0 : f.last + 1;
    const missing = [];
    for (let g = 0; g < want; g++) if (!groups.includes(g)) { missing.push(g); break; }
    if (missing.length) {
      t.incomplete = missing[0];
      this.stats.incomplete.push({ owner: t.P, reason: "world_incomplete", group: missing[0] });
    }
    this.log(`fetched the world track of ${t.P.slice(0, 8)}: ${objs.length} groups, ${rec.length} receipts`);
    this.onChange();
  }

  onReceipts(t, group, payload) {
    const r = JSON.parse(t.verifier.receipts(BigInt(group), payload, BigInt(Date.now())));
    if (!r.ok) { bump(this.stats.receipts_refused, r.reason); return; }
    this.stats.receipts++;
    bump(this.stats.chain, r.chain);
    const w = t.waiting.get(group);
    if (w) { t.waiting.delete(group); this.onObject(t, group, w.payload, w.meta, w.fetched); }
  }

  onObject(t, group, payload, meta, fetched) {
    if (t.seen.has(group)) return;
    const r = JSON.parse(t.verifier.object(BigInt(group), 0n, payload));
    if (!r.ok && r.reason === "no_receipt") {
      if (!t.waiting.has(group) && t.waiting.size < HOLD_MAX) { t.waiting.set(group, { payload, meta, fetched }); this.stats.waiting_receipts++; }
      return;
    }
    t.seen.add(group);
    if (fetched) this.stats.fetched++; else this.stats.received++;
    if (!r.ok) { bump(this.stats.refused, r.reason); this.log(`group ${group} of ${t.P.slice(0, 8)} refused: ${r.reason}`); return; }
    this.stats.verified++;
    t.ready.set(group, { payload, meta, fetched, at: performance.now() });
    this.drain(t);
  }

  // Offer the track's verified objects to the world in group order (§9.2 step 2).
  drain(t) {
    const owner = unhex(t.P);
    for (;;) {
      if (t.broken || !t.ready.size) break;
      // An i64 crosses the wasm boundary as a BigInt.
      const n = Number(this.world.nextGroup(owner));
      let g;
      if (n >= 0) g = n;
      else if (!this.private) g = 0;
      else g = Math.min(...t.ready.keys());
      const o = t.ready.get(g);
      if (!o) break;
      if (n < 0 && this.private && performance.now() - o.at < FIRST_SETTLE_MS) {
        if (!t.settling) { t.settling = true; setTimeout(() => { t.settling = false; this.drain(t); }, FIRST_SETTLE_MS); }
        break;
      }
      let batch = o.payload;
      if (this.private) {
        const v = this.e2ee.open(t.ns, "0", g, o.meta.object || 0, o.meta.priority, o.meta.props, o.payload);
        if (!v.ok) {
          if (v.pending != null || (v.reason === "epoch_not_joined" && this.e2ee.awaitingWelcome)) { this.stats.open_held++; break; }
          if (n < 0) {
            // Sealed before this member joined: not part of what it opens (§9.2 step 2).
            t.ready.delete(g);
            this.stats.open_skipped++;
            continue;
          }
          bump(this.stats.open_refused, v.reason);
          t.ready.delete(g);
          t.broken = v.reason;
          break;
        }
        this.stats.opened++;
        batch = v.payload;
      }
      t.ready.delete(g);
      const a = JSON.parse(this.world.offer(owner, BigInt(g), batch));
      if (!a.ok) {
        bump(this.stats.offer_refused, a.reason);
        t.broken = a.reason;
        this.log(`group ${g} of ${t.P.slice(0, 8)} refused by the world: ${a.reason}`);
        break;
      }
      this.stats.admitted += a.ops;
      this.stats.redelivered += a.redelivered;
    }
    this.onChange();
  }

  // The room's group moved: what waited opens now, and in a new epoch this member carries what
  // it holds before anything else it seals (§8.2).
  onKeys() {
    for (const t of this.tracks.values()) this.drain(t);
    this.pump();
    this.onChange();
  }

  // Sign a move and publish it.
  act(dx, dy) {
    if (this.role !== "publisher" || !this.signer) return null;
    this.lamport += 1;
    const op = wai.worldAction(this.seed, BigInt(this.lamport), dx | 0, dy | 0);
    this.ops.push(op);
    this.pump();
    return hex(op);
  }

  pump() {
    this.sending = this.sending.then(() => this.flush()).catch((e) => this.fail(e));
    return this.sending;
  }

  async flush() {
    if (!this.signer || !this.session || this.phase === "left") return;
    if (this.private) {
      // Nothing is sent before this member holds an epoch.
      if (!this.e2ee.joined) return;
      const epoch = Number(this.e2ee.member.epoch());
      if (epoch !== this.carriedEpoch) {
        this.carriedEpoch = epoch;
        for (const h of JSON.parse(this.world.carry())) {
          await this.send(unhex(h));
          this.stats.carried++;
        }
      }
    }
    while (this.ops.length) {
      const op = this.ops.shift();
      await this.send(op);
      this.stats.published++;
    }
  }

  // One group of this page's world track, its own copy of the world taking it too.
  async send(batch) {
    const g = this.nextOwn++;
    const now = new Date().toISOString();
    let sg;
    if (this.private) {
      const s = this.e2ee.member.seal(JSON.stringify([this.roomId, this.me, "world"]), "0", BigInt(g), 0n, PRIORITY, batch, true, false, 0n);
      sg = this.signer.signSealed(BigInt(g), s.payload(), s.properties(), now, now);
      s.free();
    } else {
      sg = this.signer.sign(BigInt(g), batch, now, now);
    }
    const r = sg.receiptsStream();
    const md = sg.mediaStream();
    sg.free();
    await this.session.sendStream(r, 3_000_001);
    await this.session.sendStream(md, 3_000_000);
    const a = JSON.parse(this.world.offer(unhex(this.me), BigInt(g), batch));
    if (!a.ok) this.fail(new Error(`this page's own group ${g} is refused by its world: ${a.reason}`));
    this.onChange();
  }

  sessionHash() {
    return this.world ? this.world.sessionHash() : null;
  }

  // The merged log, in canonical order: what a renderer replays.
  mergedLog() {
    return this.world ? this.world.log() : new Uint8Array(0);
  }

  // Stop acting and stay: PUBLISH_DONE on the world track and its receipts track, the session
  // and every subscription kept. The room's clock records the leave (§6); in a private world the
  // next epoch excludes this page, which goes on receiving and opens nothing sealed after it.
  async stopActing() {
    this.stopped = true;
    for (const r of this.published || []) await r.writer.write(this.moq.encodeControl(JSON.stringify({ message: "PUBLISH_DONE", status: 2, reason: "stopped acting" })));
    this.signer = null;
    this.onChange();
  }

  leave() {
    this.phase = "left";
    if (this.session) this.session.close(0, "left");
    this.onChange();
  }

  snapshot() {
    return {
      phase: this.phase,
      errors: this.errors,
      me: this.me,
      room: this.roomId,
      private: !!this.private,
      grant: this.grant,
      statement: this.statement,
      session_hash: this.sessionHash(),
      ops: this.world ? this.world.count() : 0,
      by_actor: this.world ? JSON.parse(this.world.byActor()) : {},
      actors_followed: [...this.tracks.keys()],
      tracks: Object.fromEntries([...this.tracks.values()].map((t) => [t.P, { next: Number(this.world.nextGroup(unhex(t.P))), live: t.live, broken: t.broken, incomplete: t.incomplete ?? null, waiting: t.waiting.size, ready: t.ready.size }])),
      own: { groups: this.nextOwn, queued: this.ops.length },
      stats: this.stats,
      e2ee: this.e2ee ? { joined: this.e2ee.joined, epoch: this.e2ee.joined ? Number(this.e2ee.member.epoch()) : null, stats: this.e2ee.stats, stopped: this.e2ee.stopped, member: this.e2ee.state() } : null,
      transport: this.session ? { kind: this.session.t.kind } : null,
    };
  }
}