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

// A MOQT client session (draft-22, or -23) over a transport of ./transport.js, the codec being
// joule-moq-web's wasm build.
//
// - SETUP on this end's control stream (its first unidirectional stream), sent as the session
//   opens, with joule-moq's receipt extension, and either a room grant or a token by value.
//   The relay reads this SETUP before it sends its own (draft-ietf-moq-privacy-pass-auth-04
//   §3.5.1; WAI live-rooms §10.5.2). A room grant is public, so it is honoured only with a
//   proof by the principal's key (signed in the wasm build) over a challenge of the relay's
//   and the `//authority/path` of the URL the transport opened. A relay's challenge comes in
//   its SETUP, so a session that holds none presents its grant unproved, is refused
//   (`GRANT_REFUSED`, `pop_missing`) and given one; MoqSession.connect then opens a second
//   session, through `redial`, whose SETUP proves over it. The relay's SETUP is read from its
//   control stream, data streams that come first held until it.
// - Requests on bidirectional streams, Request IDs even from 0 (§6.4.2): PUBLISH,
//   PUBLISH_NAMESPACE, SUBSCRIBE (with priorities, filters and FILL_PARAMETERS), FETCH,
//   SUBSCRIBE_NAMESPACE, TRACK_STATUS.
// - Every data stream read concurrently, each by a task of its own as it arrives: a
//   subscriber that reads its streams one at a time, in the order they arrive, can stall
//   itself, since the stream it waits on may have its bytes held back for flow-control
//   credit that the unread streams behind it hold (joule-moq-relay's README, "A subscriber
//   that does not read").
// - GOAWAY (§9.2): no new request is opened once one arrives; the caller is told the new
//   session URI and the timeout, and moves.

import { StreamReset } from "./transport.js";

const STEP_MS = 10000;
const ALIAS_WAIT_MS = 3000;
// A Track Alias a subgroup stream names before this end has read the SUBSCRIBE_OK that
// gives it waits up to ALIAS_WAIT_MS.

export function withTimeout(p, ms, what) {
  let t;
  return Promise.race([p, new Promise((_, no) => { t = setTimeout(() => no(new Error(`${what}: no answer in ${ms} ms`)), ms); })]).finally(() => clearTimeout(t));
}

const hexToBytes = (h) => new Uint8Array((h.match(/../g) || []).map((x) => parseInt(x, 16)));
const utf8 = (h) => new TextDecoder().decode(hexToBytes(h));

// The relay refused the session at its SETUP (WAI live-rooms §10.5.2): `reason` is the code
// its GRANT_REFUSED option carries, and `challenge` the one its SETUP gave, which a next
// session's proof is made over.
export class SetupRefused extends Error {
  constructor(reason, challenge) {
    super(`the relay refused the SETUP: ${reason}`);
    this.reason = reason;
    this.challenge = challenge;
  }
  // A refusal a proof over the challenge beside it answers; any other, no new proof changes.
  get asksForProof() { return ["pop_missing", "pop_wrong_challenge", "pop_replayed"].includes(this.reason); }
}

export class RequestRefused extends Error {
  constructor(message, first) {
    super(message);
    this.first = first;
  }
}

class Queue {
  constructor() { this.items = []; this.waiters = []; this.done = false; }
  push(x) { const w = this.waiters.shift(); if (w) w(x); else this.items.push(x); }
  end() { this.done = true; for (const w of this.waiters.splice(0)) w(null); }
  next() {
    if (this.items.length) return Promise.resolve(this.items.shift());
    if (this.done) return Promise.resolve(null);
    return new Promise((r) => this.waiters.push(r));
  }
}

export class MoqSession {
  constructor(transport, moq, { token, grant, challenge, implementation = "joule-moq-web", log = () => {}, maxPayload = 4 << 20, properties = false } = {}) {
    this.t = transport;
    // Each object's property list, encoded, as its fourth argument to a sink: what a
    // member of an end-to-end encrypted room opens an object with.
    this.properties = properties;
    this.moq = moq;
    this.token = token;
    // { record, seed }: a room grant's record and its principal's 32-byte seed.
    this.grant = grant;
    // A challenge an earlier SETUP of this relay gave (32 bytes): the grant's proof is made
    // over it. Without one the grant goes unproved, to be refused and given one.
    this.challenge = challenge;
    // The challenge this session's SETUP exchange gave, for the next session to this relay.
    this.nextChallenge = null;
    // How many sessions it took to be admitted: 1, or 2 when the first was given a challenge.
    this.sessionsToAdmit = 1;
    this.implementation = implementation;
    this.log = log;
    this.maxPayload = maxPayload;
    this.nextRequestId = 0;
    this.tracks = new Map(); // downstream Track Alias -> handler { object, end }
    this.aliasWaiters = new Map();
    this.fetches = new Map(); // request id -> fetch handler
    this.goaway = null;
    this.onGoaway = null;
    this.closedInfo = null;
    this.counters = { uni_streams: 0, subgroups: 0, fetch_streams: 0, objects: 0, bytes: 0, resets: 0, unrouted: 0, decode_errors: 0, streams_open_max: 0 };
    this.openStreams = 0;
    this.peerSetup = null;
    this.closed = transport.closed.then((i) => { this.closedInfo = i; return i; });
  }

