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

// A conversation in a page (WAI live-messaging): a member of an end-to-end encrypted room
// kept by a room service (`joule-moq-room` with `[messaging]`).
//
// The page makes its own Ed25519 key and joins with a proof of it (a proof naming no grant,
// live-rooms §10.5.3); the service admits it under a grant, and its clock records the `join`.
// The page leaves a KeyPackage and reads the conversation's log from its first entry
// (live-messaging §9.1): the clock's lifecycle records, each checked as the clock's; the
// group's handshake messages, which its MLS member (`joule_moq_web_e2ee`'s E2eeMember, a
// conversation's member, `anyMemberCommits`) checks and applies, creating the group or
// committing when the room calls on it (§7.2); the clock's epoch statements, each checked
// against the epoch the member reached (§7.3); messages, each opened by the member through the
// receiving checks and verified as §11.3 verifies it (`wai_live_web`'s Messages); and
// acknowledgements. It seals each message it sends under its latest epoch, signs it into its
// receipt chain (§4.3), and posts it; it acknowledges what it has shown (§8). Threads,
// reactions, edits and retractions are messages naming their target's link hash (§6): a sink
// shows an edited message's current text and that it was edited, and a retracted one's body
// never. The member's MLS state and every key it derives live in its memory alone: a page that
// is closed and opened again is a new member (§7.5), and opens nothing it held before.
//
// Query parameters:
//   service=<url>   the room service, on the page's own origin (./endpoint.js)
//
// Everything a test reads is on `window.msgTest`.

import initWai, * as wai from "../pkg/wai_live_web.js";
import { checkEndpoint } from "./endpoint.js";

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 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));
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
const KIND = { text: 1, reply: 2, reaction: 3, edit: 4, retraction: 5 };
const PRIORITY = 64; // live-messaging §4.2

const q = new URLSearchParams(location.search);
const logLines = [];
function log(m) {
  logLines.push(`${(performance.now() / 1000).toFixed(3)} ${m}`);
  const el = $("log");
  if (el) el.textContent = logLines.slice(-200).join("\n");
}
function status(m) {
  const el = $("status");
  if (el) el.textContent = m;
}

const state = {
  phase: "idle", // idle, joining, member, removed, left, refused
  me: null,
  room: null,
  at: 0,
  epoch: null,
  members: [],
  view: null,
  refused: {},
  posts: { ok: 0, refused: {} },
  acks: { ok: 0, refused: {} },
  opened: 0,
  unopened: 0,
  offline: false,
  error: null,
};
window.msgTest = { state, logLines };

let service = null;
{
  const v = q.get("service");
  if (v) {
    const c = checkEndpoint(v, "http", location.href);
    if (c.ok) service = c.url.replace(/\/$/, "");
    else {
      state.phase = "refused";
      status(`This link names a room service that is not this page's own (${v}: ${c.reason}). The page does not use it.`);
    }
  }
}

let moq = null;
let member = null;
let messages = null; // wai_live_web's Messages: the conversation as this member verifies it
let seed = null;
let grant = null;
let conv = null; // GET /msg's view at joining
let chain = { seq: 0, prev: "00".repeat(32) };
const challenges = []; // single-use, each from a response of the service (live-rooms §10.5.3)
const acked = {}; // author -> the last group this page acknowledged read

// One request to the service, with this page's grant (or, before it has one, a proof of its
// key alone) and a proof over a challenge the service issued; asked once more with a fresh
// challenge when refused 401.
async function request(method, path, body) {
  const u = new URL(`${service}${path}`, location.href);
  for (let i = 0; i < 2; i++) {
    let ch = challenges.pop();
    if (!ch) {
      const r = await fetch(new URL(`${service}/msg/challenge`, location.href), { cache: "no-store" });
      ch = unhex((await r.json()).challenge);
    }
    const target = `//${u.host}${u.pathname}`;
    const proof = grant ? moq.grantProofHttp(unhex(grant), seed, ch, method, target) : moq.keyProofHttp(seed, ch, method, target);
    const headers = { "wai-room-proof": b64u(proof) };
    if (grant) headers.authorization = `Bearer ${b64u(unhex(grant))}`;
    if (body) headers["content-type"] = "application/json";
    const r = await fetch(u, { method, headers, body: body ? JSON.stringify(body) : undefined, cache: "no-store" });
    const c = r.headers.get("wai-room-challenge");
    if (c) challenges.push(fromB64u(c));
    let j = null;
    try { j = await r.json(); } catch { /* not JSON */ }
    if (r.status !== 401) return { status: r.status, body: j };
  }
  return { status: 401, body: null };
}

