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

// The capsules of joule-moq-relay's WebSocket binding (its README, "The WebSocket
// binding"): those of draft-ietf-webtrans-http2-15 §6, one capsule to each binary
// message, `Type (i) Length (i) Value` with QUIC variable-length integers (RFC 9000 §16),
// not MOQT's. The MOQT bytes a stream carries ride inside WT_STREAM unchanged.
//
// Plain ECMAScript, no DOM: it runs in a page and in Node (capsule.test.mjs).

export const TY = Object.freeze({
  DATAGRAM: 0x00,
  CLOSE_SESSION: 0x2843,
  DRAIN_SESSION: 0x78ae,
  PADDING: 0x190b4d38,
  RESET_STREAM: 0x190b4d39,
  STOP_SENDING: 0x190b4d3a,
  STREAM_FIN: 0x190b4d3b, // WT_STREAM with the FIN bit
  STREAM: 0x190b4d3c, // WT_STREAM without it
  MAX_STREAMS_BIDI: 0x190b4d3f,
  MAX_STREAMS_UNI: 0x190b4d40,
  STREAMS_BLOCKED_BIDI: 0x190b4d43,
  STREAMS_BLOCKED_UNI: 0x190b4d44,
});

// The largest stream data one capsule carries, as the relay's binding chunks it.
export const CHUNK = 16 * 1024;
// WT_FLOW_CONTROL_ERROR as a 32-bit application code: what a receiver closes with when
// the peer opens past its credit.
export const FLOW_CONTROL_ERROR = 0x045d4487;
const MAX_CLOSE_MESSAGE = 1024;
const MAX_STREAMS_LIMIT = 2 ** 60;

// QUIC varints up to 2^53 - 1: every value this binding carries (stream IDs, codes,
// stream counts) fits a JavaScript number exactly.
export function qvarintLength(v) {
  if (v < 0x40) return 1;
  if (v < 0x4000) return 2;
  if (v < 0x40000000) return 4;
  return 8;
}

export function putQvarint(out, at, v) {
  if (!Number.isSafeInteger(v) || v < 0) throw new RangeError(`a QUIC varint of ${v}`);
  const n = qvarintLength(v);
  if (n === 1) out[at] = v;
  else if (n === 2) { out[at] = 0x40 | (v >> 8); out[at + 1] = v & 0xff; }
  else if (n === 4) {
    out[at] = 0x80 | ((v >>> 24) & 0x3f); out[at + 1] = (v >>> 16) & 0xff; out[at + 2] = (v >>> 8) & 0xff; out[at + 3] = v & 0xff;
  } else {
    const hi = Math.floor(v / 2 ** 32), lo = v >>> 0;
    out[at] = 0xc0 | ((hi >>> 24) & 0x3f); out[at + 1] = (hi >>> 16) & 0xff; out[at + 2] = (hi >>> 8) & 0xff; out[at + 3] = hi & 0xff;
    out[at + 4] = (lo >>> 24) & 0xff; out[at + 5] = (lo >>> 16) & 0xff; out[at + 6] = (lo >>> 8) & 0xff; out[at + 7] = lo & 0xff;
  }
  return at + n;
}

// [value, length] of the varint at the front of `b`, or null when it is not all there.
export function getQvarint(b, at = 0) {
  if (at >= b.length) return null;
  const n = 1 << (b[at] >> 6);
  if (at + n > b.length) return null;
  let v = b[at] & 0x3f;
  for (let i = 1; i < n; i++) v = v * 256 + b[at + i];
  return [v, n];
}

function frame(type, value) {
  const out = new Uint8Array(qvarintLength(type) + qvarintLength(value.length) + value.length);
  let at = putQvarint(out, 0, type);
  at = putQvarint(out, at, value.length);
  out.set(value, at);
  return out;
}

function varints(...vs) {
  const out = new Uint8Array(vs.reduce((n, v) => n + qvarintLength(v), 0));
  let at = 0;
  for (const v of vs) at = putQvarint(out, at, v);
  return out;
}

export function streamCapsule(id, data, fin) {
  const head = varints(id);
  const value = new Uint8Array(head.length + data.length);
  value.set(head, 0);
  value.set(data, head.length);
  return frame(fin ? TY.STREAM_FIN : TY.STREAM, value);
}