  // A session on `transport`. `opts.redial`, a function that opens another transport to the
  // same relay, lets a session that presents a grant and holds no challenge be given one and
  // prove over it on a second session; without it the refusal is thrown (SetupRefused).
  static async connect(transport, moq, opts = {}) {
    const s = new MoqSession(transport, moq, opts);
    try {
      await s.start();
      return s;
    } catch (e) {
      if (!(e instanceof SetupRefused) || !e.asksForProof || !e.challenge || !opts.redial || !opts.grant) throw e;
      (opts.log || (() => {}))(`session: the relay gave a challenge (${e.reason}); opening a second session with the grant's proof`);
      const again = new MoqSession(await opts.redial(), moq, { ...opts, challenge: e.challenge });
      again.sessionsToAdmit = 2;
      await again.start();
      return again;
    }
  }

  async start() {
    // The draft the transport opened the session at: every encoder and decoder of the
    // codec made from here speaks it (-23's messages read and written in -22's model).
    if (this.moq.setDraft) this.moq.setDraft(this.t.draft || 22);
    this.setupReceived = new Promise((ok, no) => { this._setupOk = ok; this._setupNo = no; });
    this.acceptUniLoop();
    this.acceptBidiLoop();
    this.control = await this.t.openUni();
    // This end's SETUP goes first, at once: the relay reads it before it sends its own.
    let setup;
    if (!this.grant) {
      setup = this.moq.clientSetup(this.token || undefined, this.implementation, true, 0n);
    } else if (!this.challenge) {
      // No challenge of this relay's is held: the grant unproved, as a token by value. The
      // relay refuses it and its SETUP gives a challenge.
      setup = this.moq.clientSetup(this.grant.record, this.implementation, true, 0n);
    } else {
      const u = new URL(this.t.url);
      this.target = `//${u.host}${u.pathname}`;
      // `proofBy`: a test's proof by another key than the principal's, which the relay refuses.
      setup = this.grant.proofBy
        ? this.moq.clientSetupWithProof(this.grant.record, this.moq.grantProofBy(this.grant.record, this.grant.proofBy, this.challenge, this.target), this.implementation, true, 0n)
        : this.moq.clientSetupWithGrant(this.grant.record, this.grant.seed, this.challenge, this.target, this.implementation, true, 0n);
    }
    await this.control.write(setup);
    this.peerSetup = await withTimeout(this.setupReceived, STEP_MS, "the relay's SETUP");
    const option = (type) => (this.peerSetup.options || []).find((o) => o.type === Number(type) && o.bytes);
    const given = option(this.moq.grantChallengeOption());
    this.nextChallenge = given ? hexToBytes(given.bytes) : null;
    const refused = option(this.moq.grantRefusedOption());
    if (refused || (this.grant && !this.challenge)) {
      // Refused, or admitted with no proof of the grant, which is not the session asked for:
      // this end leaves at once, and the relay closes the session UNAUTHORIZED.
      const reason = refused ? utf8(refused.bytes) : "pop_missing";
      this.t.close(0, "refused at SETUP");
      throw new SetupRefused(reason, this.nextChallenge);
    }
    this.log(this.grant ? `session: SETUP exchanged over ${this.t.kind}, the grant presented with its proof for ${this.target}` : `session: SETUP exchanged over ${this.t.kind}`);
  }

  // ------------------------------------------------------------------- incoming streams

  async acceptUniLoop() {
    for (;;) {
      const r = await this.t.acceptUni();
      if (!r) return;
      this.counters.uni_streams++;
      // Each stream is read by a task of its own, at once.
      this.readUni(r);
    }
  }

  async acceptBidiLoop() {
    for (;;) {
      const s = await this.t.acceptBidi();
      if (!s) return;
      // This client publishes by PUBLISH, so the relay opens no request to it; one it
      // opens anyway is refused (REQUEST_ERROR NOT_SUPPORTED).
      this.refuseIncoming(s);
    }
  }

  async refuseIncoming({ writer, reader }) {
    const ms = new this.moq.MessageStream();
    try {
      for (;;) {
        const { value, done } = await reader.read();
        if (done) break;
        ms.push(value);
        const m = ms.next();
        if (m) {
          await writer.write(this.moq.encodeControl(JSON.stringify({ message: "REQUEST_ERROR", code: 3, reason: "this client takes no requests" })));
          await writer.close();
          break;
        }
      }
    } catch { /* the stream ended */ }
    ms.free();
  }