// Every change to the member and to the verified conversation, one at a time.
let lock = Promise.resolve();
const serial = (f) => {
  const p = lock.then(f, f);
  lock = p.catch(() => {});
  return p;
};

function nsOf(author) {
  return [state.room, author, "messages"];
}

// One entry of the log, in its order.
function take(e) {
  const seq = BigInt(e.seq);
  if (e.event) {
    const v = JSON.parse(wai.eventRead(unhex(e.event)));
    if (!v.ok || v.principal !== conv.clock || v.room !== state.room) throw new Error(`a lifecycle record that is not the clock's: ${v.reason || v.principal}`);
    if (v.seq !== chain.seq || v.prev !== chain.prev) throw new Error(`the clock's chain breaks at ${v.seq}`);
    chain = { seq: v.seq + 1, prev: v.digest };
    const p = v.payload || {};
    if (v.kind_name === "join") member.event(JSON.stringify({ event: "join", subject: p.subject, keyed: true }));
    else if (v.kind_name === "leave") member.event(JSON.stringify({ event: "leave", subject: p.subject, reason: p.detail.reason }));
  } else if (e.mls) {
    const v = JSON.parse(member.receive(unhex(e.mls)));
    if (!v.ok) {
      const why = v.reason || v.error;
      state.refused[why] = (state.refused[why] || 0) + 1;
      log(`an MLS message refused: ${why}${v.detail ? ` (${v.detail})` : ""}`);
    } else if (["epoch", "joined", "created"].includes(v.received)) {
      log(`epoch ${v.epoch} (${v.received})`);
    } else if (v.received === "removed") {
      log(`removed from the group by the commit that opens epoch ${v.epoch}`);
    }
  } else if (e.epoch) {
    const v = JSON.parse(messages.offerEpoch(seq, unhex(e.epoch)));
    if (!v.ok) throw new Error(`an epoch statement does not verify: ${v.reason}`);
    // §7.3: the statement names the roster this member's group reached.
    const ms = JSON.parse(member.state());
    if (ms.epoch === v.epoch && ms.roster) {
      const mine = Object.entries(ms.roster).map(([l, p]) => [Number(l), p]).sort((a, b) => a[0] - b[0]);
      if (JSON.stringify(mine) !== JSON.stringify(v.roster)) throw new Error(`the epoch statement for epoch ${v.epoch} is not the roster this page's group reached`);
    }
    state.epoch = { generation: v.generation, epoch: v.epoch };
    state.members = v.roster.map((x) => x[1]);
  } else if (e.message) {
    const m = e.message;
    const props = unhex(m.properties);
    const payload = unhex(m.payload);
    const o = member.open(JSON.stringify(nsOf(m.author)), "0", BigInt(m.group), 0n, PRIORITY, props, payload);
    const ov = JSON.parse(o.json);
    const opened = ov.ok ? o.payload() : undefined;
    o.free();
    if (ov.ok) state.opened++;
    else { state.unopened++; log(`a message at ${e.seq} did not open: ${ov.reason || `pending ${ov.pending}`}`); }
    const r = JSON.parse(messages.offerMessage(seq, unhex(m.author), BigInt(m.group), props, payload, unhex(m.receipts), opened));
    if (r.outcome !== "accepted") log(`a message at ${e.seq}: ${r.outcome}`);
  } else if (e.ack) {
    messages.offerAck(seq, unhex(e.ack));
  }
}

