open-standards/wai/wai-web/peer/page.js

// The reference page of peer delivery (extensions/peer-delivery.md §7): an asker or a holder,
// by its query, over data channels and the HTTP holders its configuration names.
//
//   ?role=asker&id=a1&peers=h1,h2          ask those holders, in that order, then the HTTP holders
//   ?role=holder&id=h1&holds=0:0-3;1:1     hold those objects of the world (group:positions)
//
// The operator's policy comes from `config.json`, served beside the page; the viewer's setting
// is the two checkboxes, off until someone checks them. Introductions go through the mailbox
// the configuration names: one page posts its session description for another, which answers.
//
// A holder page takes three more parameters, which plant the faults run.mjs holds an asker to.
// None is anything a holder may do (§4.3): `alter=0:0` serves those objects with a byte
// changed, `lists=0:0-5` lists objects in its record beyond what it holds, and `count=1`
// states each group one object larger than it is.

import init, * as w from "./pkg/wai_peer.js";
import { PeerClient, PeerServer, cacheStore, channelLink, hex, overHttp, overLink } from "./peer.js";

const q = new URLSearchParams(location.search);
const id = q.get("id") || "page";
const role = q.get("role") || "asker";
const hook = (window.__wai_peer = { state: "loading", role, id, error: null });
const $ = (x) => document.getElementById(x);
const status = (text) => ($("status").textContent = text);

// "0:0-3;1:1" → [[0, 0], [0, 1], [0, 2], [0, 3], [1, 1]]
const objects = (text) =>
  (text || "").split(";").filter(Boolean).flatMap((part) => {
    const [g, list] = part.split(":");
    return list.split(",").flatMap((r) => {
      const [a, b = a] = r.split("-").map(Number);
      return Array.from({ length: b - a + 1 }, (_, k) => [Number(g), a + k]);
    });
  });
const sha256 = async (bytes) => hex(new Uint8Array(await crypto.subtle.digest("SHA-256", bytes)));
const seed = () => crypto.getRandomValues(new Uint8Array(32));
const json = async (url) => (await fetch(url, { cache: "no-store" })).json();

let config;
let world;

async function fromOrigin(group, index) {
  const res = await fetch(`${config.origin}object/${group}/${index}?c=${encodeURIComponent(id)}`, { cache: "no-store" });
  if (!res.ok) throw new Error(`the origin answered ${res.status}`);
  return new Uint8Array(await res.arrayBuffer());
}

// ───────────────────────────── introductions ─────────────────────────────

const mailbox = new Map();
const arrivals = [];
let polling = false;
function poll() {
  if (polling) return;
  polling = true;
  (async () => {
    while (polling) {
      try {
        for (const m of await json(`${config.signal}recv/${id}`)) {
          const wait = mailbox.get(m.from);
          if (m.description.type === "answer" && wait) wait(m);
          else arrivals.push(m);
        }
      } catch {
        // the mailbox is gone: the run is over
      }
      await new Promise((ok) => setTimeout(ok, 100));
    }
  })();
}
const post = (to, description) => fetch(`${config.signal}send/${to}`, { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ from: id, description }) });

function connection() {
  // §7: with `relay_only`, relayed candidates alone, and no link where there is no relay.
  const relays = config.relays || [];
  if (config.policy.relay_only && relays.length === 0) throw new Error("relay_only and no relay");
  return new RTCPeerConnection({ iceServers: relays, iceTransportPolicy: config.policy.relay_only ? "relay" : "all" });
}
const gathered = (pc) =>
  pc.iceGatheringState === "complete" ? Promise.resolve() : new Promise((ok) => pc.addEventListener("icegatheringstatechange", () => pc.iceGatheringState === "complete" && ok()));
const within = (promise, ms, what) => Promise.race([promise, new Promise((_, no) => setTimeout(() => no(new Error(`${what}: no answer in ${ms} ms`)), ms))]);

