From 60788bfa909205e3f4e52f225b881c11454cca40 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Luk=C3=A1=C5=A1=20Lipinsk=C3=BD?= <6032558+Mr-Tao@users.noreply.github.com> Date: Sun, 30 Aug 2026 03:09:16 +0200 Subject: [PATCH] fix(player): harden device state projection --- pkg/client/websocket.go | 313 ++++++++++---- pkg/client/websocket_test.go | 153 +++++++ .../soundtouchweb/device_projection.go | 56 ++- .../soundtouchweb/device_projection_test.go | 48 ++- pkg/service/soundtouchweb/discovery.go | 48 ++- pkg/service/soundtouchweb/discovery_test.go | 16 + .../frontend_test/devicePresentation.test.mjs | 67 +++ .../frontend_test/deviceRemoval.test.mjs | 94 ++++ pkg/service/soundtouchweb/handler.go | 34 +- pkg/service/soundtouchweb/handler_test.go | 33 ++ pkg/service/soundtouchweb/static/css/app.css | 51 ++- pkg/service/soundtouchweb/static/js/app.js | 51 ++- .../static/js/components/DeviceList.js | 62 +-- .../static/js/devicePresentation.js | 42 ++ .../soundtouchweb/static/js/deviceRemoval.js | 32 ++ pkg/service/soundtouchweb/websocket.go | 168 ++++++-- pkg/service/soundtouchweb/websocket_test.go | 146 +++++++ .../soundtouchweb/webtypes/status_test.go | 259 +++++++++++ pkg/service/soundtouchweb/webtypes/types.go | 405 +++++++++++++++++- 19 files changed, 1850 insertions(+), 228 deletions(-) create mode 100644 pkg/service/soundtouchweb/frontend_test/devicePresentation.test.mjs create mode 100644 pkg/service/soundtouchweb/frontend_test/deviceRemoval.test.mjs create mode 100644 pkg/service/soundtouchweb/static/js/devicePresentation.js create mode 100644 pkg/service/soundtouchweb/static/js/deviceRemoval.js diff --git a/pkg/client/websocket.go b/pkg/client/websocket.go index 6ea8937a..6b4ab3d3 100644 --- a/pkg/client/websocket.go +++ b/pkg/client/websocket.go @@ -5,6 +5,7 @@ import ( "encoding/xml" "fmt" "log" + "net/http" "net/url" "strings" "sync" @@ -16,17 +17,33 @@ import ( // WebSocketClient handles WebSocket connections to SoundTouch devices type WebSocketClient struct { - client *Client - conn *websocket.Conn - handlers *models.WebSocketEventHandlers - mu sync.RWMutex - writeMu sync.Mutex // serializes all writes; gorilla/websocket allows one concurrent writer - connected bool - reconnect bool - ctx context.Context - cancel context.CancelFunc - logger Logger - bufferSize int + client *Client + conn *websocket.Conn + connection *webSocketConnection + handlers *models.WebSocketEventHandlers + mu sync.RWMutex + connectMu sync.Mutex // serializes dial attempts without blocking shutdown + writeMu sync.Mutex // serializes all writes; gorilla/websocket allows one concurrent writer + connected bool + reconnect bool + ctx context.Context + cancel context.CancelFunc + logger Logger + bufferSize int + dialContext webSocketDialContext + + transportHandler func(connected bool, generation uint64) + transportGeneration uint64 +} + +type webSocketDialContext func(context.Context, string, http.Header) (*websocket.Conn, *http.Response, error) + +// webSocketConnection gives each transport generation its own lifecycle so an +// old read or ping loop cannot start using a replacement connection. +type webSocketConnection struct { + conn *websocket.Conn + ctx context.Context + cancel context.CancelFunc } // Logger interface for WebSocket logging @@ -159,6 +176,23 @@ func (ws *WebSocketClient) OnBassUpdated(handler models.TypedEventHandler[*model ws.handlers.OnBassUpdated = handler } +// OnNameUpdated sets a handler for device name update events. +func (ws *WebSocketClient) OnNameUpdated(handler models.TypedEventHandler[*models.NameUpdatedEvent]) { + ws.mu.Lock() + defer ws.mu.Unlock() + + ws.handlers.OnNameUpdated = handler +} + +// OnTransportState observes authoritative connection transitions. Generation +// numbers let consumers reject callbacks that arrive out of order. +func (ws *WebSocketClient) OnTransportState(handler func(connected bool, generation uint64)) { + ws.mu.Lock() + defer ws.mu.Unlock() + + ws.transportHandler = handler +} + // OnUnknownEvent sets a handler for unknown events func (ws *WebSocketClient) OnUnknownEvent(handler models.EventHandler) { ws.mu.Lock() @@ -198,12 +232,21 @@ func (ws *WebSocketClient) ConnectWithConfig(config *WebSocketConfig) error { } func (ws *WebSocketClient) connectWithConfig(config *WebSocketConfig) error { - ws.mu.Lock() - defer ws.mu.Unlock() + ws.connectMu.Lock() + defer ws.connectMu.Unlock() + ws.mu.RLock() if ws.connected { + ws.mu.RUnlock() return fmt.Errorf("already connected") } + ctx := ws.ctx + dialContext := ws.dialContext + ws.mu.RUnlock() + + if err := ctx.Err(); err != nil { + return fmt.Errorf("WebSocket client is closed: %w", err) + } // Build WebSocket URL // Parse the base URL to extract just the hostname @@ -220,16 +263,20 @@ func (ws *WebSocketClient) connectWithConfig(config *WebSocketConfig) error { ws.logger.Printf("Connecting to %s", sanitizeLog(wsURL.String())) - // Create dialer with custom buffer sizes and "gabbo" protocol - dialer := websocket.Dialer{ - HandshakeTimeout: 10 * time.Second, - ReadBufferSize: config.ReadBufferSize, - WriteBufferSize: config.WriteBufferSize, - Subprotocols: []string{"gabbo"}, // Required by SoundTouch API + if dialContext == nil { + // Create dialer with custom buffer sizes and "gabbo" protocol. + dialer := websocket.Dialer{ + HandshakeTimeout: 10 * time.Second, + ReadBufferSize: config.ReadBufferSize, + WriteBufferSize: config.WriteBufferSize, + Subprotocols: []string{"gabbo"}, // Required by SoundTouch API + } + dialContext = dialer.DialContext } - // Establish connection - conn, resp, err := dialer.DialContext(ws.ctx, wsURL.String(), nil) + // Dial without holding the state mutex so shutdown can cancel the context + // immediately instead of waiting for the handshake timeout. + conn, resp, err := dialContext(ctx, wsURL.String(), nil) if resp != nil && resp.Body != nil { defer func() { _ = resp.Body.Close() }() } @@ -238,8 +285,50 @@ func (ws *WebSocketClient) connectWithConfig(config *WebSocketConfig) error { return fmt.Errorf("failed to connect to WebSocket: %w", err) } + ws.mu.Lock() + if err := ctx.Err(); err != nil { + ws.mu.Unlock() + _ = conn.Close() + + return fmt.Errorf("WebSocket client closed during connect: %w", err) + } + if ws.connected { + ws.mu.Unlock() + _ = conn.Close() + + return fmt.Errorf("already connected") + } + connection, transportHandler, transportGeneration := ws.activateConnectionLocked(conn) + ws.mu.Unlock() + + notifyTransportState(transportHandler, true, transportGeneration) + go ws.readLoop(config, connection) + go ws.pingLoop(config, connection) + + ws.logger.Printf("Connected to %s", sanitizeLog(wsURL.String())) + + return nil +} + +func (ws *WebSocketClient) activateConnectionLocked( + conn *websocket.Conn, +) (*webSocketConnection, func(bool, uint64), uint64) { + if ws.connection != nil { + ws.connection.cancel() + _ = ws.connection.conn.Close() + } + + connectionCtx, connectionCancel := context.WithCancel(ws.ctx) + connection := &webSocketConnection{ + conn: conn, + ctx: connectionCtx, + cancel: connectionCancel, + } + ws.conn = conn + ws.connection = connection ws.connected = true + ws.transportGeneration++ // Extend the read deadline on every pong so the connection survives // quiet periods between speaker events. Without this, the 60-second @@ -250,41 +339,90 @@ func (ws *WebSocketClient) connectWithConfig(config *WebSocketConfig) error { return nil }) - // Start background goroutines for connection management - go ws.readLoop(config) - go ws.pingLoop(config) - - ws.logger.Printf("Connected to %s", sanitizeLog(wsURL.String())) - - return nil + return connection, ws.transportHandler, ws.transportGeneration } // Disconnect closes the WebSocket connection func (ws *WebSocketClient) Disconnect() error { + // Cancel first so an in-progress DialContext wakes without waiting for mu. + ws.cancel() + ws.mu.Lock() - defer ws.mu.Unlock() - - if !ws.connected { - return fmt.Errorf("not connected") - } - + wasConnected := ws.connected ws.reconnect = false - ws.cancel() // Cancel context to stop goroutines + if ws.connection != nil { + ws.connection.cancel() + ws.connection = nil + } + conn := ws.conn + ws.conn = nil + ws.connected = false + var ( + transportHandler func(bool, uint64) + transportGeneration uint64 + ) + if wasConnected { + ws.transportGeneration++ + transportHandler = ws.transportHandler + transportGeneration = ws.transportGeneration + } + ws.mu.Unlock() - if ws.conn != nil { - err := ws.conn.Close() - ws.conn = nil - ws.connected = false + if conn != nil { + err := conn.Close() ws.logger.Printf("Disconnected") + notifyTransportState(transportHandler, false, transportGeneration) + return err } - - ws.connected = false + notifyTransportState(transportHandler, false, transportGeneration) + if !wasConnected { + return fmt.Errorf("not connected") + } return nil } +// Close permanently stops this client and is idempotent. Device registries use +// it when removal races an initial dial or an automatic reconnect. +func (ws *WebSocketClient) Close() error { + ws.cancel() + + ws.mu.Lock() + wasConnected := ws.connected + ws.reconnect = false + if ws.connection != nil { + ws.connection.cancel() + ws.connection = nil + } + conn := ws.conn + ws.conn = nil + ws.connected = false + var ( + transportHandler func(bool, uint64) + transportGeneration uint64 + ) + if wasConnected { + ws.transportGeneration++ + transportHandler = ws.transportHandler + transportGeneration = ws.transportGeneration + } + ws.mu.Unlock() + + if conn == nil { + notifyTransportState(transportHandler, false, transportGeneration) + + return nil + } + + err := conn.Close() + ws.logger.Printf("Disconnected") + notifyTransportState(transportHandler, false, transportGeneration) + + return err +} + // IsConnected returns true if the WebSocket is connected func (ws *WebSocketClient) IsConnected() bool { ws.mu.RLock() @@ -293,45 +431,61 @@ func (ws *WebSocketClient) IsConnected() bool { return ws.connected } +func (ws *WebSocketClient) shouldReconnect() bool { + ws.mu.RLock() + defer ws.mu.RUnlock() + + return ws.reconnect +} + +func notifyTransportState(handler func(bool, uint64), connected bool, generation uint64) { + if handler != nil { + handler(connected, generation) + } +} + // readLoop continuously reads messages from the WebSocket connection -func (ws *WebSocketClient) readLoop(config *WebSocketConfig) { +func (ws *WebSocketClient) readLoop(config *WebSocketConfig, connection *webSocketConnection) { defer func() { ws.mu.Lock() + if ws.connection != connection { + ws.mu.Unlock() + connection.cancel() + _ = connection.conn.Close() - ws.connected = false - if ws.conn != nil { - _ = ws.conn.Close() - ws.conn = nil + return } + connection.cancel() + ws.connection = nil + ws.conn = nil + ws.connected = false + ws.transportGeneration++ + transportHandler := ws.transportHandler + transportGeneration := ws.transportGeneration + reconnect := ws.reconnect ws.mu.Unlock() + _ = connection.conn.Close() + notifyTransportState(transportHandler, false, transportGeneration) // Attempt reconnection if enabled - if ws.reconnect { + if reconnect { go ws.attemptReconnect(config) } }() for { select { - case <-ws.ctx.Done(): + case <-connection.ctx.Done(): return default: } - ws.mu.RLock() - conn := ws.conn - ws.mu.RUnlock() - - if conn == nil { - return - } - // Set read deadline - _ = conn.SetReadDeadline(time.Now().Add(60 * time.Second)) + _ = connection.conn.SetReadDeadline(time.Now().Add(60 * time.Second)) // Read message - messageType, data, err := conn.ReadMessage() + messageType, data, err := connection.conn.ReadMessage() if err != nil { if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) { ws.logger.Printf("WebSocket read error: %v", err) @@ -351,30 +505,20 @@ func (ws *WebSocketClient) readLoop(config *WebSocketConfig) { } // pingLoop sends periodic ping messages to keep the connection alive -func (ws *WebSocketClient) pingLoop(config *WebSocketConfig) { +func (ws *WebSocketClient) pingLoop(config *WebSocketConfig, connection *webSocketConnection) { ticker := time.NewTicker(config.PingInterval) defer ticker.Stop() for { select { - case <-ws.ctx.Done(): + case <-connection.ctx.Done(): return case <-ticker.C: - ws.mu.RLock() - conn := ws.conn - connected := ws.connected - ws.mu.RUnlock() - - if !connected || conn == nil { + active, err := ws.writePing(connection) + if !active { return } - // Set write deadline for ping - ws.writeMu.Lock() - _ = conn.SetWriteDeadline(time.Now().Add(10 * time.Second)) - err := conn.WriteMessage(websocket.PingMessage, nil) - ws.writeMu.Unlock() - if err != nil { ws.logger.Printf("Failed to send ping: %v", err) return @@ -383,10 +527,26 @@ func (ws *WebSocketClient) pingLoop(config *WebSocketConfig) { } } +func (ws *WebSocketClient) writePing(connection *webSocketConnection) (bool, error) { + ws.writeMu.Lock() + defer ws.writeMu.Unlock() + + ws.mu.RLock() + defer ws.mu.RUnlock() + + if connection.ctx.Err() != nil || ws.connection != connection || !ws.connected { + return false, nil + } + + _ = connection.conn.SetWriteDeadline(time.Now().Add(10 * time.Second)) + + return true, connection.conn.WriteMessage(websocket.PingMessage, nil) +} + // attemptReconnect attempts to reconnect to the WebSocket func (ws *WebSocketClient) attemptReconnect(config *WebSocketConfig) { attempt := 0 - for ws.reconnect && (config.MaxReconnectAttempts == 0 || attempt < config.MaxReconnectAttempts) { + for ws.shouldReconnect() && (config.MaxReconnectAttempts == 0 || attempt < config.MaxReconnectAttempts) { select { case <-ws.ctx.Done(): return @@ -533,6 +693,13 @@ func (ws *WebSocketClient) dispatchTypedEventContinued(handlers *models.WebSocke return true + case models.EventTypeNameUpdated: + if handlers.OnNameUpdated != nil && event.NameUpdated != nil { + handlers.OnNameUpdated(event.NameUpdated) + } + + return true + case models.EventTypeRecentsUpdated: return true diff --git a/pkg/client/websocket_test.go b/pkg/client/websocket_test.go index 29d149c6..297549d5 100644 --- a/pkg/client/websocket_test.go +++ b/pkg/client/websocket_test.go @@ -1,6 +1,7 @@ package client import ( + "context" "net/http" "net/http/httptest" "strings" @@ -269,6 +270,139 @@ func TestWebSocketClient_Disconnect(t *testing.T) { } } +func TestWebSocketClient_DisconnectWhileDisconnectedStopsReconnect(t *testing.T) { + client := NewClientFromHost("192.0.2.10") + wsClient := client.NewWebSocketClient(nil) + + if err := wsClient.Disconnect(); err == nil { + t.Fatal("Disconnect() while disconnected should retain its compatibility error") + } + + wsClient.mu.RLock() + reconnect := wsClient.reconnect + wsClient.mu.RUnlock() + if reconnect { + t.Fatal("Disconnect() left reconnect enabled") + } + + select { + case <-wsClient.ctx.Done(): + case <-time.After(100 * time.Millisecond): + t.Fatal("Disconnect() did not cancel the WebSocket context") + } +} + +func TestWebSocketClient_CloseCancelsInProgressDial(t *testing.T) { + client := NewClientFromHost("192.0.2.10") + wsClient := client.NewWebSocketClient(nil) + dialStarted := make(chan struct{}) + + wsClient.dialContext = func(ctx context.Context, _ string, _ http.Header) (*websocket.Conn, *http.Response, error) { + close(dialStarted) + <-ctx.Done() + + return nil, nil, ctx.Err() + } + + connectDone := make(chan error, 1) + go func() { + connectDone <- wsClient.Connect() + }() + + select { + case <-dialStarted: + case <-time.After(time.Second): + t.Fatal("WebSocket dial did not start") + } + + if err := wsClient.Close(); err != nil { + t.Fatalf("Close() failed: %v", err) + } + + select { + case err := <-connectDone: + if err == nil { + t.Fatal("Connect() succeeded after Close() canceled its dial") + } + case <-time.After(time.Second): + t.Fatal("canceled WebSocket dial did not return") + } + + if err := wsClient.Close(); err != nil { + t.Fatalf("second Close() was not idempotent: %v", err) + } +} + +func TestWebSocketClient_StaleGenerationCannotOwnPing(t *testing.T) { + client := NewClientFromHost("192.0.2.10") + wsClient := client.NewWebSocketClient(nil) + oldCtx, oldCancel := context.WithCancel(context.Background()) + currentCtx, currentCancel := context.WithCancel(context.Background()) + defer oldCancel() + defer currentCancel() + + oldConnection := &webSocketConnection{ctx: oldCtx, cancel: oldCancel} + currentConnection := &webSocketConnection{ctx: currentCtx, cancel: currentCancel} + wsClient.mu.Lock() + wsClient.connection = currentConnection + wsClient.connected = true + wsClient.mu.Unlock() + + active, err := wsClient.writePing(oldConnection) + if err != nil { + t.Fatalf("writePing() for stale generation returned error: %v", err) + } + if active { + t.Fatal("replaced connection generation retained ping ownership") + } +} + +func TestWebSocketClient_TransportCallbacksCarryMonotonicGenerations(t *testing.T) { + server, messagesChan := setupMockWebSocketServer(t) + defer server.Close() + defer close(messagesChan) + + client := NewClientFromHost("192.0.2.10") + wsClient := client.NewWebSocketClient(nil) + serverWebSocketURL := strings.Replace(server.URL, "http://", "ws://", 1) + wsClient.dialContext = func(ctx context.Context, _ string, header http.Header) (*websocket.Conn, *http.Response, error) { + return websocket.DefaultDialer.DialContext(ctx, serverWebSocketURL, header) + } + + type transportState struct { + connected bool + generation uint64 + } + states := make(chan transportState, 2) + wsClient.OnTransportState(func(connected bool, generation uint64) { + states <- transportState{connected: connected, generation: generation} + }) + nextState := func() transportState { + t.Helper() + select { + case state := <-states: + return state + case <-time.After(time.Second): + t.Fatal("timed out waiting for transport callback") + return transportState{} + } + } + + if err := wsClient.Connect(); err != nil { + t.Fatalf("Connect() failed: %v", err) + } + if state := nextState(); !state.connected || state.generation != 1 { + t.Fatalf("connected state = %+v, want connected generation 1", state) + } + + if err := wsClient.Close(); err != nil { + t.Fatalf("Close() failed: %v", err) + } + if state := nextState(); state.connected || state.generation != 2 { + t.Fatalf("disconnected state = %+v, want disconnected generation 2", state) + } +} + func TestWebSocketClient_HandleMessage(t *testing.T) { client := NewClientFromHost("192.0.2.10") wsClient := client.NewWebSocketClient(&WebSocketConfig{ @@ -278,6 +412,7 @@ func TestWebSocketClient_HandleMessage(t *testing.T) { var ( nowPlayingEvent *models.NowPlayingUpdatedEvent volumeEvent *models.VolumeUpdatedEvent + nameEvent *models.NameUpdatedEvent ) wsClient.OnNowPlaying(func(event *models.NowPlayingUpdatedEvent) { @@ -288,6 +423,10 @@ func TestWebSocketClient_HandleMessage(t *testing.T) { volumeEvent = event }) + wsClient.OnNameUpdated(func(event *models.NameUpdatedEvent) { + nameEvent = event + }) + t.Run("HandleNowPlayingEvent", func(t *testing.T) { xmlData := []byte(` @@ -343,6 +482,20 @@ func TestWebSocketClient_HandleMessage(t *testing.T) { } }) + t.Run("HandleNameEvent", func(t *testing.T) { + xmlData := []byte(`Living Room Left`) + + wsClient.handleMessage(xmlData) + + if nameEvent == nil { + t.Fatal("Name event handler was not called") + } + + if nameEvent.DeviceID != "689E19B8BB8A" || nameEvent.Name.Value != "Living Room Left" { + t.Errorf("Unexpected name event: %+v", nameEvent) + } + }) + t.Run("HandleInvalidXML", func(t *testing.T) { logger := &mockLogger{} wsClient.logger = logger diff --git a/pkg/service/soundtouchweb/device_projection.go b/pkg/service/soundtouchweb/device_projection.go index 60504d31..6167f7a0 100644 --- a/pkg/service/soundtouchweb/device_projection.go +++ b/pkg/service/soundtouchweb/device_projection.go @@ -1,6 +1,7 @@ package soundtouchweb import ( + "net" "strings" "time" @@ -99,7 +100,7 @@ func captureDeviceProjectionEntries(snapshot []DeviceEntry) []deviceProjectionEn captured = append(captured, deviceProjectionEntry{ ID: entry.ID, - Info: entry.Device.DeviceInfo, + Info: entry.Device.Info(), Status: entry.Device.Status(), LastSeen: entry.LastSeen, }) @@ -157,7 +158,7 @@ func projectCapturedDeviceEntries(snapshot []deviceProjectionEntry) map[string]d pair := masters[entry.ID] devices[entry.ID] = deviceView{ - Info: projectedDeviceInfo(entry.Info, pair), + Info: projectedDeviceInfo(entry.ID, entry.Info, pair), Status: entry.Status, LastSeen: entry.LastSeen, StereoPair: pair, @@ -236,12 +237,10 @@ func newStereoPairView(group *models.Group, byDeviceID map[string][]deviceProjec if entry, ok := uniqueDeviceEntry(byDeviceID, role.DeviceID); ok { if entry.Info != nil { member.Name = entry.Info.Name - if entry.Info.IPAddress != "" { - member.IPAddress = entry.Info.IPAddress - } + member.IPAddress = projectedIPAddress(entry.ID, entry.Info, member.IPAddress) } - member.Available = entry.Status != nil && entry.Status.IsConnected + member.Available = projectedConnectivity(entry.Status) != webtypes.ConnectivityOffline if member.Available { available++ } @@ -262,17 +261,56 @@ func newStereoPairView(group *models.Group, byDeviceID map[string][]deviceProjec } } -func projectedDeviceInfo(info *models.DeviceInfo, pair *stereoPairView) *models.DeviceInfo { - if info == nil || pair == nil || pair.Name == "" || pair.Name == info.Name { +func projectedDeviceInfo(controlID string, info *models.DeviceInfo, pair *stereoPairView) *models.DeviceInfo { + if info == nil { + return info + } + + address := projectedIPAddress(controlID, info, "") + name := info.Name + if pair != nil && pair.Name != "" { + name = pair.Name + } + if address == info.IPAddress && name == info.Name { return info } projected := *info - projected.Name = pair.Name + projected.IPAddress = address + projected.Name = name return &projected } +func projectedIPAddress(controlID string, info *models.DeviceInfo, fallback string) string { + candidates := []string{controlID, fallback} + if info != nil { + candidates = append([]string{info.IPAddress}, candidates...) + } + + for _, candidate := range candidates { + if ip := net.ParseIP(strings.TrimSpace(candidate)); ip != nil { + return ip.String() + } + } + + return "" +} + +func projectedConnectivity(status *webtypes.DeviceStatus) webtypes.Connectivity { + if status == nil { + return webtypes.ConnectivityOffline + } + if status.Connectivity != "" { + return status.Connectivity + } + if status.IsConnected { + return webtypes.ConnectivityOnline + } + + return webtypes.ConnectivityOffline +} + func logicalPairName(groupName string, members []stereoPairMemberView) string { commonName := "" diff --git a/pkg/service/soundtouchweb/device_projection_test.go b/pkg/service/soundtouchweb/device_projection_test.go index 4ab9cedd..1e5c71a8 100644 --- a/pkg/service/soundtouchweb/device_projection_test.go +++ b/pkg/service/soundtouchweb/device_projection_test.go @@ -14,14 +14,18 @@ import ( ) func projectionDevice(host, deviceID, name string, connected bool, group *models.Group) DeviceEntry { + return projectionDeviceAt(host, host, deviceID, name, connected, group) +} + +func projectionDeviceAt(controlID, address, deviceID, name string, connected bool, group *models.Group) DeviceEntry { conn := webtypes.NewDeviceConnection(nil, &models.DeviceInfo{ DeviceID: deviceID, Name: name, - IPAddress: host, + IPAddress: address, }) conn.SetStatus(&webtypes.DeviceStatus{IsConnected: connected, Group: group}) - return DeviceEntry{ID: host, Device: conn, LastSeen: conn.LastSeen} + return DeviceEntry{ID: controlID, Device: conn, LastSeen: conn.LastSeen} } func testStereoGroup() *models.Group { @@ -106,6 +110,46 @@ func TestProjectDeviceEntriesNormalizesMemberRoleAndDeviceID(t *testing.T) { } } +func TestProjectDeviceEntriesKeepsHostnameControlIDSeparateFromAddress(t *testing.T) { + entry := projectionDeviceAt("kitchen.local", "192.0.2.10", "kitchen-id", "Kitchen", true, nil) + got := projectDeviceEntries([]DeviceEntry{entry}) + + view, ok := got["kitchen.local"] + if !ok || len(got) != 1 { + t.Fatalf("hostname-keyed projection = %+v", got) + } + if view.Info == nil || view.Info.IPAddress != "192.0.2.10" { + t.Fatalf("presentation address = %+v, want canonical numeric IP", view.Info) + } +} + +func TestProjectDeviceEntriesUsesLatestInfoName(t *testing.T) { + entry := projectionDevice("192.0.2.10", "kitchen-id", "Old Kitchen", true, nil) + entry.Device.ApplyNameEvent("Kitchen") + + view := projectDeviceEntries([]DeviceEntry{entry})["192.0.2.10"] + if view.Info == nil || view.Info.Name != "Kitchen" { + t.Fatalf("projected info = %+v, want latest event name", view.Info) + } +} + +func TestProjectDeviceEntriesTreatsStaleStereoMemberAsAvailable(t *testing.T) { + group := testStereoGroup() + left := projectionDevice("192.0.2.10", "left-id", "Living Room", false, group) + left.Device.UpdateStatus(func(status *webtypes.DeviceStatus) { + status.Connectivity = webtypes.ConnectivityStale + }) + right := projectionDevice("192.0.2.11", "right-id", "Living Room", true, group) + right.Device.UpdateStatus(func(status *webtypes.DeviceStatus) { + status.Connectivity = webtypes.ConnectivityOnline + }) + + pair := projectDeviceEntries([]DeviceEntry{left, right})["192.0.2.10"].StereoPair + if pair == nil || pair.AvailableMemberCount != 2 || pair.Degraded { + t.Fatalf("stale logical member was treated as offline: %+v", pair) + } +} + func TestProjectDeviceEntriesShowsDegradedPairWhenMemberIsMissing(t *testing.T) { got := projectDeviceEntries([]DeviceEntry{ projectionDevice("192.0.2.10", "left-id", "Living Room", true, testStereoGroup()), diff --git a/pkg/service/soundtouchweb/discovery.go b/pkg/service/soundtouchweb/discovery.go index 3dcbd077..ce3104da 100644 --- a/pkg/service/soundtouchweb/discovery.go +++ b/pkg/service/soundtouchweb/discovery.go @@ -3,6 +3,7 @@ package soundtouchweb import ( "context" "log" + "net" "strings" "sync" "time" @@ -10,6 +11,7 @@ import ( "github.com/gesellix/bose-soundtouch/pkg/client" "github.com/gesellix/bose-soundtouch/pkg/config" "github.com/gesellix/bose-soundtouch/pkg/discovery" + "github.com/gesellix/bose-soundtouch/pkg/models" "github.com/gesellix/bose-soundtouch/pkg/service/soundtouchweb/webtypes" "github.com/gesellix/bose-soundtouch/pkg/speaker" ) @@ -85,12 +87,12 @@ func (app *WebApp) addDeviceByHost(host string, port int, source string) *webtyp return nil } - // Ensure IPAddress is set for the web UI - if info.IPAddress == "" { - info.IPAddress = host - } + // Keep the registry key stable for controls, but expose a canonical numeric + // address separately for presentation and sorting. + info.IPAddress = resolvedDeviceIPAddress(host, info) conn := webtypes.NewDeviceConnection(c, info) + conn.MarkHTTPSuccess(time.Now()) if !app.AddDevice(host, conn) { // Lost a race — another goroutine inserted the same host // between TouchDevice and AddDevice. AddDevice bumped LastSeen @@ -122,6 +124,44 @@ func (app *WebApp) addDeviceByHost(host string, port int, source string) *webtyp return conn } +func resolvedDeviceIPAddress(host string, info *models.DeviceInfo) string { + if info != nil { + if address := numericIPAddress(info.IPAddress); address != "" { + return address + } + + for _, network := range info.NetworkInfo { + if address := numericIPAddress(network.IPAddress); address != "" { + return address + } + } + } + + bareHost := hostOnly(host) + if address := numericIPAddress(bareHost); address != "" { + return address + } + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + addresses, err := net.DefaultResolver.LookupNetIP(ctx, "ip4", bareHost) + if err == nil && len(addresses) > 0 { + return addresses[0].String() + } + + return "" +} + +func numericIPAddress(address string) string { + ip := net.ParseIP(strings.TrimSpace(address)) + if ip == nil { + return "" + } + + return ip.String() +} + // SeedExtraDevices registers any devices reported by the ExtraDeviceHosts hook // (if set) via AddDeviceByHost, and prunes any previously-seeded host that no // longer appears in that set. Idempotent: already-known hosts are skipped. diff --git a/pkg/service/soundtouchweb/discovery_test.go b/pkg/service/soundtouchweb/discovery_test.go index 5364bf15..b798bee0 100644 --- a/pkg/service/soundtouchweb/discovery_test.go +++ b/pkg/service/soundtouchweb/discovery_test.go @@ -102,6 +102,22 @@ func TestClassifySource(t *testing.T) { } } +func TestResolvedDeviceIPAddressSeparatesHostnameFromReportedAddress(t *testing.T) { + info := &models.DeviceInfo{NetworkInfo: []models.NetworkInfo{ + {Type: "SCM", IPAddress: "192.0.2.42"}, + }} + + if got := resolvedDeviceIPAddress("kitchen.local", info); got != "192.0.2.42" { + t.Fatalf("resolved address = %q, want reported speaker address", got) + } +} + +func TestResolvedDeviceIPAddressPreservesLiteralIP(t *testing.T) { + if got := resolvedDeviceIPAddress("192.0.2.20", &models.DeviceInfo{}); got != "192.0.2.20" { + t.Fatalf("resolved literal address = %q, want unchanged literal", got) + } +} + func TestRetryUntilReadyStopsAfterSuccess(t *testing.T) { attempts := 0 retryUntilReady(context.Background(), time.Millisecond, func() bool { diff --git a/pkg/service/soundtouchweb/frontend_test/devicePresentation.test.mjs b/pkg/service/soundtouchweb/frontend_test/devicePresentation.test.mjs new file mode 100644 index 00000000..d3a2326e --- /dev/null +++ b/pkg/service/soundtouchweb/frontend_test/devicePresentation.test.mjs @@ -0,0 +1,67 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; + +import { + connectivityLabel, + connectivityState, + sortDeviceEntries, +} from '../static/js/devicePresentation.js'; + +test('uses tri-state connectivity before the compatibility flag', () => { + assert.equal(connectivityState({ status: { connectivity: 'online', isConnected: false } }), 'online'); + assert.equal(connectivityState({ status: { connectivity: 'stale', isConnected: true } }), 'stale'); + assert.equal(connectivityState({ status: { connectivity: 'offline', isConnected: true } }), 'offline'); +}); + +test('falls back to the legacy connected flag and supplies a label', () => { + assert.equal(connectivityState({ status: { isConnected: true } }), 'online'); + assert.equal(connectivityState({ status: { isConnected: false } }), 'offline'); + assert.equal(connectivityState(undefined), 'offline'); + assert.equal(connectivityLabel({ status: { connectivity: 'stale' } }), 'Stale'); +}); + +test('sorts hostname control IDs by their numeric presentation IP', () => { + const entries = [ + ['speaker-a', { info: { ip_address: '192.0.2.10' } }], + ['speaker-b', { info: { ip_address: '192.0.2.2' } }], + ]; + + assert.deepEqual(sortDeviceEntries(entries, 'ip').map(([id]) => id), [ + 'speaker-b', + 'speaker-a', + ]); +}); + +test('sorts literal IP keys numerically and falls back when an address is missing', () => { + const entries = [ + ['192.0.2.10', { info: {} }], + ['192.0.2.2', {}], + ['192.0.2.1', { info: { ip_address: '' } }], + ]; + + assert.deepEqual(sortDeviceEntries(entries, 'ip').map(([id]) => id), [ + '192.0.2.1', + '192.0.2.2', + '192.0.2.10', + ]); +}); + +test('uses stable control-ID tie breaks for equal names and addresses', () => { + const byAddress = [ + ['speaker-b', { info: { ip_address: '192.0.2.4' } }], + ['speaker-a', { info: { ip_address: '192.0.2.4' } }], + ]; + const byName = [ + ['speaker-b', { info: { name: ' Kitchen ' } }], + ['speaker-a', { info: { name: 'kitchen' } }], + ]; + + assert.deepEqual(sortDeviceEntries(byAddress, 'ip').map(([id]) => id), [ + 'speaker-a', + 'speaker-b', + ]); + assert.deepEqual(sortDeviceEntries(byName, 'name').map(([id]) => id), [ + 'speaker-a', + 'speaker-b', + ]); +}); diff --git a/pkg/service/soundtouchweb/frontend_test/deviceRemoval.test.mjs b/pkg/service/soundtouchweb/frontend_test/deviceRemoval.test.mjs new file mode 100644 index 00000000..45cf2ec3 --- /dev/null +++ b/pkg/service/soundtouchweb/frontend_test/deviceRemoval.test.mjs @@ -0,0 +1,94 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; + +import { removeDeviceAndRefresh } from '../static/js/deviceRemoval.js'; + +function operation(overrides = {}) { + const calls = []; + return { + calls, + options: { + id: '192.0.2.10', + name: 'Kitchen', + remove: async (id) => { + calls.push(['remove', id]); + return { success: true }; + }, + refresh: async () => calls.push(['refresh']), + showDeviceList: () => calls.push(['showDeviceList']), + notify: (message) => calls.push(['notify', message]), + ...overrides, + }, + }; +} + +test('waits for successful removal before navigating and refreshing', async () => { + let resolveRemoval; + const pendingRemoval = new Promise(resolve => { resolveRemoval = resolve; }); + const { calls, options } = operation({ + remove: (id) => { + calls.push(['remove', id]); + return pendingRemoval; + }, + }); + + const result = removeDeviceAndRefresh(options); + await Promise.resolve(); + assert.deepEqual(calls, [['remove', '192.0.2.10']]); + + resolveRemoval({ success: true }); + assert.equal(await result, true); + assert.deepEqual(calls, [ + ['remove', '192.0.2.10'], + ['showDeviceList'], + ['refresh'], + ['notify', 'Removed "Kitchen"'], + ]); +}); + +test('keeps the current view and state when removal is rejected', async () => { + const { calls, options } = operation({ + remove: async (id) => { + calls.push(['remove', id]); + return { success: false, error: 'Device is busy' }; + }, + }); + + assert.equal(await removeDeviceAndRefresh(options), false); + assert.deepEqual(calls, [ + ['remove', '192.0.2.10'], + ['notify', 'Device is busy'], + ]); +}); + +test('reports transport failures without navigating or refreshing', async () => { + const { calls, options } = operation({ + remove: async (id) => { + calls.push(['remove', id]); + throw new Error('network failure'); + }, + }); + + assert.equal(await removeDeviceAndRefresh(options), false); + assert.deepEqual(calls, [ + ['remove', '192.0.2.10'], + ['notify', 'Failed to remove device'], + ]); +}); + +test('retains successful removal when the authoritative refresh fails', async () => { + const { calls, options } = operation({ + refresh: async () => { + calls.push(['refresh']); + throw new Error('refresh failure'); + }, + }); + + assert.equal(await removeDeviceAndRefresh(options), true); + assert.deepEqual(calls, [ + ['remove', '192.0.2.10'], + ['showDeviceList'], + ['refresh'], + ['notify', 'Removed "Kitchen", but failed to refresh devices'], + ]); +}); diff --git a/pkg/service/soundtouchweb/handler.go b/pkg/service/soundtouchweb/handler.go index 769c9924..fc02da21 100644 --- a/pkg/service/soundtouchweb/handler.go +++ b/pkg/service/soundtouchweb/handler.go @@ -398,7 +398,7 @@ func (app *WebApp) HandleAPIDevice(w http.ResponseWriter, r *http.Request) { app.UpdateDeviceStatus(deviceID, device) // Connect WebSocket for real-time updates if not already connected - if device.WebSocket == nil { + if device.CurrentWebSocket() == nil { go app.ConnectDeviceWebSocket(deviceID, device) } @@ -448,8 +448,8 @@ func (app *WebApp) HandleDeleteDevice(w http.ResponseWriter, r *http.Request) { // only prunes the in-memory registry below. if app.RemoveDeviceHook != nil { deviceID := "" - if conn.DeviceInfo != nil { - deviceID = conn.DeviceInfo.DeviceID + if info := conn.Info(); info != nil { + deviceID = info.DeviceID } if err := app.RemoveDeviceHook(deviceID); err != nil { @@ -460,7 +460,11 @@ func (app *WebApp) HandleDeleteDevice(w http.ResponseWriter, r *http.Request) { } } - app.RemoveDevice(host) + if !app.removeDeviceIfMatch(host, conn) { + app.sendError(w, "Device changed during removal", http.StatusConflict) + + return + } app.BroadcastDeviceList() w.Header().Set("Content-Type", "application/json") @@ -487,7 +491,7 @@ func (app *WebApp) HandleAPIControl(w http.ResponseWriter, r *http.Request) { } // Connect WebSocket for real-time updates if not already connected - if device.WebSocket == nil { + if device.CurrentWebSocket() == nil { go app.ConnectDeviceWebSocket(deviceID, device) } @@ -730,7 +734,7 @@ func (app *WebApp) HandleDeviceKey(w http.ResponseWriter, r *http.Request) { } // Connect WebSocket for real-time updates if not already connected - if device.WebSocket == nil { + if device.CurrentWebSocket() == nil { go app.ConnectDeviceWebSocket(deviceID, device) } @@ -762,7 +766,7 @@ func (app *WebApp) HandleDirectVolumeControl(w http.ResponseWriter, r *http.Requ } // Connect WebSocket for real-time updates if not already connected - if device.WebSocket == nil { + if device.CurrentWebSocket() == nil { go app.ConnectDeviceWebSocket(deviceID, device) } @@ -788,7 +792,7 @@ func (app *WebApp) HandleDevicePower(w http.ResponseWriter, r *http.Request) { } // Connect WebSocket for real-time updates if not already connected - if device.WebSocket == nil { + if device.CurrentWebSocket() == nil { go app.ConnectDeviceWebSocket(deviceID, device) } @@ -1014,7 +1018,7 @@ func (app *WebApp) HandleTuneInNavigate(w http.ResponseWriter, r *http.Request) // Returns "" when no match is found. func (app *WebApp) findIPByHwID(hwID string) string { for _, entry := range app.DeviceSnapshot() { - if entry.Device.DeviceInfo != nil && entry.Device.DeviceInfo.DeviceID == hwID { + if info := entry.Device.Info(); info != nil && info.DeviceID == hwID { return entry.ID } } @@ -1053,8 +1057,10 @@ func (app *WebApp) HandleGetZone(w http.ResponseWriter, r *http.Request) { masterIP := app.findIPByHwID(zone.Master) masterName := "" - if conn, ok := app.GetDevice(masterIP); ok && conn.DeviceInfo != nil { - masterName = conn.DeviceInfo.Name + if conn, ok := app.GetDevice(masterIP); ok { + if info := conn.Info(); info != nil { + masterName = info.Name + } } type memberInfo struct { @@ -1067,8 +1073,10 @@ func (app *WebApp) HandleGetZone(w http.ResponseWriter, r *http.Request) { for _, m := range zone.Members { name := "" - if conn, ok := app.GetDevice(m.IP); ok && conn.DeviceInfo != nil { - name = conn.DeviceInfo.Name + if conn, ok := app.GetDevice(m.IP); ok { + if info := conn.Info(); info != nil { + name = info.Name + } } members = append(members, memberInfo{IP: m.IP, HwID: m.DeviceID, Name: name}) diff --git a/pkg/service/soundtouchweb/handler_test.go b/pkg/service/soundtouchweb/handler_test.go index 67e5f413..a4103ab2 100644 --- a/pkg/service/soundtouchweb/handler_test.go +++ b/pkg/service/soundtouchweb/handler_test.go @@ -117,6 +117,39 @@ func TestHandleAPIDevices(t *testing.T) { } } +func TestHandleDeleteDevicePreservesConcurrentReplacement(t *testing.T) { + app := NewWebApp() + host := "speaker.local" + original := webtypes.NewDeviceConnection(nil, &models.DeviceInfo{DeviceID: "ORIGINAL"}) + replacement := webtypes.NewDeviceConnection(nil, &models.DeviceInfo{DeviceID: "REPLACEMENT"}) + app.AddDevice(host, original) + app.RemoveDeviceHook = func(deviceID string) error { + if deviceID != "ORIGINAL" { + t.Fatalf("RemoveDeviceHook deviceID = %q, want ORIGINAL", deviceID) + } + app.RemoveDevice(host) + app.AddDevice(host, replacement) + + return nil + } + + req := httptest.NewRequest(http.MethodDelete, "/api/control/devices/"+host, nil) + req = withChiParams(req, map[string]string{"id": host}) + w := httptest.NewRecorder() + + app.HandleDeleteDevice(w, req) + + if w.Code != http.StatusConflict { + t.Fatalf("status = %d, want %d", w.Code, http.StatusConflict) + } + got, ok := app.GetDevice(host) + if !ok || got != replacement { + t.Fatal("concurrent replacement was removed") + } + + app.RemoveDevice(host) +} + func TestHandleAPIDevice(t *testing.T) { app := createTestApp() diff --git a/pkg/service/soundtouchweb/static/css/app.css b/pkg/service/soundtouchweb/static/css/app.css index bca70b8d..a1dc4748 100644 --- a/pkg/service/soundtouchweb/static/css/app.css +++ b/pkg/service/soundtouchweb/static/css/app.css @@ -10,7 +10,9 @@ --accent: #000000; --accent-fg: #ffffff; --online: #22c55e; + --stale: #f59e0b; --offline: #9ca3af; + --danger: #b42318; --radius: 8px; --shadow: 0 1px 3px rgba(0,0,0,.10), 0 1px 2px rgba(0,0,0,.06); --nav-icon-filter: brightness(0) invert(1); @@ -25,6 +27,7 @@ --text-dim: #ccc; --accent: #e0e0e0; --accent-fg:#111; + --danger: #ff8a80; --nav-icon-filter: none; } } @@ -356,10 +359,15 @@ img { display: block; max-width: 100%; } } .device-card { + appearance: none; + width: 100%; background: var(--surface); border: 1px solid var(--border); border-radius: var(--radius); padding: 1rem; + color: inherit; + font: inherit; + text-align: left; cursor: pointer; transition: box-shadow .15s, transform .1s; box-shadow: var(--shadow); @@ -368,30 +376,12 @@ img { display: block; max-width: 100%; } min-width: 0; } .device-card:hover { box-shadow: 0 4px 12px rgba(0,0,0,.12); transform: translateY(-1px); } +.device-card:focus-visible { outline: 2px solid var(--accent); outline-offset: 2px; } .device-header { display: flex; align-items: center; justify-content: space-between; margin-bottom: .25rem; gap: .5rem; } .device-name { font-weight: 600; font-size: .95rem; min-width: 0; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; } .device-header-right { display: flex; align-items: center; gap: .5rem; flex-shrink: 0; } -/* Quiet remove affordance: invisible until the card is hovered, then dim, - warming to red only on its own hover. Keeps the grid relaxed. */ -.device-remove { - border: 0; - background: none; - padding: 0; - width: 1.1rem; - height: 1.1rem; - line-height: 1; - font-size: .8rem; - color: var(--text-dim); - cursor: pointer; - opacity: 0; - transition: opacity .15s, color .15s; -} -.device-card:hover .device-remove { opacity: .55; } -.device-remove:hover { opacity: 1; color: var(--offline); } -.device-remove:focus-visible { opacity: 1; outline: 2px solid var(--offline); outline-offset: 2px; } - .device-list-note { margin: .9rem .15rem 0; font-size: .78rem; @@ -408,6 +398,7 @@ img { display: block; max-width: 100%; } flex-shrink: 0; } .device-indicator.online { background: var(--online); } +.device-indicator.stale { background: var(--stale); } .device-indicator.offline { background: var(--offline); } .now-playing-mini { font-size: .8rem; color: var(--text-dim); margin-top: .25rem; white-space: nowrap; overflow: hidden; text-overflow: ellipsis; } @@ -417,6 +408,28 @@ img { display: block; max-width: 100%; } /* ── Device detail ───────────────────────────────────────────────────────── */ .device-detail { max-width: 560px; } +.device-management-section { + margin-top: 1.5rem; + padding-top: 1.25rem; + border-top: 1px solid var(--border); +} + +.device-remove-action { + display: inline-flex; + align-items: center; + gap: .45rem; + min-height: 44px; + color: var(--text-dim); +} +.device-remove-action:hover { + color: var(--danger); + border-color: var(--danger); +} +.device-remove-action:focus-visible { + outline: 2px solid var(--danger); + outline-offset: 2px; +} + /* ── Now playing ─────────────────────────────────────────────────────────── */ .now-playing { display: flex; diff --git a/pkg/service/soundtouchweb/static/js/app.js b/pkg/service/soundtouchweb/static/js/app.js index a75d363f..ff453b8d 100644 --- a/pkg/service/soundtouchweb/static/js/app.js +++ b/pkg/service/soundtouchweb/static/js/app.js @@ -17,10 +17,11 @@ import { TTS } from './components/TTS.js'; import { Announcements } from './components/Announcements.js'; import { api } from './api.js'; import { isSoundTouch10StereoPair } from './stereoPresentation.mjs'; +import { removeDeviceAndRefresh } from './deviceRemoval.js'; const html = htm.bind(h); -function DeviceDetail({ deviceId, devices, onBack, onDevicesChanged, notify }) { +function DeviceDetail({ deviceId, devices, onBack, onDevicesChanged, notify, onRemove }) { const device = devices[deviceId]; if (!device) { @@ -62,6 +63,26 @@ function DeviceDetail({ deviceId, devices, onBack, onDevicesChanged, notify }) { ` : null} <${Zone} deviceId=${deviceId} devices=${devices} /> <${Recents} deviceId=${deviceId} /> + ${!device.stereoPair ? html` +
+
Device management
+ +
+ ` : null} `; } @@ -178,26 +199,23 @@ function App() { async function refreshDevices() { const resp = await api.devices(); - if (resp?.success) setDevices(resp.data || {}); + if (!resp?.success) throw new Error(resp?.error || 'Failed to refresh devices'); + setDevices(resp.data || {}); } async function removeDevice(id) { const name = devices[id]?.info?.name || id; - if (!confirm(`Remove "${name}"?\n\nThis clears it from AfterTouch. A device still online may reappear after the next discovery scan.`)) { + if (!confirm(`Remove "${name}" from AfterTouch?\n\nThis does not reset the speaker. A device still online may reappear after the next discovery scan.`)) { return; } - // Optimistically drop it; the server's devices broadcast reconciles. - setDevices(prev => { - const next = { ...prev }; - delete next[id]; - return next; + await removeDeviceAndRefresh({ + id, + name, + remove: api.removeDevice, + refresh: refreshDevices, + showDeviceList: () => navigate('devices'), + notify: showToast, }); - try { - const resp = await api.removeDevice(id); - showToast(resp?.success ? `Removed "${name}"` : (resp?.error || 'Failed to remove device')); - } catch (err) { - showToast('Failed to remove device'); - } } return html` @@ -274,7 +292,6 @@ function App() { isDiscovering=${isDiscovering} onSelect=${(id) => navigate('device', id)} onDiscover=${discover} - onRemove=${removeDevice} /> ` : page === 'device' ? html` <${DeviceDetail} @@ -284,6 +301,7 @@ function App() { onBack=${() => navigate('devices')} onDevicesChanged=${refreshDevices} notify=${showToast} + onRemove=${removeDevice} /> ` : page === 'tunein' ? html` <${TuneInBrowser} key="tunein-browser" devices=${devices} /> @@ -310,7 +328,8 @@ function App() { ` : null} - ${toast ? html`
${toast}
` : null} + ${toast ? html`
${toast}
` : null} `; } diff --git a/pkg/service/soundtouchweb/static/js/components/DeviceList.js b/pkg/service/soundtouchweb/static/js/components/DeviceList.js index c2ad0e4f..3756f72f 100644 --- a/pkg/service/soundtouchweb/static/js/components/DeviceList.js +++ b/pkg/service/soundtouchweb/static/js/components/DeviceList.js @@ -1,45 +1,33 @@ import { h } from 'preact'; import { useState } from 'preact/hooks'; import htm from 'htm'; +import { + connectivityLabel, + connectivityState, + sortDeviceEntries, +} from '../devicePresentation.js'; const html = htm.bind(h); const SORT_LS_KEY = 'aftertouch_device_sort'; -function sortEntries(entries, mode) { - const copy = [...entries]; - if (mode === 'name') { - // Sort by the speaker's display name, falling back to the map key (its IP) - // when a device has no name yet. - copy.sort(([idA, a], [idB, b]) => - (a?.info?.name || idA).localeCompare(b?.info?.name || idB, undefined, { sensitivity: 'base' })); - } else { - // Default: by IP (the map key), ordered numerically so .2 precedes .10. - copy.sort(([idA], [idB]) => - idA.localeCompare(idB, undefined, { numeric: true, sensitivity: 'base' })); - } - return copy; -} - -function DeviceCard({ id, device, onSelect, onRemove }) { +function DeviceCard({ id, device, onSelect }) { const { info, status } = device; const stereoPair = device.stereoPair; const np = status?.nowPlaying; const isPlaying = np?.PlayStatus === 'PLAY_STATE'; const isStandby = !np || np.Source === 'STANDBY'; + const connectivity = connectivityState(device); + const statusLabel = `Connectivity: ${connectivityLabel(device)}`; return html` -
onSelect(id)}> -
+ ` : null} - -
-
+ + + ${info?.type || ''} ${info?.ip_address ? html`(${info.ip_address})` : null} ${stereoPair ? html` @@ -47,20 +35,20 @@ function DeviceCard({ id, device, onSelect, onRemove }) { Stereo pair ${stereoPair.availableMemberCount}/${stereoPair.memberCount} ` : null} -
+ ${!isStandby ? html` -
+ ${isPlaying ? '▶' : '⏸'} ${np.Track || np.StationName || np.Source} ${np.Artist ? html` — ${np.Artist}` : null} -
+ ` : null} - ${isStandby ? html`
Standby
` : null} -
+ ${isStandby ? html`Standby` : null} + `; } -export function DeviceList({ devices, isDiscovering, onSelect, onDiscover, onRemove }) { +export function DeviceList({ devices, isDiscovering, onSelect, onDiscover }) { const [sortMode, setSortMode] = useState(() => localStorage.getItem(SORT_LS_KEY) || 'ip'); function changeSort(mode) { @@ -68,7 +56,7 @@ export function DeviceList({ devices, isDiscovering, onSelect, onDiscover, onRem localStorage.setItem(SORT_LS_KEY, mode); } - const entries = sortEntries(Object.entries(devices), sortMode); + const entries = sortDeviceEntries(Object.entries(devices), sortMode); return html`
@@ -91,13 +79,9 @@ export function DeviceList({ devices, isDiscovering, onSelect, onDiscover, onRem
${entries.map(([id, device]) => html` - <${DeviceCard} key=${id} id=${id} device=${device} onSelect=${onSelect} onRemove=${onRemove} /> + <${DeviceCard} key=${id} id=${id} device=${device} onSelect=${onSelect} /> `)} -
-

- Removing a device clears it here. One that is still online may - reappear after the next discovery scan. -

` + ` } `; diff --git a/pkg/service/soundtouchweb/static/js/devicePresentation.js b/pkg/service/soundtouchweb/static/js/devicePresentation.js new file mode 100644 index 00000000..8cceca6c --- /dev/null +++ b/pkg/service/soundtouchweb/static/js/devicePresentation.js @@ -0,0 +1,42 @@ +const connectivityStates = new Set(['online', 'stale', 'offline']); + +export function connectivityState(device) { + const reported = device?.status?.connectivity; + if (connectivityStates.has(reported)) return reported; + + return device?.status?.isConnected ? 'online' : 'offline'; +} + +export function connectivityLabel(device) { + const state = connectivityState(device); + return state.charAt(0).toUpperCase() + state.slice(1); +} + +function compareText(a, b) { + return String(a).localeCompare(String(b), undefined, { + numeric: true, + sensitivity: 'base', + }); +} + +function displayName(id, device) { + return String(device?.info?.name || '').trim() || id; +} + +function presentationAddress(id, device) { + return String(device?.info?.ip_address || '').trim() || id; +} + +export function sortDeviceEntries(entries, mode) { + const sorted = [...entries]; + + sorted.sort(([idA, deviceA], [idB, deviceB]) => { + const primary = mode === 'name' + ? compareText(displayName(idA, deviceA), displayName(idB, deviceB)) + : compareText(presentationAddress(idA, deviceA), presentationAddress(idB, deviceB)); + + return primary || compareText(idA, idB); + }); + + return sorted; +} diff --git a/pkg/service/soundtouchweb/static/js/deviceRemoval.js b/pkg/service/soundtouchweb/static/js/deviceRemoval.js new file mode 100644 index 00000000..d3ea6fea --- /dev/null +++ b/pkg/service/soundtouchweb/static/js/deviceRemoval.js @@ -0,0 +1,32 @@ +export async function removeDeviceAndRefresh({ + id, + name, + remove, + refresh, + showDeviceList, + notify, +}) { + let response; + try { + response = await remove(id); + } catch (_) { + notify('Failed to remove device'); + return false; + } + + if (!response?.success) { + notify(response?.error || 'Failed to remove device'); + return false; + } + + showDeviceList(); + try { + await refresh(); + } catch (_) { + notify(`Removed "${name}", but failed to refresh devices`); + return true; + } + + notify(`Removed "${name}"`); + return true; +} diff --git a/pkg/service/soundtouchweb/websocket.go b/pkg/service/soundtouchweb/websocket.go index ed227239..74a02fe0 100644 --- a/pkg/service/soundtouchweb/websocket.go +++ b/pkg/service/soundtouchweb/websocket.go @@ -13,6 +13,7 @@ import ( "sync" "time" + "github.com/gesellix/bose-soundtouch/pkg/client" "github.com/gesellix/bose-soundtouch/pkg/models" "github.com/gesellix/bose-soundtouch/pkg/service/soundtouchweb/webtypes" "github.com/go-chi/chi/v5" @@ -482,22 +483,19 @@ func (app *WebApp) HandleAPIDiscover(w http.ResponseWriter, r *http.Request) { } } -// ConnectDeviceWebSocket establishes a WebSocket connection to a device -// and keeps it alive: on disconnect or connect failure, it reconnects -// with exponential backoff (1 s → 30 s cap, reset after each successful -// connect). The goroutine runs for the lifetime of the device entry, -// so status flows from the speaker keep streaming through transient -// network blips, speaker reboots, and idle timeouts. -// -// conn.WebSocket is only updated on a successful (re)connect, never -// cleared, so the duplicate-spawn guards at the callsites (which check -// `if device.WebSocket == nil`) stay correct — once this goroutine is -// running for a device, no second one is needed. +// ConnectDeviceWebSocket starts the single event-transport supervisor for a +// device. Initial connection failures are retried here; after the first +// success WebSocketClient owns transport reconnects and this supervisor +// observes their state until the device is removed. func (app *WebApp) ConnectDeviceWebSocket(deviceID string, conn *webtypes.DeviceConnection) { // Skip WebSocket connection if client is not available (e.g., in tests) if conn.Client == nil { return } + if !conn.TryStartWebSocketLoop() { + return + } + defer conn.FinishWebSocketLoop() const ( initialBackoff = 1 * time.Second @@ -524,6 +522,7 @@ func (app *WebApp) ConnectDeviceWebSocket(deviceID string, conn *webtypes.Device // UpdateStatus so concurrent events and the periodic poller // (UpdateDeviceStatus) cannot lose each other's writes. wsClient.OnNowPlaying(func(event *models.NowPlayingUpdatedEvent) { + activity := time.Now() np := &event.NowPlaying // A /select returns 200 even when the source is rejected; the @@ -536,29 +535,77 @@ func (app *WebApp) ConnectDeviceWebSocket(deviceID string, conn *webtypes.Device prevSource = np.Source app.applyNowPlayingEvent(conn, np) + conn.MarkEventStreamActivity(activity) }) wsClient.OnVolumeUpdated(func(event *models.VolumeUpdatedEvent) { + activity := time.Now() app.applyVolumeEvent(conn, &event.Volume) + conn.MarkEventStreamActivity(activity) }) wsClient.OnConnectionState(func(event *models.ConnectionStateUpdatedEvent) { + if !speakerConnectionEventMatches(conn, event.DeviceID) { + log.Printf("Ignoring connection state for mismatched device %s on %s", + sanitizeLog(event.DeviceID), sanitizeLog(deviceID)) + + return + } + app.applyConnectionStateEvent(conn, event.ConnectionState.IsConnected()) + conn.ApplySpeakerConnectionEvent(webtypes.SpeakerConnectionState{ + State: event.ConnectionState.State, + Signal: event.ConnectionState.Signal, + }, time.Now()) }) wsClient.OnPresetUpdated(func(event *models.PresetUpdatedEvent) { + activity := time.Now() app.applyPresetEvent(conn, &event.Presets) + conn.MarkEventStreamActivity(activity) }) wsClient.OnBassUpdated(func(event *models.BassUpdatedEvent) { + activity := time.Now() app.applyBassEvent(conn, &event.Bass) + conn.MarkEventStreamActivity(activity) }) wsClient.OnGroupUpdated(func(event *models.GroupUpdatedEvent) { app.applyGroupUpdatedEvent(conn, event) }) - if err := wsClient.Connect(); err != nil { + wsClient.OnNameUpdated(func(event *models.NameUpdatedEvent) { + conn.MarkEventStreamActivity(time.Now()) + conn.ApplyNameEvent(event.Name.Value) + }) + + wsClient.OnTransportState(func(connected bool, generation uint64) { + if !conn.ObserveEventStreamTransport(generation, connected, time.Now()) { + return + } + + // Drives the same FieldConnectivity fencing the old inline + // connect/disconnect sites used to update directly -- ordered by + // ObserveEventStreamTransport's own generation check above, so a + // reordered transport callback can no longer apply here either. + app.applyConnectionStateEvent(conn, connected) + + if connected { + if generation > 1 { + log.Printf("WebSocket reconnected for device %s", sanitizeLog(deviceID)) + go app.UpdateDeviceStatus(deviceID, conn) + } + } else { + log.Printf("WebSocket transport disconnected for device %s", sanitizeLog(deviceID)) + } + }) + + published, err := publishAndConnectDeviceWebSocket(conn, wsClient, wsClient.Connect) + if !published { + return + } + if err != nil { log.Printf("Failed to connect WebSocket for device %s: %v (retrying in %s)", sanitizeLog(deviceID), err, backoff) if sleepOrDone(conn, backoff) { @@ -573,10 +620,6 @@ func (app *WebApp) ConnectDeviceWebSocket(deviceID string, conn *webtypes.Device continue } - conn.WebSocket = wsClient - - app.applyConnectionStateEvent(conn, true) - log.Printf("WebSocket connected for device %s", sanitizeLog(deviceID)) // Fetch current state immediately: speakers do not replay events on @@ -588,24 +631,45 @@ func (app *WebApp) ConnectDeviceWebSocket(deviceID string, conn *webtypes.Device // starts at the lowest cadence again. backoff = initialBackoff - // Block until the device-side WebSocket disconnects. - wsClient.Wait() + <-conn.Done() - app.applyConnectionStateEvent(conn, false) - - log.Printf("WebSocket disconnected for device %s — reconnecting in %s", sanitizeLog(deviceID), backoff) - - if sleepOrDone(conn, backoff) { - return - } - - backoff *= 2 - if backoff > maxBackoff { - backoff = maxBackoff - } + return } } +func publishAndConnectDeviceWebSocket( + conn *webtypes.DeviceConnection, + wsClient *client.WebSocketClient, + connect func() error, +) (bool, error) { + if !conn.SetWebSocket(wsClient) { + return false, nil + } + + if err := connect(); err != nil { + conn.ClearWebSocket(wsClient) + _ = wsClient.Close() + + return true, err + } + + return true, nil +} + +func speakerConnectionEventMatches(conn *webtypes.DeviceConnection, eventDeviceID string) bool { + eventDeviceID = strings.TrimSpace(eventDeviceID) + if eventDeviceID == "" { + return true + } + + info := conn.Info() + if info == nil || strings.TrimSpace(info.DeviceID) == "" { + return false + } + + return strings.EqualFold(eventDeviceID, strings.TrimSpace(info.DeviceID)) +} + // sleepOrDone waits for d to elapse or for the connection to be closed, // whichever comes first. It returns true if the connection was closed // (the caller should stop), false if the timer fired normally. @@ -624,13 +688,17 @@ func sleepOrDone(conn *webtypes.DeviceConnection, d time.Duration) bool { // UpdateDeviceStatus fetches current status from the device. // // Network calls run outside any atomic merge so slow I/O never blocks a -// concurrent CAS retry. Each field (NowPlaying/Volume/Presets/Sources/Bass, -// plus derived connectivity) is merged and ordered independently via its own -// StatusField generation (BeginFieldPoll/CompleteFieldPoll/ApplyFieldEvent in -// webtypes) -- a real-time push event, or a concurrent poll, for one field -// can supersede only that field. A slow-but-successful fetch for one field -// is never discarded merely because a DIFFERENT field's event or poll -// completion happened to land first. +// concurrent CAS retry. Each field (NowPlaying/Name/Volume/Presets/Sources/ +// Bass, plus derived connectivity) is merged and ordered independently via +// its own StatusField generation (BeginFieldPoll/CompleteFieldPoll/ +// ApplyFieldEvent in webtypes) -- a real-time push event, or a concurrent +// poll, for one field can supersede only that field. A slow-but-successful +// fetch for one field is never discarded merely because a DIFFERENT field's +// event or poll completion happened to land first. Independently, the same +// round also feeds BeginHTTPPoll/CompleteHTTPPoll, which derives HTTP +// reachability and the Online/Stale/Offline connectivity classification -- +// CompleteHTTPPoll is called with a nil merge func here since field merging +// is already handled per-field above; it only records health/connectivity. func (app *WebApp) UpdateDeviceStatus(deviceID string, conn *webtypes.DeviceConnection) { app.updateDeviceStatus(deviceID, conn, nil) } @@ -658,6 +726,7 @@ func (app *WebApp) updateDeviceStatus(_ string, conn *webtypes.DeviceConnection, sourcesGen := conn.BeginFieldPoll(webtypes.FieldSources) bassGen := conn.BeginFieldPoll(webtypes.FieldBass) connectivityGen := conn.BeginFieldPoll(webtypes.FieldConnectivity) + pollGeneration := conn.BeginHTTPPoll() // /getGroup must be gated to ST10 models -- see Client.GetGroup's doc // comment (verified against real hardware: a ST20 never replies at all, @@ -668,11 +737,13 @@ func (app *WebApp) updateDeviceStatus(_ string, conn *webtypes.DeviceConnection, if stereoCapable && groupBaseline == nil { groupGeneration = conn.BeginGroupRefresh() } + nameGeneration := conn.BeginNameRefresh() // Phase 1: slow network fetches. Local vars only, no shared state // is touched yet. Errors are recorded so the merge below can tell // "field N stayed unchanged" apart from "field N got refreshed". nowPlaying, nowPlayingErr := conn.Client.GetNowPlaying() + name, nameErr := conn.Client.GetName() volume, volumeErr := conn.Client.GetVolume() presets, presetsErr := conn.Client.GetPresets() sources, sourcesErr := conn.Client.GetSources() @@ -737,6 +808,12 @@ func (app *WebApp) updateDeviceStatus(_ string, conn *webtypes.DeviceConnection, }) } + if nameErr == nil { + anyFetchSucceeded = true + + conn.ApplyPolledName(nameGeneration, name.Value) + } + // Mark as connected if we successfully got at least one status from // this round. Mirrors prior behaviour: deliberately does NOT fold // groupErr in here. GetGroup is gated to stereo-capable models and @@ -749,6 +826,15 @@ func (app *WebApp) updateDeviceStatus(_ string, conn *webtypes.DeviceConnection, s.LastActivity = time.Now() }) + // Independently of the per-field merges above, also record this round + // against the health/connectivity generation. merge is nil: field data + // was already merged field-by-field above, so this call only derives + // HTTPReachable/WebSocketConnected/Connectivity (Online/Stale/Offline) + // from anyFetchSucceeded -- it must never re-apply payload fields, or a + // concurrent speaker event landing between BeginHTTPPoll and here could + // cause this call to silently discard part of the merge above. + conn.CompleteHTTPPoll(pollGeneration, anyFetchSucceeded, time.Now(), nil) + if stereoCapable && groupErr == nil { if groupBaseline != nil { conn.ApplyPolledGroupIfBaseline(*groupBaseline, group) @@ -762,6 +848,8 @@ func (app *WebApp) applyGroupUpdatedEvent( conn *webtypes.DeviceConnection, event *models.GroupUpdatedEvent, ) bool { + conn.MarkEventStreamActivity(time.Now()) + return app.queueBroadcastIfChanged(conn.ApplyGroupEvent(&event.Group, time.Now())) } @@ -797,7 +885,7 @@ func (app *WebApp) HandleDeviceWebSocket(w http.ResponseWriter, r *http.Request) Type: "device_status", DeviceID: deviceID, Data: map[string]interface{}{ - "info": device.DeviceInfo, + "info": device.Info(), "status": device.Status(), }, }) @@ -857,14 +945,14 @@ func (app *WebApp) writeDeviceWebSocketUpdate( Type: "device_status", DeviceID: deviceID, Data: map[string]interface{}{ - "info": device.DeviceInfo, + "info": device.Info(), "status": status, }, }); err != nil { return err } - if device.WebSocket == nil || !status.IsConnected { + if device.CurrentWebSocket() == nil || !status.IsConnected { return nil } diff --git a/pkg/service/soundtouchweb/websocket_test.go b/pkg/service/soundtouchweb/websocket_test.go index 707db0b2..8e4a13ef 100644 --- a/pkg/service/soundtouchweb/websocket_test.go +++ b/pkg/service/soundtouchweb/websocket_test.go @@ -1,6 +1,7 @@ package soundtouchweb import ( + "context" "errors" "net/http" "net/http/httptest" @@ -16,6 +17,54 @@ import ( "github.com/gorilla/websocket" ) +func TestPublishAndConnectDeviceWebSocketLetsRemovalCancelInitialDial(t *testing.T) { + conn := webtypes.NewDeviceConnection(nil, nil) + wsClient := client.NewClientFromHost("192.0.2.10").NewWebSocketClient(nil) + connectStarted := make(chan struct{}) + connectDone := make(chan struct { + published bool + err error + }, 1) + + go func() { + published, err := publishAndConnectDeviceWebSocket(conn, wsClient, func() error { + close(connectStarted) + wsClient.Wait() + + return context.Canceled + }) + connectDone <- struct { + published bool + err error + }{published: published, err: err} + }() + + select { + case <-connectStarted: + case <-time.After(time.Second): + t.Fatal("initial dial did not start") + } + + if got := conn.CurrentWebSocket(); got != wsClient { + t.Fatalf("CurrentWebSocket() = %p during initial dial, want %p", got, wsClient) + } + + conn.Close() + + select { + case result := <-connectDone: + if !result.published || result.err == nil { + t.Fatalf("publish/connect result = (%v, %v), want (true, error)", result.published, result.err) + } + case <-time.After(time.Second): + t.Fatal("device removal did not cancel the initial dial") + } + + if got := conn.CurrentWebSocket(); got != nil { + t.Fatalf("CurrentWebSocket() = %p after removal, want nil", got) + } +} + func TestUpdateDeviceStatusRefreshesGroup(t *testing.T) { server := newStatusTestServer(t, http.StatusOK, ` Living Room @@ -69,6 +118,101 @@ func TestUpdateDeviceStatusPreservesGroupOnError(t *testing.T) { } } +func TestSpeakerConnectionEventMatchesRegisteredHardwareID(t *testing.T) { + conn := webtypes.NewDeviceConnection(nil, &models.DeviceInfo{DeviceID: "AA11BB22CC33"}) + + for _, eventDeviceID := range []string{"", "AA11BB22CC33", "aa11bb22cc33"} { + if !speakerConnectionEventMatches(conn, eventDeviceID) { + t.Fatalf("event device ID %q should match", eventDeviceID) + } + } + if speakerConnectionEventMatches(conn, "DEADBEEF0000") { + t.Fatal("mismatched speaker connection event was accepted") + } + if speakerConnectionEventMatches(webtypes.NewDeviceConnection(nil, nil), "AA11BB22CC33") { + t.Fatal("identified event matched a connection without device identity") + } +} + +func TestUpdateDeviceStatusRefreshesDeviceName(t *testing.T) { + server := newStatusTestServer(t, http.StatusOK, ``) + defer server.Close() + + conn := webtypes.NewDeviceConnection( + client.NewClientFromHost(server.URL), + &models.DeviceInfo{Name: "Old Name", DeviceID: "device-1"}, + ) + NewWebApp().UpdateDeviceStatus("device-1", conn) + + if info := conn.Info(); info == nil || info.Name != "Living Room Left" { + t.Fatalf("refreshed device info = %+v, want Living Room Left", info) + } +} + +func TestUpdateDeviceStatusDoesNotOverwriteNewerNameEvent(t *testing.T) { + nameRequestStarted := make(chan struct{}) + releaseNameResponse := make(chan struct{}) + responses := map[string]string{ + "/now_playing": `STOP_STATE`, + "/volume": `1010false`, + "/presets": ``, + "/sources": ``, + "/bass": `00`, + "/getGroup": ``, + } + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/xml") + if r.URL.Path == "/name" { + close(nameRequestStarted) + <-releaseNameResponse + _, _ = w.Write([]byte(`Old Poll Result`)) + + return + } + + body, ok := responses[r.URL.Path] + if !ok { + t.Errorf("unexpected status endpoint %q", r.URL.Path) + w.WriteHeader(http.StatusNotFound) + + return + } + _, _ = w.Write([]byte(body)) + })) + defer server.Close() + + conn := webtypes.NewDeviceConnection( + client.NewClientFromHost(server.URL), + &models.DeviceInfo{Name: "Initial Name", DeviceID: "device-1", Type: "SoundTouch 10"}, + ) + refreshDone := make(chan struct{}) + go func() { + NewWebApp().UpdateDeviceStatus("device-1", conn) + close(refreshDone) + }() + + select { + case <-nameRequestStarted: + case <-time.After(time.Second): + close(releaseNameResponse) + t.Fatal("name poll did not start") + } + + conn.MarkEventStreamActivity(time.Now()) + conn.ApplyNameEvent("New Event Name") + close(releaseNameResponse) + + select { + case <-refreshDone: + case <-time.After(time.Second): + t.Fatal("status refresh did not finish") + } + + if info := conn.Info(); info == nil || info.Name != "New Event Name" { + t.Fatalf("device info after stale poll = %+v, want newer event name", info) + } +} + func TestUpdateDeviceStatusSkipsGroupForNonStereoModel(t *testing.T) { var groupRequested atomic.Bool server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -81,6 +225,7 @@ func TestUpdateDeviceStatusSkipsGroupForNonStereoModel(t *testing.T) { responses := map[string]string{ "/now_playing": `STOP_STATE`, + "/name": `Living Room Left`, "/volume": `1010false`, "/presets": ``, "/sources": ``, @@ -278,6 +423,7 @@ func newStatusTestServer(t *testing.T, groupStatus int, groupBody string) *httpt responses := map[string]string{ "/now_playing": `STOP_STATE`, + "/name": `Living Room Left`, "/volume": `1010false`, "/presets": ``, "/sources": ``, diff --git a/pkg/service/soundtouchweb/webtypes/status_test.go b/pkg/service/soundtouchweb/webtypes/status_test.go index 083f920d..a3e7ad30 100644 --- a/pkg/service/soundtouchweb/webtypes/status_test.go +++ b/pkg/service/soundtouchweb/webtypes/status_test.go @@ -9,6 +9,7 @@ import ( "testing" "time" + "github.com/gesellix/bose-soundtouch/pkg/client" "github.com/gesellix/bose-soundtouch/pkg/models" ) @@ -23,12 +24,270 @@ func TestNewDeviceConnection_InitialStatus(t *testing.T) { if status.IsConnected { t.Error("IsConnected should default to false") } + if status.Connectivity != ConnectivityOffline { + t.Errorf("Connectivity = %q, want %q", status.Connectivity, ConnectivityOffline) + } if status.LastActivity.IsZero() { t.Error("LastActivity should be initialised, got zero time") } } +func TestDeviceConnectionRejectsWebSocketAfterClose(t *testing.T) { + conn := NewDeviceConnection(nil, nil) + conn.Close() + + ws := client.NewClientFromHost("192.0.2.10").NewWebSocketClient(nil) + if conn.SetWebSocket(ws) { + t.Fatal("SetWebSocket() accepted a transport after Close()") + } + if got := conn.CurrentWebSocket(); got != nil { + t.Fatalf("CurrentWebSocket() = %p after Close(), want nil", got) + } + + waitDone := make(chan struct{}) + go func() { + ws.Wait() + close(waitDone) + }() + + select { + case <-waitDone: + case <-time.After(100 * time.Millisecond): + t.Fatal("rejected WebSocket transport was not stopped") + } +} + +func TestDeviceConnectionWebSocketAccessConcurrentWithClose(t *testing.T) { + conn := NewDeviceConnection(nil, nil) + ws := client.NewClientFromHost("192.0.2.10").NewWebSocketClient(nil) + start := make(chan struct{}) + var wg sync.WaitGroup + + wg.Add(2) + go func() { + defer wg.Done() + <-start + for range 100 { + conn.SetWebSocket(ws) + _ = conn.CurrentWebSocket() + } + }() + go func() { + defer wg.Done() + <-start + conn.Close() + }() + + close(start) + wg.Wait() + + if got := conn.CurrentWebSocket(); got != nil { + t.Fatalf("CurrentWebSocket() = %p after concurrent Close(), want nil", got) + } +} + +func TestDeviceConnectionClearWebSocketPreservesReplacement(t *testing.T) { + conn := NewDeviceConnection(nil, nil) + soundTouchClient := client.NewClientFromHost("192.0.2.10") + original := soundTouchClient.NewWebSocketClient(nil) + replacement := soundTouchClient.NewWebSocketClient(nil) + + conn.SetWebSocket(original) + conn.SetWebSocket(replacement) + if conn.ClearWebSocket(original) { + t.Fatal("ClearWebSocket() cleared a replacement transport") + } + if got := conn.CurrentWebSocket(); got != replacement { + t.Fatalf("CurrentWebSocket() = %p, want replacement %p", got, replacement) + } + + _ = original.Close() + conn.Close() +} + +func TestDeviceConnectionInfoReflectsUpdatedName(t *testing.T) { + discovered := &models.DeviceInfo{Name: "Living Room", DeviceID: "DEVICE01"} + conn := NewDeviceConnection(nil, discovered) + + conn.ApplyNameEvent("Living Room Left") + info := conn.Info() + + if info == nil || info.Name != "Living Room Left" || info.DeviceID != "DEVICE01" { + t.Fatalf("Info() = %+v, want updated name with original metadata", info) + } + if discovered.Name != "Living Room" { + t.Fatalf("discovery snapshot was mutated: %+v", discovered) + } +} + +func TestNameEventSupersedesInFlightPoll(t *testing.T) { + conn := NewDeviceConnection(nil, &models.DeviceInfo{Name: "Initial"}) + generation := conn.BeginNameRefresh() + + conn.ApplyNameEvent("Event Name") + if conn.ApplyPolledName(generation, "Stale Poll Name") { + t.Fatal("stale name poll was accepted") + } + if got := conn.Info().Name; got != "Event Name" { + t.Fatalf("Info().Name = %q, want newer event name", got) + } +} + +func TestConnectivityAggregatesHTTPAndEventStreamEvidence(t *testing.T) { + conn := NewDeviceConnection(nil, &models.DeviceInfo{Name: "test"}) + started := time.Date(2026, time.August, 28, 12, 0, 0, 0, time.UTC) + + conn.ObserveEventStream(true, started) + status := conn.Status() + if status.Connectivity != ConnectivityOnline || status.HTTPReachable || + !status.WebSocketConnected || !status.IsConnected { + t.Fatalf("stream-only success = %+v", status) + } + + failure := conn.BeginHTTPPoll() + conn.CompleteHTTPPoll(failure, false, started.Add(30*time.Second), nil) + status = conn.Status() + if status.Connectivity != ConnectivityOnline || status.HTTPReachable || + !status.WebSocketConnected || !status.IsConnected { + t.Fatalf("HTTP failure over live stream = %+v", status) + } + + conn.ObserveEventStream(false, started.Add(30*time.Second)) + status = conn.Status() + if status.Connectivity != ConnectivityStale || status.WebSocketConnected || !status.IsConnected { + t.Fatalf("direct-path loss within grace = %+v", status) + } + + secondFailure := conn.BeginHTTPPoll() + conn.CompleteHTTPPoll(secondFailure, false, started.Add(60*time.Second), nil) + status = conn.Status() + if status.Connectivity != ConnectivityOffline || status.IsConnected { + t.Fatalf("sustained direct-path loss = %+v", status) + } + + recovery := conn.BeginHTTPPoll() + conn.CompleteHTTPPoll(recovery, true, started.Add(61*time.Second), nil) + status = conn.Status() + if status.Connectivity != ConnectivityOnline || !status.HTTPReachable || !status.IsConnected { + t.Fatalf("HTTP recovery = %+v", status) + } +} + +func TestOlderHTTPFailureCannotDemoteNewerStreamActivity(t *testing.T) { + conn := NewDeviceConnection(nil, &models.DeviceInfo{Name: "test"}) + started := time.Date(2026, time.August, 28, 12, 0, 0, 0, time.UTC) + conn.MarkHTTPSuccess(started) + + older := conn.BeginHTTPPoll() + conn.ObserveEventStream(true, started.Add(61*time.Second)) + if !conn.CompleteHTTPPoll(older, false, started.Add(62*time.Second), nil) { + t.Fatal("latest HTTP-channel observation was unexpectedly rejected") + } + + status := conn.Status() + if status.Connectivity != ConnectivityOnline || status.HTTPReachable || + !status.WebSocketConnected || !status.IsConnected { + t.Fatalf("older HTTP failure demoted newer stream success: %+v", status) + } +} + +func TestOlderHTTPPayloadCannotOverwriteNewerSpeakerEvent(t *testing.T) { + conn := NewDeviceConnection(nil, &models.DeviceInfo{Name: "test"}) + started := time.Date(2026, time.August, 28, 12, 0, 0, 0, time.UTC) + + poll := conn.BeginHTTPPoll() + conn.ApplySpeakerEventAt(started.Add(time.Second), func(status *DeviceStatus) { + status.Volume = &models.Volume{ActualVolume: 99} + }) + conn.CompleteHTTPPoll(poll, true, started.Add(2*time.Second), func(status *DeviceStatus) { + status.Volume = &models.Volume{ActualVolume: 42} + }) + + status := conn.Status() + if status.Volume == nil || status.Volume.ActualVolume != 99 { + t.Fatalf("speaker event was overwritten by older poll data: %+v", status.Volume) + } + if status.Connectivity != ConnectivityOnline || !status.HTTPReachable || !status.IsConnected { + t.Fatalf("poll health was not retained: %+v", status) + } +} + +func TestOlderHTTPPollCannotOverwriteNewerSuccess(t *testing.T) { + conn := NewDeviceConnection(nil, &models.DeviceInfo{Name: "test"}) + started := time.Date(2026, time.August, 28, 12, 0, 0, 0, time.UTC) + + older := conn.BeginHTTPPoll() + newer := conn.BeginHTTPPoll() + if !conn.CompleteHTTPPoll(newer, true, started.Add(time.Second), nil) { + t.Fatal("newer successful poll was unexpectedly rejected") + } + if conn.CompleteHTTPPoll(older, false, started.Add(2*time.Second), nil) { + t.Fatal("older failed poll was unexpectedly accepted") + } + + status := conn.Status() + if status.Connectivity != ConnectivityOnline || !status.HTTPReachable || !status.IsConnected { + t.Fatalf("older poll overwrote newer success: %+v", status) + } +} + +func TestOlderTransportCallbackCannotOverwriteNewerGeneration(t *testing.T) { + conn := NewDeviceConnection(nil, &models.DeviceInfo{Name: "test"}) + started := time.Date(2026, time.August, 28, 12, 0, 0, 0, time.UTC) + + if !conn.ObserveEventStreamTransport(3, true, started) { + t.Fatal("newer connected transport generation was unexpectedly rejected") + } + conn.ApplySpeakerEventAt(started.Add(time.Second), func(status *DeviceStatus) { + status.Volume = &models.Volume{ActualVolume: 42} + }) + if conn.ObserveEventStreamTransport(2, false, started.Add(2*time.Second)) { + t.Fatal("older disconnected transport callback was unexpectedly accepted") + } + + status := conn.Status() + if status.Connectivity != ConnectivityOnline || !status.WebSocketConnected || !status.IsConnected { + t.Fatalf("older transport callback demoted newer success: %+v", status) + } + if status.Volume == nil || status.Volume.ActualVolume != 42 { + t.Fatalf("speaker event payload was lost: %+v", status.Volume) + } +} + +func TestInitialHTTPFailureStaysOfflineWithoutPriorSuccess(t *testing.T) { + conn := NewDeviceConnection(nil, &models.DeviceInfo{Name: "test"}) + started := time.Date(2026, time.August, 28, 12, 0, 0, 0, time.UTC) + + first := conn.BeginHTTPPoll() + conn.CompleteHTTPPoll(first, false, started, nil) + if status := conn.Status(); status.Connectivity != ConnectivityOffline || status.IsConnected { + t.Fatalf("first initial failure = %+v", status) + } + + second := conn.BeginHTTPPoll() + conn.CompleteHTTPPoll(second, false, started.Add(time.Second), nil) + if status := conn.Status(); status.Connectivity != ConnectivityOffline || status.IsConnected { + t.Fatalf("second initial failure = %+v", status) + } +} + +func TestWebSocketLoopHasSingleOwner(t *testing.T) { + conn := NewDeviceConnection(nil, &models.DeviceInfo{Name: "test"}) + + if !conn.TryStartWebSocketLoop() { + t.Fatal("first supervisor did not acquire ownership") + } + if conn.TryStartWebSocketLoop() { + t.Fatal("second supervisor acquired duplicate ownership") + } + + conn.FinishWebSocketLoop() + if !conn.TryStartWebSocketLoop() { + t.Fatal("supervisor ownership was not released") + } +} + func TestSetStatus_ReplacesEntireStatus(t *testing.T) { conn := NewDeviceConnection(nil, &models.DeviceInfo{Name: "test"}) conn.SetStatus(&DeviceStatus{ diff --git a/pkg/service/soundtouchweb/webtypes/types.go b/pkg/service/soundtouchweb/webtypes/types.go index cef8a0be..c47cbee5 100644 --- a/pkg/service/soundtouchweb/webtypes/types.go +++ b/pkg/service/soundtouchweb/webtypes/types.go @@ -2,6 +2,7 @@ package webtypes import ( + "strings" "sync" "sync/atomic" "time" @@ -40,12 +41,40 @@ type SoundTouchClient interface { // / UpdateStatus rather than the private field; construct connections // via NewDeviceConnection to guarantee the status is initialised. type DeviceConnection struct { - Client *client.Client - WebSocket *client.WebSocketClient + Client *client.Client + // WebSocket is retained for source compatibility with callers that build + // DeviceConnection values directly. Concurrent code must use + // CurrentWebSocket and SetWebSocket. + WebSocket *client.WebSocketClient + // DeviceInfo is the immutable discovery snapshot. Use Info for player-facing + // output so later nameUpdated events are reflected without racing readers. DeviceInfo *models.DeviceInfo LastSeen time.Time - status atomic.Pointer[DeviceStatus] + deviceName atomic.Pointer[string] + status atomic.Pointer[DeviceStatus] + + webSocketMu sync.RWMutex + webSocketLoopRunning atomic.Bool + + nameMu sync.Mutex + nameGen uint64 + + healthMu sync.Mutex + + nextPollGeneration uint64 + lastPollGeneration uint64 + httpReachable bool + consecutiveFailures int + speakerEventGen uint64 + pollEventGen map[uint64]uint64 + + lastTransportGeneration uint64 + eventStreamConnected bool + lastDirectSuccess time.Time + speakerConnectionKnown bool + speakerConnectionConnected bool + speakerConnectionObserved time.Time // fieldGenMu guards fieldGen, the per-field generation ordering used by // BeginFieldPoll/CompleteFieldPoll/ApplyFieldEvent. Each StatusField gets @@ -78,14 +107,40 @@ type DeviceConnection struct { // DeviceStatus represents the current device state type DeviceStatus struct { - NowPlaying *models.NowPlaying `json:"nowPlaying,omitempty"` - Volume *models.Volume `json:"volume,omitempty"` - Presets *models.Presets `json:"presets,omitempty"` - Sources *models.Sources `json:"sources,omitempty"` - Bass *models.Bass `json:"bass,omitempty"` - Group *models.Group `json:"group,omitempty"` - IsConnected bool `json:"isConnected"` - LastActivity time.Time `json:"lastActivity"` + NowPlaying *models.NowPlaying `json:"nowPlaying,omitempty"` + Volume *models.Volume `json:"volume,omitempty"` + Presets *models.Presets `json:"presets,omitempty"` + Sources *models.Sources `json:"sources,omitempty"` + Bass *models.Bass `json:"bass,omitempty"` + Group *models.Group `json:"group,omitempty"` + Connectivity Connectivity `json:"connectivity"` + HTTPReachable bool `json:"httpReachable"` + WebSocketConnected bool `json:"webSocketConnected"` + SpeakerConnectionState *SpeakerConnectionState `json:"speakerConnectionState,omitempty"` + IsConnected bool `json:"isConnected"` + LastActivity time.Time `json:"lastActivity"` +} + +// Connectivity is the player's aggregate view of HTTP and event-stream +// reachability. Speaker-reported connection state is retained as supporting +// evidence but cannot override a current direct-path success. +type Connectivity string + +const ( + ConnectivityOnline Connectivity = "online" + ConnectivityStale Connectivity = "stale" + ConnectivityOffline Connectivity = "offline" +) + +const ( + offlineFailureThreshold = 2 + offlineGracePeriod = 60 * time.Second +) + +// SpeakerConnectionState is the network state reported by the speaker. +type SpeakerConnectionState struct { + State string `json:"state"` + Signal string `json:"signal,omitempty"` } // StatusField identifies one independently-racing field of DeviceStatus for @@ -120,13 +175,71 @@ func NewDeviceConnection(c *client.Client, info *models.DeviceInfo) *DeviceConne done: make(chan struct{}), } conn.status.Store(&DeviceStatus{ + Connectivity: ConnectivityOffline, IsConnected: false, LastActivity: time.Now(), }) + if info != nil { + conn.storeDeviceName(info.Name) + } return conn } +func (c *DeviceConnection) storeDeviceName(name string) { + c.deviceName.Store(&name) +} + +// BeginNameRefresh starts a generation for an asynchronous /name request. +func (c *DeviceConnection) BeginNameRefresh() uint64 { + c.nameMu.Lock() + defer c.nameMu.Unlock() + + c.nameGen++ + + return c.nameGen +} + +// ApplyPolledName stores a /name result unless a newer poll or event won. +func (c *DeviceConnection) ApplyPolledName(generation uint64, name string) bool { + c.nameMu.Lock() + defer c.nameMu.Unlock() + + if generation != c.nameGen { + return false + } + + c.storeDeviceName(name) + + return true +} + +// ApplyNameEvent stores a nameUpdated event and invalidates in-flight polls. +func (c *DeviceConnection) ApplyNameEvent(name string) { + c.nameMu.Lock() + defer c.nameMu.Unlock() + + c.nameGen++ + c.storeDeviceName(name) +} + +// Info returns a read-only metadata snapshot with the latest device name. +func (c *DeviceConnection) Info() *models.DeviceInfo { + if c.DeviceInfo == nil { + return nil + } + + name := c.deviceName.Load() + if name == nil || *name == c.DeviceInfo.Name { + return c.DeviceInfo + } + + info := *c.DeviceInfo + info.Name = *name + + return &info +} + // Status returns a snapshot of the current device status. The returned // pointer is read-only from the caller's perspective and must not be // mutated. Use UpdateStatus or SetStatus to apply changes. Never returns @@ -151,12 +264,70 @@ func (c *DeviceConnection) Close() { c.closeOnce.Do(func() { close(c.done) - if c.WebSocket != nil { - _ = c.WebSocket.Disconnect() + c.webSocketMu.Lock() + ws := c.WebSocket + c.WebSocket = nil + c.webSocketMu.Unlock() + + if ws != nil { + _ = ws.Close() } }) } +// CurrentWebSocket returns the current device event transport, if any. +func (c *DeviceConnection) CurrentWebSocket() *client.WebSocketClient { + c.webSocketMu.RLock() + defer c.webSocketMu.RUnlock() + + return c.WebSocket +} + +// SetWebSocket publishes a device event transport unless the connection was +// already closed. A transport that loses that race is stopped immediately. +func (c *DeviceConnection) SetWebSocket(ws *client.WebSocketClient) bool { + c.webSocketMu.Lock() + select { + case <-c.done: + c.webSocketMu.Unlock() + if ws != nil { + _ = ws.Close() + } + + return false + default: + c.WebSocket = ws + c.webSocketMu.Unlock() + + return true + } +} + +// ClearWebSocket clears only the expected transport, preserving a replacement +// that may already have been published under the same device entry. +func (c *DeviceConnection) ClearWebSocket(expected *client.WebSocketClient) bool { + c.webSocketMu.Lock() + defer c.webSocketMu.Unlock() + + if c.WebSocket != expected { + return false + } + + c.WebSocket = nil + + return true +} + +// TryStartWebSocketLoop claims the single event supervisor for this device. +func (c *DeviceConnection) TryStartWebSocketLoop() bool { + return c.webSocketLoopRunning.CompareAndSwap(false, true) +} + +// FinishWebSocketLoop releases event-supervisor ownership. +func (c *DeviceConnection) FinishWebSocketLoop() { + c.webSocketLoopRunning.Store(false) +} + // SetStatus atomically replaces the entire status. Use sparingly — // UpdateStatus is the preferred entry point because it preserves // concurrent changes from other goroutines. @@ -235,6 +406,214 @@ func (c *DeviceConnection) UpdateStatus(mut func(*DeviceStatus)) { } } +// BeginHTTPPoll reserves an ordering generation for a status poll. +func (c *DeviceConnection) BeginHTTPPoll() uint64 { + c.healthMu.Lock() + defer c.healthMu.Unlock() + + c.nextPollGeneration++ + if c.pollEventGen == nil { + c.pollEventGen = make(map[uint64]uint64) + } + c.pollEventGen[c.nextPollGeneration] = c.speakerEventGen + + return c.nextPollGeneration +} + +// ApplySpeakerEventAt applies event payload and records a live event stream. +func (c *DeviceConnection) ApplySpeakerEventAt(at time.Time, mut func(*DeviceStatus)) { + c.healthMu.Lock() + defer c.healthMu.Unlock() + + c.speakerEventGen++ + c.markEventStreamActivityLocked(at) + c.UpdateStatus(func(status *DeviceStatus) { + if mut != nil { + mut(status) + } + c.applyConnectivityLocked(status, at) + }) +} + +// ApplySpeakerConnectionEvent stores speaker-reported diagnostic state. The +// event itself proves the event stream is live. +func (c *DeviceConnection) ApplySpeakerConnectionEvent(state SpeakerConnectionState, at time.Time) { + c.healthMu.Lock() + defer c.healthMu.Unlock() + + c.speakerEventGen++ + c.markEventStreamActivityLocked(at) + c.speakerConnectionObserved = at + + switch strings.ToUpper(strings.TrimSpace(state.State)) { + case string(models.ConnectionStateConnected): + c.speakerConnectionKnown = true + c.speakerConnectionConnected = true + case string(models.ConnectionStateDisconnected): + c.speakerConnectionKnown = true + c.speakerConnectionConnected = false + default: + c.speakerConnectionKnown = false + c.speakerConnectionConnected = false + } + + c.UpdateStatus(func(status *DeviceStatus) { + reported := state + status.SpeakerConnectionState = &reported + status.LastActivity = at + c.applyConnectivityLocked(status, at) + }) +} + +// ObserveEventStreamTransport applies an authoritative client transport +// transition. Its generation originates in WebSocketClient, so reordered +// callbacks cannot overwrite a newer transport state. +func (c *DeviceConnection) ObserveEventStreamTransport( + generation uint64, + connected bool, + at time.Time, +) bool { + c.healthMu.Lock() + defer c.healthMu.Unlock() + + if generation <= c.lastTransportGeneration { + return false + } + + c.lastTransportGeneration = generation + c.eventStreamConnected = connected + if connected { + c.recordDirectSuccessLocked(at) + } + c.UpdateStatus(func(status *DeviceStatus) { + c.applyConnectivityLocked(status, at) + }) + + return true +} + +// ObserveEventStream records a transition when no transport generation is +// available, primarily for non-client callers and deterministic tests. +func (c *DeviceConnection) ObserveEventStream(connected bool, at time.Time) { + c.healthMu.Lock() + defer c.healthMu.Unlock() + + c.eventStreamConnected = connected + if connected { + c.recordDirectSuccessLocked(at) + } + c.UpdateStatus(func(status *DeviceStatus) { + c.applyConnectivityLocked(status, at) + }) +} + +// MarkEventStreamActivity records an event handled by a field-specific path. +func (c *DeviceConnection) MarkEventStreamActivity(at time.Time) { + c.ObserveEventStream(true, at) +} + +// CompleteHTTPPoll records health and merges payload only if no newer poll or +// speaker event superseded it. +func (c *DeviceConnection) CompleteHTTPPoll( + generation uint64, + success bool, + at time.Time, + merge func(*DeviceStatus), +) bool { + c.healthMu.Lock() + defer c.healthMu.Unlock() + + pollEventGeneration, knownGeneration := c.pollEventGen[generation] + delete(c.pollEventGen, generation) + + if generation <= c.lastPollGeneration { + return false + } + + c.lastPollGeneration = generation + for olderGeneration := range c.pollEventGen { + if olderGeneration < generation { + delete(c.pollEventGen, olderGeneration) + } + } + + if success { + c.httpReachable = true + c.consecutiveFailures = 0 + c.recordDirectSuccessLocked(at) + } else { + c.httpReachable = false + c.consecutiveFailures++ + } + + c.UpdateStatus(func(status *DeviceStatus) { + if merge != nil && knownGeneration && pollEventGeneration == c.speakerEventGen { + merge(status) + } + c.applyConnectivityLocked(status, at) + if success { + status.LastActivity = at + } + }) + + return true +} + +func (c *DeviceConnection) markEventStreamActivityLocked(at time.Time) { + c.eventStreamConnected = true + c.recordDirectSuccessLocked(at) +} + +func (c *DeviceConnection) recordDirectSuccessLocked(at time.Time) { + if c.lastDirectSuccess.IsZero() || at.After(c.lastDirectSuccess) { + c.lastDirectSuccess = at + } +} + +func (c *DeviceConnection) applyConnectivityLocked(status *DeviceStatus, at time.Time) { + connectivity := c.connectivityLocked(at) + status.Connectivity = connectivity + status.HTTPReachable = c.httpReachable + status.WebSocketConnected = c.eventStreamConnected + status.IsConnected = connectivity != ConnectivityOffline +} + +func (c *DeviceConnection) connectivityLocked(at time.Time) Connectivity { + if c.httpReachable || c.eventStreamConnected { + return ConnectivityOnline + } + + if c.speakerConnectionKnown && c.speakerConnectionConnected && + withinConnectivityGrace(at, c.speakerConnectionObserved) { + return ConnectivityStale + } + + if !c.lastDirectSuccess.IsZero() && + (c.consecutiveFailures < offlineFailureThreshold || + withinConnectivityGrace(at, c.lastDirectSuccess)) { + return ConnectivityStale + } + + return ConnectivityOffline +} + +func withinConnectivityGrace(at, success time.Time) bool { + if success.IsZero() { + return false + } + if at.Before(success) { + return true + } + + return at.Sub(success) < offlineGracePeriod +} + +// MarkHTTPSuccess records a successful out-of-band request such as /info. +func (c *DeviceConnection) MarkHTTPSuccess(at time.Time) { + generation := c.BeginHTTPPoll() + c.CompleteHTTPPoll(generation, true, at, nil) +} + // BeginGroupRefresh starts a new generation for an asynchronous /getGroup // request. Only the latest started request may later update Group. func (c *DeviceConnection) BeginGroupRefresh() uint64 {