  async readUni(reader) {
    const dec = new this.moq.UniDecoder(BigInt(this.maxPayload));
    let sink = null;
    let kind = null;
    this.openStreams++;
    this.counters.streams_open_max = Math.max(this.counters.streams_open_max, this.openStreams);
    try {
      for (;;) {
        let chunk;
        try {
          chunk = await reader.read();
        } catch (e) {
          this.counters.resets++;
          if (sink && sink.reset) sink.reset(e instanceof StreamReset ? e.code : -1);
          return;
        }
        if (chunk.done) {
          try { dec.finish(); } catch (e) { this.counters.decode_errors++; this.log(`a stream ended inside an item: ${e.message || e}`); }
          if (sink && sink.end) sink.end();
          return;
        }
        this.counters.bytes += chunk.value.length;
        dec.push(chunk.value);
        for (;;) {
          let item;
          try { item = dec.next(); } catch (e) {
            this.counters.decode_errors++;
            this.log(`a data stream does not decode: ${e.message || e}`);
            reader.cancel(0);
            if (sink && sink.reset) sink.reset(-2);
            return;
          }
          if (!item) break;
          const k = item.kind;
          const j = JSON.parse(item.json);
          if (k === "header") {
            kind = j.header;
            sink = await this.route(j, dec);
            if (!sink) { this.counters.unrouted++; reader.cancel(0); item.free(); return; }
          } else if (k === "control") {
            this.onControl(j.control);
          } else if (k === "object" || k === "fetch_object") {
            this.counters.objects++;
            const payload = item.payload();
            const props = this.properties ? item.properties() : null;
            sink.object && sink.object(j, payload, k, props);
          } else if (k === "end_of_range") {
            sink.endOfRange && sink.endOfRange(j.end_of_range);
          }
          item.free();
        }
      }
    } finally {
      this.openStreams--;
      dec.free();
      void kind;
    }
  }

  async route(h, dec) {
    if (h.header === "control") {
      return { object() {}, end: () => { if (!this.closedInfo) this.log("the relay's control stream ended"); } };
    }
    if (h.header === "subgroup") {
      this.counters.subgroups++;
      let t = this.tracks.get(h.track_alias);
      if (!t) t = await this.waitAlias(h.track_alias);
      if (!t) return null;
      return t.stream(h);
    }
    if (h.header === "fetch") {
      this.counters.fetch_streams++;
      const f = this.fetches.get(h.request_id);
      if (!f) return null;
      dec.setGroupOrder(f.order || 1);
      return f.stream(h);
    }
    return null;
  }

  waitAlias(alias) {
    return new Promise((r) => {
      const t = setTimeout(() => { done(null); }, ALIAS_WAIT_MS);
      const done = (x) => {
        clearTimeout(t);
        const ws = (this.aliasWaiters.get(alias) || []).filter((w) => w !== done);
        if (ws.length) this.aliasWaiters.set(alias, ws); else this.aliasWaiters.delete(alias);
        r(x);
      };
      this.aliasWaiters.set(alias, [...(this.aliasWaiters.get(alias) || []), done]);
    });
  }

  setTrack(alias, handler) {
    this.tracks.set(alias, handler);
    for (const w of this.aliasWaiters.get(alias) || []) w(handler);
  }

  onControl(m) {
    if (m.message === "SETUP") {
      this._setupOk(m);
    } else if (m.message === "GOAWAY") {
      this.goaway = { uri: utf8(m.new_session_uri), timeout_ms: m.timeout_ms };
      this.log(`GOAWAY: new session URI "${this.goaway.uri}", timeout ${this.goaway.timeout_ms} ms`);
      if (this.onGoaway) this.onGoaway(this.goaway);
    }
  }

  // ---------------------------------------------------------------------- requests

  // `answerMs`: how long the answer is waited for once the request is written.
  async request(desc, answerMs = STEP_MS) {
    if (this.goaway) throw new Error(`GOAWAY received: no new ${desc.message}`);
    if (this.closedInfo) throw new Error("the session is closed");
    const id = this.nextRequestId;
    this.nextRequestId += 2;
    // The authorization token rides in SETUP, where the relay's grant rules read it for
    // every request of the session.
    const d = { ...desc, request_id: id };
    const { writer, reader } = await this.t.openBidi();
    const bytes = this.moq.encodeControl(JSON.stringify(d));
    await writer.write(bytes);
    const q = new Queue();
    const ms = new this.moq.MessageStream();
    // At -23 a FETCH is answered by REQUEST_OK, which the codec reads as FETCH_OK once it
    // knows the stream is a FETCH's.
    if (ms.forRequest) ms.forRequest(JSON.stringify(d));
    (async () => {
      try {
        for (;;) {
          const { value, done } = await reader.read();
          if (done) break;
          ms.push(value);
          for (let m; (m = ms.next()); ) q.push(JSON.parse(m));
        }
      } catch (e) {
        q.push({ message: "STREAM_RESET", code: e.code });
      } finally {
        ms.free();
        q.end();
      }
    })();
    const first = await withTimeout(q.next(), answerMs, desc.message);
    if (!first) throw new RequestRefused(`${desc.message}: the stream ended before its answer`, null);
    return { id, first, messages: q, writer, reader };
  }

