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)) + } +}