From a431dfa10b3b05b9b617dde4e28f513a638f9f42 Mon Sep 17 00:00:00 2001 From: John O'Keefe Date: Mon, 2 Nov 2026 18:06:45 -0500 Subject: [PATCH] =?UTF-8?q?fix(sync):=20websocket=20events=20are=20per-use?= =?UTF-8?q?r=20=E2=80=94=20chokepoint=20the=20fan-out?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Owner-flagged after the dev-docs audit: BroadcastProgressUpdate, BroadcastAnnotationUpdate, and BroadcastConflictNotification fanned out through the untargeted Broadcast, delivering one user's book progress, conflict notices, and — worst — annotation payloads (which can carry note/highlight text) to EVERY connected client on the instance. A family-lens violation of exactly the kind the RLS move eliminated for database reads: user B's client received events for user A's books, including books in libraries hidden from B. Fix is the RLS philosophy applied to the fan-out layer: enforcement at one choke point. The untargeted Broadcast method and its channel/loop machinery are deleted — BroadcastToUser is now the only door, and the compiler forces every future emitter to name its audience instead of relying on review. The three helpers take the acting user's ID; the four sync call sites (progress.go x2, annotations.go x2) already had the acting user in scope. Scan events were already per-triggering- identity and collections already per-user — unchanged. Logging is untouched: the only log line in the diff is the deleted method's own, so docker logs remains the complete administrator audit trail. TestBroadcastIsPerUser pins the invariant (alice's two devices receive her three events; bob's client receives nothing — any message there is a cross-user leak). Both websocket docs updated in the same commit: the wire doc's fan-out section now states per-user delivery and the chokepoint, and the gated endpoint doc's intro matches. Full make test green. --- docs/developer/api/websocket/protocol.md | 3 +- docs/developer/websocket-api.md | 28 ++++----- internal/sync/annotations.go | 2 + internal/sync/progress.go | 9 ++- internal/sync/websocket.go | 67 ++++++++------------- internal/sync/websocket_test.go | 75 ++++++++++++++++++++++++ 6 files changed, 122 insertions(+), 62 deletions(-) create mode 100644 internal/sync/websocket_test.go diff --git a/docs/developer/api/websocket/protocol.md b/docs/developer/api/websocket/protocol.md index 624559a2..5cd193d2 100644 --- a/docs/developer/api/websocket/protocol.md +++ b/docs/developer/api/websocket/protocol.md @@ -1,6 +1,7 @@ # WebSocket Protocol -Real-time sync events broadcast to connected clients. +Real-time sync events, delivered only to the acting user's connected +devices. ## Connect to WebSocket diff --git a/docs/developer/websocket-api.md b/docs/developer/websocket-api.md index 5e7b7faa..e8856731 100644 --- a/docs/developer/websocket-api.md +++ b/docs/developer/websocket-api.md @@ -62,10 +62,9 @@ All ten types defined by `internal/sync/websocket.go`: | `heartbeat` | *Defined, not currently emitted* | — | | `sync_complete` | *Defined, not currently emitted* | — | -(`collection_updated` is the one per-user-targeted event — it's emitted -with the connecting user's id directly. It's also currently the only -message type defined as a raw string in the collections handler rather -than in the sync package's constants block.) +(`collection_updated` is currently the only message type defined as a raw +string in the collections handler rather than in the sync package's +constants block.) ## Keepalive @@ -76,14 +75,15 @@ nothing to implement. A connection that stops answering is removed. ## Fan-out and backpressure — read this before building a client -- Sync and scan events fan out **instance-wide**: every connected client - receives progress, annotation, conflict, and scan events. The - exception is `collection_updated`, which is targeted to one user's - connections. (A per-user targeting helper exists and is used for - collections only — the sync/scan emitters do not use it.) +- **Every event is delivered only to the acting user's connections.** + Progress, annotation, and conflict events go to the user whose sync + activity produced them; scan events go to the identity that triggered + the job; collection events to the collection's owner. The + ConnectionManager has no untargeted broadcast — the compiler enforces + an audience on every emitter (2026-11-02, after cross-user delivery + was found and fixed). - Every connection has a bounded send queue. **If a client falls behind, - messages for that connection are skipped, not queued** (and the global - broadcast ring drops messages when saturated). Clients must therefore - treat the socket as a *hint* — on reconnect (or any gap), refetch real - state over REST. The fresh `initial_state` on reconnect is designed to - be that gap-healer. + messages for that connection are skipped, not queued.** Clients must + therefore treat the socket as a *hint* — on reconnect (or any gap), + refetch real state over REST. The fresh `initial_state` on reconnect is + designed to be that gap-healer. diff --git a/internal/sync/annotations.go b/internal/sync/annotations.go index b012fc91..e6dad177 100644 --- a/internal/sync/annotations.go +++ b/internal/sync/annotations.go @@ -985,6 +985,7 @@ func (s *AnnotationService) recordConflict( conflictIDStr = uuid.UUID(conflict.ID.Bytes).String() } s.connMgr.BroadcastConflictNotification( + uuid.UUID(userID.Bytes).String(), uuid.UUID(mediaItemID.Bytes), "annotation_conflict", conflictIDStr, @@ -1001,6 +1002,7 @@ func (s *AnnotationService) broadcast( } src := SourceDevice{Type: source} s.connMgr.BroadcastAnnotationUpdate( + uuid.UUID(userID.Bytes).String(), uuid.UUID(mediaItemID.Bytes), annotationType, map[string]interface{}{ diff --git a/internal/sync/progress.go b/internal/sync/progress.go index 4fd0265b..54499bd5 100644 --- a/internal/sync/progress.go +++ b/internal/sync/progress.go @@ -598,9 +598,11 @@ func (s *ProgressService) SaveProgress(ctx context.Context, req SaveProgressRequ }) if err != nil { log.Printf("Failed to record sync conflict: %v", err) - } else if s.connManager != nil { - s.connManager.BroadcastConflictNotification(req.MediaItemID.Bytes, "detection", "") - } + } else if s.connManager != nil { + s.connManager.BroadcastConflictNotification( + uuid.UUID(req.UserID.Bytes).String(), + req.MediaItemID.Bytes, "detection", "") + } } if s.connManager != nil && req.Broadcast { @@ -613,6 +615,7 @@ func (s *ProgressService) SaveProgress(ctx context.Context, req SaveProgressRequ pct = result.Percentage.Float64 } s.connManager.BroadcastProgressUpdate( + uuid.UUID(req.UserID.Bytes).String(), uuid.UUID(req.MediaItemID.Bytes), pct, SourceDevice{ diff --git a/internal/sync/websocket.go b/internal/sync/websocket.go index 5d861713..06aa21be 100644 --- a/internal/sync/websocket.go +++ b/internal/sync/websocket.go @@ -51,49 +51,23 @@ type DeviceConnection struct { } // ConnectionManager manages WebSocket connections +// +// CHOKEPOINT (2026-11-02): there is deliberately NO untargeted broadcast +// method. Every send must name its audience — the user whose data the +// event carries. The pre-fix Broadcast fanned progress, annotation, and +// conflict events out to every connected client on the instance, +// crossing user boundaries (a family-lens violation: hidden-library +// book progress and annotation payloads reached unrelated clients). +// The compiler now enforces what review used to. type ConnectionManager struct { connections map[string]*DeviceConnection - broadcast chan BroadcastMessage mu sync.RWMutex } // NewConnectionManager creates a new connection manager func NewConnectionManager() *ConnectionManager { - cm := &ConnectionManager{ + return &ConnectionManager{ connections: make(map[string]*DeviceConnection), - broadcast: make(chan BroadcastMessage, 100), - } - - // Start broadcast goroutine - go cm.broadcastLoop() - - return cm -} - -// broadcastLoop handles broadcasting messages to all connected clients -func (m *ConnectionManager) broadcastLoop() { - for { - select { - case msg := <-m.broadcast: - m.mu.RLock() - for _, conn := range m.connections { - select { - case conn.Send <- msg: - default: - // Channel full, skip this connection - } - } - m.mu.RUnlock() - } - } -} - -// Broadcast sends a message to all connected clients -func (m *ConnectionManager) Broadcast(msg BroadcastMessage) { - select { - case m.broadcast <- msg: - default: - log.Println("Broadcast channel full, message dropped") } } @@ -114,8 +88,10 @@ func (m *ConnectionManager) BroadcastToUser(userID string, msg BroadcastMessage) } } -// BroadcastProgressUpdate broadcasts a progress update to all connected clients -func (m *ConnectionManager) BroadcastProgressUpdate(bookID uuid.UUID, percentage float64, source SourceDevice) { +// BroadcastProgressUpdate broadcasts a progress update to the acting +// user's connections. Progress is per-user data — nobody else's client +// receives it. +func (m *ConnectionManager) BroadcastProgressUpdate(userID string, bookID uuid.UUID, percentage float64, source SourceDevice) { msg := BroadcastMessage{ Type: MessageTypeProgressUpdate, Timestamp: time.Now().Format(time.RFC3339), @@ -125,11 +101,13 @@ func (m *ConnectionManager) BroadcastProgressUpdate(bookID uuid.UUID, percentage }, Source: &source, } - m.Broadcast(msg) + m.BroadcastToUser(userID, msg) } -// BroadcastAnnotationUpdate broadcasts an annotation update to all connected clients -func (m *ConnectionManager) BroadcastAnnotationUpdate(bookID uuid.UUID, annotationType string, data interface{}, source SourceDevice) { +// BroadcastAnnotationUpdate broadcasts an annotation update to the acting +// user's connections. Annotation payloads can carry note and highlight +// content — strictly per-user. +func (m *ConnectionManager) BroadcastAnnotationUpdate(userID string, bookID uuid.UUID, annotationType string, data interface{}, source SourceDevice) { msg := BroadcastMessage{ Type: MessageTypeAnnotationUpdate, Timestamp: time.Now().Format(time.RFC3339), @@ -140,11 +118,12 @@ func (m *ConnectionManager) BroadcastAnnotationUpdate(bookID uuid.UUID, annotati }, Source: &source, } - m.Broadcast(msg) + m.BroadcastToUser(userID, msg) } -// BroadcastConflictNotification broadcasts a conflict notification to all connected clients -func (m *ConnectionManager) BroadcastConflictNotification(bookID [16]byte, notificationType string, conflictID string) { +// BroadcastConflictNotification broadcasts a conflict notification to the +// acting user's connections. +func (m *ConnectionManager) BroadcastConflictNotification(userID string, bookID [16]byte, notificationType string, conflictID string) { msg := BroadcastMessage{ Type: MessageTypeConflict, Timestamp: time.Now().Format(time.RFC3339), @@ -154,7 +133,7 @@ func (m *ConnectionManager) BroadcastConflictNotification(bookID [16]byte, notif "conflict_id": conflictID, }, } - m.Broadcast(msg) + m.BroadcastToUser(userID, msg) } // AddConnection adds a new WebSocket connection diff --git a/internal/sync/websocket_test.go b/internal/sync/websocket_test.go new file mode 100644 index 00000000..7d851d97 --- /dev/null +++ b/internal/sync/websocket_test.go @@ -0,0 +1,75 @@ +package sync + +import ( + "testing" + + "github.com/google/uuid" +) + +// TestBroadcastIsPerUser pins the 2026-11-02 chokepoint: sync events are +// private to the acting user. The pre-fix helpers fanned out through the +// untargeted Broadcast, delivering one user's book progress — and +// annotation payloads, which can carry note text — to every connected +// client on the instance. The untargeted method no longer exists (the +// compiler enforces an audience on every emitter); this test enforces the +// routing at runtime, in the spirit of TestRlsInvariant: no event may +// cross a user boundary. +func TestBroadcastIsPerUser(t *testing.T) { + cm := NewConnectionManager() + + mkConn := func(user string) *DeviceConnection { + conn := &DeviceConnection{ + ID: uuid.NewString(), + UserID: user, + DeviceType: "test", + DeviceName: user + "-device", + Send: make(chan BroadcastMessage, 8), + } + cm.AddConnection(conn) + return conn + } + + // Alice has two devices connected; Bob has one. + a1 := mkConn("alice") + a2 := mkConn("alice") + b1 := mkConn("bob") + + source := SourceDevice{ID: "dev-1", Name: "alice-kobo", Type: "kobo"} + + cm.BroadcastProgressUpdate("alice", uuid.New(), 42.5, source) + cm.BroadcastAnnotationUpdate("alice", uuid.New(), "highlight", map[string]any{"highlight_id": "h1"}, source) + cm.BroadcastConflictNotification("alice", uuid.New(), "detection", "conflict-1") + + drain := func(conn *DeviceConnection) []BroadcastMessage { + var got []BroadcastMessage + for { + select { + case msg := <-conn.Send: + got = append(got, msg) + default: + return got + } + } + } + + // Alice's devices each received all three of her events... + for _, conn := range []*DeviceConnection{a1, a2} { + got := drain(conn) + if len(got) != 3 { + t.Fatalf("%s got %d messages, want 3", conn.DeviceName, len(got)) + } + for _, msg := range got { + // Conflict events are server-detected and carry no source; + // progress and annotation events identify the originating + // device so clients can ignore their own echo. + if msg.Type != "conflict" && (msg.Source == nil || msg.Source.Name != "alice-kobo") { + t.Errorf("%s: unexpected event source on %q: %+v", conn.DeviceName, msg.Type, msg.Source) + } + } + } + + // ...and Bob received nothing. Any message here is a cross-user leak. + if leaked := drain(b1); len(leaked) != 0 { + t.Fatalf("bob's client received %d message(s) from alice's sync activity — cross-user leak", len(leaked)) + } +}