// Read the log from where this page stopped; `wait` holds the request while nothing is new.
async function readOnce(wait) {
  const r = await request("GET", `/msg/log?from=${state.at}&wait_ms=${wait}`);
  if (r.status === 403) {
    state.phase = state.phase === "left" ? "left" : "removed";
    status(state.phase === "removed" ? "You are no longer a member of this conversation: nothing sent after your removal reaches this page." : "You left the conversation.");
    return false;
  }
  if (r.status !== 200) throw new Error(`the log answered ${r.status}`);
  await serial(async () => {
    for (const e of r.body.entries) {
      if (e.seq < state.at) continue;
      take(e);
      state.at = e.seq + 1;
    }
    state.at = Math.max(state.at, r.body.next);
    await act();
    render();
  });
  return true;
}

// Create the group, or commit, when the room calls on this member; the log's order decides
// whether it is applied (confidentiality §15.2).
async function act() {
  if (member.owesCreate()) {
    await postMls([member.stageCreate()], "GroupInfo");
  } else if (member.shouldCommit()) {
    const c = member.stageCommit();
    const w = c.welcome();
    const frames = [c.commit(), ...(w ? [w] : [])];
    const e = c.epoch;
    c.free();
    await postMls(frames, `commit opening epoch ${e}`);
  }
}

async function postMls(frames, what) {
  const r = await request("POST", "/msg/mls", { frames: frames.map(hex) });
  if (r.status === 200) log(`posted the ${what}`);
  else {
    member.withdraw();
    log(`the ${what} was not carried: ${r.status}${r.body && r.body.error ? ` (${r.body.error})` : ""}`);
  }
}

async function follow() {
  let back = 500;
  while (state.phase === "member") {
    if (state.offline) { await sleep(200); continue; }
    try {
      if (!(await readOnce(member.staged() ? 0 : 4000))) break;
      await ackRead();
      back = 500;
    } catch (e) {
      log(`reading the log: ${e.message || e}`);
      await sleep(back);
      back = Math.min(back * 2, 4000);
    }
  }
}

function roomNow() {
  return BigInt(Date.now()) * 1000n - BigInt(conv.t0_us);
}

// Seal, sign into this page's receipt chain, and post one message (§4.2, §4.3, §5); on a
// stale epoch or a group already taken, read the log and try once more.
async function send(kind, target, text) {
  if (state.phase !== "member") throw new Error("not a member");
  for (let attempt = 0; attempt < 3; attempt++) {
    const r = await serial(async () => {
      if (member.epoch() === undefined) return { status: 0, body: { error: "no epoch yet" } };
      const nx = JSON.parse(messages.nextOf(unhex(state.me)));
      const group = BigInt(nx.group);
      const rec = wai.messageRecord(seed, unhex(state.room), group, roomNow(), KIND[kind], target ? unhex(target) : new Uint8Array(0), text);
      const s = member.seal(JSON.stringify(nsOf(state.me)), "0", group, 0n, PRIORITY, rec, false, false, 0n);
      const props = s.properties();
      const payload = s.payload();
      s.free();
      const rc = JSON.parse(wai.messageReceipt(seed, group, payload, nx.parent ? unhex(nx.parent) : new Uint8Array(0), new Date().toISOString().replace(/\.\d+Z$/, "Z")));
      return request("POST", "/msg/post", { group: nx.group, properties: hex(props), payload: hex(payload), receipts: rc.receipts }).then((x) => ({ ...x, link: rc.link }));
    });
    if (r.status === 200) {
      state.posts.ok++;
      await readOnce(0);
      return r.link;
    }
    const why = (r.body && r.body.error) || String(r.status);
    state.posts.refused[why] = (state.posts.refused[why] || 0) + 1;
    log(`a message was not taken: ${why}`);
    if (!["epoch_stale", "message_gap", "no epoch yet"].includes(why)) throw new Error(`the conversation refused the message: ${why}`);
    await readOnce(0);
    await sleep(200);
  }
  throw new Error("the message was not taken");
}