/** The asker's side: a data channel to the holder page `peer`, as a connection. */
async function connect(peer) {
  const pc = connection();
  const channel = pc.createDataChannel("wai-peer", { ordered: true, protocol: "wai-peer-1" });
  await pc.setLocalDescription(await pc.createOffer());
  await gathered(pc);
  const answer = new Promise((ok) => mailbox.set(peer, ok));
  await post(peer, pc.localDescription);
  try {
    await pc.setRemoteDescription((await within(answer, 20000, `an answer from ${peer}`)).description);
    if (channel.readyState !== "open") await within(new Promise((ok) => channel.addEventListener("open", ok, { once: true })), 20000, `the channel to ${peer}`);
  } catch (e) {
    // What the connection had to work with, for whoever reads the failure: how many candidates
    // this end offered and of which kinds, never the addresses themselves.
    const kinds = (pc.localDescription?.sdp.match(/^a=candidate:.*$/gm) || []).map((c) => `${c.split(" ")[7]}${/\.local /.test(c) ? " (a local name)" : ""}`);
    pc.close();
    throw new Error(`${e.message}; connection ${pc.iceConnectionState}; candidates offered: ${kinds.join(", ") || "none"}`);
  }
  const link = channelLink(channel);
  const conn = overLink(w, link);
  return { ...conn, close: () => (conn.close(), pc.close()) };
}

/** The holder's side: answer each offer, and serve the channel it opens. */
async function accept(server) {
  for (;;) {
    const m = arrivals.shift();
    if (!m) {
      await new Promise((ok) => setTimeout(ok, 50));
      continue;
    }
    if (m.description.type !== "offer") continue;
    const pc = connection();
    pc.addEventListener("datachannel", (e) => {
      if (e.channel.label !== "wai-peer" || e.channel.protocol !== "wai-peer-1") return e.channel.close();
      hook.links++;
      server.serve(channelLink(e.channel)).finally(() => pc.close());
    });
    await pc.setRemoteDescription(m.description);
    await pc.setLocalDescription(await pc.createAnswer());
    await gathered(pc);
    await post(m.from, pc.localDescription);
  }
}

// ───────────────────────────── the holder ─────────────────────────────

async function holder(setting) {
  // Pages of one origin share its caches: this page's objects are in a cache of its own.
  const inner = await cacheStore(`wai-peer-1-${id}`);
  const altered = objects(q.get("alter")).map(([g, i]) => `${world.groups[g].link}/${i}`);
  const store = {
    ...inner,
    get: async (group, index) => {
      const bytes = await inner.get(group, index);
      if (bytes && altered.includes(`${group}/${index}`)) bytes[bytes.length - 1] ^= 1;
      return bytes;
    },
  };
  const server = new PeerServer(w, { seed: seed(), policy: config.policy, setting, store });
  Object.assign(hook, { mayServe: server.mayServe, links: 0, stats: server.stats, key: server.mayServe ? server.key : null });
  if (!server.mayServe) {
    // §3.3: it signs no record, accepts no link and opens no connection.
    hook.state = "off";
    return status("Serving is off: this page tells no neighbour what it holds.");
  }
  const held = objects(q.get("holds"));
  for (const [g, i] of held) {
    const group = world.groups[g];
    const bytes = await fromOrigin(group.link, i);
    const refusal = w.peerVerifyObject(JSON.stringify(group.receipt), JSON.stringify(group.objects[i].receipt), bytes);
    if (refusal) throw new Error(`the origin's object ${g}:${i} fails its receipt: ${refusal}`);
    await inner.put(group.link, i, bytes);
  }
  const listed = [...held, ...objects(q.get("lists"))];
  const entries = world.groups
    .map((group, g) => ({ group: group.link, leaf_count: group.objects.length + (q.get("count") ? 1 : 0), held: listed.filter(([x]) => x === g).map(([, i]) => i) }))
    .filter((e) => e.held.length);
  const now = Date.now();
  hook.record = hex(server.publish(now - 5000, now + 3600000, entries));
  $("caption").textContent = `Holding ${held.length} objects as ${server.key.slice(0, 12)}…`;
  poll();
  accept(server);
  hook.state = "serving";
  status("Serving.");
}

