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

// How long a request waits for its answer (`node --test web/moq/session.test.mjs`), over a
// transport and a codec of the test's own, the clock mocked.
import test from "node:test";
import assert from "node:assert/strict";
import { MoqSession, RequestRefused, SetupRefused } from "./session.js";

const enc = new TextEncoder();
const hex = (s) => [...enc.encode(s)].map((b) => b.toString(16).padStart(2, "0")).join("");

// Control messages as JSON text, one per read.
const moq = {
  encodeControl: (s) => enc.encode(s),
  MessageStream: class {
    constructor() { this.q = []; }
    push(v) { this.q.push(new TextDecoder().decode(v)); }
    next() { return this.q.shift(); }
    free() {}
  },
};

// Each request stream's reader gives what `answer(i, message)` hands it, and nothing before.
function transport() {
  const answers = [];
  return {
    kind: "test",
    closed: new Promise(() => {}),
    async openBidi() {
      let give;
      const first = new Promise((r) => { give = r; });
      let read = 0;
      answers.push(give);
      return {
        writer: { async write() {}, reset() {} },
        reader: { read: () => (read++ === 0 ? first : new Promise(() => {})) },
      };
    },
    answer(i, message) { answers[i]({ value: enc.encode(JSON.stringify(message)), done: false }); },
  };
}

const settle = async () => { for (let i = 0; i < 8; i++) await new Promise((r) => setImmediate(r)); };
const refused = { message: "REQUEST_ERROR", request_id: 0, code: 2, reason: hex("no publisher within RENDEZVOUS_TIMEOUT") };

test("a SUBSCRIBE that asks for a rendezvous takes the relay's answer when the rendezvous ends", async (t) => {
  t.mock.timers.enable({ apis: ["setTimeout"] });
  const tr = transport();
  const s = new MoqSession(tr, moq);
  const p = s.subscribe(["room", "a", "world"], "0", { rendezvous_timeout: 10000 }, {});
  p.catch(() => {});
  await settle();
  // The relay holds the request for the whole rendezvous, and its answer arrives after it.
  t.mock.timers.tick(10_500);
  await settle();
  tr.answer(0, refused);
  await assert.rejects(p, (e) => e instanceof RequestRefused && /REQUEST_ERROR 2 no publisher within RENDEZVOUS_TIMEOUT/.test(e.message));
});

test("a SUBSCRIBE that asks for a rendezvous waits no longer than the step past it", async (t) => {
  t.mock.timers.enable({ apis: ["setTimeout"] });
  const s = new MoqSession(transport(), moq);
  const p = s.subscribe(["room", "a", "world"], "0", { rendezvous_timeout: 10000 }, {});
  p.catch(() => {});
  await settle();
  t.mock.timers.tick(20_000);
  await assert.rejects(p, (e) => !(e instanceof RequestRefused) && /SUBSCRIBE: no answer in 20000 ms/.test(e.message));
});

test("a SUBSCRIBE without a rendezvous waits one step for its answer", async (t) => {
  t.mock.timers.enable({ apis: ["setTimeout"] });
  const s = new MoqSession(transport(), moq);
  const p = s.subscribe(["room", "a", "world"], "0", { forward: true }, {});
  p.catch(() => {});
  await settle();
  t.mock.timers.tick(10_000);
  await assert.rejects(p, /SUBSCRIBE: no answer in 10000 ms/);
});

// ---- SETUP: this end's first, and a grant proved on a second session (live-rooms §10.5.2) ----

const CHALLENGE = 0x3a47;
const REFUSED = 0x3a53;
const bytesHex = (b) => [...b].map((x) => x.toString(16).padStart(2, "0")).join("");

// A codec that writes a SETUP as JSON text saying what it presents, and reads a control
// stream's items as JSON text.
const setupMoq = {
  ...moq,
  setDraft() {},
  grantChallengeOption: () => BigInt(CHALLENGE),
  grantRefusedOption: () => BigInt(REFUSED),
  clientSetup: (token) => enc.encode(JSON.stringify({ presents: token ? "token" : "nothing", token: token ? bytesHex(token) : null })),
  clientSetupWithGrant: (record, seed, challenge, target) => enc.encode(JSON.stringify({ presents: "grant+proof", over: bytesHex(challenge), target })),
  UniDecoder: class {
    constructor() { this.q = []; }
    push(v) { this.q.push(new TextDecoder().decode(v)); }
    next() { const m = this.q.shift(); return m ? { kind: "control", json: JSON.stringify({ control: JSON.parse(m) }), free() {} } : null; }
    finish() {}
  },
};