// Acknowledge, as read, every author's track up to what this page shows (§8).
async function ackRead() {
  if (document.visibilityState === "hidden") return;
  const v = JSON.parse(messages.view());
  const last = {};
  for (const m of v.messages) if (m.outcome === "accepted" && m.author !== state.me) last[m.author] = m;
  for (const [author, m] of Object.entries(last)) {
    if (acked[author] != null && acked[author] >= m.group) continue;
    const a = wai.ackRecord(seed, unhex(state.room), unhex(author), BigInt(m.group), unhex(m.link), 2, roomNow());
    const r = await request("POST", "/msg/ack", { ack: hex(a) });
    if (r.status === 200) { acked[author] = m.group; state.acks.ok++; }
    else { const why = (r.body && r.body.error) || String(r.status); state.acks.refused[why] = (state.acks.refused[why] || 0) + 1; acked[author] = m.group; }
  }
}

// --------------------------------------------------------------------------- the view

let replyTo = null;
let editing = null;

function short(p) {
  return p === state.me ? "you" : p.slice(0, 8);
}

function render() {
  if (!messages) return;
  const v = JSON.parse(messages.view());
  state.view = v;
  const list = $("messages");
  if (!list) return;
  const byLink = new Map(v.messages.filter((m) => m.outcome === "accepted").map((m) => [m.link, m]));
  const roots = v.messages.filter((m) => m.outcome === "accepted" && m.kind === "text");
  list.replaceChildren(...roots.map((m) => item(m, byLink)));
  $("who").textContent = `you · ${state.me.slice(0, 12)}`;
  if (state.epoch) $("epoch").textContent = `Epoch ${state.epoch.epoch} of the room's group`;
  $("members").replaceChildren(...state.members.map((p) => { const li = document.createElement("li"); li.textContent = short(p); return li; }));
}

function item(m, byLink) {
  const li = document.createElement("li");
  li.dataset.link = m.link;
  if (m.author === state.me) li.classList.add("mine");
  const body = document.createElement("div");
  body.className = "body";
  const who = document.createElement("span");
  who.className = "meta";
  who.textContent = `${short(m.author)} `;
  body.append(who);
  const text = document.createElement("span");
  if (m.status === "retracted") { text.className = "retracted"; text.textContent = "retracted"; }
  else text.textContent = m.text;
  body.append(text);
  if (m.status === "edited") { const e = document.createElement("span"); e.className = "meta"; e.textContent = " (edited)"; body.append(e); }
  if (m.status !== "retracted") body.append(actions(m));
  li.append(body);
  if (m.reactions && m.reactions.length) {
    const counts = {};
    for (const [w, r] of m.reactions) counts[r] = (counts[r] || 0) + 1;
    const rs = document.createElement("div");
    rs.className = "reactions";
    for (const [r, n] of Object.entries(counts)) { const s = document.createElement("span"); s.textContent = `${r} ${n}`; rs.append(s); }
    li.append(rs);
  }
  if (m.replies && m.replies.length) {
    const ul = document.createElement("ul");
    ul.className = "thread";
    for (const l of m.replies) { const r = byLink.get(l); if (r) ul.append(item(r, byLink)); }
    li.append(ul);
  }
  return li;
}

function button(label, f) {
  const b = document.createElement("button");
  b.type = "button";
  b.textContent = label;
  b.addEventListener("click", () => f().catch(fail));
  return b;
}

function actions(m) {
  const s = document.createElement("span");
  s.className = "acts";
  if (m.kind === "text") s.append(button("Reply", async () => { replyTo = m.link; editing = null; showTarget(`Replying to ${short(m.author)}`); }));
  const mine = m.reactions && m.reactions.some(([w, r]) => w === state.me && r === "+1");
  if (!mine && m.author !== state.me) s.append(button("+1", () => send("reaction", m.link, "+1")));
  if (m.author === state.me) {
    s.append(button("Edit", async () => { editing = m.link; replyTo = null; $("text").value = m.text || ""; showTarget("Editing your message"); }));
    s.append(button("Retract", () => send("retraction", m.link, "")));
  }
  return s;
}