  static refusal(what, first) {
    if (first.message === "REQUEST_ERROR") return new RequestRefused(`${what}: REQUEST_ERROR ${first.code} ${utf8(first.reason || "")}`, first);
    return new RequestRefused(`${what}: an unexpected ${first.message}`, first);
  }

  async publish(namespace, name, alias, trackProperties) {
    const r = await this.request({ message: "PUBLISH", namespace, name, track_alias: alias, track_properties: trackProperties });
    if (r.first.message !== "REQUEST_OK") throw MoqSession.refusal(`PUBLISH ${namespace.join("/")}--${name}`, r.first);
    return r;
  }

  // SUBSCRIBE: `handler` gets each subgroup stream of the track (`stream(header)` gives
  // its object sink) and, with `params.fill`, the fill fetch stream named by the request.
  async subscribe(namespace, name, params, handler) {
    const id = this.nextRequestId;
    if (params && params.fill) this.fetches.set(id, { order: params.group_order || 1, stream: (h) => handler.fill(h) });
    let r;
    try {
      // The relay may hold a SUBSCRIBE that asks for a rendezvous until its RENDEZVOUS_TIMEOUT
      // ends with no publisher (§9.20.6), and answers then: that answer is waited for a step
      // past the rendezvous, so that it is read rather than raced.
      r = await this.request({ message: "SUBSCRIBE", namespace, name, params }, STEP_MS + Number((params && params.rendezvous_timeout) || 0));
    } catch (e) {
      this.fetches.delete(id);
      throw e;
    }
    if (r.first.message !== "SUBSCRIBE_OK") {
      this.fetches.delete(id);
      throw MoqSession.refusal(`SUBSCRIBE ${namespace.join("/")}--${name}`, r.first);
    }
    const alias = r.first.track_alias;
    this.setTrack(alias, handler);
    (async () => {
      for (let m; (m = await r.messages.next()); ) {
        if (m.message === "PUBLISH_DONE") { handler.done && handler.done(m); }
      }
    })();
    return { ...r, alias };
  }

  // The subscriber ends a subscription by resetting its request stream (moq-transport-22
  // §6.4.2.3), with CANCELLED (0x1): the relay drops the subscription and sends nothing more
  // of the track; data streams already in flight for its alias find no handler here.
  unsubscribe(sub, code = 0x1) {
    if (!sub || sub.unsubscribed) return;
    sub.unsubscribed = true;
    try { sub.writer.reset(code); } catch { /* already closed */ }
    try { sub.reader.cancel(code); } catch { /* already closed */ }
    if (sub.alias != null) this.tracks.delete(sub.alias);
    this.counters.unsubscribed = (this.counters.unsubscribed || 0) + 1;
  }

  // FETCH: every entry of the fetch stream, until its FIN.
  async fetch(namespace, name, params, onObject) {
    const id = this.nextRequestId;
    let finished;
    const ended = new Promise((r) => { finished = r; });
    const entries = { objects: 0, ranges: [] };
    this.fetches.set(id, {
      order: (params && params.group_order) || 1,
      stream: () => ({
        object: (j, payload, k, props) => { entries.objects++; onObject(j, payload, props); },
        endOfRange: (e) => entries.ranges.push(e),
        end: () => finished("fin"),
        reset: (code) => finished(`reset ${code}`),
      }),
    });
    let r;
    try {
      r = await this.request({ message: "FETCH", namespace, name, params });
    } catch (e) {
      this.fetches.delete(id);
      throw e;
    }
    if (r.first.message !== "FETCH_OK") {
      this.fetches.delete(id);
      throw MoqSession.refusal(`FETCH ${namespace.join("/")}--${name}`, r.first);
    }
    const how = await withTimeout(ended, STEP_MS * 2, "the fetch stream");
    this.fetches.delete(id);
    return { ok: r.first, how, ...entries };
  }

  // One subgroup stream of `bytes`, at `sendOrder` where the transport has it.
  async sendStream(bytes, sendOrder) {
    const w = await this.t.openUni({ sendOrder });
    await w.write(bytes);
    await w.close();
  }

  async openStream(sendOrder) {
    return this.t.openUni({ sendOrder });
  }

  close(code = 0, reason = "") {
    this.t.close(code, reason);
  }
}