// ───────────────────────────── the asker ─────────────────────────────

async function asker(setting) {
  const started = Date.now();
  const client = new PeerClient(w, { seed: seed(), policy: config.policy, setting, origin: fromOrigin });
  Object.assign(hook, { mayAsk: client.mayAsk, key: client.key });
  for (const g of world.groups) client.holdGroup(g.receipt, g.objects.map((o) => o.receipt));
  if (client.mayAsk) poll();
  const peers = {};
  hook.failures = [];
  const reported = (peer, open) => () =>
    open().catch((e) => {
      hook.failures.push(`${peer}: ${e?.message || e}`);
      throw e;
    });
  for (const peer of (q.get("peers") || "").split(",").filter(Boolean)) peers[peer] = await client.addNeighbour(reported(peer, () => connect(peer)));
  for (const base of config.http_holders || []) peers[base] = await client.addNeighbour(() => overHttp(w, base));
  hook.peers = peers;
  hook.before = client.records();
  const name = (key) => Object.keys(peers).find((p) => peers[p] === key) || key;
  $("caption").textContent = client.mayAsk ? `Asking ${Object.keys(peers).length} neighbours as ${client.key.slice(0, 12)}…` : "Fetching from peers is off: every object is the origin's.";
  hook.objects = [];
  const wanted = q.get("want") ? objects(q.get("want")) : world.groups.flatMap((g, k) => g.objects.map((_, i) => [k, i]));
  for (const [g, i] of wanted) {
    status(`Object ${g}:${i}…`);
    const r = await client.get(world.groups[g].link, i);
    const row = { group: g, index: i, from: r.from === "origin" ? "origin" : name(r.from), route: r.route, attempts: r.attempts.map((a) => ({ holder: name(a.holder), outcome: a.outcome })), verified: r.verified, length: r.bytes ? r.bytes.length : 0, sha256: r.bytes ? await sha256(r.bytes) : null };
    hook.objects.push(row);
    const tr = $("rows").insertRow();
    tr.insertCell().textContent = `${g}:${i}`;
    tr.insertCell().textContent = row.from;
    tr.insertCell().textContent = row.attempts.map((a) => `${a.holder}: ${a.outcome}`).join("; ") || row.route;
    const cell = tr.insertCell();
    cell.textContent = row.verified ? "yes" : "no";
    cell.className = row.verified ? "ok" : "no";
  }
  status("Signing for what the peers served…");
  hook.acks = (await client.acknowledge(started * 1000, Date.now() * 1000 + 1)).map((a) => ({ ...a, upstream_name: name(a.upstream) }));
  hook.after = client.records();
  hook.asks = Object.fromEntries(Object.entries(client.asks()).map(([k, n]) => [name(k), n]));
  polling = false;
  hook.state = "done";
  status(`Done: ${hook.objects.filter((o) => o.from !== "origin").length} of ${hook.objects.length} objects from peers, every one verified before use.`);
}

async function start() {
  $("start").disabled = true;
  $("ask").disabled = $("serve").disabled = true;
  const setting = { ask: $("ask").checked, serve: $("serve").checked };
  try {
    await (role === "holder" ? holder(setting) : asker(setting));
  } catch (e) {
    hook.error = String(e?.message || e);
    hook.state = "failed";
    status(`Failed: ${hook.error}`);
  }
}

try {
  await init();
  config = await json("config.json");
  world = await json(`${config.origin}world.json`);
  $("start").addEventListener("click", start);
  hook.state = "ready";
} catch (e) {
  hook.error = String(e?.message || e);
  hook.state = "failed";
  status(`Failed: ${hook.error}`);
}