function showTarget(t) {
  $("replying").hidden = !t;
  $("replying").textContent = t || "";
  $("cancel").hidden = !t;
  $("text").focus();
}

function fail(e) {
  state.error = String(e && e.message || e);
  log(`error: ${state.error}`);
  status(`Something went wrong: ${state.error}`);
}

// --------------------------------------------------------------------------- joining

async function join() {
  if (state.phase !== "idle") return state.phase;
  state.phase = "joining";
  const m = await import("../pkg/joule_moq_web_e2ee.js");
  await m.default();
  moq = m;
  await initWai();
  seed = crypto.getRandomValues(new Uint8Array(32));
  state.me = hex(moq.publicKeyOf(seed));
  status("Asking to join…");
  const r = await request("POST", "/msg/join", {});
  if (r.status !== 200) {
    state.phase = "refused";
    status(`The conversation did not admit this page: ${(r.body && r.body.error) || r.status}`);
    return state.phase;
  }
  grant = r.body.grant;
  conv = r.body.conversation;
  state.room = conv.room;
  // The room's confidentiality record, checked against its descriptor (confidentiality §3.3).
  const d = JSON.parse(wai.roomDescriptor(unhex(conv.descriptor)));
  const c = JSON.parse(moq.e2eeCheckRecord(unhex(conv.record), unhex(conv.room), unhex(d.authority), d.recording_mode));
  if (!c.ok) throw new Error(`the conversation's confidentiality record is refused: ${c.reason}`);
  messages = new wai.Messages(unhex(conv.descriptor), unhex(conv.record), unhex(conv.statement));
  member = new moq.E2eeMember(unhex(conv.room), unhex(conv.clock), seed, c.suite);
  member.anyMemberCommits();
  const kp = await request("POST", "/msg/key-package", { key_package: hex(member.keyPackage()) });
  if (kp.status !== 200) throw new Error(`the conversation kept no KeyPackage: ${kp.status}`);
  state.phase = "member";
  status("You are a member. Messages are sealed on this page before they are sent.");
  $("compose").hidden = false;
  $("join").hidden = true;
  $("leave").hidden = false;
  render();
  follow().catch(fail);
  return state.phase;
}

async function leave() {
  if (state.phase !== "member") return;
  state.phase = "left";
  await request("POST", "/msg/leave", {});
  status("You left the conversation.");
}

Object.assign(window.msgTest, {
  join,
  leave,
  send: (text) => send("text", null, text),
  reply: (target, text) => send("reply", target, text),
  react: (target, r) => send("reaction", target, r),
  edit: (target, text) => send("edit", target, text),
  retract: (target) => send("retraction", target, ""),
  remove: (principal) => request("POST", "/msg/remove", { principal }),
  setOffline: (v) => { state.offline = !!v; },
  memberState: () => (member ? JSON.parse(member.state()) : null),
  view: () => (messages ? JSON.parse(messages.view()) : null),
});

if (service && state.phase === "idle") {
  $("join").addEventListener("click", () => join().catch(fail));
  $("leave").addEventListener("click", () => leave().catch(fail));
  $("cancel").addEventListener("click", () => { replyTo = null; editing = null; showTarget(null); });
  $("compose").addEventListener("submit", (e) => {
    e.preventDefault();
    const t = $("text").value.trim();
    if (!t) return;
    $("text").value = "";
    const go = editing ? send("edit", editing, t) : replyTo ? send("reply", replyTo, t) : send("text", null, t);
    replyTo = null;
    editing = null;
    showTarget(null);
    go.catch(fail);
  });
  status("Join to read and write in this conversation.");
} else if (!service && state.phase === "idle") {
  $("join").disabled = true;
  status("No room service: open this page with ?service=<url> on its own origin.");
}