relay: broadcast as a special case of collaboration

.github/workflows/ci.yml

@@ -847,6 +847,22 @@ jobs:
         # in people's files. This diffs only what goes on the wire.
         run: node scripts/test-relay-protocol.ts
 
+      - name: relay broadcast rig
+        # A show is a special case of collaboration: an audience member holds
+        # an owner-signed `audience` invite and a per-show key, and the relay
+        # routes by stream. What breaks silently if this regresses: an audience
+        # socket admitted on the room token (it holds the show key, not the
+        # room key, so the compare must be skipped for the role and only the
+        # role); an audience socket that can send anything (the invite is on the
+        # same chain as a writer's, so the check is by role); an aud frame that
+        # reaches the room stream or the room's op log (show-key ciphertext
+        # persisted forever, replayed to every future collaborator as noise); a
+        # late joiner served the room's op log; show state kept in memory,
+        # which hibernation evicts mid-show; and the presenter-loss grace timer
+        # firing the DO's single alarm as the 30-day room WIPE. Each is a
+        # mutation this rig turns red on.
+        run: node scripts/test-relay-broadcast.ts
+
       - name: relay authorization rig
         # Drives the real Room durable object with fake sockets and storage.
         # What it guards: the relay stamps a fanned-out frame with exactly what

docs/DECISIONS.md

@@ -6390,3 +6390,55 @@ chance to run and it is cheap. Reconciliation for this cycle: 41 commits, 40
 mapped, 1 correctly absent, run by bento-team-slides.
 
 Claude-Session: https://claude.ai/code/session_01Jcfdy8A69nonyATtm8vRy8
+
+## 2026-09-13 — Broadcast is a special case of collaboration: the relay half
+
+**Decision.** A live show is not a second transport. An audience member is a
+collaborator holding a TICKET — an owner-signed invite with role `audience`,
+on the same chain as "Invite to edit" — whose `collab.key` is a per-show SHOW
+KEY rather than the room key. The relay (`server/sync-worker/`) implements
+the show as five rules, each guarded by `scripts/test-relay-broadcast.ts`:
+
+1. **Admission is the invite, not the token.** `?tok=` is a hash of the room
+   key and an audience copy cannot derive it, so the token compare is skipped
+   for the `audience` role and only for it. `w` rooms only; only while live
+   (else close `4002 not-live`); refused on a revoked invite; refused `4003
+   show-full` when the held show has outgrown its cap.
+2. **Audience sockets are receive-only.** Every frame from one is dropped.
+   Control verbs verify by ROLE — the socket's pinned writer key — never by
+   chain membership, because the audience invite is on the same chain.
+3. **Two streams.** A frame tagged `s:'aud'` goes to audience sockets only,
+   is never persisted whatever else it carries, and comes only from writer
+   sockets while live. The room stream never reaches an audience socket —
+   including presence, in either direction. `nav`/`black` are signed with the
+   stream in the text and retained as latest state; `laser` is unsigned and
+   never retained.
+4. **The relay holds the show, in DO storage.** One presenter `audsnap` on
+   `live` and at checkpoints; aud ops since; nav/black. A late joiner is served
+   that and never the op log or the room's persisted snapshot. Storage, not
+   memory: the Hibernation API evicts the object mid-connection, and an
+   in-memory show would vanish silently. Past 256 KB of held ops the presenter
+   is asked to checkpoint once; past twice that, joiners are refused until it
+   does.
+5. **`end` is any writer's; grace is the only unsigned path.** Presenter
+   socket loss starts 60 s; a writer's `live` cancels it; audience activity
+   never extends it. When it fires the show ends and **the room survives**.
+
+**The Durable Object has one alarm, and it used to mean "wipe the room".**
+Every timer now goes through one multiplexer (`schedule`/`rearm`/`alarm`):
+each kind stores its due time, the DO alarm is armed to the earliest, and the
+handler runs whichever are due. Arming the grace timer with a bare `setAlarm`
+would have replaced the idle alarm and, on firing, run the wipe on a live
+room. The rig asserts the room survives a grace expiry and that the idle
+alarm still evaporates it.
+
+**`ready` carries `v` (relay protocol version, 2) and `bc:1`.** `v` is the
+one read-only way to tell a deployed relay from the last one; every other
+discriminator is a write or needs an owner key. Clients feature-detect
+broadcast on `bc`.
+
+Deploy: after #452's relay (deployed 2026-09-13 as `62a12ffa`, from
+`b1b4a67`), as its own deploy. Additive to every shipped client — none sends
+`ivr=audience` or `s:'aud'`, and `v`/`bc` on `ready` are ignored by clients
+that do not read them. The client half (slides) feature-detects and lands
+separately. Design note: private until the client ships, then promoted.

