amadan
Carry each peer's mute and camera-off state in the room
A peer reports {muted, cameraOff} with a new `media` signal. The hub
keeps it on the participation, puts it in every roster and peer event,
and fans out only a change, so late joiners, resumes and restarts need
no replay rule. Unreported peers carry no media key: unknown, never a
guessed "on". callscore reports the host's state on admission, whenever
a roster disagrees, and on each change; the bridge exposes SetMedia.
Spec §15.19.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
12 files changed,
+996
−10
callscore/media_test.go+177 −0callscore/room.go+66 −2docs/superpowers/specs/2026-09-07-meet-design.md+68 −0internal/calls/hub/admission.go+2 −2internal/calls/hub/hub.go+4 −0internal/calls/hub/media.go+63 −0internal/calls/hub/media_test.go+361 −0internal/calls/hub/messages.go+2 −0internal/calls/signal.go+58 −6internal/callstest/media_test.go+170 −0native/bridge/bridge.go+5 −0protocol/messages.go+20 −0
diff --git a/callscore/media_test.go b/callscore/media_test.go| new file mode 100644 |
| index 0000000..53c1d39 |
| --- /dev/null |
| +++ b/callscore/media_test.go |
| @@ -0,0 +1,177 @@ |
| +package callscore |
| + |
| +import ( |
| + "encoding/json" |
| + "net/http" |
| + "net/http/httptest" |
| + "testing" |
| + "time" |
| + |
| + "amadan.net/oficina/calls/protocol" |
| +) |
| + |
| +// mediaRoom is a Room seeded as connected-but-not-admitted, with every |
| +// POST it makes captured, so the tests below can drive receive() by hand |
| +// and see exactly which reports go out. |
| +func mediaRoom(t *testing.T) (*Room, chan protocol.Signal) { |
| + t.Helper() |
| + posted := make(chan protocol.Signal, 16) |
| + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { |
| + var s protocol.Signal |
| + _ = json.NewDecoder(req.Body).Decode(&s) |
| + posted <- s |
| + })) |
| + t.Cleanup(srv.Close) |
| + r, err := New(srv.URL, "room", srv.Client(), func(Event) {}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + r.started = true |
| + r.state.Peer = "self" |
| + t.Cleanup(r.Close) |
| + return r, posted |
| +} |
| + |
| +func nextMedia(t *testing.T, posted chan protocol.Signal) protocol.MediaState { |
| + t.Helper() |
| + select { |
| + case s := <-posted: |
| + if s.Type != protocol.TypeMedia || s.To != "" || s.Seq != 0 || s.Gen != 0 || s.Peer != "self" { |
| + t.Fatalf("posted %+v, want an untargeted media report", s) |
| + } |
| + var m protocol.MediaState |
| + if err := json.Unmarshal(s.Payload, &m); err != nil { |
| + t.Fatal(err) |
| + } |
| + return m |
| + case <-time.After(time.Second): |
| + t.Fatal("no media report was posted") |
| + } |
| + return protocol.MediaState{} |
| +} |
| + |
| +func nothingPosted(t *testing.T, posted chan protocol.Signal, why string) { |
| + t.Helper() |
| + select { |
| + case s := <-posted: |
| + t.Fatalf("%s, but %+v was posted", why, s) |
| + case <-time.After(150 * time.Millisecond): |
| + } |
| +} |
| + |
| +func roster(peers ...protocol.RosterPeer) protocol.Message { |
| + return protocol.Message{Type: protocol.TypeRoster, Roster: &protocol.Roster{Peers: peers}} |
| +} |
| + |
| +// §15.19: the host's own state is reported when it is admitted, after a |
| +// resume whose roster disagrees, and on every change while admitted — |
| +// and not while waiting, where there is nobody to tell. |
| +func TestOwnMediaIsReportedOnAdmissionAndOnChange(t *testing.T) { |
| + r, posted := mediaRoom(t) |
| + |
| + // Muted in the lobby: remembered, not sent. |
| + if err := r.SetMedia(true, false); err != nil { |
| + t.Fatal(err) |
| + } |
| + nothingPosted(t, posted, "a client that is not admitted has nobody to tell") |
| + |
| + // Admitted, with no media for ourselves in the roster: the hub has |
| + // never heard from this participation, so the report goes now. |
| + if err := r.receive(roster(protocol.RosterPeer{Peer: "self"}, protocol.RosterPeer{Peer: "other", Gen: 1})); err != nil { |
| + t.Fatal(err) |
| + } |
| + if m := nextMedia(t, posted); !m.Muted || m.CameraOff { |
| + t.Fatalf("admission reported %+v, want muted", m) |
| + } |
| + |
| + // A later roster that already says what we say: nothing to send. |
| + if err := r.receive(roster(protocol.RosterPeer{Peer: "self", Media: &protocol.MediaState{Muted: true}}, protocol.RosterPeer{Peer: "other", Gen: 1})); err != nil { |
| + t.Fatal(err) |
| + } |
| + nothingPosted(t, posted, "a roster that agrees needs no report") |
| + |
| + // A change while admitted goes at once. |
| + if err := r.SetMedia(true, true); err != nil { |
| + t.Fatal(err) |
| + } |
| + if m := nextMedia(t, posted); !m.Muted || !m.CameraOff { |
| + t.Fatalf("change reported %+v, want muted and camera off", m) |
| + } |
| + |
| + // A resume whose roster is behind — the report was lost while the |
| + // stream was down — is corrected. |
| + if err := r.receive(roster(protocol.RosterPeer{Peer: "self", Media: &protocol.MediaState{Muted: true}}, protocol.RosterPeer{Peer: "other", Gen: 1})); err != nil { |
| + t.Fatal(err) |
| + } |
| + if m := nextMedia(t, posted); !m.Muted || !m.CameraOff { |
| + t.Fatalf("resume reported %+v, want the host's current state", m) |
| + } |
| +} |
| + |
| +// Other peers' state arrives two ways — in the roster and as `media` |
| +// events — and both land on the snapshot's peers, where the native UI |
| +// reads it. A peer that never reported stays nil: unknown, not "on". |
| +func TestRemoteMediaReachesTheSnapshot(t *testing.T) { |
| + r, posted := mediaRoom(t) |
| + if err := r.receive(roster( |
| + protocol.RosterPeer{Peer: "self"}, |
| + protocol.RosterPeer{Peer: "old", Gen: 1}, |
| + protocol.RosterPeer{Peer: "new", Gen: 1, Media: &protocol.MediaState{Muted: true}}, |
| + )); err != nil { |
| + t.Fatal(err) |
| + } |
| + <-posted // our own first report; covered above |
| + before := r.Snapshot() |
| + |
| + if err := r.receive(protocol.Message{Type: protocol.TypeMedia, From: "new", Payload: json.RawMessage(`{"muted":false,"cameraOff":true}`)}); err != nil { |
| + t.Fatal(err) |
| + } |
| + // A malformed one, and one from nobody, change nothing. |
| + _ = r.receive(protocol.Message{Type: protocol.TypeMedia, From: "old", Payload: json.RawMessage(`"nope"`)}) |
| + _ = r.receive(protocol.Message{Type: protocol.TypeMedia, Payload: json.RawMessage(`{"muted":true,"cameraOff":true}`)}) |
| + |
| + after := r.Snapshot() |
| + byID := map[protocol.PeerID]*protocol.MediaState{} |
| + for _, p := range after.Peers { |
| + byID[p.Peer] = p.Media |
| + } |
| + if m := byID["new"]; m == nil || m.Muted || !m.CameraOff { |
| + t.Fatalf("new = %+v, want unmuted with camera off", m) |
| + } |
| + if byID["old"] != nil { |
| + t.Fatalf("a peer that never reported must stay unknown, got %+v", byID["old"]) |
| + } |
| + // The snapshot taken before is untouched: the event replaced the |
| + // pointer rather than writing through the one it had handed out. |
| + for _, p := range before.Peers { |
| + if p.Peer == "new" && (p.Media == nil || !p.Media.Muted || p.Media.CameraOff) { |
| + t.Fatalf("an earlier snapshot changed under its holder: %+v", p.Media) |
| + } |
| + } |
| + nothingPosted(t, posted, "hearing about somebody else sends nothing") |
| +} |
| + |
| +// A fresh join after a restart starts a new participation the hub knows |
| +// nothing about, so the host's state — kept across it — is reported again. |
| +func TestOwnMediaSurvivesAFreshJoin(t *testing.T) { |
| + r, posted := mediaRoom(t) |
| + _ = r.SetMedia(true, false) |
| + _ = r.receive(roster(protocol.RosterPeer{Peer: "self"})) |
| + nextMedia(t, posted) |
| + |
| + r.fresh() |
| + r.mu.Lock() |
| + r.state.Peer = "self2" |
| + r.mu.Unlock() |
| + _ = r.receive(roster(protocol.RosterPeer{Peer: "self2"})) |
| + select { |
| + case s := <-posted: |
| + var m protocol.MediaState |
| + _ = json.Unmarshal(s.Payload, &m) |
| + if s.Type != protocol.TypeMedia || s.Peer != "self2" || !m.Muted { |
| + t.Fatalf("after a restart posted %+v (%+v), want the kept muted state from the new peer", s, m) |
| + } |
| + case <-time.After(time.Second): |
| + t.Fatal("no report after the fresh join") |
| + } |
| +} |
diff --git a/callscore/room.go b/callscore/room.go| index 01d240d..a02c121 100644 |
| --- a/callscore/room.go |
| +++ b/callscore/room.go |
| @@ -56,6 +56,10 @@ type Room struct { |
| lanes map[protocol.PeerID]*lane |
| blocked map[protocol.PeerID]bool |
| iceWake chan struct{} |
| + // media is this client's own microphone and camera, as the host last |
| + // set it (§15.19). It outlives a fresh join: a restart does not |
| + // unmute anybody. |
| + media protocol.MediaState |
| } |
| func New(origin, room string, authenticated *http.Client, callback func(Event)) (*Room, error) { |
| @@ -226,6 +230,60 @@ func pause(ctx context.Context, d time.Duration) bool { |
| return true |
| } |
| } |
| + |
| +// SetMedia records this client's own microphone and camera state and, |
| +// once admitted, reports it (§15.19). Before admission there is nobody |
| +// to tell; the roster that admits this client is compared against it |
| +// and the report goes then, as it does after every resume and restart. |
| +func (r *Room) SetMedia(muted, cameraOff bool) error { |
| + r.mu.Lock() |
| + r.media = protocol.MediaState{Muted: muted, CameraOff: cameraOff} |
| + report := r.state.Phase == "admitted" && !r.stopped |
| + r.mu.Unlock() |
| + if !report { |
| + return nil |
| + } |
| + return r.sendMedia() |
| +} |
| + |
| +func (r *Room) sendMedia() error { |
| + r.mu.Lock() |
| + b, err := json.Marshal(r.media) |
| + r.mu.Unlock() |
| + if err != nil { |
| + return err |
| + } |
| + return r.Send(protocol.TypeMedia, "", 0, string(b)) |
| +} |
| + |
| +// mediaStaleLocked says whether the hub's record of this client — its |
| +// own entry in the roster just applied — differs from what the host last |
| +// set, or is missing, which is how a fresh participation, a restarted |
| +// server, or a report lost while the stream was down all look. |
| +func (r *Room) mediaStaleLocked() bool { |
| + for _, p := range r.state.Peers { |
| + if p.Peer == r.state.Peer { |
| + return p.Media == nil || *p.Media != r.media |
| + } |
| + } |
| + return false |
| +} |
| + |
| +// applyMediaLocked is a `media` event: one peer's new state, replacing |
| +// its roster entry's pointer rather than writing through it, because |
| +// snapshots already handed out share the old one. |
| +func (r *Room) applyMediaLocked(m protocol.Message) { |
| + var next protocol.MediaState |
| + if m.From == "" || json.Unmarshal(m.Payload, &next) != nil { |
| + return |
| + } |
| + for i, p := range r.state.Peers { |
| + if p.Peer == m.From { |
| + r.state.Peers[i].Media = &next |
| + } |
| + } |
| +} |
| + |
| func (r *Room) generationLocked(peer protocol.PeerID) uint64 { |
| for _, p := range r.state.Peers { |
| if p.Peer == peer { |
| @@ -244,7 +302,7 @@ func (r *Room) receive(m protocol.Message) error { |
| r.mu.Unlock() |
| return context.Canceled |
| } |
| - join, ice, reset := false, false, protocol.PeerID("") |
| + join, ice, reset, report := false, false, protocol.PeerID(""), false |
| switch m.Type { |
| case protocol.TypeHello: |
| join = r.state.Peer == "" |
| @@ -278,6 +336,7 @@ func (r *Room) receive(m protocol.Message) error { |
| r.state.Waiting = append([]protocol.RosterPeer{}, m.Roster.Waiting...) |
| r.state.Phase = "admitted" |
| ice = true |
| + report = r.mediaStaleLocked() |
| case protocol.TypePeer: |
| if m.Roster != nil { |
| for _, p := range m.Roster.Peers { |
| @@ -307,6 +366,8 @@ func (r *Room) receive(m protocol.Message) error { |
| delete(r.blocked, m.Peer) |
| } |
| } |
| + case protocol.TypeMedia: |
| + r.applyMediaLocked(m) |
| case protocol.TypeGap: |
| reset = m.From |
| case protocol.TypeLost: |
| @@ -354,6 +415,9 @@ func (r *Room) receive(m protocol.Message) error { |
| if reset != "" { |
| _ = r.Send(protocol.TypeReset, string(reset), 0, "") |
| } |
| + if report { |
| + _ = r.sendMedia() |
| + } |
| return nil |
| } |
| func without(peers []protocol.RosterPeer, id protocol.PeerID) []protocol.RosterPeer { |
| @@ -380,7 +444,7 @@ func (r *Room) Send(typ, to string, expectedGen uint64, payload string) error { |
| return errors.New("room is not connected") |
| } |
| switch typ { |
| - case protocol.TypeJoin, protocol.TypeAdmit, protocol.TypeReset, protocol.TypeChat, protocol.TypeShare, protocol.TypeOffer, protocol.TypeAnswer, protocol.TypeCandidate: |
| + case protocol.TypeJoin, protocol.TypeAdmit, protocol.TypeReset, protocol.TypeChat, protocol.TypeShare, protocol.TypeMedia, protocol.TypeOffer, protocol.TypeAnswer, protocol.TypeCandidate: |
| default: |
| return errors.New("unsupported room command") |
| } |
diff --git a/docs/superpowers/specs/2026-09-07-meet-design.md b/docs/superpowers/specs/2026-09-07-meet-design.md| index 67a4798..9e3a680 100644 |
| --- a/docs/superpowers/specs/2026-09-07-meet-design.md |
| +++ b/docs/superpowers/specs/2026-09-07-meet-design.md |
| @@ -567,6 +567,7 @@ a multi-host fleet; that is noted, not solved. |
| | `reset {to}` | an admitted peer | both peers of the pair | Under the hub lock, increment the pair's generation and send `pair {peer, gen}` to both, into their replay rings. Idempotent in effect: two resets yield one higher generation both sides converge on. | |
| | `chat {text}` | an admitted peer | all admitted peers | Fan out with the sender's name. 2 KB. Not stored. | |
| | `share {on}` | an admitted peer | all admitted peers | Fan out so tiles can be labelled. | |
| +| `media {muted, cameraOff}` (2026-09-23, §15.19) | an admitted peer | all other admitted peers | Record on the participation, carry in every later `roster` and `peer`; fan out only a change. | |
| | `hello`, `roster`, `peer`, `gone`, `knock`, `knock-gone`, `full`, `ended`, `lost`, `pair`, `gap`, `state` | server | as named | `roster`, `peer`, `pair` and `gap` go only to admitted peers; a waiting guest receives `hello`, `state`, then `roster` on admission, or `lost`, or `ended`, and nothing else. | |
| `offer`, `answer` and `candidate` also carry `seq` and `gen` (§4.6); |
| @@ -2227,3 +2228,70 @@ match on 2026-09-14, in the same commit. |
| state as of that observation, not asserted as current here. This |
| slice does not touch that queue or announce a deploy of its own, |
| which is checkable from this diff alone. |
| + |
| +### 15.19 §4.3, §4.5 — media state: who is muted, whose camera is off (2026-09-23) |
| + |
| +**Why.** Paul, 2026-09-23: "add mute indicators so that folks can see |
| +when someone is muted." Until now muting disabled the local audio track |
| +and sent silence, and nothing on the wire said so; nobody else could |
| +tell a muted person from a broken microphone. Camera off had the same |
| +gap, and a remote tile showed a black or frozen frame for it. |
| + |
| +**Decision.** A new message type, `media`, that goes both ways and is |
| +**state, not an event**: |
| + |
| +- Upstream, `POST /signal {"peer","type":"media","payload": |
| + {"muted":bool,"cameraOff":bool}}`. Both keys, spelled exactly, each |
| + once, both booleans, nothing else, judged on the raw bytes as §15.15 |
| + judges chat and share (a token walk, not a map). The raw payload is |
| + capped at 128 bytes before decoding (`413 {"error":"too-large"}`); |
| + any other shape, a `to`, a `seq` or a `gen` is |
| + `400 {"error":"bad-request"}`. The sender's authority is `Fanout`'s |
| + (§15.15): admitted, bound to this room and session, a member's |
| + standing re-derived first; a waiting or unjoined participation gets |
| + the same 404 an unknown peer gets. |
| +- The hub records the value on the participation (`Participation.Media`, |
| + nil until the first report). **Every roster and every `peer` event |
| + carries it**, as `media` on the `RosterPeer`, for every admitted peer |
| + including the recipient itself. That is what makes a late joiner, a |
| + resume (§4.5) and a restart correct without any new replay rule: |
| + nothing about `media` goes in the replay ring, because the roster a |
| + resume is answered with is already the authority, exactly as it is |
| + for who is present. |
| +- Downstream, a **change** is fanned out to every other admitted peer as |
| + `{type:"media", from, name, payload:{"muted","cameraOff"}}`, the |
| + envelope rebuilt from the sender's participation and the payload |
| + re-encoded from the two validated booleans. A report equal to what the |
| + hub already holds is accepted (200) and sends nothing, so clients may |
| + repeat it freely. |
| + |
| +**What a client does.** It keeps its own state locally and reports it |
| +whenever a roster's entry for itself disagrees or is missing — which is |
| +admission, a resume after a report was lost, and a fresh join after a |
| +restart — and at once on each toggle while admitted. Before admission it |
| +sends nothing. `muted` means the person's microphone is not being sent |
| +(toggled off, or there is no microphone track at all); `cameraOff` the |
| +same for the camera. Both are the sender's own claim about what it is |
| +choosing to send. The server never sees media (§0.1) and cannot check |
| +them, and nothing depends on them being true: a peer that lies about |
| +being muted is heard anyway. |
| + |
| +**Unknown is not muted.** A `RosterPeer` with no `media` key is a peer |
| +that has not reported — every client that predates this section, and |
| +every peer for the moment between admission and its first report. |
| +Clients show that as neither muted nor camera-off. The zero value is |
| +not used for it, because `{"muted":false,"cameraOff":false}` is a real |
| +report and must be told apart from silence. |
| + |
| +**Mixed versions.** An old client ignores the `media` key on a roster |
| +and the `media` event type (the web client's dispatch and the native |
| +core both drop unknown types), so it never breaks and simply never |
| +shows or sends indicators; the people talking to it see it as unknown. |
| +A new client against an old server has its report refused |
| +`400 {"error":"bad-request"}` as an unknown type; it does not retry a |
| +400 and shows nothing. Server and clients should still deploy together |
| +for the feature to exist, but no order of deployment breaks a call. |
| + |
| +**Not logged, not stored.** Like every accepted message, a report logs |
| +nothing; a refused one logs type, room, peer and outcome only (§4.4). |
| +The state lives on the in-memory participation and dies with it (§2.2). |
diff --git a/internal/calls/hub/admission.go b/internal/calls/hub/admission.go| index f736614..0bc7871 100644 |
| --- a/internal/calls/hub/admission.go |
| +++ b/internal/calls/hub/admission.go |
| @@ -31,7 +31,7 @@ func (h *Hub) rosterFor(r *room, viewer *Participation) *Roster { |
| if !p.Admitted { |
| continue |
| } |
| - rp := RosterPeer{Peer: p.Peer, Name: p.Name, Kind: p.Kind.String()} |
| + rp := RosterPeer{Peer: p.Peer, Name: p.Name, Kind: p.Kind.String(), Media: p.Media} |
| if p.Peer != viewer.Peer { |
| rp.Gen = h.genLocked(r, viewer.Peer, p.Peer) |
| } |
| @@ -77,7 +77,7 @@ func (h *Hub) admitLocked(r *room, p *Participation) bool { |
| // generation with the newcomer; browsers read it from |
| // roster.peers[0]. |
| q.sub.send(Message{Type: TypePeer, Peer: p.Peer, Name: p.Name, |
| - Roster: &Roster{Peers: []RosterPeer{{Peer: p.Peer, Name: p.Name, Kind: p.Kind.String(), Gen: h.genLocked(r, q.Peer, p.Peer)}}}}) |
| + Roster: &Roster{Peers: []RosterPeer{{Peer: p.Peer, Name: p.Name, Kind: p.Kind.String(), Gen: h.genLocked(r, q.Peer, p.Peer), Media: p.Media}}}}) |
| } |
| } |
| if wasWaiting { |
diff --git a/internal/calls/hub/hub.go b/internal/calls/hub/hub.go| index 0ef0aed..59c3619 100644 |
| --- a/internal/calls/hub/hub.go |
| +++ b/internal/calls/hub/hub.go |
| @@ -98,6 +98,10 @@ type Participation struct { |
| Joined bool // has posted join (waiting or admitted) |
| JoinedAt time.Time |
| OpenedAt time.Time |
| + // Media is the peer's last report of its own microphone and camera |
| + // (§15.19), nil until it makes one. It lives and dies with the |
| + // participation, so a resume keeps it and a fresh join starts unknown. |
| + Media *MediaState |
| sub *Sub |
| resumeUntil time.Time // zero while a stream is attached |
diff --git a/internal/calls/hub/media.go b/internal/calls/hub/media.go| new file mode 100644 |
| index 0000000..6504a0d |
| --- /dev/null |
| +++ b/internal/calls/hub/media.go |
| @@ -0,0 +1,63 @@ |
| +package hub |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| +) |
| + |
| +// SetMedia is §15.19: an admitted peer reports its own microphone and |
| +// camera state. The hub records it on the participation, where every |
| +// later roster — admission, resume, a kind change — carries it, and |
| +// fans a change out to every other admitted peer as |
| +// {type: "media", from, name, payload: MediaState}, the envelope |
| +// rebuilt from the sender's participation exactly as Fanout's is. |
| +// |
| +// It is state, not an event. A report that changes nothing is accepted |
| +// and sends nothing, so a client may repeat it freely — on every roster, |
| +// after every resume — and the room hears only changes. Nothing goes in |
| +// a replay ring: a peer that missed a `media` while detached learns the |
| +// current state from the roster its resume is answered with (§4.5), |
| +// which is the authority a ring would only have approximated. |
| +// |
| +// The sender's authority is Fanout's: a member re-derives standing |
| +// first (§15.12) and one whose standing has gone is ended here and |
| +// answered ErrNotYours; a guest never consults standing (ruling 14); a |
| +// participation not yet admitted is refused ErrNotAdmitted, because a |
| +// waiting guest has nobody to tell and learns nothing about the room by |
| +// asking (ruling 9). |
| +func (h *Hub) SetMedia(ctx context.Context, sqid, session string, from PeerID, m MediaState) error { |
| + gone, err := h.memberStandingGone(ctx, sqid, session, from, true) |
| + if err != nil { |
| + return err |
| + } |
| + if gone { |
| + return ErrNotYours |
| + } |
| + 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 |
| + } |
| + if p.Media != nil && *p.Media == m { |
| + return nil |
| + } |
| + // A fresh value, never a write through the old pointer: rosters |
| + // already sent hold that one. |
| + next := m |
| + p.Media = &next |
| + payload, err := json.Marshal(next) |
| + if err != nil { |
| + return err |
| + } |
| + out := Message{Type: TypeMedia, From: p.Peer, Name: p.Name, Payload: payload} |
| + h.broadcastLocked(r, out, func(q *Participation) bool { return q.Admitted && q.Peer != p.Peer }) |
| + return nil |
| +} |
diff --git a/internal/calls/hub/media_test.go b/internal/calls/hub/media_test.go| new file mode 100644 |
| index 0000000..6890c14 |
| --- /dev/null |
| +++ b/internal/calls/hub/media_test.go |
| @@ -0,0 +1,361 @@ |
| +package hub |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "io" |
| + "log/slog" |
| + "sync" |
| + "testing" |
| + "time" |
| +) |
| + |
| +// rosterMedia finds one peer's media in a roster, and whether the peer |
| +// is in it at all. |
| +func rosterMedia(ro *Roster, peer PeerID) (*MediaState, bool) { |
| + if ro == nil { |
| + return nil, false |
| + } |
| + for _, rp := range ro.Peers { |
| + if rp.Peer == peer { |
| + return rp.Media, true |
| + } |
| + } |
| + return nil, false |
| +} |
| + |
| +// TestSetMediaFansOutAChangeAndNothingElse is §15.19's delivery: every |
| +// other admitted peer hears a change, in an envelope the hub built, and |
| +// nobody else hears anything — not the sender, not a waiting guest, and |
| +// nothing in any ring. A repeat of the same state is accepted and silent. |
| +func TestSetMediaFansOutAChangeAndNothingElse(t *testing.T) { |
| + now := time.Date(2026, 9, 23, 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) |
| + g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest) |
| + h.Join(ctx, "r1", "sg", g.Peer) // waiting |
| + drain(a.Sub) |
| + drain(b.Sub) |
| + drain(g.Sub) |
| + |
| + if err := h.SetMedia(ctx, "r1", "sa", a.Peer, MediaState{Muted: true}); err != nil { |
| + t.Fatal(err) |
| + } |
| + got := drain(b.Sub) |
| + if len(got) != 1 { |
| + t.Fatalf("Bob got %d events, want 1: %v", len(got), types(got)) |
| + } |
| + m := got[0] |
| + if m.Type != TypeMedia || m.From != a.Peer || m.Name != "Ann" { |
| + t.Fatalf("Bob got %+v, want media from Ann", m) |
| + } |
| + if string(m.Payload) != `{"muted":true,"cameraOff":false}` { |
| + t.Fatalf("payload %s, want the canonical two booleans", m.Payload) |
| + } |
| + if m.To != "" || m.Seq != 0 || m.Gen != 0 || m.Roster != nil { |
| + t.Fatalf("Bob got %+v, want nothing but type, from, name and payload", m) |
| + } |
| + if n := len(drain(a.Sub)); n != 0 { |
| + t.Fatalf("the sender must not hear its own report, got %d", n) |
| + } |
| + if n := len(drain(g.Sub)); n != 0 { |
| + t.Fatalf("a waiting guest must hear nothing, got %d", n) |
| + } |
| + if n := ringLen(h, "r1", "sb", b.Peer); n != 0 { |
| + t.Fatalf("media is state, never ringed; ring has %d", n) |
| + } |
| + |
| + // The same state again: accepted, and the room hears nothing. |
| + if err := h.SetMedia(ctx, "r1", "sa", a.Peer, MediaState{Muted: true}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if n := len(drain(b.Sub)); n != 0 { |
| + t.Fatalf("a report that changes nothing must send nothing, got %d", n) |
| + } |
| + // A different state is a change again. |
| + if err := h.SetMedia(ctx, "r1", "sa", a.Peer, MediaState{Muted: true, CameraOff: true}); err != nil { |
| + t.Fatal(err) |
| + } |
| + got = drain(b.Sub) |
| + if len(got) != 1 || string(got[0].Payload) != `{"muted":true,"cameraOff":true}` { |
| + t.Fatalf("camera off should be one more media event, got %v", got) |
| + } |
| +} |
| + |
| +// TestRostersCarryMediaAndUnknownIsAbsent is the "state, not event" |
| +// half: a peer admitted after the report learns it from its roster, the |
| +// room learns a newcomer's from the peer event once it has one, and a |
| +// peer that has never reported — every client older than §15.19 — has |
| +// no media at all rather than a guessed false. |
| +func TestRostersCarryMediaAndUnknownIsAbsent(t *testing.T) { |
| + now := time.Date(2026, 9, 23, 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) // never reports: an old client |
| + if err := h.SetMedia(ctx, "r1", "sa", a.Peer, MediaState{Muted: true}); err != nil { |
| + t.Fatal(err) |
| + } |
| + drain(a.Sub) |
| + drain(b.Sub) |
| + |
| + c, _ := h.Open("r1", "sc", 3, "Cy", KindMember) |
| + h.Join(ctx, "r1", "sc", c.Peer) |
| + cev := drain(c.Sub) |
| + var ro *Roster |
| + for _, m := range cev { |
| + if m.Type == TypeRoster { |
| + ro = m.Roster |
| + } |
| + } |
| + am, ok := rosterMedia(ro, a.Peer) |
| + if !ok || am == nil || !am.Muted || am.CameraOff { |
| + t.Fatalf("Cy's roster should say Ann is muted, got %+v (present %v)", am, ok) |
| + } |
| + bm, ok := rosterMedia(ro, b.Peer) |
| + if !ok || bm != nil { |
| + t.Fatalf("Bob never reported, so his roster entry must carry no media, got %+v (present %v)", bm, ok) |
| + } |
| + // And on the wire, absent rather than a zero object: that is what |
| + // an old client ignores and a new one reads as unknown. |
| + raw, _ := json.Marshal(RosterPeer{Peer: b.Peer, Name: "Bob", Kind: "member"}) |
| + if string(raw) != `{"peer":"`+string(b.Peer)+`","name":"Bob","kind":"member"}` { |
| + t.Fatalf("an unreported peer marshals as %s, want no media key", raw) |
| + } |
| + |
| + // The newcomer's own first report reaches everyone already there, |
| + // the old client included: it ignores the type, and it is not the |
| + // hub's business to know that. |
| + drain(a.Sub) |
| + drain(b.Sub) |
| + if err := h.SetMedia(ctx, "r1", "sc", c.Peer, MediaState{CameraOff: true}); err != nil { |
| + t.Fatal(err) |
| + } |
| + for _, s := range []*Sub{a.Sub, b.Sub} { |
| + ev := drain(s) |
| + if len(ev) != 1 || ev[0].Type != TypeMedia || ev[0].From != c.Peer { |
| + t.Fatalf("Cy's report should reach both others once, got %v", types(ev)) |
| + } |
| + } |
| +} |
| + |
| +// TestPeerEventCarriesMediaFromTheParticipation: the `peer` event is |
| +// built from the participation, like the roster, so whatever media the |
| +// participation holds at admission goes with it. Today a report needs |
| +// admission first, so this is nil in practice — the test pins the |
| +// contract so that loosening that rule later cannot forget the event. |
| +func TestPeerEventCarriesMediaFromTheParticipation(t *testing.T) { |
| + now := time.Date(2026, 9, 23, 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) |
| + drain(a.Sub) |
| + b, _ := h.Open("r1", "sb", 2, "Bob", KindMember) |
| + // Bob's media is set directly under the lock, as if he had reported |
| + // before this admission; the hub's own paths only allow a report once |
| + // admitted, so this is the peer event's contract, not a flow. |
| + h.mu.Lock() |
| + p, _ := h.findLocked("r1", "sb", b.Peer) |
| + p.Media = &MediaState{Muted: true} |
| + h.mu.Unlock() |
| + h.Join(ctx, "r1", "sb", b.Peer) |
| + ev := drain(a.Sub) |
| + if len(ev) != 1 || ev[0].Type != TypePeer { |
| + t.Fatalf("Ann should get one peer event, got %v", types(ev)) |
| + } |
| + bm, ok := rosterMedia(ev[0].Roster, b.Peer) |
| + if !ok || bm == nil || !bm.Muted { |
| + t.Fatalf("the peer event should carry Bob's media, got %+v", bm) |
| + } |
| +} |
| + |
| +// TestResumeRosterCarriesMediaMissedWhileDetached: a change made while a |
| +// peer's stream is down is not ringed and not replayed, and it does not |
| +// need to be — the roster the resume is answered with says it. |
| +func TestResumeRosterCarriesMediaMissedWhileDetached(t *testing.T) { |
| + now := time.Date(2026, 9, 23, 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) |
| + if err := h.SetMedia(ctx, "r1", "sb", b.Peer, MediaState{Muted: true}); err != nil { |
| + t.Fatal(err) |
| + } |
| + drain(a.Sub) |
| + drain(b.Sub) |
| + |
| + h.Detach("r1", b.Peer, b.Sub) |
| + if err := h.SetMedia(ctx, "r1", "sa", a.Peer, MediaState{Muted: true}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if n := ringLen(h, "r1", "sb", b.Peer); n != 0 { |
| + t.Fatalf("a media report to a detached peer must not be ringed, ring has %d", n) |
| + } |
| + res, err := h.Resume(ctx, "r1", "sb", b.Peer, map[PeerID]uint64{}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(res.Replay) != 0 || len(res.Gap) != 0 { |
| + t.Fatalf("nothing to replay and no gap, got %d replayed and %d gaps", len(res.Replay), len(res.Gap)) |
| + } |
| + am, _ := rosterMedia(res.Roster, a.Peer) |
| + if am == nil || !am.Muted { |
| + t.Fatalf("the resume roster should say Ann is muted, got %+v", am) |
| + } |
| + // Bob's own state survived his detach: the participation holds it. |
| + bm, _ := rosterMedia(res.Roster, b.Peer) |
| + if bm == nil || !bm.Muted { |
| + t.Fatalf("Bob's own report should survive his resume, got %+v", bm) |
| + } |
| + |
| + // A leave ends the participation, and with it the state: a fresh |
| + // join starts unknown. |
| + if err := h.Leave("r1", "sb", b.Peer); err != nil { |
| + t.Fatal(err) |
| + } |
| + b2, _ := h.Open("r1", "sb", 2, "Bob", KindMember) |
| + h.Join(ctx, "r1", "sb", b2.Peer) |
| + ev := drain(a.Sub) |
| + var fresh *Roster |
| + for _, m := range ev { |
| + if m.Type == TypePeer && m.Peer == b2.Peer { |
| + fresh = m.Roster |
| + } |
| + } |
| + if bm, ok := rosterMedia(fresh, b2.Peer); !ok || bm != nil { |
| + t.Fatalf("a rejoin must start with no media, got %+v (present %v)", bm, ok) |
| + } |
| +} |
| + |
| +// TestSetMediaAuthority is Fanout's authority, reused: bound to this |
| +// room and session, admitted, and a member's standing re-derived. |
| +func TestSetMediaAuthority(t *testing.T) { |
| + now := time.Date(2026, 9, 23, 9, 0, 0, 0, time.UTC) |
| + standing := map[int64]bool{1: true, 2: true, 100: true} |
| + h := New(Config{ |
| + Now: func() time.Time { return now }, |
| + Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), |
| + StandingFor: func(ctx context.Context, room string, id int64) (Kind, bool, error) { |
| + if id >= 100 { |
| + return KindGuest, standing[id], nil |
| + } |
| + return KindMember, standing[id], nil |
| + }, |
| + }) |
| + ctx := context.Background() |
| + m := MediaState{Muted: true} |
| + a, _ := h.Open("r1", "sa", 1, "Ann", KindMember) |
| + if err := h.SetMedia(ctx, "r1", "sa", a.Peer, m); !errors.Is(err, ErrNotAdmitted) { |
| + t.Fatalf("unjoined: %v", err) |
| + } |
| + h.Join(ctx, "r1", "sa", a.Peer) |
| + if err := h.SetMedia(ctx, "r1", "sb", a.Peer, m); !errors.Is(err, ErrUnknownPeer) { |
| + t.Fatalf("another session: %v", err) |
| + } |
| + if err := h.SetMedia(ctx, "r2", "sa", a.Peer, m); !errors.Is(err, ErrUnknownPeer) { |
| + t.Fatalf("another room: %v", err) |
| + } |
| + g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest) |
| + h.Join(ctx, "r1", "sg", g.Peer) |
| + if err := h.SetMedia(ctx, "r1", "sg", g.Peer, m); !errors.Is(err, ErrNotAdmitted) { |
| + t.Fatalf("waiting guest: %v", err) |
| + } |
| + b, _ := h.Open("r1", "sb", 2, "Bob", KindMember) |
| + h.Join(ctx, "r1", "sb", b.Peer) |
| + drain(a.Sub) |
| + drain(b.Sub) |
| + |
| + standing[1] = false |
| + if err := h.SetMedia(ctx, "r1", "sa", a.Peer, m); !errors.Is(err, ErrNotYours) { |
| + t.Fatalf("member whose standing is gone: %v", err) |
| + } |
| + ev := drain(b.Sub) |
| + if len(ev) != 1 || ev[0].Type != TypeGone { |
| + t.Fatalf("Bob should see Ann gone and no media, got %v", types(ev)) |
| + } |
| +} |
| + |
| +// TestSetMediaUnderConcurrency is for `make race`: reports, resumes, |
| +// rosters and a tick all at once, over the same participations. Media |
| +// is a pointer handed to rosters that are serialised on other |
| +// goroutines, so a write through it anywhere would show up here. |
| +func TestSetMediaUnderConcurrency(t *testing.T) { |
| + now := time.Date(2026, 9, 23, 9, 0, 0, 0, time.UTC) |
| + var clock sync.Mutex |
| + h := New(Config{ |
| + Now: func() time.Time { |
| + clock.Lock() |
| + defer clock.Unlock() |
| + return now |
| + }, |
| + Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), |
| + StandingFor: func(ctx context.Context, room string, id int64) (Kind, bool, error) { |
| + return KindMember, true, nil |
| + }, |
| + }) |
| + 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) |
| + |
| + var wg sync.WaitGroup |
| + stop := make(chan struct{}) |
| + // Readers: marshal whatever arrives, the way the stream handler does. |
| + for _, s := range []*Sub{a.Sub, b.Sub} { |
| + wg.Add(1) |
| + go func(s *Sub) { |
| + defer wg.Done() |
| + for { |
| + select { |
| + case <-stop: |
| + return |
| + case m, ok := <-s.C(): |
| + if !ok { |
| + return |
| + } |
| + _, _ = json.Marshal(m) |
| + } |
| + } |
| + }(s) |
| + } |
| + var writers sync.WaitGroup |
| + for i, who := range []struct { |
| + session string |
| + peer PeerID |
| + }{{"sa", a.Peer}, {"sb", b.Peer}} { |
| + writers.Add(1) |
| + go func(i int, session string, peer PeerID) { |
| + defer writers.Done() |
| + for n := 0; n < 200; n++ { |
| + _ = h.SetMedia(ctx, "r1", session, peer, MediaState{Muted: n%2 == 0, CameraOff: (n+i)%3 == 0}) |
| + } |
| + }(i, who.session, who.peer) |
| + } |
| + writers.Add(1) |
| + go func() { |
| + defer writers.Done() |
| + for n := 0; n < 50; n++ { |
| + h.Tick(ctx) |
| + c, err := h.Open("r1", "sc", 3, "Cy", KindMember) |
| + if err == nil { |
| + _ = h.Join(ctx, "r1", "sc", c.Peer) |
| + _, _ = json.Marshal(drain(c.Sub)) |
| + _ = h.Leave("r1", "sc", c.Peer) |
| + } |
| + } |
| + }() |
| + writers.Wait() |
| + close(stop) |
| + wg.Wait() |
| +} |
diff --git a/internal/calls/hub/messages.go b/internal/calls/hub/messages.go| index f32ab43..45f3353 100644 |
| --- a/internal/calls/hub/messages.go |
| +++ b/internal/calls/hub/messages.go |
| @@ -9,6 +9,7 @@ type Reason = protocol.Reason |
| type Message = protocol.Message |
| type Roster = protocol.Roster |
| type RosterPeer = protocol.RosterPeer |
| +type MediaState = protocol.MediaState |
| const ( |
| KindGuest = protocol.KindGuest |
| @@ -27,6 +28,7 @@ const ( |
| TypeReset = protocol.TypeReset |
| TypeChat = protocol.TypeChat |
| TypeShare = protocol.TypeShare |
| + TypeMedia = protocol.TypeMedia |
| TypeHello = protocol.TypeHello |
| TypeRoster = protocol.TypeRoster |
| TypePeer = protocol.TypePeer |
diff --git a/internal/calls/signal.go b/internal/calls/signal.go| index ef60b7a..ae2608f 100644 |
| --- a/internal/calls/signal.go |
| +++ b/internal/calls/signal.go |
| @@ -137,7 +137,17 @@ func validateFanout(typ string, raw json.RawMessage) (json.RawMessage, int, stri |
| // {"text":"b"}, and the first of those is a payload whose raw bytes say |
| // one thing and whose decoded value says another. The walk counts every |
| // member it passes, so it can refuse both. |
| -func exactKey(raw json.RawMessage, want string) bool { |
| +func exactKey(raw json.RawMessage, want string) bool { return exactKeys(raw, want) } |
| + |
| +// exactKeys is exactKey for an object with more than one member: every |
| +// name in want, each spelled byte-for-byte and appearing exactly once, |
| +// in any order, and no other member at all. |
| +func exactKeys(raw json.RawMessage, want ...string) bool { |
| + seen := make(map[string]bool, len(want)) |
| + allowed := make(map[string]bool, len(want)) |
| + for _, w := range want { |
| + allowed[w] = true |
| + } |
| dec := json.NewDecoder(bytes.NewReader(raw)) |
| open, err := dec.Token() |
| if err != nil { |
| @@ -153,13 +163,11 @@ func exactKey(raw json.RawMessage, want string) bool { |
| return false |
| } |
| name, ok := key.(string) |
| - if !ok || name != want { |
| + if !ok || !allowed[name] || seen[name] { |
| return false |
| } |
| + seen[name] = true |
| members++ |
| - if members > 1 { |
| - return false |
| - } |
| // The value, whatever shape it takes: read past it so the next |
| // token is the next key or the closing brace. Its type is the |
| // business of the decode above, not of this walk. |
| @@ -173,7 +181,36 @@ func exactKey(raw json.RawMessage, want string) bool { |
| if _, err := dec.Token(); err != nil { |
| return false |
| } |
| - return members == 1 && !dec.More() |
| + return members == len(want) && !dec.More() |
| +} |
| + |
| +// maxMediaBytes bounds §15.19's payload before it is decoded. The |
| +// longest honest one, {"muted":false,"cameraOff":false}, is 34 bytes; |
| +// the rest is room for whitespace a client's encoder might add, and |
| +// nothing a peer can use to put more on anyone's stream — what goes out |
| +// is re-encoded from two booleans. |
| +const maxMediaBytes = 128 |
| + |
| +// validateMedia is §15.19's shape: exactly {"muted": bool, "cameraOff": |
| +// bool}, both present, keys byte-exact and each once, no other member, |
| +// judged on the raw bytes the same way validateFanout judges chat and |
| +// share (Codex S4-1). It returns the decoded state, or a status and an |
| +// outcome word for the refusal log. |
| +func validateMedia(raw json.RawMessage) (protocol.MediaState, int, string) { |
| + if len(raw) > maxMediaBytes { |
| + return protocol.MediaState{}, http.StatusRequestEntityTooLarge, "size" |
| + } |
| + if !exactKeys(raw, "muted", "cameraOff") { |
| + return protocol.MediaState{}, http.StatusBadRequest, "shape" |
| + } |
| + var p struct { |
| + Muted *bool `json:"muted"` |
| + CameraOff *bool `json:"cameraOff"` |
| + } |
| + if err := json.Unmarshal(raw, &p); err != nil || p.Muted == nil || p.CameraOff == nil { |
| + return protocol.MediaState{}, http.StatusBadRequest, "shape" |
| + } |
| + return protocol.MediaState{Muted: *p.Muted, CameraOff: *p.CameraOff}, 0, "" |
| } |
| // canonicalJSON encodes one validated value the way it goes out on the |
| @@ -214,6 +251,7 @@ var knownSignalTypes = map[string]bool{ |
| hub.TypeJoin: true, hub.TypeAdmit: true, hub.TypeLeave: true, |
| hub.TypeOffer: true, hub.TypeAnswer: true, hub.TypeCandidate: true, |
| hub.TypeReset: true, hub.TypeChat: true, hub.TypeShare: true, |
| + hub.TypeMedia: true, |
| } |
| func safeSignalType(t string) string { |
| @@ -352,6 +390,20 @@ func (a *app) signal(w http.ResponseWriter, r *http.Request) { |
| return |
| } |
| herr = a.hub.Fanout(r.Context(), sqid, session, body.Peer, hub.Message{Type: body.Type, Payload: payload}) |
| + case hub.TypeMedia: |
| + // §15.19: state about the sender, never targeted, and never |
| + // carrying a seq or a generation — none of them means anything |
| + // for a message the hub records rather than relays. |
| + if body.To != "" || body.Seq != 0 || body.Gen != 0 { |
| + a.refuseSignal(w, sqid, body.Peer, body.Type, http.StatusBadRequest, "shape", map[string]any{"error": "bad-request"}) |
| + return |
| + } |
| + m, status, outcome := validateMedia(body.Payload) |
| + if status != 0 { |
| + a.refuseSignal(w, sqid, body.Peer, body.Type, status, outcome, map[string]any{"error": fanoutError(status)}) |
| + return |
| + } |
| + herr = a.hub.SetMedia(r.Context(), sqid, session, body.Peer, m) |
| default: |
| a.refuseSignal(w, sqid, body.Peer, body.Type, http.StatusBadRequest, "type", map[string]any{"error": "bad-request"}) |
| return |
diff --git a/internal/callstest/media_test.go b/internal/callstest/media_test.go| new file mode 100644 |
| index 0000000..008549f |
| --- /dev/null |
| +++ b/internal/callstest/media_test.go |
| @@ -0,0 +1,170 @@ |
| +package callstest |
| + |
| +import ( |
| + "net/http" |
| + "strings" |
| + "testing" |
| + "time" |
| +) |
| + |
| +// Spec §15.19 over the real handler: a peer's own media report reaches |
| +// the other admitted peer as `media`, from the hub's envelope, with the |
| +// payload re-encoded from the two booleans the handler judged — never |
| +// the bytes the sender chose — and the sender hears nothing back. |
| +func TestMediaReachesTheOtherPeerCanonicallyAndNotTheSender(t *testing.T) { |
| + a, sqid, pa, sb := chatPair(t) |
| + |
| + // Whitespace, and the keys in the other order: both legal, and |
| + // neither must survive onto Bob's stream. |
| + body := `{"peer":"` + pa + `","type":"media","payload":{ "cameraOff" : false , "muted" : true }}` |
| + res, rb := a.signalRaw(sqid, body) |
| + if res.StatusCode != http.StatusOK { |
| + t.Fatalf("media: %d %s", res.StatusCode, rb) |
| + } |
| + ev, data, ok := sb.nextOf("media", 2*time.Second) |
| + if !ok || ev != "media" { |
| + t.Fatal("Bob did not receive the media report") |
| + } |
| + s := string(data) |
| + if !strings.Contains(s, `"from":"`+pa+`"`) || !strings.Contains(s, `"name":"paul"`) || |
| + !strings.Contains(s, `"payload":{"muted":true,"cameraOff":false}`) { |
| + t.Fatalf("envelope %s", s) |
| + } |
| + if strings.Contains(s, `"seq"`) || strings.Contains(s, `"to"`) || strings.Contains(s, `"gen"`) { |
| + t.Fatalf("a media envelope carries no seq, to or gen: %s", s) |
| + } |
| + |
| + // The same state again is accepted and silent. |
| + res, _ = a.signal(sqid, pa, `{"type":"media","payload":{"muted":true,"cameraOff":false}}`) |
| + if res.StatusCode != http.StatusOK { |
| + t.Fatalf("repeat: %d", res.StatusCode) |
| + } |
| + if ev, _, ok := sb.next(300 * time.Millisecond); ok && ev == "media" { |
| + t.Fatal("a report that changed nothing reached Bob") |
| + } |
| +} |
| + |
| +// §15.19's "state, not event": a peer who joins after the report reads |
| +// it from the roster it is admitted with, and a peer who has never |
| +// reported — an old client — is in that roster with no media at all. |
| +func TestMediaIsInTheRosterALateJoinerGets(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"}`) |
| + sa.nextOf("roster", 2*time.Second) |
| + if res, rb := a.signal(sqid, pa, `{"type":"media","payload":{"muted":true,"cameraOff":true}}`); res.StatusCode != http.StatusOK { |
| + t.Fatalf("media: %d %s", res.StatusCode, rb) |
| + } |
| + |
| + 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"}`) |
| + _, data, ok := sb.nextOf("roster", 2*time.Second) |
| + if !ok { |
| + t.Fatal("Bob got no roster") |
| + } |
| + s := string(data) |
| + want := `"peer":"` + pa + `","name":"paul","kind":"member","gen":1,"media":{"muted":true,"cameraOff":true}` |
| + if !strings.Contains(s, want) { |
| + t.Fatalf("Bob's roster should carry Paul's media; got %s", s) |
| + } |
| + // Bob has not reported, so his own entry carries no media key. |
| + if strings.Count(s, `"media"`) != 1 { |
| + t.Fatalf("only Paul has reported; roster %s", s) |
| + } |
| +} |
| + |
| +// The shape is exact, and judged on raw bytes the way chat and share |
| +// are (§15.15): both keys, spelled exactly, each once, booleans, and |
| +// nothing else — and never targeted. |
| +func TestMediaShapes(t *testing.T) { |
| + a, sqid, pa, _ := chatPair(t) |
| + bad := []struct { |
| + name, payload string |
| + }{ |
| + {"missing cameraOff", `{"muted":true}`}, |
| + {"missing muted", `{"cameraOff":true}`}, |
| + {"empty object", `{}`}, |
| + {"a third key", `{"muted":true,"cameraOff":false,"deafened":true}`}, |
| + {"a capitalised key", `{"Muted":true,"cameraOff":false}`}, |
| + {"snake case", `{"muted":true,"camera_off":false}`}, |
| + {"a duplicated key", `{"muted":true,"muted":false,"cameraOff":false}`}, |
| + {"a string for a bool", `{"muted":"yes","cameraOff":false}`}, |
| + {"null for a bool", `{"muted":null,"cameraOff":false}`}, |
| + {"a number for a bool", `{"muted":1,"cameraOff":0}`}, |
| + {"not an object", `[true,false]`}, |
| + {"trailing content", `{"muted":true,"cameraOff":false} {}`}, |
| + } |
| + for _, c := range bad { |
| + t.Run(c.name, func(t *testing.T) { |
| + body := `{"peer":"` + pa + `","type":"media","payload":` + c.payload + `}` |
| + res, rb := a.signalRaw(sqid, body) |
| + if res.StatusCode != http.StatusBadRequest || !strings.Contains(string(rb), `"error":"bad-request"`) { |
| + t.Fatalf("%d %s, want 400 bad-request", res.StatusCode, rb) |
| + } |
| + }) |
| + } |
| + t.Run("no payload at all", func(t *testing.T) { |
| + res, rb := a.signalRaw(sqid, `{"peer":"`+pa+`","type":"media"}`) |
| + if res.StatusCode != http.StatusBadRequest { |
| + t.Fatalf("%d %s, want 400", res.StatusCode, rb) |
| + } |
| + }) |
| + t.Run("targeted", func(t *testing.T) { |
| + res, rb := a.signalRaw(sqid, `{"peer":"`+pa+`","type":"media","to":"`+strings.Repeat("a", 32)+`","payload":{"muted":true,"cameraOff":false}}`) |
| + if res.StatusCode != http.StatusBadRequest { |
| + t.Fatalf("%d %s, want 400", res.StatusCode, rb) |
| + } |
| + }) |
| + t.Run("with a seq", func(t *testing.T) { |
| + res, rb := a.signalRaw(sqid, `{"peer":"`+pa+`","type":"media","seq":3,"payload":{"muted":true,"cameraOff":false}}`) |
| + if res.StatusCode != http.StatusBadRequest { |
| + t.Fatalf("%d %s, want 400", res.StatusCode, rb) |
| + } |
| + }) |
| + t.Run("padded past the cap", func(t *testing.T) { |
| + payload := `{"muted":true,` + strings.Repeat(" ", 128) + `"cameraOff":false}` |
| + res, rb := a.signalRaw(sqid, `{"peer":"`+pa+`","type":"media","payload":`+payload+`}`) |
| + if res.StatusCode != http.StatusRequestEntityTooLarge || !strings.Contains(string(rb), `"error":"too-large"`) { |
| + t.Fatalf("%d %s, want 413 too-large", res.StatusCode, rb) |
| + } |
| + }) |
| +} |
| + |
| +// A waiting guest has nobody to tell: its report is the usual 404, and |
| +// it hears nothing of the room's. |
| +func TestMediaFromAWaitingGuestIsTheUsual404(t *testing.T) { |
| + x, sqid, tok := memberWithMeeting(t) |
| + enableWaitingRoom(t, x, sqid) |
| + 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"}`) |
| + sa.nextOf("roster", 2*time.Second) |
| + |
| + g := x.browser(t) |
| + g.get("/j/" + tok) |
| + g.signIn("waiting-guest@example.com") |
| + gs := g.stream("/m/" + sqid + "/stream") |
| + defer gs.close() |
| + gpeer, _ := gs.hello() |
| + g.signal(sqid, gpeer, `{"type":"join"}`) |
| + sa.nextOf("knock", 2*time.Second) |
| + |
| + res, rb := g.signal(sqid, gpeer, `{"type":"media","payload":{"muted":true,"cameraOff":false}}`) |
| + if res.StatusCode != http.StatusNotFound || !strings.Contains(string(rb), `"error":"not-found"`) { |
| + t.Fatalf("a waiting guest's media must be the usual 404: %d %s", res.StatusCode, rb) |
| + } |
| + a.signal(sqid, pa, `{"type":"media","payload":{"muted":true,"cameraOff":false}}`) |
| + if ev, _, ok := gs.next(400 * time.Millisecond); ok && ev == "media" { |
| + t.Fatal("a waiting guest heard a media report") |
| + } |
| +} |
diff --git a/native/bridge/bridge.go b/native/bridge/bridge.go| index 8784a66..39bf103 100644 |
| --- a/native/bridge/bridge.go |
| +++ b/native/bridge/bridge.go |
| @@ -104,6 +104,11 @@ func (c *Client) Send(typ, to string, generation int64, payload string) error { |
| func (c *Client) Leave() { c.room.Leave() } |
| func (c *Client) Close() { c.room.Close() } |
| +// SetMedia is this app's own microphone and camera state (spec §15.19). |
| +// The core reports it once admitted, again whenever a roster says the |
| +// server has not got it, and at once on every change. |
| +func (c *Client) SetMedia(muted, cameraOff bool) error { return c.room.SetMedia(muted, cameraOff) } |
| + |
| // RegisterDevice hands the server an Apple device token so a knock can |
| // reach this device when the app is closed. Best effort: the caller |
| // ignores the error beyond logging, because a device that did not |
diff --git a/protocol/messages.go b/protocol/messages.go| index cbb5670..67cac5e 100644 |
| --- a/protocol/messages.go |
| +++ b/protocol/messages.go |
| @@ -48,6 +48,10 @@ const ( |
| TypeReset = "reset" |
| TypeChat = "chat" // slice 4 |
| TypeShare = "share" // slice 4 |
| + // TypeMedia goes both ways: a peer reports its own MediaState |
| + // upstream, and the hub fans the change out downstream as |
| + // {type, from, name, payload: MediaState} (spec §15.19). |
| + TypeMedia = "media" |
| ) |
| // Downstream types the server emits. |
| @@ -99,4 +103,20 @@ type RosterPeer struct { |
| Name string `json:"name"` |
| Kind string `json:"kind"` |
| Gen uint64 `json:"gen,omitempty"` |
| + // Media is the peer's own last report of its microphone and camera, |
| + // absent until it has made one — which is every client that predates |
| + // §15.19, and every peer for the moment between admission and its |
| + // first report. Absent means unknown, and a client shows unknown as |
| + // neither muted nor camera-off. Never written through: a snapshot |
| + // copies the pointer, so a change replaces it. |
| + Media *MediaState `json:"media,omitempty"` |
| +} |
| + |
| +// MediaState is what a peer says about its own outgoing media (spec |
| +// §15.19): whether its microphone is muted and whether its camera is |
| +// off. It is a claim about what the peer is choosing to send, not a |
| +// measurement of what arrives — the server never sees media. |
| +type MediaState struct { |
| + Muted bool `json:"muted"` |
| + CameraOff bool `json:"cameraOff"` |
| } |