plan: slice 2, holding screen and room, direct connectivity
Eleven tasks from spec revision 8 section 12 item 2: the hub, stream, signal, room and confirm pages, devices.js, room.js, the Playwright gate, the remaining section 13 gates, the copy catalogue.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CzvnuKhLzudvq7jZ7SvVt2
1 file changed,
+2831
−0
docs/superpowers/plans/2026-09-10-meet-slice-2-room.md+2831 −0
CI passed — run details
diff --git a/docs/superpowers/plans/2026-09-10-meet-slice-2-room.md b/docs/superpowers/plans/2026-09-10-meet-slice-2-room.md| new file mode 100644 |
| index 0000000..39a24d6 |
| --- /dev/null |
| +++ b/docs/superpowers/plans/2026-09-10-meet-slice-2-room.md |
| @@ -0,0 +1,2831 @@ |
| +# Meet slice 2: holding screen and room, direct connectivity — Implementation Plan |
| + |
| +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. |
| + |
| +**Goal:** Two people on the same network open a meeting link, set up camera and microphone on the holding screen, and see and hear each other in the room, with a guest waiting to be admitted by a member; every server-side rule of spec §2.4, §2.5, §4 and §5.1 enforced and tested, and a Playwright gate proving a real call. |
| + |
| +**Architecture:** An in-process hub (`internal/meet/hub`) owns rooms, participations, admission, the cap, pair generations, replay rings and the tick, and knows nothing about HTTP. Two handlers wrap it: an SSE stream per peer and a JSON `POST /signal`. Two hand-written scripts do the rest in the browser: `devices.js` for the holding screen and `room.js` for signalling, perfect negotiation, resume and reconciliation. Media never touches the server. `CARLOS_RELAY` is absent in this slice, so calls connect on host candidates only, which is what the localhost browser gate exercises. |
| + |
| +**Tech Stack:** Go 1.26, rastrillo v0.26.0, chi, gorm/SQLite (unchanged schema), `html/template`, vanilla JavaScript (ES2020, no bundler), Playwright 1.62 driven from a Go test through `node`. |
| + |
| +**Spec:** `docs/superpowers/specs/2026-09-07-meet-design.md` revision 8: §2.4, §2.5, §3.3, §3.4, §3.5, §4.1 to §4.6, §5.1, §9, §12 item 2, §13 slice-2 lines. The programme: `docs/superpowers/plans/2026-09-09-meet-slices-2-6-programme.md` §6. |
| + |
| +## Global Constraints |
| + |
| +- Module `amadan.net/oficina/meet`; framework `amadan.net/rastrillo/rastrillo v0.26.0`; `CGO_ENABLED=0`; `make ci` green at the end of every task; `make browser` (Playwright) green at the end of Tasks 8, 9 and 10. |
| +- Media never touches the server: no handler accepts a media byte; the hub reads a message's `type`, `from`, `to`, `seq`, `gen` and byte length and never its payload. |
| +- No payload, chat text, SDP, candidate, display name, title or link token is ever a log argument (spec §4.4); the captured-log test proves it. |
| +- Authorization per endpoint is exactly spec §2.5's table: entry (`join`, fresh stream) checks standing; everything after admission checks the participation; `admit` and tick promotion re-derive member kind and check the target's entry authority under the hub lock. |
| +- Recipients per message type are exactly spec §4.3's table; a waiting guest receives `hello`, `state`, then `roster` on admission, or `lost`, or `ended`, and nothing else. |
| +- Limits (spec §4.4): 64 KB body, 30 messages per second with burst 60 per peer (429 with `Retry-After`), sixteen participations per room, eight waiting, one peer per session per room, four streams per person, 64-event subscriber buffer, unjoined and resume windows of thirty seconds, replay ring of thirty-two per participation, tombstones ten minutes. |
| +- The cap is four admitted peers, enforced at every transition into admitted under the hub lock, never on resume. |
| +- `seq` is monotonic per (sender, target) for the life of a participation; `gen` is per unordered pair, held by the hub, starting at one. |
| +- Destructive actions (remove, takeover) are confirm pages with `ui/confirm-form` and POSTs under `csrf.Protect`; a `GET` never ends a participation. |
| +- Every page still works with scripts off; the one exception, browser media participation, is written into `AGENTS.md` in Task 6. |
| +- Copy review is waived for this round (suite #5); new strings are still catalogued in `docs/superpowers/copy/2026-09-10-slice-2-strings.md` in Task 11. |
| +- Commits end with `Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>` and `Claude-Session: https://claude.ai/code/session_01CzvnuKhLzudvq7jZ7SvVt2`. |
| +- Run from the slice worktree with `GOCACHE=/home/paulca/.cache/tmp/claude-1001/gocache`; one git command per shell call; never `git add .claude/`. |
| + |
| +--- |
| + |
| +## File structure |
| + |
| +| File | Responsibility | |
| +|---|---| |
| +| `internal/meet/hub/hub.go` | `Hub`, `Room`, `Participation`, peer ids, boot id, locks, limits, tombstones, `Open`/`Resume`/`Close`. | |
| +| `internal/meet/hub/messages.go` | `Message` envelope, `Type` constants, `Roster` and its builders, recipient rules. | |
| +| `internal/meet/hub/admission.go` | `Join`, `Admit`, `Leave`, `Remove`, `Takeover`, `End`, the cap, waiting queue, kind transitions, `Tick`. | |
| +| `internal/meet/hub/relay.go` | `Relay` (targeted, `seq`/`gen` checked), `Reset`, pair generations, replay rings, `gap`. | |
| +| `internal/meet/hub/subscriber.go` | Per-participation channel with the 64-event buffer and the `lost {reason: "buffer"}` outcome. | |
| +| `internal/meet/hub/*_test.go` | The §13 hub gates, no HTTP. | |
| +| `internal/meet/stream.go` | `GET /m/{sqid}/stream`: SSE, `hello {peer, boot}`, heartbeat, TTL, resume answers 404/409/410/503. | |
| +| `internal/meet/signal.go` | `POST /m/{sqid}/signal`: body limit, rate limit, type dispatch, refusal logging. | |
| +| `internal/meet/participation.go` | `Standing`-or-participation middleware for `/m/{sqid}` pages; `standingKind` adapter the hub calls on tick. | |
| +| `internal/meet/handlers_room.go` | Room page, removal and takeover confirm pages and POSTs. | |
| +| `internal/meet/templates/room.html`, `confirm.html` (reused), `join.html` (extended) | Server-rendered halves. | |
| +| `internal/meet/static/devices.js` | Holding screen: enumerate, preview, pickers, mute state, local storage, join button. | |
| +| `internal/meet/static/room.js` | SSE client, outbound queue, perfect negotiation, roster reconciliation, resume, terminal states, admission UI. | |
| +| `internal/meet/static/meet.css` | The grid, tiles and controls; vendored beside rastrillo's sheets. | |
| +| `test/browser/call.mjs`, `package.json`, `Makefile` `browser` target, `internal/meettest/browser_test.go` (`//go:build browser`) | The Playwright gate. | |
| +| `AGENTS.md`, `docs/superpowers/copy/2026-09-10-slice-2-strings.md` | The scriptless exception and the copy catalogue. | |
| + |
| +--- |
| + |
| +### Task 1: Hub core: rooms, participations, peer ids, limits |
| + |
| +**Files:** |
| +- Create: `internal/meet/hub/hub.go`, `internal/meet/hub/subscriber.go`, `internal/meet/hub/messages.go`, `internal/meet/hub/hub_test.go` |
| + |
| +**Interfaces:** |
| +- Produces: |
| + ```go |
| + package hub |
| + type PeerID string // 16 random bytes, hex, issued by Open |
| + type Kind int // KindGuest, KindMember |
| + type Reason string // "buffer","replaced","standing","expired","removed","ended" |
| + type Config struct{ Now func() time.Time; Logger *slog.Logger; StandingFor func(ctx context.Context, room string, personID int64) (Kind, bool, error) } |
| + func New(cfg Config) *Hub |
| + func (h *Hub) Boot() string |
| + type Participation struct{ Peer PeerID; Room string; Session string; PersonID int64; Name string; Kind Kind; Admitted bool; JoinedAt time.Time /* unexported: sub, resumeUntil, seqOut map[PeerID]uint64, ring, lastSeen map[PeerID]uint64 */ } |
| + type Message struct{ Type string `json:"type"`; From PeerID `json:"from,omitempty"`; Name string `json:"name,omitempty"`; To PeerID `json:"to,omitempty"`; Seq uint64 `json:"seq,omitempty"`; Gen uint64 `json:"gen,omitempty"`; Payload json.RawMessage `json:"payload,omitempty"`; Reason Reason `json:"reason,omitempty"`; Roster *Roster `json:"roster,omitempty"`; Boot string `json:"boot,omitempty"`; Peer PeerID `json:"peer,omitempty"`; Waiting bool `json:"waiting,omitempty"` } |
| + type Roster struct{ Peers []RosterPeer `json:"peers"`; Waiting []RosterPeer `json:"waiting,omitempty"` } |
| + type RosterPeer struct{ Peer PeerID `json:"peer"`; Name string `json:"name"`; Kind string `json:"kind"`; Gen uint64 `json:"gen,omitempty"` } |
| + type Sub struct{ /* C() <-chan Message; Lost() bool; Close() */ } |
| + var ErrRoomFull, ErrSessionHasPeer, ErrPersonStreams, ErrUnknownPeer, ErrNotYours error |
| + type OpenResult struct{ Peer PeerID; Sub *Sub; Existing PeerID /* set when ErrSessionHasPeer */ } |
| + func (h *Hub) Open(room, session string, personID int64, name string, kind Kind) (OpenResult, error) // fresh stream: creates an unjoined participation |
| + type ResumeResult struct{ Sub *Sub; Admitted bool; Replay []Message; Gap []PeerID; Roster *Roster } |
| + type ResumeError struct{ Status int; Reason Reason; Boot string } // 404 with Boot, 410 with Reason |
| + func (h *Hub) Resume(room, session string, peer PeerID, last map[PeerID]uint64) (ResumeResult, error) |
| + func (h *Hub) Detach(room string, peer PeerID) // stream closed: start the resume window |
| + func (h *Hub) Find(room, session string, peer PeerID) (*Participation, bool) // bound to session |
| + func (h *Hub) LiveForSession(room, session string) (PeerID, bool) // attached or in resume window |
| + ``` |
| +- The 64-event buffer and `lost {reason: "buffer"}`; unjoined participations dropped after thirty seconds; resume window thirty seconds; sixteen participations per room (503 before allocation); one peer per session per room (409 with the existing id); four streams per person; tombstones for ten minutes keyed by peer id with their reason. |
| + |
| +- [ ] **Step 1: Write the failing tests** |
| + |
| +```go |
| +package hub |
| + |
| +import ( |
| + "context" |
| + "io" |
| + "log/slog" |
| + "testing" |
| + "time" |
| +) |
| + |
| +func testHub(now *time.Time) *Hub { |
| + return New(Config{ |
| + Now: func() time.Time { return *now }, |
| + Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), |
| + StandingFor: func(ctx context.Context, room string, personID int64) (Kind, bool, error) { |
| + if personID < 100 { |
| + return KindMember, true, nil |
| + } |
| + return KindGuest, true, nil |
| + }, |
| + }) |
| +} |
| + |
| +func TestOpenIssuesADistinctPeerBoundToTheSession(t *testing.T) { |
| + now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC) |
| + h := testHub(&now) |
| + a, err := h.Open("r1", "sessA", 1, "Ann", KindMember) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + b, _ := h.Open("r1", "sessB", 2, "Bob", KindMember) |
| + if a.Peer == b.Peer || len(a.Peer) != 32 { |
| + t.Fatalf("peer ids %q %q", a.Peer, b.Peer) |
| + } |
| + if _, ok := h.Find("r1", "sessB", a.Peer); ok { |
| + t.Fatal("a peer was found through another session") |
| + } |
| + if _, ok := h.Find("r2", "sessA", a.Peer); ok { |
| + t.Fatal("a peer was found through another room") |
| + } |
| + if p, ok := h.Find("r1", "sessA", a.Peer); !ok || p.Admitted { |
| + t.Fatal("fresh participation should exist and be unjoined") |
| + } |
| +} |
| + |
| +func TestOneLivePeerPerSessionPerRoom(t *testing.T) { |
| + now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC) |
| + h := testHub(&now) |
| + a, _ := h.Open("r1", "sessA", 1, "Ann", KindMember) |
| + _, err := h.Open("r1", "sessA", 1, "Ann", KindMember) |
| + if err != ErrSessionHasPeer { |
| + t.Fatalf("second stream: %v", err) |
| + } |
| + h.Detach("r1", a.Peer) |
| + if _, err := h.Open("r1", "sessA", 1, "Ann", KindMember); err != ErrSessionHasPeer { |
| + t.Fatal("a detached peer inside its resume window should still block a fresh stream") |
| + } |
| + now = now.Add(31 * time.Second) |
| + h.Tick(context.Background()) |
| + if _, err := h.Open("r1", "sessA", 1, "Ann", KindMember); err != nil { |
| + t.Fatalf("after the window: %v", err) |
| + } |
| +} |
| + |
| +func TestSixteenParticipationsPerRoomThenRefuse(t *testing.T) { |
| + now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC) |
| + h := testHub(&now) |
| + for i := 0; i < 16; i++ { |
| + if _, err := h.Open("r1", "s"+string(rune('a'+i)), int64(100+i), "g", KindGuest); err != nil { |
| + t.Fatalf("open %d: %v", i, err) |
| + } |
| + } |
| + if _, err := h.Open("r1", "sZ", 200, "g", KindGuest); err != ErrRoomFull { |
| + t.Fatalf("seventeenth: %v", err) |
| + } |
| +} |
| + |
| +func TestUnjoinedParticipationDropsAfterThirtySeconds(t *testing.T) { |
| + now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC) |
| + h := testHub(&now) |
| + a, _ := h.Open("r1", "sessA", 1, "Ann", KindMember) |
| + now = now.Add(31 * time.Second) |
| + h.Tick(context.Background()) |
| + if _, ok := h.Find("r1", "sessA", a.Peer); ok { |
| + t.Fatal("unjoined participation survived the window") |
| + } |
| + if !a.Sub.Lost() { |
| + t.Fatal("its subscriber was not closed") |
| + } |
| +} |
| + |
| +func TestFourStreamsPerPerson(t *testing.T) { |
| + now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC) |
| + h := testHub(&now) |
| + for i := 0; i < 4; i++ { |
| + if _, err := h.Open("r"+string(rune('1'+i)), "s"+string(rune('1'+i)), 1, "Ann", KindMember); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if _, err := h.Open("r9", "s9", 1, "Ann", KindMember); err != ErrPersonStreams { |
| + t.Fatalf("fifth stream: %v", err) |
| + } |
| +} |
| + |
| +func TestSlowSubscriberIsDeclaredLost(t *testing.T) { |
| + now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC) |
| + h := testHub(&now) |
| + a, _ := h.Open("r1", "sessA", 1, "Ann", KindMember) |
| + for i := 0; i < 70; i++ { |
| + h.deliver("r1", a.Peer, Message{Type: "peer"}) |
| + } |
| + if !a.Sub.Lost() { |
| + t.Fatal("65th undelivered event did not declare the subscriber lost") |
| + } |
| + var last Message |
| + for m := range a.Sub.C() { |
| + last = m |
| + } |
| + if last.Type != "lost" || last.Reason != "buffer" { |
| + t.Fatalf("last event %+v, want lost/buffer", last) |
| + } |
| +} |
| +``` |
| + |
| +- [ ] **Step 2: Run to verify they fail** |
| + |
| +Run: `GOCACHE=/home/paulca/.cache/tmp/claude-1001/gocache go test ./internal/meet/hub/ 2>&1 | head -5` |
| +Expected: build failure, package does not exist. |
| + |
| +- [ ] **Step 3: Implement the core** |
| + |
| +`subscriber.go`: |
| + |
| +```go |
| +package hub |
| + |
| +import "sync/atomic" |
| + |
| +const subBuffer = 64 |
| + |
| +// Sub is one stream's outbound channel. A subscriber that falls 64 |
| +// events behind is declared lost and told so (spec §4.4), as in Sheets. |
| +type Sub struct { |
| + ch chan Message |
| + lost atomic.Bool |
| + done chan struct{} |
| +} |
| + |
| +func newSub() *Sub { return &Sub{ch: make(chan Message, subBuffer), done: make(chan struct{})} } |
| + |
| +func (s *Sub) C() <-chan Message { return s.ch } |
| +func (s *Sub) Lost() bool { return s.lost.Load() } |
| + |
| +// send is non-blocking; a full buffer declares the subscriber lost, |
| +// drains room for the lost event, and closes. |
| +func (s *Sub) send(m Message) { |
| + select { |
| + case <-s.done: |
| + return |
| + default: |
| + } |
| + select { |
| + case s.ch <- m: |
| + default: |
| + s.terminate(Message{Type: "lost", Reason: ReasonBuffer}) |
| + } |
| +} |
| + |
| +func (s *Sub) terminate(final Message) { |
| + if s.lost.Swap(true) { |
| + return |
| + } |
| + close(s.done) |
| + // Make room for the final event, then close so the range ends. |
| + select { |
| + case <-s.ch: |
| + default: |
| + } |
| + s.ch <- final |
| + close(s.ch) |
| +} |
| + |
| +func (s *Sub) Close() { s.terminate(Message{Type: "lost", Reason: ReasonBuffer}) } |
| +``` |
| + |
| +`messages.go`: |
| + |
| +```go |
| +package hub |
| + |
| +import "encoding/json" |
| + |
| +type PeerID string |
| +type Kind int |
| +type Reason string |
| + |
| +const ( |
| + KindGuest Kind = iota |
| + KindMember |
| +) |
| + |
| +func (k Kind) String() string { |
| + if k == KindMember { |
| + return "member" |
| + } |
| + return "guest" |
| +} |
| + |
| +const ( |
| + ReasonBuffer Reason = "buffer" |
| + ReasonReplaced Reason = "replaced" |
| + ReasonStanding Reason = "standing" |
| + ReasonExpired Reason = "expired" |
| + ReasonRemoved Reason = "removed" |
| + ReasonEnded Reason = "ended" |
| +) |
| + |
| +// Upstream types a browser may send. |
| +const ( |
| + TypeJoin = "join" |
| + TypeAdmit = "admit" |
| + TypeLeave = "leave" |
| + TypeOffer = "offer" |
| + TypeAnswer = "answer" |
| + TypeCandidate = "candidate" |
| + TypeReset = "reset" |
| + TypeChat = "chat" // slice 4 |
| + TypeShare = "share" // slice 4 |
| +) |
| + |
| +// Downstream types the server emits. |
| +const ( |
| + TypeHello = "hello" |
| + TypeRoster = "roster" |
| + TypePeer = "peer" |
| + TypeGone = "gone" |
| + TypeKnock = "knock" |
| + TypeKnockGone = "knock-gone" |
| + TypeFull = "full" |
| + TypeEnded = "ended" |
| + TypeLost = "lost" |
| + TypePair = "pair" |
| + TypeGap = "gap" |
| + TypeState = "state" |
| +) |
| + |
| +// Message is the one envelope, built by the server. Payload is opaque. |
| +type Message struct { |
| + Type string `json:"type"` |
| + From PeerID `json:"from,omitempty"` |
| + Name string `json:"name,omitempty"` |
| + To PeerID `json:"to,omitempty"` |
| + Seq uint64 `json:"seq,omitempty"` |
| + Gen uint64 `json:"gen,omitempty"` |
| + Payload json.RawMessage `json:"payload,omitempty"` |
| + Reason Reason `json:"reason,omitempty"` |
| + Roster *Roster `json:"roster,omitempty"` |
| + Boot string `json:"boot,omitempty"` |
| + Peer PeerID `json:"peer,omitempty"` |
| + Waiting bool `json:"waiting,omitempty"` |
| +} |
| + |
| +type Roster struct { |
| + Peers []RosterPeer `json:"peers"` |
| + Waiting []RosterPeer `json:"waiting,omitempty"` |
| +} |
| + |
| +type RosterPeer struct { |
| + Peer PeerID `json:"peer"` |
| + Name string `json:"name"` |
| + Kind string `json:"kind"` |
| + Gen uint64 `json:"gen,omitempty"` |
| +} |
| +``` |
| + |
| +`hub.go` (the data model and the four lifecycle entry points; admission and relay come in Tasks 2 and 3 but their fields are declared here): |
| + |
| +```go |
| +package hub |
| + |
| +import ( |
| + "context" |
| + "crypto/rand" |
| + "encoding/hex" |
| + "errors" |
| + "log/slog" |
| + "sync" |
| + "time" |
| +) |
| + |
| +const ( |
| + MaxParticipations = 16 |
| + MaxWaiting = 8 |
| + MaxAdmitted = 4 |
| + MaxStreamsPerPerson = 4 |
| + UnjoinedWindow = 30 * time.Second |
| + ResumeWindow = 30 * time.Second |
| + TombstoneLife = 10 * time.Minute |
| + RingSize = 32 |
| +) |
| + |
| +var ( |
| + ErrRoomFull = errors.New("hub: room holds sixteen participations") |
| + ErrSessionHasPeer = errors.New("hub: this session already has a peer in the room") |
| + ErrPersonStreams = errors.New("hub: this person has four streams open") |
| + ErrUnknownPeer = errors.New("hub: no such peer") |
| + ErrNotYours = errors.New("hub: peer is not bound to this session") |
| + ErrNotAdmitted = errors.New("hub: peer is not admitted") |
| + ErrNotMember = errors.New("hub: peer is not a member") |
| + ErrNotWaiting = errors.New("hub: target is not waiting") |
| + ErrStaleGen = errors.New("hub: stale generation") |
| + ErrTooManyWaiting = errors.New("hub: eight already waiting") |
| +) |
| + |
| +type Config struct { |
| + Now func() time.Time |
| + Logger *slog.Logger |
| + // StandingFor re-derives a person's kind and whether they still hold |
| + // entry authority for the room. ok=false means standing none. |
| + StandingFor func(ctx context.Context, room string, personID int64) (kind Kind, ok bool, err error) |
| +} |
| + |
| +type Participation struct { |
| + Peer PeerID |
| + Room string |
| + Session string |
| + PersonID int64 |
| + Name string |
| + Kind Kind |
| + Admitted bool |
| + Joined bool // has posted join (waiting or admitted) |
| + JoinedAt time.Time |
| + OpenedAt time.Time |
| + |
| + sub *Sub |
| + resumeUntil time.Time // zero while a stream is attached |
| + seqOut map[PeerID]uint64 // next seq per target (unused by the hub; kept by browsers) |
| + lastSeen map[PeerID]uint64 |
| + ring []Message // targeted messages delivered to this peer, newest last, cap RingSize |
| +} |
| + |
| +type pairKey struct{ a, b PeerID } |
| + |
| +func keyFor(x, y PeerID) pairKey { |
| + if x < y { |
| + return pairKey{x, y} |
| + } |
| + return pairKey{y, x} |
| +} |
| + |
| +type room struct { |
| + sqid string |
| + peers map[PeerID]*Participation |
| + gens map[pairKey]uint64 |
| + ended bool |
| +} |
| + |
| +type tombstone struct { |
| + reason Reason |
| + at time.Time |
| +} |
| + |
| +type Hub struct { |
| + cfg Config |
| + boot string |
| + |
| + mu sync.Mutex |
| + rooms map[string]*room |
| + bySession map[string]map[string]PeerID // room -> session -> peer |
| + byPerson map[int64]int // open streams per person |
| + tombstones map[PeerID]tombstone |
| +} |
| + |
| +func New(cfg Config) *Hub { |
| + if cfg.Now == nil { |
| + cfg.Now = time.Now |
| + } |
| + return &Hub{ |
| + cfg: cfg, boot: randomID(), |
| + rooms: map[string]*room{}, bySession: map[string]map[string]PeerID{}, |
| + byPerson: map[int64]int{}, tombstones: map[PeerID]tombstone{}, |
| + } |
| +} |
| + |
| +func randomID() string { |
| + b := make([]byte, 16) |
| + if _, err := rand.Read(b); err != nil { |
| + panic(err) |
| + } |
| + return hex.EncodeToString(b) |
| +} |
| + |
| +func (h *Hub) Boot() string { return h.boot } |
| + |
| +func (h *Hub) roomLocked(sqid string) *room { |
| + r, ok := h.rooms[sqid] |
| + if !ok { |
| + r = &room{sqid: sqid, peers: map[PeerID]*Participation{}, gens: map[pairKey]uint64{}} |
| + h.rooms[sqid] = r |
| + } |
| + return r |
| +} |
| + |
| +type OpenResult struct { |
| + Peer PeerID |
| + Sub *Sub |
| + Existing PeerID |
| +} |
| + |
| +// Open is a fresh stream: a new unjoined participation, subject to the |
| +// room, session and person limits, all checked before allocation. |
| +func (h *Hub) Open(sqid, session string, personID int64, name string, kind Kind) (OpenResult, error) { |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + r := h.roomLocked(sqid) |
| + if r.ended { |
| + return OpenResult{}, ErrUnknownPeer |
| + } |
| + if existing, ok := h.bySession[sqid][session]; ok { |
| + return OpenResult{Existing: existing}, ErrSessionHasPeer |
| + } |
| + if len(r.peers) >= MaxParticipations { |
| + return OpenResult{}, ErrRoomFull |
| + } |
| + if h.byPerson[personID] >= MaxStreamsPerPerson { |
| + return OpenResult{}, ErrPersonStreams |
| + } |
| + p := &Participation{ |
| + Peer: PeerID(randomID()), Room: sqid, Session: session, PersonID: personID, Name: name, Kind: kind, |
| + OpenedAt: h.cfg.Now(), sub: newSub(), seqOut: map[PeerID]uint64{}, lastSeen: map[PeerID]uint64{}, |
| + } |
| + r.peers[p.Peer] = p |
| + if h.bySession[sqid] == nil { |
| + h.bySession[sqid] = map[string]PeerID{} |
| + } |
| + h.bySession[sqid][session] = p.Peer |
| + h.byPerson[personID]++ |
| + return OpenResult{Peer: p.Peer, Sub: p.sub}, nil |
| +} |
| + |
| +// Find answers a peer only through its own room and session. |
| +func (h *Hub) Find(sqid, session string, peer PeerID) (*Participation, bool) { |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + p, ok := h.findLocked(sqid, session, peer) |
| + return p, ok |
| +} |
| + |
| +func (h *Hub) findLocked(sqid, session string, peer PeerID) (*Participation, bool) { |
| + r, ok := h.rooms[sqid] |
| + if !ok { |
| + return nil, false |
| + } |
| + p, ok := r.peers[peer] |
| + if !ok || p.Session != session { |
| + return nil, false |
| + } |
| + return p, true |
| +} |
| + |
| +func (h *Hub) LiveForSession(sqid, session string) (PeerID, bool) { |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + id, ok := h.bySession[sqid][session] |
| + return id, ok |
| +} |
| + |
| +// Detach starts the resume window for a participation whose stream |
| +// closed. Its subscriber is closed; a resume attaches a new one. |
| +func (h *Hub) Detach(sqid string, peer PeerID) { |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + r, ok := h.rooms[sqid] |
| + if !ok { |
| + return |
| + } |
| + p, ok := r.peers[peer] |
| + if !ok { |
| + return |
| + } |
| + if p.sub != nil { |
| + p.sub.terminate(Message{Type: TypeLost, Reason: ReasonBuffer}) |
| + p.sub = nil |
| + } |
| + p.resumeUntil = h.cfg.Now().Add(ResumeWindow) |
| +} |
| + |
| +// endLocked removes a participation for a terminal reason, closes its |
| +// stream with lost{reason}, leaves a tombstone, and returns whether it |
| +// was admitted or waiting so the caller can announce the departure. |
| +func (h *Hub) endLocked(r *room, p *Participation, reason Reason) (wasAdmitted, wasWaiting bool) { |
| + wasAdmitted, wasWaiting = p.Admitted, p.Joined && !p.Admitted |
| + if p.sub != nil { |
| + p.sub.terminate(Message{Type: TypeLost, Reason: reason}) |
| + } |
| + delete(r.peers, p.Peer) |
| + if h.bySession[r.sqid][p.Session] == p.Peer { |
| + delete(h.bySession[r.sqid], p.Session) |
| + } |
| + if h.byPerson[p.PersonID] > 0 { |
| + h.byPerson[p.PersonID]-- |
| + } |
| + h.tombstones[p.Peer] = tombstone{reason: reason, at: h.cfg.Now()} |
| + for k := range r.gens { |
| + if k.a == p.Peer || k.b == p.Peer { |
| + delete(r.gens, k) |
| + } |
| + } |
| + return |
| +} |
| + |
| +// deliver is the one place an event reaches a subscriber; targeted |
| +// messages are also kept in the recipient's replay ring (Task 3). |
| +func (h *Hub) deliver(sqid string, to PeerID, m Message) { |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + h.deliverLocked(sqid, to, m) |
| +} |
| + |
| +func (h *Hub) deliverLocked(sqid string, to PeerID, m Message) { |
| + r, ok := h.rooms[sqid] |
| + if !ok { |
| + return |
| + } |
| + p, ok := r.peers[to] |
| + if !ok { |
| + return |
| + } |
| + if p.sub != nil { |
| + p.sub.send(m) |
| + } |
| +} |
| +``` |
| + |
| +`Tick` (in `admission.go`, Task 2) is referenced by the tests here; for this task add a minimal `Tick` in `hub.go` that drops unjoined participations past `UnjoinedWindow` and resume-held ones past `resumeUntil` (as a leave, no tombstone), and expires tombstones. Task 2 extends it with kind re-derivation. |
| + |
| +```go |
| +// Tick runs the fifteen-second housekeeping: unjoined and lapsed |
| +// participations are dropped, tombstones expire. Task 2 adds kind |
| +// re-derivation and promotion. |
| +func (h *Hub) Tick(ctx context.Context) { |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + now := h.cfg.Now() |
| + for _, r := range h.rooms { |
| + for _, p := range r.peers { |
| + switch { |
| + case !p.Joined && p.sub != nil && now.Sub(p.OpenedAt) > UnjoinedWindow: |
| + h.endLocked(r, p, ReasonBuffer) // dropped quietly; the browser sees a closed stream |
| + delete(h.tombstones, p.Peer) |
| + case p.sub == nil && !p.resumeUntil.IsZero() && now.After(p.resumeUntil): |
| + wasAdmitted, wasWaiting := h.endLocked(r, p, ReasonBuffer) |
| + delete(h.tombstones, p.Peer) |
| + h.announceDepartureLocked(r, p, wasAdmitted, wasWaiting) |
| + } |
| + } |
| + } |
| + for id, t := range h.tombstones { |
| + if now.Sub(t.at) > TombstoneLife { |
| + delete(h.tombstones, id) |
| + } |
| + } |
| +} |
| +``` |
| + |
| +with `announceDepartureLocked` stubbed in this task (`gone` to admitted peers, `knock-gone` to admitted members; Task 2 fills the recipient helpers): |
| + |
| +```go |
| +func (h *Hub) announceDepartureLocked(r *room, p *Participation, wasAdmitted, wasWaiting bool) { |
| + switch { |
| + case wasAdmitted: |
| + h.broadcastLocked(r, Message{Type: TypeGone, From: p.Peer}, admittedPeers) |
| + case wasWaiting: |
| + h.broadcastLocked(r, Message{Type: TypeKnockGone, From: p.Peer}, admittedMembers) |
| + } |
| +} |
| + |
| +type audience func(*Participation) bool |
| + |
| +func admittedPeers(p *Participation) bool { return p.Admitted } |
| +func admittedMembers(p *Participation) bool { return p.Admitted && p.Kind == KindMember } |
| + |
| +func (h *Hub) broadcastLocked(r *room, m Message, who audience) { |
| + for _, q := range r.peers { |
| + if who(q) && q.sub != nil { |
| + q.sub.send(m) |
| + } |
| + } |
| +} |
| +``` |
| + |
| +- [ ] **Step 4: Run the tests** |
| + |
| +Run: `GOCACHE=/home/paulca/.cache/tmp/claude-1001/gocache go test ./internal/meet/hub/ -v 2>&1 | tail -12` |
| +Expected: six tests PASS. |
| + |
| +- [ ] **Step 5: Gate and commit** |
| + |
| +Run: `GOCACHE=/home/paulca/.cache/tmp/claude-1001/gocache make ci 2>&1 | tail -3` |
| + |
| +```bash |
| +git add internal/meet/hub |
| +git -c user.name="Paul Campbell" -c user.email="paul@rslw.com" commit -m "hub: rooms, participations, peer ids, limits, tombstones |
| + |
| +Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> |
| +Claude-Session: https://claude.ai/code/session_01CzvnuKhLzudvq7jZ7SvVt2" |
| +``` |
| + |
| +--- |
| + |
| +### Task 2: Admission: join, admit, leave, remove, takeover, end, the cap, the tick |
| + |
| +**Files:** |
| +- Create: `internal/meet/hub/admission.go`, `internal/meet/hub/admission_test.go` |
| +- Modify: `internal/meet/hub/hub.go` (`Tick` extended) |
| + |
| +**Interfaces:** |
| +- Produces: |
| + ```go |
| + func (h *Hub) Join(ctx, sqid, session string, peer PeerID) error // standing re-checked via cfg.StandingFor; member: admit under cap or ErrRoomFull→`full` to the joiner; guest: waiting or ErrTooManyWaiting |
| + func (h *Hub) Admit(ctx, sqid, session string, actor, target PeerID) error // actor admitted member (kind re-derived now); target waiting with entry authority now; cap under lock |
| + func (h *Hub) Leave(sqid, session string, peer PeerID) error |
| + func (h *Hub) Remove(ctx, sqid, session string, actor, target PeerID) error // §3.5, not a signalling message |
| + func (h *Hub) Takeover(sqid, session string) (old PeerID, err error) // §4.2: ends the session's live participation with the state-dependent departure |
| + func (h *Hub) End(sqid string) // meeting ended: `ended` to all, room dropped |
| + func (h *Hub) rosterFor(r *room, viewer *Participation) *Roster // waiting queue only for members; per-pair gen for the viewer |
| + ``` |
| +- The cap check `admittedCountLocked(r) >= MaxAdmitted` at both transitions; `full` to the actor; `roster` to the newly admitted; `peer` to the others; `knock` to admitted members on a guest join; `knock-gone` on any waiting departure; `Tick` re-derives kind for every joined participation through `cfg.StandingFor`, ends members who lost standing with `ReasonStanding`, ends waiting guests whose authority lapsed with `ReasonExpired`, refreshes `roster` to admitted peers on a kind change, and attempts promotion of a waiting peer that became a member. |
| + |
| +- [ ] **Step 1: Write the failing tests** |
| + |
| +```go |
| +package hub |
| + |
| +import ( |
| + "context" |
| + "testing" |
| + "time" |
| +) |
| + |
| +// drain returns every event currently buffered on a sub, non-blocking. |
| +func drain(s *Sub) []Message { |
| + var out []Message |
| + for { |
| + select { |
| + case m, ok := <-s.C(): |
| + if !ok { |
| + return out |
| + } |
| + out = append(out, m) |
| + default: |
| + return out |
| + } |
| + } |
| +} |
| + |
| +func types(ms []Message) []string { |
| + out := make([]string, len(ms)) |
| + for i, m := range ms { |
| + out[i] = m.Type |
| + } |
| + return out |
| +} |
| + |
| +func TestMemberJoinsAndGuestWaits(t *testing.T) { |
| + now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC) |
| + h := testHub(&now) |
| + ctx := context.Background() |
| + m, _ := h.Open("r1", "sm", 1, "Mia", KindMember) |
| + if err := h.Join(ctx, "r1", "sm", m.Peer); err != nil { |
| + t.Fatal(err) |
| + } |
| + ev := drain(m.Sub) |
| + if len(ev) != 1 || ev[0].Type != TypeRoster || len(ev[0].Roster.Peers) != 1 { |
| + t.Fatalf("member join events %v", types(ev)) |
| + } |
| + g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest) |
| + if err := h.Join(ctx, "r1", "sg", g.Peer); err != nil { |
| + t.Fatal(err) |
| + } |
| + if ev := drain(g.Sub); len(ev) != 0 { |
| + t.Fatalf("waiting guest received %v", types(ev)) |
| + } |
| + ev = drain(m.Sub) |
| + if len(ev) != 1 || ev[0].Type != TypeKnock || ev[0].From != g.Peer || ev[0].Name != "Gus" { |
| + t.Fatalf("member did not get the knock: %v", types(ev)) |
| + } |
| +} |
| + |
| +func TestAdmitSendsRosterToGuestPeerToOthersAndKnockGone(t *testing.T) { |
| + now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC) |
| + h := testHub(&now) |
| + ctx := context.Background() |
| + m, _ := h.Open("r1", "sm", 1, "Mia", KindMember) |
| + h.Join(ctx, "r1", "sm", m.Peer) |
| + g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest) |
| + h.Join(ctx, "r1", "sg", g.Peer) |
| + drain(m.Sub) |
| + if err := h.Admit(ctx, "r1", "sm", m.Peer, g.Peer); err != nil { |
| + t.Fatal(err) |
| + } |
| + gev := drain(g.Sub) |
| + if len(gev) != 1 || gev[0].Type != TypeRoster || len(gev[0].Roster.Peers) != 2 || gev[0].Roster.Waiting != nil { |
| + t.Fatalf("guest got %v (waiting queue must not reach a guest)", types(gev)) |
| + } |
| + mev := types(drain(m.Sub)) |
| + if len(mev) != 2 || mev[0] != TypePeer || mev[1] != TypeKnockGone { |
| + t.Fatalf("member got %v", mev) |
| + } |
| +} |
| + |
| +func TestOnlyAnAdmittedMemberMayAdmit(t *testing.T) { |
| + now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC) |
| + h := testHub(&now) |
| + ctx := context.Background() |
| + m, _ := h.Open("r1", "sm", 1, "Mia", KindMember) |
| + h.Join(ctx, "r1", "sm", m.Peer) |
| + g1, _ := h.Open("r1", "s1", 100, "G1", KindGuest) |
| + h.Join(ctx, "r1", "s1", g1.Peer) |
| + g2, _ := h.Open("r1", "s2", 101, "G2", KindGuest) |
| + h.Join(ctx, "r1", "s2", g2.Peer) |
| + h.Admit(ctx, "r1", "sm", m.Peer, g1.Peer) |
| + if err := h.Admit(ctx, "r1", "s1", g1.Peer, g2.Peer); err != ErrNotMember { |
| + t.Fatalf("admitted guest admitting: %v", err) |
| + } |
| + if err := h.Admit(ctx, "r1", "s2", g2.Peer, g2.Peer); err != ErrNotAdmitted { |
| + t.Fatalf("waiting guest admitting: %v", err) |
| + } |
| +} |
| + |
| +func TestCapAtJoinAndAtAdmit(t *testing.T) { |
| + now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC) |
| + h := testHub(&now) |
| + ctx := context.Background() |
| + var members []OpenResult |
| + for i := 0; i < 5; i++ { |
| + o, _ := h.Open("r1", "s"+string(rune('a'+i)), int64(1+i), "M", KindMember) |
| + members = append(members, o) |
| + h.Join(ctx, "r1", "s"+string(rune('a'+i)), o.Peer) |
| + } |
| + fifth := drain(members[4].Sub) |
| + if len(fifth) != 1 || fifth[0].Type != TypeFull { |
| + t.Fatalf("fifth member got %v, want full", types(fifth)) |
| + } |
| + // One leaves; four queued guests and the freed slot yield one admission. |
| + h.Leave("r1", "sd", members[3].Peer) |
| + var guests []OpenResult |
| + for i := 0; i < 4; i++ { |
| + o, _ := h.Open("r1", "g"+string(rune('a'+i)), int64(100+i), "G", KindGuest) |
| + guests = append(guests, o) |
| + h.Join(ctx, "r1", "g"+string(rune('a'+i)), o.Peer) |
| + } |
| + drain(members[0].Sub) |
| + if err := h.Admit(ctx, "r1", "sa", members[0].Peer, guests[0].Peer); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := h.Admit(ctx, "r1", "sa", members[0].Peer, guests[1].Peer); err != nil { |
| + t.Fatal(err) // Admit itself returns nil; the outcome is the `full` event |
| + } |
| + ev := types(drain(members[0].Sub)) |
| + if ev[len(ev)-1] != TypeFull { |
| + t.Fatalf("second admit over the cap: %v, want full last", ev) |
| + } |
| + if p, _ := h.Find("r1", "gb", guests[1].Peer); p.Admitted { |
| + t.Fatal("guest admitted over the cap") |
| + } |
| +} |
| + |
| +func TestEightWaitingThenFull(t *testing.T) { |
| + now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC) |
| + h := testHub(&now) |
| + ctx := context.Background() |
| + for i := 0; i < 8; i++ { |
| + o, _ := h.Open("r1", "g"+string(rune('a'+i)), int64(100+i), "G", KindGuest) |
| + h.Join(ctx, "r1", "g"+string(rune('a'+i)), o.Peer) |
| + } |
| + o, _ := h.Open("r1", "gz", 200, "G", KindGuest) |
| + if err := h.Join(ctx, "r1", "gz", o.Peer); err != ErrTooManyWaiting { |
| + t.Fatalf("ninth waiting: %v", err) |
| + } |
| + if ev := drain(o.Sub); len(ev) != 1 || ev[0].Type != TypeFull { |
| + t.Fatalf("ninth got %v", types(ev)) |
| + } |
| +} |
| + |
| +func TestGuestFirstArrivalIsInTheMembersRoster(t *testing.T) { |
| + now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC) |
| + h := testHub(&now) |
| + ctx := context.Background() |
| + g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest) |
| + h.Join(ctx, "r1", "sg", g.Peer) |
| + m, _ := h.Open("r1", "sm", 1, "Mia", KindMember) |
| + h.Join(ctx, "r1", "sm", m.Peer) |
| + ev := drain(m.Sub) |
| + if len(ev) != 1 || ev[0].Type != TypeRoster || len(ev[0].Roster.Waiting) != 1 || ev[0].Roster.Waiting[0].Peer != g.Peer { |
| + t.Fatalf("member roster %+v", ev) |
| + } |
| +} |
| + |
| +func TestAdmitRefusesAnExpiredGrantBeforeTheTick(t *testing.T) { |
| + now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC) |
| + authority := map[int64]bool{100: true} |
| + h := New(Config{Now: func() time.Time { return now }, StandingFor: func(ctx context.Context, room string, id int64) (Kind, bool, error) { |
| + if id < 100 { |
| + return KindMember, true, nil |
| + } |
| + return KindGuest, authority[id], nil |
| + }}) |
| + ctx := context.Background() |
| + m, _ := h.Open("r1", "sm", 1, "Mia", KindMember) |
| + h.Join(ctx, "r1", "sm", m.Peer) |
| + g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest) |
| + h.Join(ctx, "r1", "sg", g.Peer) |
| + authority[100] = false // grant expired, no tick yet |
| + drain(m.Sub) |
| + if err := h.Admit(ctx, "r1", "sm", m.Peer, g.Peer); err != nil { |
| + t.Fatal(err) |
| + } |
| + if p, ok := h.Find("r1", "sg", g.Peer); ok && p.Admitted { |
| + t.Fatal("expired guest was admitted") |
| + } |
| + gev := drain(g.Sub) |
| + if len(gev) == 0 || gev[len(gev)-1].Type != TypeLost || gev[len(gev)-1].Reason != ReasonExpired { |
| + t.Fatalf("expired guest got %v", types(gev)) |
| + } |
| + if ev := types(drain(m.Sub)); len(ev) != 1 || ev[0] != TypeKnockGone { |
| + t.Fatalf("member got %v", ev) |
| + } |
| +} |
| + |
| +func TestTickDropsAMemberWhoLostStandingAndPromotesAWaitingGuest(t *testing.T) { |
| + now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC) |
| + kinds := map[int64]Kind{1: KindMember, 100: KindGuest} |
| + ok := map[int64]bool{1: true, 100: true} |
| + h := New(Config{Now: func() time.Time { return now }, StandingFor: func(ctx context.Context, room string, id int64) (Kind, bool, error) { |
| + return kinds[id], ok[id], nil |
| + }}) |
| + ctx := context.Background() |
| + m, _ := h.Open("r1", "sm", 1, "Mia", KindMember) |
| + h.Join(ctx, "r1", "sm", m.Peer) |
| + g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest) |
| + h.Join(ctx, "r1", "sg", g.Peer) |
| + drain(m.Sub) |
| + kinds[100] = KindMember // promoted while waiting |
| + h.Tick(ctx) |
| + if p, _ := h.Find("r1", "sg", g.Peer); !p.Admitted || p.Kind != KindMember { |
| + t.Fatal("waiting guest promoted to member was not admitted on the tick") |
| + } |
| + ok[1] = false // member removed from the allowlist |
| + drain(g.Sub) |
| + h.Tick(ctx) |
| + if _, found := h.Find("r1", "sm", m.Peer); found { |
| + t.Fatal("removed member still in the room") |
| + } |
| + mev := drain(m.Sub) |
| + if len(mev) == 0 || mev[len(mev)-1].Reason != ReasonStanding { |
| + t.Fatalf("removed member got %v", types(mev)) |
| + } |
| + if ev := types(drain(g.Sub)); len(ev) == 0 || ev[len(ev)-1] != TypeGone { |
| + t.Fatalf("remaining peer got %v, want gone last", ev) |
| + } |
| +} |
| + |
| +func TestTakeoverDepartureDependsOnState(t *testing.T) { |
| + now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC) |
| + h := testHub(&now) |
| + ctx := context.Background() |
| + m, _ := h.Open("r1", "sm", 1, "Mia", KindMember) |
| + h.Join(ctx, "r1", "sm", m.Peer) |
| + ag, _ := h.Open("r1", "sag", 100, "Adm", KindGuest) |
| + h.Join(ctx, "r1", "sag", ag.Peer) |
| + h.Admit(ctx, "r1", "sm", m.Peer, ag.Peer) |
| + w, _ := h.Open("r1", "sw", 101, "Wait", KindGuest) |
| + h.Join(ctx, "r1", "sw", w.Peer) |
| + u, _ := h.Open("r1", "su", 102, "Unj", KindGuest) |
| + drain(m.Sub) |
| + drain(ag.Sub) |
| + // waiting takeover: knock-gone to members only |
| + if _, err := h.Takeover("r1", "sw"); err != nil { |
| + t.Fatal(err) |
| + } |
| + if ev := types(drain(ag.Sub)); len(ev) != 0 { |
| + t.Fatalf("admitted guest saw a waiting departure: %v", ev) |
| + } |
| + if ev := types(drain(m.Sub)); len(ev) != 1 || ev[0] != TypeKnockGone { |
| + t.Fatalf("member got %v", ev) |
| + } |
| + // unjoined takeover: nothing |
| + h.Takeover("r1", "su") |
| + _ = u |
| + if ev := types(drain(m.Sub)); len(ev) != 0 { |
| + t.Fatalf("unjoined takeover announced %v", ev) |
| + } |
| + // admitted takeover: gone to admitted peers, old stream lost/replaced |
| + h.Takeover("r1", "sag") |
| + agev := drain(ag.Sub) |
| + if len(agev) == 0 || agev[len(agev)-1].Reason != ReasonReplaced { |
| + t.Fatalf("replaced peer got %v", types(agev)) |
| + } |
| + if ev := types(drain(m.Sub)); len(ev) != 1 || ev[0] != TypeGone { |
| + t.Fatalf("member got %v", ev) |
| + } |
| + if _, err := h.Open("r1", "sag", 100, "Adm", KindGuest); err != nil { |
| + t.Fatalf("fresh stream after takeover: %v", err) |
| + } |
| +} |
| + |
| +func TestEndSendsEndedToEveryoneAndDropsTheRoom(t *testing.T) { |
| + now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC) |
| + h := testHub(&now) |
| + ctx := context.Background() |
| + m, _ := h.Open("r1", "sm", 1, "Mia", KindMember) |
| + h.Join(ctx, "r1", "sm", m.Peer) |
| + g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest) |
| + h.Join(ctx, "r1", "sg", g.Peer) |
| + h.End("r1") |
| + for _, s := range []*Sub{m.Sub, g.Sub} { |
| + ev := drain(s) |
| + if len(ev) == 0 || ev[len(ev)-1].Type != TypeLost || ev[len(ev)-1].Reason != ReasonEnded { |
| + t.Fatalf("got %v, want lost/ended last", types(ev)) |
| + } |
| + } |
| + if _, err := h.Open("r1", "sx", 5, "X", KindMember); err == nil { |
| + t.Fatal("ended room accepted a stream") |
| + } |
| +} |
| +``` |
| + |
| +- [ ] **Step 2: Run to verify they fail** |
| + |
| +Run: `GOCACHE=/home/paulca/.cache/tmp/claude-1001/gocache go test ./internal/meet/hub/ 2>&1 | grep -c 'undefined'` |
| +Expected: non-zero (Join, Admit, Leave, Takeover, End undefined). |
| + |
| +- [ ] **Step 3: Implement admission** |
| + |
| +`admission.go`: |
| + |
| +```go |
| +package hub |
| + |
| +import "context" |
| + |
| +func admittedCountLocked(r *room) int { |
| + n := 0 |
| + for _, p := range r.peers { |
| + if p.Admitted { |
| + n++ |
| + } |
| + } |
| + return n |
| +} |
| + |
| +func waitingCountLocked(r *room) int { |
| + n := 0 |
| + for _, p := range r.peers { |
| + if p.Joined && !p.Admitted { |
| + n++ |
| + } |
| + } |
| + return n |
| +} |
| + |
| +// rosterFor is the authoritative snapshot for one viewer: admitted peers |
| +// with the viewer's pair generation, and the waiting queue only when the |
| +// viewer is a member. |
| +func (h *Hub) rosterFor(r *room, viewer *Participation) *Roster { |
| + ro := &Roster{} |
| + for _, p := range r.peers { |
| + if !p.Admitted { |
| + continue |
| + } |
| + rp := RosterPeer{Peer: p.Peer, Name: p.Name, Kind: p.Kind.String()} |
| + if p.Peer != viewer.Peer { |
| + rp.Gen = h.genLocked(r, viewer.Peer, p.Peer) |
| + } |
| + ro.Peers = append(ro.Peers, rp) |
| + } |
| + if viewer.Kind == KindMember { |
| + for _, p := range r.peers { |
| + if p.Joined && !p.Admitted { |
| + ro.Waiting = append(ro.Waiting, RosterPeer{Peer: p.Peer, Name: p.Name, Kind: "guest"}) |
| + } |
| + } |
| + } |
| + return ro |
| +} |
| + |
| +// genLocked returns the pair's generation, starting it at one on first sight. |
| +func (h *Hub) genLocked(r *room, a, b PeerID) uint64 { |
| + k := keyFor(a, b) |
| + if r.gens[k] == 0 { |
| + r.gens[k] = 1 |
| + } |
| + return r.gens[k] |
| +} |
| + |
| +// admitLocked is the one transition into the admitted state: cap under |
| +// the lock, roster to the newcomer, peer to the others, knock-gone to |
| +// members if they were waiting. Returns false when the room is full. |
| +func (h *Hub) admitLocked(r *room, p *Participation) bool { |
| + if admittedCountLocked(r) >= MaxAdmitted { |
| + return false |
| + } |
| + wasWaiting := p.Joined && !p.Admitted |
| + p.Admitted = true |
| + p.Joined = true |
| + p.JoinedAt = h.cfg.Now() |
| + if p.sub != nil { |
| + p.sub.send(Message{Type: TypeRoster, Roster: h.rosterFor(r, p)}) |
| + } |
| + for _, q := range r.peers { |
| + if q.Admitted && q.Peer != p.Peer && q.sub != nil { |
| + q.sub.send(Message{Type: TypePeer, Peer: p.Peer, Name: p.Name, Payload: nil, |
| + Roster: &Roster{Peers: []RosterPeer{{Peer: p.Peer, Name: p.Name, Kind: p.Kind.String(), Gen: h.genLocked(r, q.Peer, p.Peer)}}}}) |
| + } |
| + } |
| + if wasWaiting { |
| + h.broadcastLocked(r, Message{Type: TypeKnockGone, From: p.Peer}, admittedMembers) |
| + } |
| + return true |
| +} |
| + |
| +// Join is the entry. Standing is re-derived now (spec §2.5). |
| +func (h *Hub) Join(ctx context.Context, sqid, session string, peer PeerID) error { |
| + kind, ok, err := h.standing(ctx, sqid, session, peer) |
| + if err != nil { |
| + return err |
| + } |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + r := h.roomLocked(sqid) |
| + p, found := h.findLocked(sqid, session, peer) |
| + if !found { |
| + return ErrUnknownPeer |
| + } |
| + if !ok { |
| + h.endLocked(r, p, ReasonStanding) |
| + return ErrNotYours |
| + } |
| + p.Kind = kind |
| + if p.Joined { |
| + return nil // idempotent |
| + } |
| + if kind == KindMember { |
| + if !h.admitLocked(r, p) { |
| + if p.sub != nil { |
| + p.sub.send(Message{Type: TypeFull}) |
| + } |
| + return ErrRoomFull |
| + } |
| + return nil |
| + } |
| + if waitingCountLocked(r) >= MaxWaiting { |
| + if p.sub != nil { |
| + p.sub.send(Message{Type: TypeFull}) |
| + } |
| + return ErrTooManyWaiting |
| + } |
| + p.Joined = true |
| + p.JoinedAt = h.cfg.Now() |
| + h.broadcastLocked(r, Message{Type: TypeKnock, From: p.Peer, Name: p.Name}, admittedMembers) |
| + return nil |
| +} |
| + |
| +// standing looks the peer up to find its person, then asks the app. |
| +func (h *Hub) standing(ctx context.Context, sqid, session string, peer PeerID) (Kind, bool, error) { |
| + h.mu.Lock() |
| + p, found := h.findLocked(sqid, session, peer) |
| + var personID int64 |
| + if found { |
| + personID = p.PersonID |
| + } |
| + h.mu.Unlock() |
| + if !found { |
| + return 0, false, ErrUnknownPeer |
| + } |
| + return h.cfg.StandingFor(ctx, sqid, personID) |
| +} |
| + |
| +// Admit: the actor must be admitted with current member kind; the target |
| +// waiting with entry authority now; the cap under the lock. |
| +func (h *Hub) Admit(ctx context.Context, sqid, session string, actor, target PeerID) error { |
| + actorKind, actorOK, err := h.standing(ctx, sqid, session, actor) |
| + if err != nil { |
| + return err |
| + } |
| + h.mu.Lock() |
| + tp, found := h.rooms[sqid].peers[target] |
| + var targetPerson int64 |
| + if found { |
| + targetPerson = tp.PersonID |
| + } |
| + h.mu.Unlock() |
| + if !found { |
| + return ErrNotWaiting |
| + } |
| + targetKind, targetOK, err := h.cfg.StandingFor(ctx, sqid, targetPerson) |
| + if err != nil { |
| + return err |
| + } |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + r := h.roomLocked(sqid) |
| + a, ok := h.findLocked(sqid, session, actor) |
| + if !ok { |
| + return ErrUnknownPeer |
| + } |
| + if !actorOK { |
| + h.endLocked(r, a, ReasonStanding) |
| + return ErrNotYours |
| + } |
| + a.Kind = actorKind |
| + if !a.Admitted { |
| + return ErrNotAdmitted |
| + } |
| + if a.Kind != KindMember { |
| + return ErrNotMember |
| + } |
| + tp, ok = r.peers[target] |
| + if !ok || !tp.Joined || tp.Admitted { |
| + return ErrNotWaiting |
| + } |
| + if !targetOK { |
| + wasAdmitted, wasWaiting := h.endLocked(r, tp, ReasonExpired) |
| + h.announceDepartureLocked(r, tp, wasAdmitted, wasWaiting) |
| + return nil |
| + } |
| + tp.Kind = targetKind |
| + if !h.admitLocked(r, tp) { |
| + if a.sub != nil { |
| + a.sub.send(Message{Type: TypeFull}) |
| + } |
| + } |
| + return nil |
| +} |
| + |
| +func (h *Hub) Leave(sqid, session string, peer PeerID) error { |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + r, ok := h.rooms[sqid] |
| + if !ok { |
| + return ErrUnknownPeer |
| + } |
| + p, found := h.findLocked(sqid, session, peer) |
| + if !found { |
| + return ErrUnknownPeer |
| + } |
| + wasAdmitted, wasWaiting := h.endLocked(r, p, ReasonBuffer) |
| + delete(h.tombstones, p.Peer) // a leave is not a tombstoned ending |
| + h.announceDepartureLocked(r, p, wasAdmitted, wasWaiting) |
| + return nil |
| +} |
| + |
| +// Remove is §3.5: the actor holds an admitted member participation in |
| +// this room (kind re-derived), the target is any other participation. |
| +func (h *Hub) Remove(ctx context.Context, sqid, session string, actor, target PeerID) error { |
| + kind, ok, err := h.standing(ctx, sqid, session, actor) |
| + if err != nil { |
| + return err |
| + } |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + r := h.roomLocked(sqid) |
| + a, found := h.findLocked(sqid, session, actor) |
| + if !found || !ok { |
| + return ErrUnknownPeer |
| + } |
| + a.Kind = kind |
| + if !a.Admitted { |
| + return ErrNotAdmitted |
| + } |
| + if a.Kind != KindMember { |
| + return ErrNotMember |
| + } |
| + tp, found := r.peers[target] |
| + if !found || tp.Peer == a.Peer { |
| + return ErrUnknownPeer |
| + } |
| + wasAdmitted, wasWaiting := h.endLocked(r, tp, ReasonRemoved) |
| + h.announceDepartureLocked(r, tp, wasAdmitted, wasWaiting) |
| + return nil |
| +} |
| + |
| +// Takeover ends the session's live participation in the room with the |
| +// state-dependent departure (spec §4.2), leaving a replaced tombstone. |
| +func (h *Hub) Takeover(sqid, session string) (PeerID, error) { |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + r, ok := h.rooms[sqid] |
| + if !ok { |
| + return "", ErrUnknownPeer |
| + } |
| + id, ok := h.bySession[sqid][session] |
| + if !ok { |
| + return "", ErrUnknownPeer |
| + } |
| + p := r.peers[id] |
| + wasAdmitted, wasWaiting := h.endLocked(r, p, ReasonReplaced) |
| + h.announceDepartureLocked(r, p, wasAdmitted, wasWaiting) |
| + return id, nil |
| +} |
| + |
| +// End is the meeting ending: ended to every stream, room dropped. |
| +func (h *Hub) End(sqid string) { |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + r, ok := h.rooms[sqid] |
| + if !ok { |
| + r = h.roomLocked(sqid) |
| + } |
| + for _, p := range r.peers { |
| + h.endLocked(r, p, ReasonEnded) |
| + } |
| + r.ended = true |
| + delete(h.bySession, sqid) |
| +} |
| +``` |
| + |
| +Extend `Tick` in `hub.go`: after the drops, for every joined participation call `cfg.StandingFor` (outside the lock: collect `(room, peer, person)` triples under the lock, call, then re-lock and apply) and apply the §2.4 table: member → none: end with `ReasonStanding` and announce; waiting guest → not ok: end with `ReasonExpired` and announce; kind changed on an admitted peer: set and send `roster` to admitted peers; waiting peer now member: `admitLocked` or `full` to it. Admitted guest with ok=false: nothing (ruling 14). |
| + |
| +- [ ] **Step 4: Run the tests** |
| + |
| +Run: `GOCACHE=/home/paulca/.cache/tmp/claude-1001/gocache go test ./internal/meet/hub/ -v 2>&1 | grep -E '^(=== RUN|--- (PASS|FAIL)|FAIL|ok)' | tail -20` |
| +Expected: all PASS. |
| + |
| +- [ ] **Step 5: Gate and commit** |
| + |
| +```bash |
| +git add internal/meet/hub |
| +git -c user.name="Paul Campbell" -c user.email="paul@rslw.com" commit -m "hub: join, admit, leave, remove, takeover, end, the cap and the tick |
| + |
| +Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> |
| +Claude-Session: https://claude.ai/code/session_01CzvnuKhLzudvq7jZ7SvVt2" |
| +``` |
| + |
| +--- |
| + |
| +### Task 3: Relay: targeted messages, seq and gen, replay ring, reset and pair, resume and gap |
| + |
| +**Files:** |
| +- Create: `internal/meet/hub/relay.go`, `internal/meet/hub/relay_test.go` |
| +- Modify: `internal/meet/hub/hub.go` (`Resume` implemented; `deliverLocked` records targeted messages into the ring) |
| + |
| +**Interfaces:** |
| +- Produces: |
| + ```go |
| + func (h *Hub) Relay(sqid, session string, from PeerID, m Message) error // m.Type in offer/answer/candidate; from admitted; m.To admitted in this room; m.Gen == current gen else ErrStaleGen; delivers envelope {Type, From, Name, To, Seq, Gen, Payload} and rings it on the target |
| + func (h *Hub) Reset(sqid, session string, from, to PeerID) error // both admitted; gen++ under lock; `pair {peer, gen}` to both, ringed |
| + func (h *Hub) Resume(sqid, session string, peer PeerID, last map[PeerID]uint64) (ResumeResult, error) |
| + ``` |
| +- Ring: newest last, capped at `RingSize`; replay on resume = every ringed message with `From == sender && Seq > last[sender]`, plus every `pair` whose `Gen` exceeds the roster's for that peer; if the ring has dropped a message a sender sent above `last` (detected by the ring's oldest `Seq` for that sender being greater than `last[sender]+1` while messages exist), emit `gap {from}` for that sender instead. |
| +- `Resume` answers: unknown peer and no tombstone → `ResumeError{Status: 404, Boot: h.boot}`; tombstone → `ResumeError{Status: 410, Reason}`; peer bound to another session → 404 with boot (never leak); found → attach a new `Sub`, clear `resumeUntil`; admitted: `hello` is the handler's, then the hub returns `Roster`, `Replay`, `Gap`; waiting/unjoined: `Admitted=false`, nothing else. |
| + |
| +- [ ] **Step 1: Write the failing tests** |
| + |
| +```go |
| +package hub |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "testing" |
| + "time" |
| +) |
| + |
| +func twoAdmitted(t *testing.T) (*Hub, *time.Time, OpenResult, OpenResult) { |
| + t.Helper() |
| + now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC) |
| + h := testHub(&now) |
| + ctx := context.Background() |
| + a, _ := h.Open("r1", "sa", 1, "Ann", KindMember) |
| + h.Join(ctx, "r1", "sa", a.Peer) |
| + b, _ := h.Open("r1", "sb", 2, "Bob", KindMember) |
| + h.Join(ctx, "r1", "sb", b.Peer) |
| + drain(a.Sub) |
| + drain(b.Sub) |
| + return h, &now, a, b |
| +} |
| + |
| +func TestRelayGoesOnlyToTheNamedAdmittedPeerWithServerBuiltEnvelope(t *testing.T) { |
| + h, _, a, b := twoAdmitted(t) |
| + ctx := context.Background() |
| + g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest) |
| + h.Join(ctx, "r1", "sg", g.Peer) |
| + drain(a.Sub) |
| + drain(b.Sub) |
| + err := h.Relay("r1", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 1, Gen: 1, From: "forged", Name: "Forged", Payload: json.RawMessage(`{"sdp":"x"}`)}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + bev := drain(b.Sub) |
| + if len(bev) != 1 || bev[0].From != a.Peer || bev[0].Name != "Ann" || string(bev[0].Payload) != `{"sdp":"x"}` { |
| + t.Fatalf("b got %+v", bev) |
| + } |
| + if ev := drain(a.Sub); len(ev) != 0 { |
| + t.Fatalf("sender echoed %v", types(ev)) |
| + } |
| + if ev := drain(g.Sub); len(ev) != 0 { |
| + t.Fatalf("waiting guest received a relay: %v", types(ev)) |
| + } |
| + if err := h.Relay("r1", "sa", a.Peer, Message{Type: TypeOffer, To: g.Peer, Seq: 2, Gen: 1}); err == nil { |
| + t.Fatal("relay to a waiting peer accepted") |
| + } |
| + if err := h.Relay("r1", "sg", g.Peer, Message{Type: TypeOffer, To: a.Peer, Seq: 1, Gen: 1}); err != ErrNotAdmitted { |
| + t.Fatalf("waiting guest relaying: %v", err) |
| + } |
| + if err := h.Relay("r1", "sb", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 3, Gen: 1}); err != ErrUnknownPeer { |
| + t.Fatalf("peer through another session: %v", err) |
| + } |
| +} |
| + |
| +func TestStaleGenIsRefusedAndResetConverges(t *testing.T) { |
| + h, _, a, b := twoAdmitted(t) |
| + if err := h.Relay("r1", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 1, Gen: 2}); err != ErrStaleGen { |
| + t.Fatalf("gen 2 before any reset: %v", err) |
| + } |
| + if err := h.Reset("r1", "sa", a.Peer, b.Peer); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := h.Reset("r1", "sb", b.Peer, a.Peer); err != nil { // simultaneous resets converge on the higher gen |
| + t.Fatal(err) |
| + } |
| + aev, bev := drain(a.Sub), drain(b.Sub) |
| + if len(aev) != 2 || aev[1].Type != TypePair || aev[1].Gen != 3 || aev[1].Peer != b.Peer { |
| + t.Fatalf("a got %+v", aev) |
| + } |
| + if len(bev) != 2 || bev[1].Gen != 3 || bev[1].Peer != a.Peer { |
| + t.Fatalf("b got %+v", bev) |
| + } |
| + if err := h.Relay("r1", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 1, Gen: 1}); err != ErrStaleGen { |
| + t.Fatal("old gen accepted after reset") |
| + } |
| + if err := h.Relay("r1", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 1, Gen: 3}); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| +func TestResumeReplaysAboveLastAndSendsRosterOnlyWhenAdmitted(t *testing.T) { |
| + h, now, a, b := twoAdmitted(t) |
| + ctx := context.Background() |
| + for i := uint64(1); i <= 5; i++ { |
| + h.Relay("r1", "sb", b.Peer, Message{Type: TypeCandidate, To: a.Peer, Seq: i, Gen: 1, Payload: json.RawMessage(`{}`)}) |
| + } |
| + drain(a.Sub) |
| + h.Detach("r1", a.Peer) |
| + *now = now.Add(5 * time.Second) |
| + res, err := h.Resume("r1", "sa", a.Peer, map[PeerID]uint64{b.Peer: 3}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if !res.Admitted || res.Roster == nil || len(res.Replay) != 2 || res.Replay[0].Seq != 4 || res.Replay[1].Seq != 5 { |
| + t.Fatalf("resume %+v", res) |
| + } |
| + // A waiting guest resumes and learns nothing. |
| + g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest) |
| + h.Join(ctx, "r1", "sg", g.Peer) |
| + h.Detach("r1", g.Peer) |
| + gres, err := h.Resume("r1", "sg", g.Peer, nil) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if gres.Admitted || gres.Roster != nil || len(gres.Replay) != 0 { |
| + t.Fatalf("waiting guest resume leaked %+v", gres) |
| + } |
| +} |
| + |
| +func TestResumeGapWhenTheRingRolledOver(t *testing.T) { |
| + h, now, a, b := twoAdmitted(t) |
| + for i := uint64(1); i <= 40; i++ { |
| + h.Relay("r1", "sb", b.Peer, Message{Type: TypeCandidate, To: a.Peer, Seq: i, Gen: 1}) |
| + } |
| + drain(a.Sub) |
| + h.Detach("r1", a.Peer) |
| + *now = now.Add(time.Second) |
| + res, _ := h.Resume("r1", "sa", a.Peer, map[PeerID]uint64{b.Peer: 2}) |
| + if len(res.Gap) != 1 || res.Gap[0] != b.Peer || len(res.Replay) != 0 { |
| + t.Fatalf("expected a gap for b, got replay %d gap %v", len(res.Replay), res.Gap) |
| + } |
| +} |
| + |
| +func TestResumeAnswers(t *testing.T) { |
| + h, now, a, _ := twoAdmitted(t) |
| + if _, err := h.Resume("r1", "sb", a.Peer, nil); err == nil || err.(*ResumeError).Status != 404 || err.(*ResumeError).Boot != h.Boot() { |
| + t.Fatalf("resume through another session: %v", err) |
| + } |
| + h.Takeover("r1", "sa") |
| + if _, err := h.Resume("r1", "sa", a.Peer, nil); err == nil || err.(*ResumeError).Status != 410 || err.(*ResumeError).Reason != ReasonReplaced { |
| + t.Fatalf("resume after takeover: %v", err) |
| + } |
| + *now = now.Add(TombstoneLife + time.Minute) |
| + h.Tick(context.Background()) |
| + if _, err := h.Resume("r1", "sa", a.Peer, nil); err == nil || err.(*ResumeError).Status != 404 { |
| + t.Fatalf("resume after tombstone expiry: %v", err) |
| + } |
| + if _, err := h.Open("r1", "sa", 1, "Ann", KindMember); err != nil { |
| + t.Fatalf("fresh stream after tombstone: %v", err) |
| + } |
| +} |
| + |
| +func TestResumeAfterWindowIsA404WithTheSameBoot(t *testing.T) { |
| + h, now, a, _ := twoAdmitted(t) |
| + h.Detach("r1", a.Peer) |
| + *now = now.Add(ResumeWindow + time.Second) |
| + h.Tick(context.Background()) |
| + _, err := h.Resume("r1", "sa", a.Peer, nil) |
| + re, ok := err.(*ResumeError) |
| + if !ok || re.Status != 404 || re.Boot != h.Boot() { |
| + t.Fatalf("lapsed resume: %v", err) |
| + } |
| +} |
| +``` |
| + |
| +- [ ] **Step 2: Run to verify they fail** |
| + |
| +Run: `GOCACHE=/home/paulca/.cache/tmp/claude-1001/gocache go test ./internal/meet/hub/ 2>&1 | grep -c undefined` |
| +Expected: non-zero. |
| + |
| +- [ ] **Step 3: Implement relay and resume** |
| + |
| +`relay.go`: |
| + |
| +```go |
| +package hub |
| + |
| +import "fmt" |
| + |
| +func isTargeted(t string) bool { return t == TypeOffer || t == TypeAnswer || t == TypeCandidate } |
| + |
| +// Relay delivers one targeted message. The envelope is rebuilt from the |
| +// binding; the payload is never read. |
| +func (h *Hub) Relay(sqid, session string, from PeerID, m Message) error { |
| + if !isTargeted(m.Type) { |
| + return fmt.Errorf("hub: %q is not a targeted type", m.Type) |
| + } |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + r, ok := h.rooms[sqid] |
| + if !ok { |
| + return ErrUnknownPeer |
| + } |
| + p, found := h.findLocked(sqid, session, from) |
| + if !found { |
| + return ErrUnknownPeer |
| + } |
| + if !p.Admitted { |
| + return ErrNotAdmitted |
| + } |
| + tp, ok := r.peers[m.To] |
| + if !ok || !tp.Admitted { |
| + return ErrUnknownPeer |
| + } |
| + if m.Gen != h.genLocked(r, p.Peer, tp.Peer) { |
| + return ErrStaleGen |
| + } |
| + out := Message{Type: m.Type, From: p.Peer, Name: p.Name, To: tp.Peer, Seq: m.Seq, Gen: m.Gen, Payload: m.Payload} |
| + h.ringLocked(tp, out) |
| + if tp.sub != nil { |
| + tp.sub.send(out) |
| + } |
| + return nil |
| +} |
| + |
| +func (h *Hub) ringLocked(p *Participation, m Message) { |
| + p.ring = append(p.ring, m) |
| + if len(p.ring) > RingSize { |
| + p.ring = p.ring[len(p.ring)-RingSize:] |
| + } |
| +} |
| + |
| +// Reset bumps the pair's generation and tells both sides. |
| +func (h *Hub) Reset(sqid, session string, from, to PeerID) error { |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + r, ok := h.rooms[sqid] |
| + if !ok { |
| + return ErrUnknownPeer |
| + } |
| + p, found := h.findLocked(sqid, session, from) |
| + if !found { |
| + return ErrUnknownPeer |
| + } |
| + if !p.Admitted { |
| + return ErrNotAdmitted |
| + } |
| + tp, ok := r.peers[to] |
| + if !ok || !tp.Admitted { |
| + return ErrUnknownPeer |
| + } |
| + k := keyFor(p.Peer, tp.Peer) |
| + h.genLocked(r, p.Peer, tp.Peer) |
| + r.gens[k]++ |
| + gen := r.gens[k] |
| + for _, pair := range [][2]*Participation{{p, tp}, {tp, p}} { |
| + m := Message{Type: TypePair, Peer: pair[1].Peer, Gen: gen, To: pair[0].Peer} |
| + h.ringLocked(pair[0], m) |
| + if pair[0].sub != nil { |
| + pair[0].sub.send(m) |
| + } |
| + } |
| + return nil |
| +} |
| + |
| +type ResumeResult struct { |
| + Sub *Sub |
| + Admitted bool |
| + Roster *Roster |
| + Replay []Message |
| + Gap []PeerID |
| +} |
| + |
| +type ResumeError struct { |
| + Status int |
| + Reason Reason |
| + Boot string |
| +} |
| + |
| +func (e *ResumeError) Error() string { return fmt.Sprintf("hub: resume refused %d %s", e.Status, e.Reason) } |
| + |
| +// Resume reattaches a stream to a participation in its window. |
| +func (h *Hub) Resume(sqid, session string, peer PeerID, last map[PeerID]uint64) (ResumeResult, error) { |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + if t, ok := h.tombstones[peer]; ok { |
| + return ResumeResult{}, &ResumeError{Status: 410, Reason: t.reason} |
| + } |
| + r, ok := h.rooms[sqid] |
| + if !ok { |
| + return ResumeResult{}, &ResumeError{Status: 404, Boot: h.boot} |
| + } |
| + p, found := h.findLocked(sqid, session, peer) |
| + if !found { |
| + return ResumeResult{}, &ResumeError{Status: 404, Boot: h.boot} |
| + } |
| + if p.sub != nil { |
| + p.sub.terminate(Message{Type: TypeLost, Reason: ReasonReplaced}) // a second stream for the same peer replaces the first attachment only |
| + } |
| + p.sub = newSub() |
| + p.resumeUntil = time.Time{} |
| + res := ResumeResult{Sub: p.sub, Admitted: p.Admitted} |
| + if !p.Admitted { |
| + return res, nil |
| + } |
| + res.Roster = h.rosterFor(r, p) |
| + // Replay: per sender, everything above last; a gap if the ring no |
| + // longer holds the first message the browser has not seen. |
| + oldest := map[PeerID]uint64{} |
| + for _, m := range p.ring { |
| + if isTargeted(m.Type) { |
| + if o, ok := oldest[m.From]; !ok || m.Seq < o { |
| + oldest[m.From] = m.Seq |
| + } |
| + } |
| + } |
| + gapped := map[PeerID]bool{} |
| + for from, o := range oldest { |
| + if o > last[from]+1 { |
| + gapped[from] = true |
| + res.Gap = append(res.Gap, from) |
| + } |
| + } |
| + for _, m := range p.ring { |
| + switch { |
| + case isTargeted(m.Type): |
| + if !gapped[m.From] && m.Seq > last[m.From] { |
| + res.Replay = append(res.Replay, m) |
| + } |
| + case m.Type == TypePair: |
| + if m.Gen > h.currentGenSeenLocked(res.Roster, m.Peer) { |
| + res.Replay = append(res.Replay, m) |
| + } |
| + } |
| + } |
| + return res, nil |
| +} |
| + |
| +func (h *Hub) currentGenSeenLocked(ro *Roster, peer PeerID) uint64 { |
| + for _, rp := range ro.Peers { |
| + if rp.Peer == peer { |
| + return rp.Gen |
| + } |
| + } |
| + return 0 |
| +} |
| +``` |
| + |
| +Note: since the roster already carries the current generation for every admitted peer, a `pair` in the ring is only replayed when its `Gen` exceeds what the roster says, which by construction never happens for a current roster; keep the branch for correctness and let the browser also reconcile from the roster (spec §4.6). Add `"time"` to the imports. |
| + |
| +- [ ] **Step 4: Run the tests, gate, commit** |
| + |
| +Run: `GOCACHE=/home/paulca/.cache/tmp/claude-1001/gocache go test ./internal/meet/hub/ -race 2>&1 | tail -3` then `make ci`. |
| + |
| +```bash |
| +git add internal/meet/hub |
| +git -c user.name="Paul Campbell" -c user.email="paul@rslw.com" commit -m "hub: targeted relay with seq and gen, replay ring, reset and pair, resume and gap |
| + |
| +Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> |
| +Claude-Session: https://claude.ai/code/session_01CzvnuKhLzudvq7jZ7SvVt2" |
| +``` |
| + |
| +--- |
| + |
| +### Task 4: The stream: SSE, hello with boot, heartbeat and TTL, resume answers |
| + |
| +**Files:** |
| +- Create: `internal/meet/stream.go`, `internal/meet/participation.go` |
| +- Modify: `internal/meet/app.go` (hub field, tick goroutine, routes), `internal/meettest/harness_test.go` (SSE client helper) |
| +- Test: `internal/meettest/stream_test.go` |
| + |
| +**Interfaces:** |
| +- Produces: `app.hub *hub.Hub` constructed in `build` with `StandingFor` adapting `standingFor` (look up the meeting by sqid and the person by id; `KindMember` when `StandingMember`, `KindGuest` when `StandingGuest`, `ok=false` when none or the meeting ended); a fifteen-second tick goroutine started by `App` only, plus `TestHandle.TickForTest()`; `GET /m/{sqid}/stream` inside the `requirePerson` group; SSE events `event: <type>\ndata: <json>\n\n`; `hello {peer, boot}` first; heartbeat every 15 s as a comment line `: hb\n\n`; TTL five minutes then the handler returns (the browser resumes); `Detach` on any exit; resume via `?peer=<id>&last=<json>`; answers per spec §4.5 with JSON bodies; test hooks `TestHandle.SetStreamTimings(heartbeat, ttl time.Duration)`. |
| +- Harness: `func (c *client) stream(target string) *sseReader` with `next(timeout) (event string, data []byte, ok bool)`, `close()`, and `hello() (peer, boot string)`. |
| + |
| +- [ ] **Step 1: Write the failing tests** |
| + |
| +```go |
| +package meettest |
| + |
| +import ( |
| + "encoding/json" |
| + "net/http" |
| + "testing" |
| + "time" |
| +) |
| + |
| +func TestStreamSaysHelloWithPeerAndBootAndRefusesTheSecondFreshStream(t *testing.T) { |
| + x, sqid, _ := memberWithMeeting(t) |
| + c := x.browser(t) |
| + c.signIn("paul@example.com") |
| + s := c.stream("/m/" + sqid + "/stream") |
| + defer s.close() |
| + peer, boot := s.hello() |
| + if len(peer) != 32 || len(boot) != 32 { |
| + t.Fatalf("hello peer=%q boot=%q", peer, boot) |
| + } |
| + rec := c.get("/m/" + sqid + "/stream") |
| + if rec.Code != http.StatusConflict { |
| + t.Fatalf("second fresh stream: %d, want 409", rec.Code) |
| + } |
| + var body struct{ Peer string } |
| + json.Unmarshal(rec.Body.Bytes(), &body) |
| + if body.Peer != peer { |
| + t.Fatalf("409 names %q, want %q", body.Peer, peer) |
| + } |
| +} |
| + |
| +func TestStreamNeedsStandingAndAResumeNeedsTheSession(t *testing.T) { |
| + x, sqid, tok := memberWithMeeting(t) |
| + stranger := x.browser(t) |
| + // stranger has no standing: no grant, not a member |
| + stranger2 := newHarness(t).browser(t) |
| + _ = stranger2 |
| + if rec := stranger.get("/m/" + sqid + "/stream"); rec.Code != http.StatusSeeOther { |
| + t.Fatalf("signed-out stream: %d", rec.Code) |
| + } |
| + c := x.browser(t) |
| + c.signIn("paul@example.com") |
| + s := c.stream("/m/" + sqid + "/stream") |
| + peer, _ := s.hello() |
| + s.close() |
| + other := x.browser(t) |
| + other.get("/j/" + tok) |
| + other.signIn("guest@example.com") |
| + rec := other.get("/m/" + sqid + "/stream?peer=" + peer) |
| + if rec.Code != http.StatusNotFound { |
| + t.Fatalf("resume through another session: %d, want 404", rec.Code) |
| + } |
| +} |
| + |
| +func TestResumeAfterTheStreamClosesKeepsAdmission(t *testing.T) { |
| + x, sqid, _ := memberWithMeeting(t) |
| + c := x.browser(t) |
| + c.signIn("paul@example.com") |
| + s := c.stream("/m/" + sqid + "/stream") |
| + peer, boot := s.hello() |
| + c.signal(sqid, peer, `{"type":"join"}`) |
| + if ev, _, ok := s.next(time.Second); !ok || ev != "roster" { |
| + t.Fatalf("after join: %s ok=%v", ev, ok) |
| + } |
| + s.close() |
| + s2 := c.stream("/m/" + sqid + "/stream?peer=" + peer) |
| + defer s2.close() |
| + p2, b2 := s2.hello() |
| + if p2 != peer || b2 != boot { |
| + t.Fatalf("resume hello %s %s", p2, b2) |
| + } |
| + if ev, _, ok := s2.next(time.Second); !ok || ev != "roster" { |
| + t.Fatalf("resume of an admitted peer should send roster, got %s", ev) |
| + } |
| +} |
| + |
| +func TestTTLClosesTheStreamAndResumeWorksWithinTheWindow(t *testing.T) { |
| + x, sqid, _ := memberWithMeeting(t) |
| + x.ctl.SetStreamTimings(50*time.Millisecond, 300*time.Millisecond) |
| + c := x.browser(t) |
| + c.signIn("paul@example.com") |
| + s := c.stream("/m/" + sqid + "/stream") |
| + peer, _ := s.hello() |
| + c.signal(sqid, peer, `{"type":"join"}`) |
| + s.next(time.Second) |
| + if !s.waitClosed(2 * time.Second) { |
| + t.Fatal("stream did not close at the TTL") |
| + } |
| + s2 := c.stream("/m/" + sqid + "/stream?peer=" + peer) |
| + defer s2.close() |
| + if p, _ := s2.hello(); p != peer { |
| + t.Fatalf("resume after TTL: %s", p) |
| + } |
| +} |
| + |
| +func TestLapsedResumeIs404WithTheSameBootAndAFreshJoinWorks(t *testing.T) { |
| + x, sqid, _ := memberWithMeeting(t) |
| + c := x.browser(t) |
| + c.signIn("paul@example.com") |
| + s := c.stream("/m/" + sqid + "/stream") |
| + peer, boot := s.hello() |
| + c.signal(sqid, peer, `{"type":"join"}`) |
| + s.next(time.Second) |
| + s.close() |
| + x.advance(31 * time.Second) |
| + x.ctl.TickForTest() |
| + rec := c.get("/m/" + sqid + "/stream?peer=" + peer) |
| + var body struct{ Boot string } |
| + json.Unmarshal(rec.Body.Bytes(), &body) |
| + if rec.Code != http.StatusNotFound || body.Boot != boot { |
| + t.Fatalf("lapsed resume: %d %+v", rec.Code, body) |
| + } |
| + s3 := c.stream("/m/" + sqid + "/stream") |
| + defer s3.close() |
| + if p, _ := s3.hello(); p == "" || p == peer { |
| + t.Fatalf("fresh join after lapse: %q", p) |
| + } |
| +} |
| +``` |
| + |
| +The harness additions (`harness_test.go`): an `sseReader` that performs the request with `httptest.NewRecorder` cannot stream; use a real `httptest.Server` wrapping `x.h` for stream tests. Add to `harness`: `srv *httptest.Server` started lazily by `x.server()`; `client.stream` issues a real HTTP GET with the client's cookies and `Origin`, reads lines in a goroutine, parses `event:`/`data:` pairs into a channel; `next(timeout)` receives; `hello()` reads the first event and unmarshals `peer` and `boot`; `waitClosed` waits for EOF; `close()` cancels the request. `client.signal(sqid, peer, body string) *http.Response` POSTs JSON `{"peer": peer, ...body fields}` with `Content-Type: application/json` and `Origin` through the same server. Because the CSRF middleware checks `Origin` against `testOrigin` (`http://app.test`), the `httptest.Server`'s origin differs; set the harness `Config.Origin` to the server's URL when the server is started (start it in `newHarness` before building the app, so `Origin` is right from the first request) and make `testOrigin` a per-harness value `x.origin`. Update `linkInPage` to match `x.origin`. This is a harness refactor the first step of this task does before the tests. |
| + |
| +- [ ] **Step 2: Implement** |
| + |
| +`participation.go`: |
| + |
| +```go |
| +package meet |
| + |
| +import ( |
| + "context" |
| + |
| + "amadan.net/oficina/meet/internal/meet/hub" |
| +) |
| + |
| +// hubStanding adapts standingFor for the hub: kind and whether the |
| +// person still holds entry authority for the room. |
| +func (a *app) hubStanding(ctx context.Context, sqid string, personID int64) (hub.Kind, bool, error) { |
| + m, ok, err := a.store.meetingBySqid(ctx, sqid) |
| + if err != nil || !ok || m.EndedAt != nil { |
| + return hub.KindGuest, false, err |
| + } |
| + p, err := a.store.personByID(ctx, personID) |
| + if err != nil { |
| + return hub.KindGuest, false, err |
| + } |
| + st, err := a.standingFor(ctx, p, m) |
| + if err != nil { |
| + return hub.KindGuest, false, err |
| + } |
| + switch st { |
| + case StandingMember: |
| + return hub.KindMember, true, nil |
| + case StandingGuest: |
| + return hub.KindGuest, true, nil |
| + } |
| + return hub.KindGuest, false, nil |
| +} |
| +``` |
| + |
| +Add `store.personByID(ctx, id int64) (Person, error)`. |
| + |
| +`stream.go`: |
| + |
| +```go |
| +package meet |
| + |
| +import ( |
| + "encoding/json" |
| + "fmt" |
| + "net/http" |
| + "time" |
| + |
| + "github.com/go-chi/chi/v5" |
| + |
| + "amadan.net/oficina/meet/internal/meet/hub" |
| + "amadan.net/rastrillo/rastrillo/sessions" |
| +) |
| + |
| +// sessionKey is the session's identity for the hub: the cookie token |
| +// hash is not exposed by rastrillo, so the subject plus a per-boot salt |
| +// stands in. Two tabs of one browser share it; two people never do. |
| +func (a *app) sessionKey(r *http.Request) string { |
| + s, _ := sessions.Current(r) |
| + return s.Subject + "|" + s.At.UTC().Format(time.RFC3339Nano) |
| +} |
| + |
| +func writeEvent(w http.ResponseWriter, flusher http.Flusher, m hub.Message) error { |
| + b, err := json.Marshal(m) |
| + if err != nil { |
| + return err |
| + } |
| + if _, err := fmt.Fprintf(w, "event: %s\ndata: %s\n\n", m.Type, b); err != nil { |
| + return err |
| + } |
| + flusher.Flush() |
| + return nil |
| +} |
| + |
| +// stream is GET /m/{sqid}/stream: fresh (standing) or resume (participation). |
| +func (a *app) stream(w http.ResponseWriter, r *http.Request) { |
| + sqid := chi.URLParam(r, "sqid") |
| + p, _ := currentPerson(r) |
| + session := a.sessionKey(r) |
| + flusher, ok := w.(http.Flusher) |
| + if !ok { |
| + ErrorPage(w, r, http.StatusInternalServerError, "") |
| + return |
| + } |
| + |
| + var sub *hub.Sub |
| + var peer hub.PeerID |
| + var first []hub.Message |
| + |
| + if resume := r.URL.Query().Get("peer"); resume != "" { |
| + var last map[hub.PeerID]uint64 |
| + if raw := r.URL.Query().Get("last"); raw != "" { |
| + if err := json.Unmarshal([]byte(raw), &last); err != nil { |
| + http.Error(w, `{"error":"bad last"}`, http.StatusBadRequest) |
| + return |
| + } |
| + } |
| + res, err := a.hub.Resume(sqid, session, hub.PeerID(resume), last) |
| + if err != nil { |
| + re, ok := err.(*hub.ResumeError) |
| + if !ok { |
| + ErrorPage(w, r, http.StatusInternalServerError, "") |
| + return |
| + } |
| + w.Header().Set("Content-Type", "application/json") |
| + w.WriteHeader(re.Status) |
| + json.NewEncoder(w).Encode(map[string]any{"reason": re.Reason, "boot": re.Boot}) |
| + return |
| + } |
| + sub, peer = res.Sub, hub.PeerID(resume) |
| + first = append(first, hub.Message{Type: hub.TypeHello, Peer: peer, Boot: a.hub.Boot()}) |
| + if res.Admitted { |
| + first = append(first, hub.Message{Type: hub.TypeRoster, Roster: res.Roster}) |
| + first = append(first, res.Replay...) |
| + for _, from := range res.Gap { |
| + first = append(first, hub.Message{Type: hub.TypeGap, From: from}) |
| + } |
| + } else { |
| + first = append(first, hub.Message{Type: hub.TypeState, Waiting: true}) |
| + } |
| + } else { |
| + m, ok, err := a.store.meetingBySqid(r.Context(), sqid) |
| + if err != nil || !ok || m.EndedAt != nil { |
| + ErrorPage(w, r, http.StatusNotFound, "") |
| + return |
| + } |
| + st, err := a.standingFor(r.Context(), p, m) |
| + if err != nil || st == StandingNone { |
| + ErrorPage(w, r, http.StatusNotFound, "") |
| + return |
| + } |
| + kind := hub.KindGuest |
| + if st == StandingMember { |
| + kind = hub.KindMember |
| + } |
| + res, err := a.hub.Open(sqid, session, p.ID, p.Name, kind) |
| + switch err { |
| + case nil: |
| + case hub.ErrSessionHasPeer: |
| + w.Header().Set("Content-Type", "application/json") |
| + w.WriteHeader(http.StatusConflict) |
| + json.NewEncoder(w).Encode(map[string]any{"peer": res.Existing}) |
| + return |
| + case hub.ErrRoomFull, hub.ErrPersonStreams: |
| + http.Error(w, `{"error":"full"}`, http.StatusServiceUnavailable) |
| + return |
| + default: |
| + ErrorPage(w, r, http.StatusNotFound, "") |
| + return |
| + } |
| + sub, peer = res.Sub, res.Peer |
| + first = append(first, hub.Message{Type: hub.TypeHello, Peer: peer, Boot: a.hub.Boot()}) |
| + } |
| + |
| + w.Header().Set("Content-Type", "text/event-stream") |
| + w.Header().Set("Cache-Control", "no-store") |
| + w.Header().Set("X-Accel-Buffering", "no") |
| + w.WriteHeader(http.StatusOK) |
| + defer a.hub.Detach(sqid, peer) |
| + |
| + rc := http.NewResponseController(w) |
| + for _, m := range first { |
| + if err := writeEvent(w, flusher, m); err != nil { |
| + return |
| + } |
| + } |
| + heartbeat := time.NewTicker(a.streamHeartbeat) |
| + defer heartbeat.Stop() |
| + ttl := time.NewTimer(a.streamTTL) |
| + defer ttl.Stop() |
| + for { |
| + _ = rc.SetWriteDeadline(time.Now().Add(30 * time.Second)) |
| + select { |
| + case <-r.Context().Done(): |
| + return |
| + case <-ttl.C: |
| + return |
| + case <-heartbeat.C: |
| + if _, err := fmt.Fprint(w, ": hb\n\n"); err != nil { |
| + return |
| + } |
| + flusher.Flush() |
| + case m, ok := <-sub.C(): |
| + if !ok { |
| + return |
| + } |
| + if err := writeEvent(w, flusher, m); err != nil { |
| + return |
| + } |
| + if m.Type == hub.TypeLost || m.Type == hub.TypeEnded { |
| + return |
| + } |
| + } |
| + } |
| +} |
| +``` |
| + |
| +In `app.go`: fields `hub *hub.Hub`, `streamHeartbeat`, `streamTTL` (defaults 15 s and 5 min); construct the hub in `build` with `Now: st.now`, `Logger`, `StandingFor: a.hubStanding`; `App` starts `go a.tickForever(ctx)` (15 s ticker calling `a.hub.Tick(ctx)`); `TestHandle.TickForTest()` and `SetStreamTimings`. Route inside the `requirePerson` group: `r.Get("/m/{sqid}/stream", a.stream)`. Also wire `endMeeting` to call `a.hub.End(m.Sqid)` after the store update. |
| + |
| +- [ ] **Step 3: Run tests, gate, commit** |
| + |
| +Run: `GOCACHE=/home/paulca/.cache/tmp/claude-1001/gocache go test ./internal/meettest/ -run 'Stream|Resume|TTL|Lapsed' -v 2>&1 | tail -15` then `make ci`. |
| + |
| +```bash |
| +git add internal/meet internal/meettest |
| +git -c user.name="Paul Campbell" -c user.email="paul@rslw.com" commit -m "stream: SSE per peer with hello and boot, heartbeat, TTL, resume answers |
| + |
| +Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> |
| +Claude-Session: https://claude.ai/code/session_01CzvnuKhLzudvq7jZ7SvVt2" |
| +``` |
| + |
| +--- |
| + |
| +### Task 5: Signal: POST with limits, rate limit, dispatch, refusal logging, captured-log test |
| + |
| +**Files:** |
| +- Create: `internal/meet/signal.go`, `internal/meet/ratelimit.go` |
| +- Test: `internal/meettest/signal_test.go` |
| + |
| +**Interfaces:** |
| +- Produces: `POST /m/{sqid}/signal` inside the `requirePerson` group; body `{"peer": "<id>", "type": "...", "to": "...", "seq": n, "gen": n, "payload": {...}}`; refused above 64 KB with 413; 429 with `Retry-After: 1` above thirty per second (burst sixty) per peer, a token bucket keyed by peer id in `a.signalLimiter`; dispatch: `join` → `hub.Join` (standing re-derived by the hub); `admit` → `hub.Admit(actor, payload.peer)`; `leave` → `hub.Leave`; `offer|answer|candidate` → `hub.Relay`; `reset` → `hub.Reset(from, payload.to)`; `chat|share` → 400 in this slice (slice 4); unknown → 400. Error mapping: `ErrUnknownPeer`/`ErrNotYours` → 404; `ErrNotAdmitted`/`ErrNotMember`/`ErrNotWaiting` → 403; `ErrStaleGen` → 409 `{"error":"stale","gen":<current>}`; `ErrRoomFull`/`ErrTooManyWaiting` → 200 (the `full` event carries the outcome). Every refusal logs `type`, `room`, `peer`, `outcome` and nothing else; an accepted message logs nothing. |
| + |
| +- [ ] **Step 1: Write the failing tests** |
| + |
| +```go |
| +package meettest |
| + |
| +import ( |
| + "bytes" |
| + "log/slog" |
| + "net/http" |
| + "strings" |
| + "testing" |
| + "time" |
| +) |
| + |
| +func TestSignalNeedsAPeerBoundToThisSessionAndRoom(t *testing.T) { |
| + x, sqid, tok := memberWithMeeting(t) |
| + c := x.browser(t) |
| + c.signIn("paul@example.com") |
| + s := c.stream("/m/" + sqid + "/stream") |
| + defer s.close() |
| + peer, _ := s.hello() |
| + if resp := c.signal(sqid, "0123456789abcdef0123456789abcdef", `{"type":"join"}`); resp.StatusCode != http.StatusNotFound { |
| + t.Fatalf("foreign peer: %d", resp.StatusCode) |
| + } |
| + other := x.browser(t) |
| + other.get("/j/" + tok) |
| + other.signIn("guest@example.com") |
| + if resp := other.signal(sqid, peer, `{"type":"join"}`); resp.StatusCode != http.StatusNotFound { |
| + t.Fatalf("peer through another session: %d", resp.StatusCode) |
| + } |
| + if resp := c.signal(sqid, peer, `{"type":"join"}`); resp.StatusCode != http.StatusOK { |
| + t.Fatalf("own join: %d", resp.StatusCode) |
| + } |
| +} |
| + |
| +func TestSignalLimits(t *testing.T) { |
| + x, sqid, _ := memberWithMeeting(t) |
| + c := x.browser(t) |
| + c.signIn("paul@example.com") |
| + s := c.stream("/m/" + sqid + "/stream") |
| + defer s.close() |
| + peer, _ := s.hello() |
| + c.signal(sqid, peer, `{"type":"join"}`) |
| + big := `{"type":"candidate","to":"` + peer + `","seq":1,"gen":1,"payload":{"x":"` + strings.Repeat("a", 70*1024) + `"}}` |
| + if resp := c.signal(sqid, peer, big); resp.StatusCode != http.StatusRequestEntityTooLarge { |
| + t.Fatalf("64 KB: %d", resp.StatusCode) |
| + } |
| + var got429 bool |
| + for i := 0; i < 100; i++ { |
| + if resp := c.signal(sqid, peer, `{"type":"leave"}`); resp.StatusCode == http.StatusTooManyRequests { |
| + if resp.Header.Get("Retry-After") == "" { |
| + t.Fatal("429 without Retry-After") |
| + } |
| + got429 = true |
| + break |
| + } |
| + } |
| + if !got429 { |
| + t.Fatal("100 messages in a burst never hit the rate limit") |
| + } |
| +} |
| + |
| +func TestSignalStaleGenAnswers409WithTheCurrentGen(t *testing.T) { |
| + x, sqid, _ := memberWithMeeting(t) |
| + a := x.browser(t) |
| + a.signIn("paul@example.com") |
| + sa := a.stream("/m/" + sqid + "/stream") |
| + defer sa.close() |
| + pa, _ := sa.hello() |
| + a.signal(sqid, pa, `{"type":"join"}`) |
| + b := newMemberBrowser(t, x, "paul@rslw.com") |
| + sb := b.stream("/m/" + sqid + "/stream") |
| + defer sb.close() |
| + pb, _ := sb.hello() |
| + b.signal(sqid, pb, `{"type":"join"}`) |
| + resp := a.signal(sqid, pa, `{"type":"offer","to":"`+pb+`","seq":1,"gen":7,"payload":{}}`) |
| + if resp.StatusCode != http.StatusConflict { |
| + t.Fatalf("stale gen: %d", resp.StatusCode) |
| + } |
| +} |
| + |
| +// The §0.1 and §4.4 promise: no payload, name, title or token in any log |
| +// line, on accepted and refused messages and on an error path. |
| +func TestSignallingLogsCarryNoContent(t *testing.T) { |
| + var buf bytes.Buffer |
| + x := newHarnessWithLogger(t, slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug})), "paul@example.com", "paul@rslw.com") |
| + c := x.browser(t) |
| + c.signIn("paul@example.com") |
| + sqid, tok := createMeeting(t, c, "Secret title", "") |
| + s := c.stream("/m/" + sqid + "/stream") |
| + defer s.close() |
| + peer, _ := s.hello() |
| + c.signal(sqid, peer, `{"type":"join"}`) |
| + b := newMemberBrowser(t, x, "paul@rslw.com") |
| + sb := b.stream("/m/" + sqid + "/stream") |
| + defer sb.close() |
| + pb, _ := sb.hello() |
| + b.signal(sqid, pb, `{"type":"join"}`) |
| + c.signal(sqid, peer, `{"type":"offer","to":"`+pb+`","seq":1,"gen":1,"payload":{"sdp":"v=0 SDPSECRET a=candidate 203.0.113.9"}}`) |
| + c.signal(sqid, peer, `{"type":"candidate","to":"`+pb+`","seq":2,"gen":1,"payload":{"candidate":"candidate:1 1 udp 2 203.0.113.9 5000 typ host"}}`) |
| + c.signal(sqid, peer, `{"type":"offer","to":"`+pb+`","seq":3,"gen":9,"payload":{"sdp":"STALESECRET"}}`) |
| + c.signal(sqid, peer, `{"type":"bogus"}`) |
| + x.advance(time.Minute) |
| + x.ctl.TickForTest() |
| + logs := buf.String() |
| + for _, needle := range []string{"Secret title", "SDPSECRET", "203.0.113.9", "STALESECRET", tok, "paul@example.com"} { |
| + if strings.Contains(logs, needle) { |
| + t.Errorf("log carries %q", needle) |
| + } |
| + } |
| +} |
| +``` |
| + |
| +Add `newHarnessWithLogger(t, logger, members...)` and `newMemberBrowser(t, x, address) *client` (a browser that signs in as an allowlisted member) to the harness. |
| + |
| +- [ ] **Step 2: Implement** |
| + |
| +`ratelimit.go`: a per-key token bucket, `rate 30/s, burst 60`, `allow(key string, now time.Time) (ok bool, retryAfter time.Duration)`, entries evicted when idle for a minute (sweep on each call every thousand calls or on the tick). |
| + |
| +`signal.go`: |
| + |
| +```go |
| +package meet |
| + |
| +import ( |
| + "encoding/json" |
| + "errors" |
| + "io" |
| + "net/http" |
| + "strconv" |
| + |
| + "github.com/go-chi/chi/v5" |
| + |
| + "amadan.net/oficina/meet/internal/meet/hub" |
| +) |
| + |
| +const maxSignalBody = 64 * 1024 |
| + |
| +type signalBody struct { |
| + Peer hub.PeerID `json:"peer"` |
| + Type string `json:"type"` |
| + To hub.PeerID `json:"to,omitempty"` |
| + Seq uint64 `json:"seq,omitempty"` |
| + Gen uint64 `json:"gen,omitempty"` |
| + Payload json.RawMessage `json:"payload,omitempty"` |
| +} |
| + |
| +func (a *app) signal(w http.ResponseWriter, r *http.Request) { |
| + sqid := chi.URLParam(r, "sqid") |
| + session := a.sessionKey(r) |
| + r.Body = http.MaxBytesReader(w, r.Body, maxSignalBody) |
| + raw, err := io.ReadAll(r.Body) |
| + if err != nil { |
| + var tooBig *http.MaxBytesError |
| + if errors.As(err, &tooBig) { |
| + a.refuse(w, sqid, "", "", http.StatusRequestEntityTooLarge, "too-large") |
| + return |
| + } |
| + a.refuse(w, sqid, "", "", http.StatusBadRequest, "read") |
| + return |
| + } |
| + var body signalBody |
| + if err := json.Unmarshal(raw, &body); err != nil || body.Peer == "" || body.Type == "" { |
| + a.refuse(w, sqid, body.Peer, body.Type, http.StatusBadRequest, "shape") |
| + return |
| + } |
| + if ok, retry := a.signalLimiter.allow(string(body.Peer), a.store.now()); !ok { |
| + w.Header().Set("Retry-After", strconv.Itoa(int(retry.Seconds())+1)) |
| + a.refuse(w, sqid, body.Peer, body.Type, http.StatusTooManyRequests, "rate") |
| + return |
| + } |
| + if _, ok := a.hub.Find(sqid, session, body.Peer); !ok { |
| + a.refuse(w, sqid, body.Peer, body.Type, http.StatusNotFound, "binding") |
| + return |
| + } |
| + var herr error |
| + switch body.Type { |
| + case hub.TypeJoin: |
| + herr = a.hub.Join(r.Context(), sqid, session, body.Peer) |
| + case hub.TypeAdmit: |
| + var p struct{ Peer hub.PeerID `json:"peer"` } |
| + json.Unmarshal(body.Payload, &p) |
| + herr = a.hub.Admit(r.Context(), sqid, session, body.Peer, p.Peer) |
| + case hub.TypeLeave: |
| + herr = a.hub.Leave(sqid, session, body.Peer) |
| + case hub.TypeOffer, hub.TypeAnswer, hub.TypeCandidate: |
| + herr = a.hub.Relay(sqid, session, body.Peer, hub.Message{Type: body.Type, To: body.To, Seq: body.Seq, Gen: body.Gen, Payload: body.Payload}) |
| + case hub.TypeReset: |
| + herr = a.hub.Reset(sqid, session, body.Peer, body.To) |
| + default: |
| + a.refuse(w, sqid, body.Peer, body.Type, http.StatusBadRequest, "type") |
| + return |
| + } |
| + switch { |
| + case herr == nil, errors.Is(herr, hub.ErrRoomFull), errors.Is(herr, hub.ErrTooManyWaiting): |
| + w.WriteHeader(http.StatusOK) |
| + case errors.Is(herr, hub.ErrStaleGen): |
| + w.Header().Set("Content-Type", "application/json") |
| + w.WriteHeader(http.StatusConflict) |
| + json.NewEncoder(w).Encode(map[string]any{"error": "stale"}) |
| + a.logRefusal(sqid, body.Peer, body.Type, "stale-gen") |
| + case errors.Is(herr, hub.ErrNotAdmitted), errors.Is(herr, hub.ErrNotMember), errors.Is(herr, hub.ErrNotWaiting): |
| + a.refuse(w, sqid, body.Peer, body.Type, http.StatusForbidden, "authority") |
| + case errors.Is(herr, hub.ErrUnknownPeer), errors.Is(herr, hub.ErrNotYours): |
| + a.refuse(w, sqid, body.Peer, body.Type, http.StatusNotFound, "binding") |
| + default: |
| + a.logger.Error("meet: signal", "room", sqid, "type", body.Type, "err", herr.Error()) |
| + w.WriteHeader(http.StatusInternalServerError) |
| + } |
| +} |
| + |
| +func (a *app) refuse(w http.ResponseWriter, sqid string, peer hub.PeerID, typ string, status int, outcome string) { |
| + a.logRefusal(sqid, peer, typ, outcome) |
| + w.WriteHeader(status) |
| +} |
| + |
| +// logRefusal is the only signalling log line, and it carries no content. |
| +func (a *app) logRefusal(sqid string, peer hub.PeerID, typ, outcome string) { |
| + a.logger.Warn("meet: signal refused", "room", sqid, "peer", string(peer), "type", typ, "outcome", outcome) |
| +} |
| +``` |
| + |
| +Route: `r.Post("/m/{sqid}/signal", a.signal)` in the `requirePerson` group. `a.signalLimiter = newRateLimiter(30, 60)` in `build`. Note the `hub.Join` error for a `full` room is a 200 because the event is the answer; `ErrNotYours` on join (lost standing) is 404. |
| + |
| +- [ ] **Step 3: Run tests, gate, commit** |
| + |
| +```bash |
| +git add internal/meet internal/meettest |
| +git -c user.name="Paul Campbell" -c user.email="paul@rslw.com" commit -m "signal: POST with body and rate limits, dispatch, refusal logging without content |
| + |
| +Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> |
| +Claude-Session: https://claude.ai/code/session_01CzvnuKhLzudvq7jZ7SvVt2" |
| +``` |
| + |
| +--- |
| + |
| +### Task 6: Room page, removal and takeover confirm pages, page authorization, AGENTS.md exception |
| + |
| +**Files:** |
| +- Create: `internal/meet/handlers_room.go`, `internal/meet/templates/room.html` |
| +- Modify: `internal/meet/handlers_join.go` (holding authorized by standing or participation), `internal/meet/templates/join.html`, `internal/meet/app.go` (routes), `AGENTS.md` |
| +- Test: `internal/meettest/room_test.go` |
| + |
| +**Interfaces:** |
| +- Produces: `GET /m/{sqid}` (room page; standing or live participation; scriptless output is the holding header, the paragraph, and a link to the holding screen; scripted output is the grid container, controls, admission queue, all rendered by `room.js`); `GET /m/{sqid}/remove/{peer}` and `POST` (actor must hold an admitted member participation, checked at execution through `hub.Remove`); `GET /m/{sqid}/takeover` and `POST` (standing and a live participation for this session, `hub.Takeover`, then 303 to the room); page data for the room: `Sqid, Title, Starts, Name, Standing, IsMember, Floor, Boot`; a `data-*` block on the room page carrying `sqid`, `boot`, `self-name`, `is-member`, `peer` (empty) for `room.js`. |
| +- `AGENTS.md` "Screens work with JavaScript disabled" line becomes: "Screens work with JavaScript disabled, with one exception: browser media participation (the holding screen's camera preview and join button, and the call itself in the room) needs script; both pages render their title, standing, floor notice and navigation without it." |
| + |
| +- [ ] **Step 1: Write the failing tests** |
| + |
| +```go |
| +package meettest |
| + |
| +import ( |
| + "net/http" |
| + "net/url" |
| + "strings" |
| + "testing" |
| +) |
| + |
| +func TestRoomPageIsScriptlessHonestAndStandingGated(t *testing.T) { |
| + x, sqid, _ := memberWithMeeting(t) |
| + c := x.browser(t) |
| + c.signIn("paul@example.com") |
| + rec := c.get("/m/" + sqid) |
| + body := rec.Body.String() |
| + if rec.Code != http.StatusOK || !strings.Contains(body, "Board meeting") || !strings.Contains(body, `data-sqid="`+sqid+`"`) || !strings.Contains(body, "/m/"+sqid+"/join") { |
| + t.Fatalf("room page: %d\n%s", rec.Code, body) |
| + } |
| + if !strings.Contains(body, "<noscript>") || !strings.Contains(body, "needs JavaScript") { |
| + t.Fatal("room page lacks its scriptless paragraph") |
| + } |
| + stranger := newHarness(t).browser(t) |
| + stranger.signIn("nobody@example.com") |
| + if rec := stranger.get("/m/" + sqid); rec.Code != http.StatusNotFound { |
| + t.Fatalf("no standing: %d", rec.Code) |
| + } |
| +} |
| + |
| +func TestRemoveNeedsAConfirmPageAndAnAdmittedMemberActor(t *testing.T) { |
| + x, sqid, tok := memberWithMeeting(t) |
| + m := x.browser(t) |
| + m.signIn("paul@example.com") |
| + sm := m.stream("/m/" + sqid + "/stream") |
| + defer sm.close() |
| + pm, _ := sm.hello() |
| + m.signal(sqid, pm, `{"type":"join"}`) |
| + g := x.browser(t) |
| + g.get("/j/" + tok) |
| + g.signIn("guest@example.com") |
| + sg := g.stream("/m/" + sqid + "/stream") |
| + defer sg.close() |
| + pg, _ := sg.hello() |
| + g.signal(sqid, pg, `{"type":"join"}`) |
| + if rec := m.get("/m/" + sqid + "/remove/" + pg); rec.Code != http.StatusOK || !strings.Contains(rec.Body.String(), "Remove") { |
| + t.Fatalf("confirm page: %d", rec.Code) |
| + } |
| + if rec := g.post("/m/"+sqid+"/remove/"+pm, nil); rec.Code != http.StatusNotFound { |
| + t.Fatalf("waiting guest removing the member: %d", rec.Code) |
| + } |
| + if rec := m.post("/m/"+sqid+"/remove/"+pg, nil); rec.Code != http.StatusSeeOther { |
| + t.Fatalf("member removing: %d", rec.Code) |
| + } |
| + ev, data, ok := sg.nextOf("lost", 2) |
| + if !ok || !strings.Contains(string(data), `"removed"`) { |
| + t.Fatalf("removed guest got %s %s", ev, data) |
| + } |
| +} |
| + |
| +func TestTakeoverIsAConfirmedPostAndAGetEndsNothing(t *testing.T) { |
| + x, sqid, _ := memberWithMeeting(t) |
| + m := x.browser(t) |
| + m.signIn("paul@example.com") |
| + s1 := m.stream("/m/" + sqid + "/stream") |
| + defer s1.close() |
| + p1, _ := s1.hello() |
| + m.signal(sqid, p1, `{"type":"join"}`) |
| + if rec := m.get("/m/" + sqid + "/stream?takeover=1"); rec.Code != http.StatusConflict { |
| + t.Fatalf("a GET with a query parameter must not take over: %d", rec.Code) |
| + } |
| + if rec := m.get("/m/" + sqid + "/takeover"); rec.Code != http.StatusOK || !strings.Contains(rec.Body.String(), "another tab") { |
| + t.Fatalf("takeover confirm: %d", rec.Code) |
| + } |
| + req := httptestPost("/m/"+sqid+"/takeover", url.Values{}) |
| + req.Header.Set("Origin", "https://evil.example") |
| + if rec := m.do(req); rec.Code != http.StatusForbidden { |
| + t.Fatalf("cross-origin takeover: %d", rec.Code) |
| + } |
| + if rec := m.post("/m/"+sqid+"/takeover", nil); rec.Code != http.StatusSeeOther { |
| + t.Fatalf("takeover: %d", rec.Code) |
| + } |
| + if ev, data, ok := s1.nextOf("lost", 2); !ok || !strings.Contains(string(data), `"replaced"`) { |
| + t.Fatalf("old stream got %s %s", ev, data) |
| + } |
| + if rec := m.get("/m/" + sqid + "/stream?peer=" + p1); rec.Code != http.StatusGone { |
| + t.Fatalf("resume of the replaced peer: %d, want 410", rec.Code) |
| + } |
| + s2 := m.stream("/m/" + sqid + "/stream") |
| + defer s2.close() |
| + if p2, _ := s2.hello(); p2 == "" || p2 == p1 { |
| + t.Fatalf("fresh stream after takeover: %q", p2) |
| + } |
| +} |
| + |
| +func TestEndingTheMeetingEndsTheRoom(t *testing.T) { |
| + x, sqid, _ := memberWithMeeting(t) |
| + m := x.browser(t) |
| + m.signIn("paul@example.com") |
| + s := m.stream("/m/" + sqid + "/stream") |
| + defer s.close() |
| + p, _ := s.hello() |
| + m.signal(sqid, p, `{"type":"join"}`) |
| + m.post("/meetings/"+sqid+"/end", nil) |
| + if ev, data, ok := s.nextOf("lost", 2); !ok || !strings.Contains(string(data), `"ended"`) { |
| + t.Fatalf("got %s %s", ev, data) |
| + } |
| +} |
| +``` |
| + |
| +`sseReader.nextOf(event string, timeoutSeconds int)` skips events until the named one arrives. |
| + |
| +- [ ] **Step 2: Implement** |
| + |
| +`handlers_room.go`: |
| + |
| +```go |
| +package meet |
| + |
| +import ( |
| + "net/http" |
| + |
| + "github.com/go-chi/chi/v5" |
| + |
| + "amadan.net/oficina/meet/internal/meet/hub" |
| +) |
| + |
| +// roomAccess is §2.5's row for the two pages: standing, or a live |
| +// participation for this session in this room. |
| +func (a *app) roomAccess(w http.ResponseWriter, r *http.Request) (Meeting, Person, Standing, bool) { |
| + sqid := chi.URLParam(r, "sqid") |
| + m, ok, err := a.store.meetingBySqid(r.Context(), sqid) |
| + if err != nil || !ok || m.EndedAt != nil { |
| + ErrorPage(w, r, http.StatusNotFound, "") |
| + return Meeting{}, Person{}, StandingNone, false |
| + } |
| + p, _ := currentPerson(r) |
| + st, err := a.standingFor(r.Context(), p, m) |
| + if err != nil { |
| + ErrorPage(w, r, http.StatusInternalServerError, "") |
| + return Meeting{}, Person{}, StandingNone, false |
| + } |
| + if st == StandingNone { |
| + if _, live := a.hub.LiveForSession(sqid, a.sessionKey(r)); !live { |
| + ErrorPage(w, r, http.StatusNotFound, "") |
| + return Meeting{}, Person{}, StandingNone, false |
| + } |
| + st = StandingGuest |
| + } |
| + return m, p, st, true |
| +} |
| + |
| +func (a *app) room(w http.ResponseWriter, r *http.Request) { |
| + m, p, st, ok := a.roomAccess(w, r) |
| + if !ok { |
| + return |
| + } |
| + render(w, "room", a.page(r, map[string]any{ |
| + "Sqid": m.Sqid, "Title": a.store.title(m), "Starts": a.fmtStart(m.StartsAt), |
| + "Name": p.Name, "Standing": st.String(), "IsMember": st == StandingMember, |
| + "Floor": !a.cfg.RelayConfigured, "Boot": a.hub.Boot(), |
| + })) |
| +} |
| + |
| +func (a *app) confirmRemove(w http.ResponseWriter, r *http.Request) { |
| + m, _, st, ok := a.roomAccess(w, r) |
| + if !ok { |
| + return |
| + } |
| + if st != StandingMember { |
| + ErrorPage(w, r, http.StatusNotFound, "") |
| + return |
| + } |
| + target := chi.URLParam(r, "peer") |
| + render(w, "confirm", a.page(r, map[string]any{ |
| + "Title": "Remove this participant?", "Body": "They will be disconnected from \"" + a.store.title(m) + "\" and can only come back through the meeting link.", |
| + "Action": "/m/" + m.Sqid + "/remove/" + target, "Label": "Remove", "CancelHref": "/m/" + m.Sqid, "Sqid": m.Sqid, |
| + })) |
| +} |
| + |
| +func (a *app) remove(w http.ResponseWriter, r *http.Request) { |
| + m, _, _, ok := a.roomAccess(w, r) |
| + if !ok { |
| + return |
| + } |
| + session := a.sessionKey(r) |
| + actor, live := a.hub.LiveForSession(m.Sqid, session) |
| + if !live { |
| + ErrorPage(w, r, http.StatusNotFound, "") |
| + return |
| + } |
| + if err := a.hub.Remove(r.Context(), m.Sqid, session, actor, hub.PeerID(chi.URLParam(r, "peer"))); err != nil { |
| + ErrorPage(w, r, http.StatusNotFound, "") |
| + return |
| + } |
| + http.Redirect(w, r, "/m/"+m.Sqid, http.StatusSeeOther) |
| +} |
| + |
| +func (a *app) confirmTakeover(w http.ResponseWriter, r *http.Request) { |
| + m, _, _, ok := a.roomAccess(w, r) |
| + if !ok { |
| + return |
| + } |
| + render(w, "confirm", a.page(r, map[string]any{ |
| + "Title": "This meeting is open in another tab", "Body": "Use this tab instead? The other tab will leave the call.", |
| + "Action": "/m/" + m.Sqid + "/takeover", "Label": "Use this tab", "CancelHref": "/m/" + m.Sqid + "/join", "Sqid": m.Sqid, |
| + })) |
| +} |
| + |
| +func (a *app) takeover(w http.ResponseWriter, r *http.Request) { |
| + m, _, _, ok := a.roomAccess(w, r) |
| + if !ok { |
| + return |
| + } |
| + if _, err := a.hub.Takeover(m.Sqid, a.sessionKey(r)); err != nil { |
| + ErrorPage(w, r, http.StatusNotFound, "") |
| + return |
| + } |
| + http.Redirect(w, r, "/m/"+m.Sqid, http.StatusSeeOther) |
| +} |
| +``` |
| + |
| +Routes in the `requirePerson` group: `r.Get("/m/{sqid}", a.room)`, `r.Get("/m/{sqid}/remove/{peer}", a.confirmRemove)`, `r.Post("/m/{sqid}/remove/{peer}", a.remove)`, `r.Get("/m/{sqid}/takeover", a.confirmTakeover)`, `r.Post("/m/{sqid}/takeover", a.takeover)`. Change `holding` to use `roomAccess` too, and give `join.html` the `data-*` block and the scripted containers (Task 7 fills them). `room.html`: |
| + |
| +```html |
| +{{define "title"}}{{.Title}}{{end}} |
| +{{define "content"}} |
| +{{if .IsMember}}{{template "back-nav" dict "Href" "/" "Label" "Meetings"}}{{end}} |
| +{{template "page-header" dict "Title" .Title}} |
| +{{if .Starts}}<p>Starts {{.Starts}}.</p>{{end}} |
| +<p>You are {{.Name}}, in the call as a {{.Standing}}.</p> |
| +{{if .Floor}}<p rst-notice>This team's Meet has no relay yet. Calls usually connect only between people on the same network.</p>{{end}} |
| +<noscript><p>The call needs JavaScript to reach your camera and the other participants; everything else on this page works without it. <a href="/m/{{.Sqid}}/join">Back to the holding screen</a>.</p></noscript> |
| +<div id="room" data-sqid="{{.Sqid}}" data-boot="{{.Boot}}" data-self-name="{{.Name}}" data-is-member="{{if .IsMember}}1{{end}}" hidden> |
| + <div id="tiles" class="meet-grid"></div> |
| + <div id="controls" class="meet-controls"> |
| + <button type="button" id="mute" rst-btn>Mute</button> |
| + <button type="button" id="camera" rst-btn>Camera off</button> |
| + <button type="button" id="leave" rst-btn="danger">Leave</button> |
| + </div> |
| + <div id="queue" class="meet-queue" hidden><h2>Waiting to join</h2><ul id="knocks"></ul></div> |
| + <p id="status" class="meet-status" role="status"></p> |
| +</div> |
| +{{if .IsMember}}<p><a href="/meetings/{{.Sqid}}/end" target="_blank" rel="noopener">End the meeting</a></p>{{end}} |
| +<link rel="stylesheet" href="{{asset "static/meet.css"}}"> |
| +<script defer src="{{asset "static/room.js"}}"></script> |
| +{{end}} |
| +``` |
| + |
| +Add `"room"` to `pageNames`, create `static/meet.css` with the grid, tile, controls and queue rules (small, `rst-*` tokens for colour), and a placeholder `static/room.js` that reveals `#room` (Task 8 replaces it). `AGENTS.md` line as specified above. |
| + |
| +- [ ] **Step 3: Run tests, gate, commit** |
| + |
| +```bash |
| +git add internal AGENTS.md |
| +git -c user.name="Paul Campbell" -c user.email="paul@rslw.com" commit -m "room: page, removal and takeover confirm pages, participation-or-standing access, the scriptless exception |
| + |
| +Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> |
| +Claude-Session: https://claude.ai/code/session_01CzvnuKhLzudvq7jZ7SvVt2" |
| +``` |
| + |
| +--- |
| + |
| +### Task 7: devices.js: the holding screen's scripted half |
| + |
| +**Files:** |
| +- Create: `internal/meet/static/devices.js` |
| +- Modify: `internal/meet/templates/join.html` |
| + |
| +**Interfaces:** |
| +- Produces: on the holding page with script, a `<video id="preview" muted autoplay playsinline>`, three `<select>`s (camera, microphone, speaker; speaker only where `setSinkId` exists), two toggles (mute, camera off), and a "Join" button that navigates to `/m/{sqid}` with the chosen device ids and mute state stored under `localStorage["meet.devices"]` as `{camera, mic, speaker, muted, cameraOff}`; without script, the paragraph and no button. Behaviour: `navigator.mediaDevices.getUserMedia({audio: true, video: true})` on load, then `enumerateDevices`, populate selects (labels appear only after permission), restart the preview stream on selection change, stop all tracks on join. Errors (`NotAllowedError`, `NotFoundError`) show a sentence in `#status` and disable Join. |
| + |
| +- [ ] **Step 1: Template** |
| + |
| +Replace the Task 7-relevant part of `join.html`'s content: |
| + |
| +```html |
| +<noscript><p>The call needs JavaScript to reach your camera and the other participants; everything else on this page works without it.</p></noscript> |
| +<div id="devices" data-sqid="{{.Sqid}}" hidden> |
| + <video id="preview" muted autoplay playsinline class="meet-preview"></video> |
| + <div class="meet-pickers"> |
| + <label>Camera <select id="pick-camera"></select></label> |
| + <label>Microphone <select id="pick-mic"></select></label> |
| + <label id="speaker-row" hidden>Speaker <select id="pick-speaker"></select></label> |
| + </div> |
| + <div class="meet-controls"> |
| + <button type="button" id="toggle-mute" rst-btn aria-pressed="false">Mute</button> |
| + <button type="button" id="toggle-camera" rst-btn aria-pressed="false">Camera off</button> |
| + <button type="button" id="join" rst-btn="primary lg" disabled>Join the call</button> |
| + </div> |
| + <p id="status" class="meet-status" role="status"></p> |
| +</div> |
| +<link rel="stylesheet" href="{{asset "static/meet.css"}}"> |
| +<script defer src="{{asset "static/devices.js"}}"></script> |
| +``` |
| + |
| +Remove the slice-1 "Calls coming soon" paragraph. |
| + |
| +- [ ] **Step 2: devices.js** |
| + |
| +```js |
| +// devices.js: the holding screen. Enumerates devices, previews the |
| +// camera, remembers choices in localStorage, and hands off to the room. |
| +// Nothing here reaches the server. |
| +(() => { |
| + 'use strict'; |
| + const root = document.getElementById('devices'); |
| + if (!root || !navigator.mediaDevices) return; |
| + root.hidden = false; |
| + const KEY = 'meet.devices'; |
| + const $ = (id) => document.getElementById(id); |
| + const preview = $('preview'), pickCam = $('pick-camera'), pickMic = $('pick-mic'), |
| + pickSpk = $('pick-speaker'), spkRow = $('speaker-row'), mute = $('toggle-mute'), |
| + cam = $('toggle-camera'), join = $('join'), status = $('status'); |
| + let stream = null; |
| + const prefs = load(); |
| + |
| + function load() { |
| + try { return JSON.parse(localStorage.getItem(KEY) || '{}') || {}; } catch { return {}; } |
| + } |
| + function save() { |
| + try { localStorage.setItem(KEY, JSON.stringify(prefs)); } catch {} |
| + } |
| + function say(text) { status.textContent = text; } |
| + |
| + async function start() { |
| + stop(); |
| + const constraints = { |
| + audio: prefs.mic ? { deviceId: { exact: prefs.mic } } : true, |
| + video: prefs.camera ? { deviceId: { exact: prefs.camera } } : true, |
| + }; |
| + try { |
| + stream = await navigator.mediaDevices.getUserMedia(constraints); |
| + } catch (err) { |
| + if (prefs.camera || prefs.mic) { delete prefs.camera; delete prefs.mic; save(); return start(); } |
| + say(err.name === 'NotAllowedError' |
| + ? 'Meet needs permission to use your camera and microphone.' |
| + : 'No camera or microphone was found.'); |
| + join.disabled = true; |
| + return; |
| + } |
| + preview.srcObject = stream; |
| + applyToggles(); |
| + await fill(); |
| + join.disabled = false; |
| + say(''); |
| + } |
| + |
| + function stop() { |
| + if (stream) { for (const t of stream.getTracks()) t.stop(); stream = null; } |
| + } |
| + |
| + async function fill() { |
| + const devices = await navigator.mediaDevices.enumerateDevices(); |
| + const fillSelect = (sel, kind, chosen) => { |
| + sel.replaceChildren(); |
| + let n = 0; |
| + for (const d of devices) { |
| + if (d.kind !== kind) continue; |
| + n++; |
| + const o = document.createElement('option'); |
| + o.value = d.deviceId; o.textContent = d.label || `${kind.replace('input', '').replace('output', '')} ${n}`; |
| + if (d.deviceId === chosen) o.selected = true; |
| + sel.append(o); |
| + } |
| + return n; |
| + }; |
| + fillSelect(pickCam, 'videoinput', prefs.camera || currentId('video')); |
| + fillSelect(pickMic, 'audioinput', prefs.mic || currentId('audio')); |
| + if (typeof HTMLMediaElement.prototype.setSinkId === 'function' && fillSelect(pickSpk, 'audiooutput', prefs.speaker) > 0) { |
| + spkRow.hidden = false; |
| + } |
| + } |
| + |
| + function currentId(kind) { |
| + const t = stream && stream.getTracks().find((t) => t.kind === kind); |
| + return t ? t.getSettings().deviceId : ''; |
| + } |
| + |
| + function applyToggles() { |
| + if (!stream) return; |
| + for (const t of stream.getAudioTracks()) t.enabled = !prefs.muted; |
| + for (const t of stream.getVideoTracks()) t.enabled = !prefs.cameraOff; |
| + mute.setAttribute('aria-pressed', String(!!prefs.muted)); |
| + mute.textContent = prefs.muted ? 'Unmute' : 'Mute'; |
| + cam.setAttribute('aria-pressed', String(!!prefs.cameraOff)); |
| + cam.textContent = prefs.cameraOff ? 'Camera on' : 'Camera off'; |
| + } |
| + |
| + pickCam.addEventListener('change', () => { prefs.camera = pickCam.value; save(); start(); }); |
| + pickMic.addEventListener('change', () => { prefs.mic = pickMic.value; save(); start(); }); |
| + pickSpk.addEventListener('change', () => { prefs.speaker = pickSpk.value; save(); }); |
| + mute.addEventListener('click', () => { prefs.muted = !prefs.muted; save(); applyToggles(); }); |
| + cam.addEventListener('click', () => { prefs.cameraOff = !prefs.cameraOff; save(); applyToggles(); }); |
| + join.addEventListener('click', () => { stop(); location.assign('/m/' + root.dataset.sqid); }); |
| + navigator.mediaDevices.addEventListener?.('devicechange', fill); |
| + window.addEventListener('pagehide', stop); |
| + start(); |
| +})(); |
| +``` |
| + |
| +- [ ] **Step 3: Test** |
| + |
| +Add to `holding_test.go`: the holding page as a member contains `id="devices"`, `devices.js`, the `<noscript>` paragraph, and no longer contains "Calls coming soon". Run `make ci`, commit: |
| + |
| +```bash |
| +git add internal |
| +git -c user.name="Paul Campbell" -c user.email="paul@rslw.com" commit -m "holding screen: devices.js, preview, pickers, mute state, join |
| + |
| +Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> |
| +Claude-Session: https://claude.ai/code/session_01CzvnuKhLzudvq7jZ7SvVt2" |
| +``` |
| + |
| +--- |
| + |
| +### Task 8: room.js: SSE client, outbound queue, perfect negotiation, reconciliation, resume, terminal states, admission UI |
| + |
| +**Files:** |
| +- Replace: `internal/meet/static/room.js` |
| +- Modify: `internal/meet/static/meet.css` |
| + |
| +**Interfaces (the client contract, spec §4.5, §4.6, §5.1):** |
| +- State: `peer`, `boot`, `gen[peerId]`, `seqOut[peerId]`, `lastIn[peerId]`, `pcs[peerId] = {pc, polite, makingOffer, ignoreOffer, pendingCandidates[]}`, `queue[peerId] = [{seq, message}]` with one POST in flight per target, `local` stream from the devices prefs, `admitted` flag, `roster`. |
| +- Stream: open `EventSource('/m/{sqid}/stream')`; on `hello` set `peer`, compare `boot` to the stored one, then POST `join` once per fresh stream; on `error` (closed), resume with `?peer=&last=` after a 500 ms backoff, capped at 30 s; handle the resume refusals by reading the response when `EventSource` fails: since `EventSource` hides status codes, do the resume with `fetch` first (`HEAD`-like `GET` with `Accept: text/event-stream` is not possible), so: on error, `fetch(url + '&probe=1')` where the server answers the same 404/409/410 JSON or `204` when the resume would succeed (add `probe=1` handling to `stream`: run the refusal checks and return 204 without attaching), then open the `EventSource`. Document this in `stream.go`. |
| +- Terminal reasons: `replaced`, `standing`, `expired`, `removed`, `ended` and any 410 close every `pc`, stop the local stream, and show the sentence for that reason with the takeover link for `replaced`. `404` with a different `boot`: close every pc, fresh stream, join again. `404` same boot: same, but a `409` on the fresh stream shows the takeover link. `409` at first open: takeover link. |
| +- Roster reconciliation: on `roster`, close pcs for ids not present, create pcs for ids present without one, and if a roster `gen` for a peer is above the local one, rebuild that pc under the new gen. On `peer`, create. On `gone`, close. On `pair`, if `gen` above local, rebuild; else ignore. On `gap`, `reset` to that peer. |
| +- Perfect negotiation (W3C reference): `polite = self < other`; `onnegotiationneeded` → `makingOffer = true; setLocalDescription(); send offer`; on `offer`: `offerCollision = makingOffer || signalingState !== 'stable'; ignoreOffer = !polite && offerCollision; if ignoreOffer return; setRemoteDescription(); if offer: setLocalDescription(); send answer`; on `candidate`: `addIceCandidate` in try/catch, swallowed when `ignoreOffer`; candidates before a remote description are buffered. |
| +- Outbound: `send(to, type, payload)` assigns `seq = ++seqOut[to]`, `gen = gen[to]`, enqueues; the pump POSTs `/m/{sqid}/signal` with `{peer, type, to, seq, gen, payload}`, waits for 2xx; on 429 wait `Retry-After`; on other failures retry with backoff up to 30 s then close that pc and mark it for recreation on the next roster; on 409 stale gen drop the queue for that target (a `pair` or roster will rebuild). |
| +- Inbound dedupe: drop a targeted message with `seq <= lastIn[from]`; drop one with `gen < gen[from]`. |
| +- Admission UI: for members, `knock` adds to `#knocks` with an "Admit" button that sends `admit {peer}`; `knock-gone` removes; `roster.waiting` fills it; `full` shows "The room is full"; `state {waiting}` shows "Waiting for someone to let you in". |
| +- Controls: mute and camera toggle track `enabled`; leave sends `leave`, closes everything, navigates to the holding screen. |
| +- Tiles: one `<video autoplay playsinline>` per remote peer with the name; local tile muted; failed connection state shows "could not connect" with a Retry button that calls `pc.restartIce()`. |
| +- Expose a test hook: `window.meet = { state() { return {peer, admitted, peers: Object.keys(pcs), frames: <Map peer→frames received via requestVideoFrameCallback or getStats>} } }` used by the Playwright gate. |
| + |
| +- [ ] **Step 1: Write room.js** |
| + |
| +The implementer writes it to the contract above, keeping each concern in a named function (`connectStream`, `resume`, `handle(event)`, `reconcile(roster)`, `createPC(id)`, `closePC(id)`, `negotiate`, `onOffer`, `onAnswer`, `onCandidate`, `send`, `pump`, `terminal(reason)`, `renderTiles`, `renderQueue`), with a header comment naming the spec sections. No library. Target size 600 to 800 lines. The Opus-tier implementer is expected to read spec §4.5, §4.6 and §5.1 before writing. |
| + |
| +- [ ] **Step 2: Add the probe branch to `stream.go`** |
| + |
| +Before attaching a resume: if `r.URL.Query().Get("probe") == "1"`, run the same `Resume` refusal checks without attaching (add `hub.CanResume(sqid, session, peer) error` returning the same `ResumeError` or nil), answer 204 on nil. For a fresh stream with `probe=1`, answer 409 if the session has a live peer, else 204. Test in `stream_test.go`: probe answers 204, 409, 410, 404 in the four situations. |
| + |
| +- [ ] **Step 3: Gate and commit** |
| + |
| +`make ci` (Go tests only; the browser gate is Task 9). |
| + |
| +```bash |
| +git add internal |
| +git -c user.name="Paul Campbell" -c user.email="paul@rslw.com" commit -m "room.js: SSE client, queue with seq and gen, perfect negotiation, reconciliation, resume, terminal states |
| + |
| +Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> |
| +Claude-Session: https://claude.ai/code/session_01CzvnuKhLzudvq7jZ7SvVt2" |
| +``` |
| + |
| +--- |
| + |
| +### Task 9: The Playwright gate |
| + |
| +**Files:** |
| +- Create: `package.json`, `test/browser/call.mjs`, `internal/meettest/browser_test.go` (`//go:build browser`) |
| +- Modify: `Makefile` (`browser` target), `.gitignore` (`node_modules/`, `package-lock.json`) |
| + |
| +**Interfaces:** |
| +- `make browser`: `npm install --no-save playwright@1.62.1` then `go test -tags browser -run TestCallInARealBrowser ./internal/meettest/`. |
| +- `browser_test.go`: starts the app on `127.0.0.1:0` with `Origin` set to that address and `AllowedAddresses` `paul@example.com,paul@rslw.com`, a recording mailer; signs two members in over HTTP and captures their session cookies (name from `rastrillo/sessions` `CookieName`, `rastrillo_session` on http); creates a meeting; runs `node test/browser/call.mjs <origin> <sqid> <cookieName> <cookieA> <cookieB> <scenario>` for each scenario with a three-minute timeout, and fails on a non-zero exit. Scenarios in this slice: `devices` (holding screen enumerates and previews, join button enabled), `call` (two contexts join, each receives the other's video track with a non-zero `frames` count within 20 s), `renewal` (with `SetStreamTimings` shortened through a test-only header `X-Meet-Test-TTL: 2s` honoured only when `Config.Env` is development; the call stays connected across a stream renewal), `guest` (a third context signs in through the link as a guest, knocks, is admitted from context A's queue, receives video; asserts through `window.meet.state()` that it had no `peers` before admission), `removed` (A removes the guest through the confirm page; the guest sees the terminal sentence), `restart` (the Go test restarts the app on the same port and asserts both contexts rejoin and reconnect), `takeover` (a second page in context A's browser gets the 409 link, confirms, the first page shows the terminal sentence and does not reconnect), `crosssite` (a page on another origin navigates to the takeover URL by GET and nothing ends), `scriptless` (a context with JavaScript disabled loads both pages and finds the paragraph). |
| +- `call.mjs`: `chromium.launch({ args: ['--use-fake-device-for-media-stream', '--use-fake-ui-for-media-stream'] })`, one `browser.newContext()` per identity with the cookie set for the origin's host, `page.waitForFunction(() => window.meet && window.meet.state().frames...)`; prints one line per assertion and exits non-zero on failure. |
| + |
| +- [ ] **Step 1: Makefile and package.json** |
| + |
| +```make |
| +# The browser gate. Not in ci: the one gate the team runs on every push |
| +# must not depend on a toolchain the app itself does not use. Run it |
| +# before every review that touches a page. |
| +browser: |
| + npm install --no-save playwright@1.62.1 |
| + go test -tags browser -run TestCallInARealBrowser -v ./internal/meettest/ |
| +``` |
| + |
| +```json |
| +{ |
| + "name": "meet-browser-harness", |
| + "private": true, |
| + "description": "Playwright harness for the Meet browser gate. Not built; not published.", |
| + "devDependencies": { "playwright": "1.62.1" }, |
| + "scripts": { "test:browser": "go test -tags browser -run TestCallInARealBrowser ./internal/meettest/" } |
| +} |
| +``` |
| + |
| +- [ ] **Step 2: The Go driver and call.mjs** |
| + |
| +Write both to the contract. Frame counting in the page: after `ontrack`, attach the stream to a hidden video and use `video.requestVideoFrameCallback` to count; fall back to `getStats()` `framesDecoded` when unavailable. The `devices` scenario asserts the camera select has at least one option and `#join` is enabled within 10 s. |
| + |
| +- [ ] **Step 3: Run the gate and commit** |
| + |
| +Run: `make browser 2>&1 | tail -20` |
| +Expected: every scenario passes; record the run's output in the branch discussion. |
| + |
| +```bash |
| +git add Makefile package.json test internal/meettest/browser_test.go .gitignore |
| +git -c user.name="Paul Campbell" -c user.email="paul@rslw.com" commit -m "browser gate: Playwright driver, two-context call, guest, removal, restart, takeover, scriptless |
| + |
| +Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> |
| +Claude-Session: https://claude.ai/code/session_01CzvnuKhLzudvq7jZ7SvVt2" |
| +``` |
| + |
| +--- |
| + |
| +### Task 10: The remaining §13 hub gates and the captured-log test extension |
| + |
| +**Files:** |
| +- Modify: `internal/meet/hub/*_test.go`, `internal/meettest/signal_test.go` |
| + |
| +Cover, by name, what Tasks 1 to 5 did not: concurrent admissions (five goroutines admitting five waiting guests into one free slot, exactly one succeeds), a `pair` at or below the local generation ignored (hub side: `Reset` twice from one side yields gens 2 and 3 and a relay with gen 2 is stale), an admitted guest continuing past grant expiry on `signal`, `stream?peer=` and (stubbed until slice 3) `/ice` while being refused fresh entry (`/m/{sqid}/join` 404 after `advance(2h+1m)`), the 64-event buffer over the HTTP stream (`lost {reason: "buffer"}` then a successful resume), and a member dropped on the tick over HTTP (allowlist change through `TestHandle.SetAllowlist(addrs...)`). Then `make ci` and commit `tests: remaining slice-2 hub and stream gates`. |
| + |
| +--- |
| + |
| +### Task 11: Copy catalogue and branch record |
| + |
| +**Files:** |
| +- Create: `docs/superpowers/copy/2026-09-10-slice-2-strings.md` |
| + |
| +Every user-facing string in Tasks 6 to 8, verbatim, with its location, marked "copy review waived for this round per suite #5; catalogued for the later pass". Commit `docs: slice 2 strings`. |
| + |
| +--- |
| + |
| +## Self-review |
| + |
| +**Spec coverage (§12 item 2):** `devices.js` (7); hub with peer binding (1), participation (1, 2), cap (2), limits (1, 5), resume with admission-branched output (3, 4), replay ring, `gap`, pair reset (3), terminal `replaced` (2, 3, 6), the tick (1, 2); stream and `hello` (4); every §4.3 message except `chat` and `share` (2, 3, 5); removal confirm page (6); `room.js` with perfect negotiation and the §4.6 queue (8); participation transitions (2, 10); `AGENTS.md` exception (6); `CARLOS_RELAY` absent (unchanged); the browser gate (9). §13 slice-2 Go gates are spread across Tasks 1, 2, 3, 4, 5, 6 and 10 and named there; the Playwright gates are Task 9's scenarios. `/ice` belongs to slice 3 and is only stubbed in Task 10's continuation test. |
| + |
| +**Placeholders:** Tasks 8 and 9 give a contract and function list rather than full code because they are Opus-tier design tasks; every behaviour they must have is named, with the spec section, and the gate that proves it is in Task 9. Everything else has code. |
| + |
| +**Type consistency:** `hub.Message` fields, `PeerID`, `Kind`, `Reason`, the error variables and `ResumeError` are declared in Task 1 and used unchanged after; `Hub` methods `Open`, `Find`, `LiveForSession`, `Detach`, `Tick` (1), `Join`, `Admit`, `Leave`, `Remove`, `Takeover`, `End`, `rosterFor`, `genLocked`, `admitLocked` (2), `Relay`, `Reset`, `Resume`, `ringLocked` (3), `CanResume` (8); app fields `hub`, `streamHeartbeat`, `streamTTL`, `signalLimiter` (4, 5); `TestHandle.TickForTest`, `SetStreamTimings`, `SetAllowlist` (4, 10); harness `stream`, `signal`, `sseReader.next/nextOf/hello/waitClosed/close`, `newHarnessWithLogger`, `newMemberBrowser`, per-harness `origin` (4, 5). |
| + |
| +**Rulings for the controller:** the session key for the hub is the session subject plus the session's creation instant, since rastrillo does not expose the cookie hash; two tabs of one browser share it, which is the spec's intent. The resume probe (`?probe=1`) is an addition to the spec's §4.5 needed because `EventSource` hides status codes; record it as a spec revision note in the branch discussion. |