scripts/lib/relay-harness.ts

@@ -0,0 +1,202 @@
+// SPDX-License-Identifier: MIT
+// Copyright (c) 2026 The Bento authors
+// A fake Durable Object world for driving the REAL `Room` from
+// server/sync-worker/src/worker.js in node: transactional storage as a Map,
+// sockets that record what they were sent and how they were closed, the
+// WebSocketPair the upgrade path constructs, and a Response shim (undici's
+// refuses status 101). Shared by test-relay-auth.ts and test-relay-broadcast.ts
+// so the two rigs cannot drift on what "the relay" is.
+//
+// Importing this module installs the globals the worker reaches for. Do it
+// BEFORE importing the worker — `Room` below is imported here for that reason
+// and re-exported.
+
+// The DO returns a 101 for the WebSocket upgrade; undici's Response refuses any
+// status outside 200–599, so the rig supplies a shim before importing.
+class FakeResponse {
+  status: number
+  body: unknown
+  webSocket: unknown
+  headers: Map<string, string>
+  constructor(body: unknown, init: { status?: number; headers?: Record<string, string>; webSocket?: unknown } = {}) {
+    this.body = body
+    this.status = init.status ?? 200
+    this.webSocket = init.webSocket
+    this.headers = new Map(Object.entries(init.headers ?? {}))
+  }
+  async text() {
+    return typeof this.body === 'string' ? this.body : ''
+  }
+}
+;(globalThis as Record<string, unknown>).Response = FakeResponse
+
+export type Sock = {
+  sent: string[]
+  closed: boolean
+  /** the code and reason the relay closed with, if it did */
+  closeCode: number | null
+  closeReason: string | null
+  send(t: string): void
+  close(code?: number, reason?: string): void
+  serializeAttachment(a: unknown): void
+  deserializeAttachment(): Record<string, unknown> | null
+}
+
+export function mkSocket(att: Record<string, unknown> | null = null): Sock {
+  let attachment = att
+  return {
+    sent: [],
+    closed: false,
+    closeCode: null,
+    closeReason: null,
+    send(t) { this.sent.push(t) },
+    close(code, reason) { this.closed = true; this.closeCode = code ?? null; this.closeReason = reason ?? null },
+    serializeAttachment(a) { attachment = JSON.parse(JSON.stringify(a)) },
+    deserializeAttachment() { return attachment },
+  }
+}
+
+let lastPair: { client: Sock; server: Sock } | null = null
+;(globalThis as Record<string, unknown>).WebSocketPair = function () {
+  const client = mkSocket()
+  const server = mkSocket()
+  lastPair = { client, server }
+  return { 0: client, 1: server }
+}
+/** The server half of the most recent upgrade — what the relay talks to. */
+export const lastServer = (): Sock => lastPair!.server
+
+/**
+ * The DO's `state`: transactional storage over a Map, the hibernation socket
+ * list, and a `setAlarm` that RECORDS rather than fires. A rig that wants an
+ * alarm to go off calls `room.alarm()` itself after putting the relevant
+ * `al:<kind>` key in the past — the worker multiplexes every timer through
+ * that one handler, and that dispatch is exactly what is worth testing.
+ */
+export function mkState() {
+  const store = new Map<string, unknown>()
+  const sockets: Sock[] = []
+  const state = {
+    store,
+    sockets,
+    /** what the DO alarm was last armed to, or null once deleted */
+    alarmAt: null as number | null,
+    storage: {
+      async get(k: string) { return store.get(k) },
+      async put(k: string, v: unknown) { store.set(k, v) },
+      async delete(k: string | string[]) {
+        for (const one of Array.isArray(k) ? k : [k]) store.delete(one)
+      },
+      async list({ start, end, prefix }: { start?: string; end?: string; prefix?: string } = {}) {
+        const out = new Map<string, unknown>()
+        for (const [k, v] of [...store.entries()].sort((a, b) => (a[0] < b[0] ? -1 : 1))) {
+          if (prefix && !k.startsWith(prefix)) continue
+          if (start && k < start) continue
+          if (end && k >= end) continue
+          out.set(k, v)
+        }
+        return out
+      },
+      async deleteAll() { store.clear() },
+      async setAlarm(at: number) { state.alarmAt = at },
+      async deleteAlarm() { state.alarmAt = null },
+    },
+    acceptWebSocket(ws: Sock) { sockets.push(ws) },
+    getWebSockets() { return sockets.filter((s) => !s.closed) },
+    setWebSocketAutoResponse() { /* keepalive, not under test */ },
+  }
+  return state
+}
+
+export const req = (url: string, headers: Record<string, string> = {}) => ({
+  url,
+  method: 'GET',
+  headers: { get: (k: string) => headers[k.toLowerCase()] ?? null },
+})
+
+export const { Room } = await import('../../server/sync-worker/src/worker.js')
+
+// --- key material -------------------------------------------------------------
+const EC = { name: 'ECDSA', namedCurve: 'P-256' } as const
+const SIGN = { name: 'ECDSA', hash: 'SHA-256' } as const
+export const b64u = {
+  enc(bytes: Uint8Array) {
+    let s = ''
+    for (const b of bytes) s += String.fromCharCode(b)
+    return btoa(s).replace(/\+/g, '-').replace(/\//g, '_').replace(/=+$/, '')
+  },
+}
+
+export type Keys = {
+  pub: string
+  /** the `w` room name this key commits to */
+  room: string
+  sign(text: string): Promise<string>
+}
+
+export async function mintKeys(): Promise<Keys> {
+  const kp = (await crypto.subtle.generateKey(EC, true, ['sign', 'verify'])) as CryptoKeyPair
+  const raw = new Uint8Array(await crypto.subtle.exportKey('raw', kp.publicKey))
+  const commit = new Uint8Array(await crypto.subtle.digest('SHA-256', raw as BufferSource))
+  return {
+    pub: b64u.enc(raw),
+    room: 'w' + b64u.enc(commit),
+    async sign(text: string) {
+      return b64u.enc(new Uint8Array(await crypto.subtle.sign(SIGN, kp.privateKey, new TextEncoder().encode(text))))
+    },
+  }
+}
+
+/** Build the query string a CHAIN member sends: owner-signed invite of `role`,
+ *  invite-signed delegation of the member key. Exactly what a real client
+ *  (kernel/src/sync/online.ts) puts on the wire. */
+export async function chainQuery(owner: Keys, invite: Keys, member: Keys, role: string, exp = 0): Promise<string> {
+  const ivs = await owner.sign(`inv.${invite.pub}.${role}.${exp}`)
+  const dg = await invite.sign(`dlg.${member.pub}`)
+  return `&w=${member.pub}&o=${owner.pub}&ivp=${invite.pub}&ivr=${role}&ive=${exp}&ivs=${ivs}&dg=${dg}`
+}
+
+export const TOK = 'tok0123456789abcd'
+/** a frame body the relay will never read — it only ever sees ciphertext */
+export const IV = 'aXZpdmluaXY'
+export const CT = 'Y2lwaGVydGV4dA'
+export const parse = (s: string) => JSON.parse(s) as Record<string, unknown>
+
+// --- a tiny check counter, shared so the summary line is uniform -------------
+export const tally = { failures: 0, checks: 0 }
+export function ok(cond: boolean, msg: string) {
+  tally.checks++
+  if (!cond) {
+    tally.failures++
+    console.error(`  ✗ ${msg}`)
+  }
+}
+export function finish(name: string): never {
+  console.log(tally.failures === 0 ? `\nALL PASS (${tally.checks} checks)` : `\n${tally.failures} FAILURES of ${tally.checks} checks`)
+  void name
+  process.exit(tally.failures ? 1 : 0)
+}
+
+/** Open a socket on `room` with the given query (after `?tok=`). Returns the
+ *  server socket and the `ready` frame it was sent, if any. */
+export const connect = async (room: InstanceType<typeof Room>, roomName: string, query: string, tok = TOK) => {
+  const res = await room.fetch(req(`https://relay/d/${roomName}?tok=${tok}${query}`, { upgrade: 'websocket' }))
+  const server = lastServer()
+  const ready = server.sent.map(parse).find((f) => f.ctl === 'ready')
+  return { status: res.status as number, server, ready }
+}
+
+/** Answer the relay's possession challenge and return the ticket, or null. */
+export const prove = async (
+  room: InstanceType<typeof Room>,
+  c: { server: Sock; ready?: Record<string, unknown> },
+  signer: { sign(text: string): Promise<string> },
+  roomName: string,
+) => {
+  const nonce = c.ready?.c
+  if (typeof nonce !== 'string') return null
+  const before = c.server.sent.length
+  await room.onMessage(c.server, JSON.stringify({ ctl: 'prove', g: await signer.sign(`prove.${nonce}.${roomName}`) }))
+  const wtFrame = c.server.sent.slice(before).map(parse).find((f) => f.ctl === 'wt')
+  return (wtFrame?.wt as string | undefined) ?? null
+}