// A relay that reads the client's SETUP before it sends its own: its control stream gives
// `relaySetup` only once this end has written. A session that waited for the relay's SETUP
// first would wait for ever.
function relayTransport(relaySetup) {
  const t = { kind: "test", url: "https://relay.example/moq", written: [], closedBy: null };
  let wrote;
  const written = new Promise((r) => { wrote = r; });
  let gave = false;
  t.closed = new Promise(() => {});
  t.close = (code, reason) => { t.closedBy = { code, reason }; };
  t.openUni = async () => ({ async write(b) { t.written.push(JSON.parse(new TextDecoder().decode(b))); wrote(); } });
  t.acceptBidi = () => new Promise(() => {});
  let streams = 0;
  t.acceptUni = () => (streams++ === 0
    ? Promise.resolve({ read: async () => { if (gave) return new Promise(() => {}); await written; gave = true; return { value: enc.encode(JSON.stringify(relaySetup)), done: false }; }, cancel() {} })
    : new Promise(() => {}));
  return t;
}

const relaySays = (challenge, refused) => ({
  message: "SETUP",
  options: [{ type: CHALLENGE, bytes: challenge }, ...(refused ? [{ type: REFUSED, bytes: hex(refused) }] : [])],
});
const record = new Uint8Array([1, 2, 3]);
const grant = { record, seed: new Uint8Array(32) };

test("a page that holds no challenge presents its grant unproved, is given one, and proves on a second session", async () => {
  const first = relayTransport(relaySays("aa".repeat(32), "pop_missing"));
  const second = relayTransport(relaySays("bb".repeat(32)));
  let redials = 0;
  const s = await MoqSession.connect(first, setupMoq, { grant, redial: async () => { redials++; return second; } });
  assert.deepEqual(first.written, [{ presents: "token", token: "010203" }], "the grant with no proof, sent before the relay's SETUP is read");
  assert.deepEqual(first.closedBy, { code: 0, reason: "refused at SETUP" }, "this end leaves a session the relay refused");
  assert.deepEqual(second.written, [{ presents: "grant+proof", over: "aa".repeat(32), target: "//relay.example/moq" }], "the proof over the challenge the first SETUP gave");
  assert.equal(redials, 1);
  assert.equal(s.sessionsToAdmit, 2);
  assert.equal(bytesHex(s.nextChallenge), "bb".repeat(32), "the admitting SETUP gives the next challenge");
});

test("a refusal no proof answers is thrown with its reason, and no second session is opened", async () => {
  let redials = 0;
  const redial = async () => { redials++; return relayTransport(relaySays("cc".repeat(32))); };
  // A proof by another key: the relay says so, and a proof over a new challenge changes nothing.
  const wrongKey = relayTransport(relaySays("aa".repeat(32), "pop_wrong_key"));
  await assert.rejects(MoqSession.connect(wrongKey, setupMoq, { grant, challenge: new Uint8Array(32), redial }), (e) => e instanceof SetupRefused && e.reason === "pop_wrong_key" && !e.asksForProof);
  // A second refusal is the last: the second session's is thrown as it is.
  const again = relayTransport(relaySays("dd".repeat(32), "pop_wrong_challenge"));
  await assert.rejects(
    MoqSession.connect(relayTransport(relaySays("aa".repeat(32), "pop_missing")), setupMoq, { grant, redial: async () => { redials++; return again; } }),
    (e) => e instanceof SetupRefused && e.reason === "pop_wrong_challenge",
  );
  assert.equal(redials, 1, "one second session, and no third");
  // With no way to open another session, the refusal is the caller's.
  await assert.rejects(MoqSession.connect(relayTransport(relaySays("aa".repeat(32), "pop_missing")), setupMoq, { grant }), (e) => e instanceof SetupRefused && e.asksForProof && bytesHex(e.challenge) === "aa".repeat(32));
  // A token, not a grant: refused is refused.
  await assert.rejects(MoqSession.connect(relayTransport(relaySays("aa".repeat(32), "pop_missing")), setupMoq, { token: record, redial }), (e) => e instanceof SetupRefused && e.reason === "pop_missing");
  assert.equal(redials, 1);
});

test("a session that presents no grant sends its SETUP first and is admitted on its first session", async () => {
  const t = relayTransport(relaySays("aa".repeat(32)));
  const s = await MoqSession.connect(t, setupMoq, {});
  assert.deepEqual(t.written, [{ presents: "nothing", token: null }]);
  assert.equal(s.sessionsToAdmit, 1);
  assert.equal(t.closedBy, null);
});