export const resetStreamCapsule = (id, code) => frame(TY.RESET_STREAM, varints(id, code, 0));
export const stopSendingCapsule = (id, code) => frame(TY.STOP_SENDING, varints(id, code));
export const maxStreamsCapsule = (bidi, max) => frame(bidi ? TY.MAX_STREAMS_BIDI : TY.MAX_STREAMS_UNI, varints(max));
export const streamsBlockedCapsule = (bidi, max) => frame(bidi ? TY.STREAMS_BLOCKED_BIDI : TY.STREAMS_BLOCKED_UNI, varints(max));

export function closeSessionCapsule(code, message) {
  let m = new TextEncoder().encode(message || "");
  if (m.length > MAX_CLOSE_MESSAGE) m = m.subarray(0, MAX_CLOSE_MESSAGE);
  const v = new Uint8Array(4 + m.length);
  new DataView(v.buffer).setUint32(0, code >>> 0);
  v.set(m, 4);
  return frame(TY.CLOSE_SESSION, v);
}

// Read a message that holds exactly one capsule. Throws on one that is not a whole
// capsule (the session then closes) or whose value does not parse.
export function decodeCapsule(msg) {
  const t = getQvarint(msg, 0);
  if (!t) throw new Error("the message is not one whole capsule");
  const l = getQvarint(msg, t[1]);
  if (!l || t[1] + l[1] + l[0] !== msg.length) throw new Error("the message is not one whole capsule");
  const value = msg.subarray(t[1] + l[1]);
  const type = t[0];
  const read = (what, count) => {
    const out = [];
    let at = 0;
    for (let i = 0; i < count; i++) {
      const x = getQvarint(value, at);
      if (!x) throw new Error(`a ${what} capsule whose value does not parse`);
      out.push(x[0]);
      at += x[1];
    }
    return [out, at];
  };
  const exact = (what, count) => {
    const [vs, at] = read(what, count);
    if (at !== value.length) throw new Error(`a ${what} capsule whose value does not parse`);
    return vs;
  };
  switch (type) {
    case TY.STREAM:
    case TY.STREAM_FIN: {
      const [[id], at] = read("WT_STREAM", 1);
      return { kind: "stream", id, data: value.subarray(at), fin: type === TY.STREAM_FIN };
    }
    case TY.RESET_STREAM: {
      const [id, code] = exact("WT_RESET_STREAM", 3);
      return { kind: "reset", id, code };
    }
    case TY.STOP_SENDING: {
      const [id, code] = exact("WT_STOP_SENDING", 2);
      return { kind: "stop", id, code };
    }
    case TY.MAX_STREAMS_BIDI:
    case TY.MAX_STREAMS_UNI: {
      const [max] = exact("WT_MAX_STREAMS", 1);
      if (max > MAX_STREAMS_LIMIT) throw new Error("a WT_MAX_STREAMS above 2^60");
      return { kind: "max_streams", bidi: type === TY.MAX_STREAMS_BIDI, max };
    }
    case TY.STREAMS_BLOCKED_BIDI:
    case TY.STREAMS_BLOCKED_UNI:
      return { kind: "blocked", bidi: type === TY.STREAMS_BLOCKED_BIDI, max: exact("WT_STREAMS_BLOCKED", 1)[0] };
    case TY.DATAGRAM:
      return { kind: "datagram", data: value };
    case TY.CLOSE_SESSION: {
      if (value.length < 4 || value.length > 4 + MAX_CLOSE_MESSAGE) throw new Error("a WT_CLOSE_SESSION capsule whose value does not parse");
      const code = new DataView(value.buffer, value.byteOffset, 4).getUint32(0);
      return { kind: "close", code, message: new TextDecoder("utf-8", { fatal: true }).decode(value.subarray(4)) };
    }
    case TY.DRAIN_SESSION:
      return { kind: "drain" };
    default:
      return { kind: "ignored", type };
  }
}

// Stream IDs as QUIC numbers them: bit 0 the initiator (0 the client), bit 1 the
// direction (0 bidirectional); the nth stream of a kind is 4n + kind.
export const streamId = Object.freeze({
  clientInitiated: (id) => id % 2 === 0,
  bidi: (id) => (id & 2) === 0,
  nth: (n, client, bidi) => n * 4 + (client ? 0 : 1) + (bidi ? 0 : 2),
  index: (id) => Math.floor(id / 4),
});