diff --git a/lib/screens/video_player_screen.dart b/lib/screens/video_player_screen.dart index 0c367d07..8924017c 100644 --- a/lib/screens/video_player_screen.dart +++ b/lib/screens/video_player_screen.dart @@ -3597,6 +3597,8 @@ class VideoPlayerScreenState extends State with WidgetsBindin ); }, ), + // Watch Together: session indicator and menu + WatchTogetherOverlay(onLeaveSession: _handleBackButton), // Watch Together: participant join/leave/buffering notifications const ParticipantNotificationOverlay(), // Watch Together: waiting for participants to load diff --git a/lib/services/base_peer_service.dart b/lib/services/base_peer_service.dart index c953bfa5..2b3c5a5a 100644 --- a/lib/services/base_peer_service.dart +++ b/lib/services/base_peer_service.dart @@ -22,7 +22,10 @@ class PeerError { final String message; final dynamic originalError; - const PeerError({required this.type, required this.message, this.originalError}); + /// Server-provided error code (e.g. 'room_not_found', 'room_full'). + final String? serverCode; + + const PeerError({required this.type, required this.message, this.originalError, this.serverCode}); @override String toString() => 'PeerError($type): $message'; diff --git a/lib/watch_together/models/sync_message.dart b/lib/watch_together/models/sync_message.dart index 421e4894..2e2dfa8f 100644 --- a/lib/watch_together/models/sync_message.dart +++ b/lib/watch_together/models/sync_message.dart @@ -280,6 +280,25 @@ class SyncMessage { /// Position as Duration (convenience getter) Duration? get position => positionMs != null ? Duration(milliseconds: positionMs!) : null; + SyncMessage copyWith({String? peerId}) { + return SyncMessage( + type: type, + timestamp: timestamp, + positionMs: positionMs, + bufferingState: bufferingState, + rate: rate, + peerId: peerId ?? this.peerId, + displayName: displayName, + isHost: isHost, + controlMode: controlMode, + pingId: pingId, + ratingKey: ratingKey, + serverId: serverId, + mediaTitle: mediaTitle, + isPlaying: isPlaying, + ); + } + /// Serialize to JSON string for sending over data channel String toJson() { final map = {'t': type.name, 'ts': timestamp}; diff --git a/lib/watch_together/models/watch_session.dart b/lib/watch_together/models/watch_session.dart index 1e5e6025..f811b13d 100644 --- a/lib/watch_together/models/watch_session.dart +++ b/lib/watch_together/models/watch_session.dart @@ -79,9 +79,6 @@ class WatchSession { /// Current connection state final SessionState state; - /// List of participants in the session - final List participants; - /// Error message if state is error final String? errorMessage; @@ -102,7 +99,6 @@ class WatchSession { required this.role, required this.controlMode, required this.state, - this.participants = const [], this.errorMessage, this.mediaRatingKey, this.mediaServerId, @@ -116,15 +112,11 @@ class WatchSession { /// Whether the session is currently connected bool get isConnected => state == SessionState.connected; - /// Number of participants (including self) - int get participantCount => participants.length; - WatchSession copyWith({ String? sessionId, SessionRole? role, ControlMode? controlMode, SessionState? state, - List? participants, String? errorMessage, String? mediaRatingKey, String? mediaServerId, @@ -136,7 +128,6 @@ class WatchSession { role: role ?? this.role, controlMode: controlMode ?? this.controlMode, state: state ?? this.state, - participants: participants ?? this.participants, errorMessage: errorMessage ?? this.errorMessage, mediaRatingKey: mediaRatingKey ?? this.mediaRatingKey, mediaServerId: mediaServerId ?? this.mediaServerId, @@ -163,7 +154,6 @@ class WatchSession { mediaRatingKey: mediaRatingKey, mediaServerId: mediaServerId, mediaTitle: mediaTitle, - participants: [], ); } @@ -174,7 +164,6 @@ class WatchSession { role: SessionRole.guest, controlMode: ControlMode.hostOnly, // Will be updated when connected state: SessionState.connecting, - participants: [], ); } } diff --git a/lib/watch_together/providers/watch_together_provider.dart b/lib/watch_together/providers/watch_together_provider.dart index 83c77ef9..91f1f82e 100644 --- a/lib/watch_together/providers/watch_together_provider.dart +++ b/lib/watch_together/providers/watch_together_provider.dart @@ -36,20 +36,22 @@ class WatchTogetherProvider with ChangeNotifier { // During Watch Together join, 4-5 notifications fire within milliseconds; // this batches them into one rebuild to avoid overwhelming low-end devices. bool _notifyScheduled = false; + bool _disposed = false; @override void notifyListeners() { - if (_notifyScheduled) return; + if (_disposed || _notifyScheduled) return; _notifyScheduled = true; scheduleMicrotask(() { _notifyScheduled = false; - super.notifyListeners(); + if (!_disposed) super.notifyListeners(); }); } // Host reconnect grace period Timer? _hostReconnectTimer; bool _isWaitingForHostReconnect = false; + bool _hostIntentionallyLeft = false; // Debounce map for action events (peerId+type → last emission timestamp) final Map _lastActionEventMs = {}; @@ -131,7 +133,6 @@ class WatchTogetherProvider with ChangeNotifier { role: session.role, controlMode: session.controlMode, state: session.state, - participants: session.participants, errorMessage: session.errorMessage, hostPeerId: session.hostPeerId, ); @@ -174,6 +175,13 @@ class WatchTogetherProvider with ChangeNotifier { } } + /// Wire up reconnection handler to re-announce join after reconnect + void _wireReconnectHandler() { + _peerService!.onReconnected = () { + _syncManager?.announceJoin(_displayName); + }; + } + /// Wire up sync manager's state change callback to update provider state void _wireSyncStateChanges() { _syncManager!.onSyncStateChanged = (isSyncing) { @@ -227,6 +235,7 @@ class WatchTogetherProvider with ChangeNotifier { ); _wireSyncStateChanges(); + _wireReconnectHandler(); notifyListeners(); appLogger.d('WatchTogether: Session created: $createdSessionId'); @@ -276,6 +285,7 @@ class WatchTogetherProvider with ChangeNotifier { }; _wireSyncStateChanges(); + _wireReconnectHandler(); // Add self to participants _participants.add(Participant(peerId: _peerService!.myPeerId!, displayName: _displayName, isHost: false)); @@ -298,25 +308,15 @@ class WatchTogetherProvider with ChangeNotifier { /// /// Returns `true` if the user became the host. Future enterRoom(String sessionId, {ControlMode controlMode = ControlMode.anyone, String? displayName}) async { - // Probe the relay to determine if the room exists, then delegate to - // the existing createSession / joinSession which handle all setup. - final customRelayUrl = SettingsService.instanceOrNull?.getCustomRelayUrl(); - final probe = WatchTogetherPeerService(customBaseUrl: customRelayUrl); try { - final becameHost = await probe.joinOrCreateSession(sessionId); - final shouldBeHost = becameHost || probe.connectedPeers.isEmpty; - await probe.disconnect(); - probe.dispose(); - - if (shouldBeHost) { + await joinSession(sessionId, displayName: displayName); + return false; // joined as guest + } on PeerError catch (e) { + // Room doesn't exist — create it as host + if (e.serverCode == 'room_not_found') { await createSession(controlMode: controlMode, displayName: displayName, sessionId: sessionId); - } else { - await joinSession(sessionId, displayName: displayName); + return true; } - return shouldBeHost; - } catch (e) { - await probe.disconnect(); - probe.dispose(); rethrow; } } @@ -358,6 +358,7 @@ class WatchTogetherProvider with ChangeNotifier { _isDeferredPlay = false; _lastHandledCurrentPlaybackKey = null; _lastActionEventMs.clear(); + _hostIntentionallyLeft = false; notifyListeners(); appLogger.d('WatchTogether: Session left'); @@ -410,8 +411,9 @@ class WatchTogetherProvider with ChangeNotifier { _participants.removeWhere((p) => p.peerId == peerId); - // If host disconnected, start grace period for reconnection - if (!isHost && peerId == _session?.hostPeerId) { + // If host disconnected unexpectedly, start grace period for reconnection. + // Skip if the host already sent a deliberate leave message. + if (!isHost && peerId == _session?.hostPeerId && !_hostIntentionallyLeft) { _startHostReconnectGracePeriod(); } else if (disconnectedName != null) { _participantEventController.add( @@ -459,16 +461,16 @@ class WatchTogetherProvider with ChangeNotifier { _participantEventController.add( ParticipantEvent(displayName: message.displayName!, type: ParticipantEventType.joined), ); - } - // Send our join info back so the new peer adds us to their - // participant list. Every peer does this (not just the host) - // so that late joiners learn about all existing participants. - if (_peerService != null) { - _peerService!.sendTo( - message.peerId!, - SyncMessage.join(peerId: _peerService!.myPeerId!, displayName: _displayName, isHost: isHost), - ); + // Send our join info back so the new peer adds us to their + // participant list. Only reply to NEW peers to avoid an + // infinite join ping-pong (A→join→B→join→A→...). + if (_peerService != null) { + _peerService!.sendTo( + message.peerId!, + SyncMessage.join(peerId: _peerService!.myPeerId!, displayName: _displayName, isHost: isHost), + ); + } } notifyListeners(); @@ -487,6 +489,14 @@ class WatchTogetherProvider with ChangeNotifier { ParticipantEvent(displayName: leavingName, type: ParticipantEventType.left), ); } + + // If the host deliberately left, end the session immediately + // instead of waiting for the 15s reconnect grace period. + if (!isHost && message.peerId == _session?.hostPeerId) { + _hostIntentionallyLeft = true; + _handleHostExitedPlayer(message); + } + notifyListeners(); } break; @@ -739,6 +749,7 @@ class WatchTogetherProvider with ChangeNotifier { @override void dispose() { + _disposed = true; _cancelHostReconnectGracePeriod(); _participantEventController.close(); leaveSession(); diff --git a/lib/watch_together/services/watch_together_peer_service.dart b/lib/watch_together/services/watch_together_peer_service.dart index 678446c5..e7287fcb 100644 --- a/lib/watch_together/services/watch_together_peer_service.dart +++ b/lib/watch_together/services/watch_together_peer_service.dart @@ -59,6 +59,9 @@ class WatchTogetherPeerService with KeepaliveMixin { static const int _maxReconnectAttempts = 3; Timer? _reconnectTimer; + /// Called after a successful reconnection so the provider can re-announce join. + void Function()? onReconnected; + // Keepalive (via KeepaliveMixin) @override Duration get pingInterval => const Duration(seconds: 15); @@ -192,10 +195,16 @@ class WatchTogetherPeerService with KeepaliveMixin { case 'message': final payload = msg['payload']; + final serverFrom = msg['from'] as String?; if (payload != null) { try { final payloadStr = payload is String ? payload : jsonEncode(payload); - final syncMsg = SyncMessage.fromJson(payloadStr); + var syncMsg = SyncMessage.fromJson(payloadStr); + // Use the server-authenticated sender ID instead of the + // self-reported peerId in the payload to prevent spoofing. + if (serverFrom != null && syncMsg.peerId != serverFrom) { + syncMsg = syncMsg.copyWith(peerId: serverFrom); + } _safeAdd(_messageReceivedController, syncMsg); } catch (e) { appLogger.e('WatchTogether: Failed to parse sync message payload', error: e); @@ -206,7 +215,7 @@ class WatchTogetherPeerService with KeepaliveMixin { final code = msg['code'] as String? ?? 'unknown'; final message = msg['message'] as String? ?? 'Unknown error'; appLogger.e('WatchTogether: Server error: $code - $message'); - final error = PeerError(type: PeerErrorType.serverError, message: '$code: $message'); + final error = PeerError(type: PeerErrorType.serverError, message: '$code: $message', serverCode: code); _safeAdd(_errorController, error); if (setupCompleter != null && !setupCompleter.isCompleted) { setupCompleter.completeError(error); @@ -292,15 +301,27 @@ class WatchTogetherPeerService with KeepaliveMixin { _listenToChannel(channel, setupCompleter: completer); startKeepalive(); - // Re-send create or join - if (_isHost) { - _sendRaw({'type': 'create', 'sessionId': _sessionId, 'peerId': _myPeerId}); - } else { - _sendRaw({'type': 'join', 'sessionId': _sessionId, 'peerId': _myPeerId}); + // Always try join first — the room may still have peers (e.g. host + // reconnecting while guests remain). Fall back to create only if + // the room no longer exists and we were the host. + _sendRaw({'type': 'join', 'sessionId': _sessionId, 'peerId': _myPeerId}); + + try { + await completer.future.namedTimeout(const Duration(seconds: 10), operation: 'WatchTogether reconnect'); + } on PeerError catch (e) { + if (_isHost && e.serverCode == 'room_not_found') { + appLogger.d('WatchTogether: Room gone, re-creating as host'); + final createCompleter = Completer(); + _listenToChannel(channel, setupCompleter: createCompleter); + _sendRaw({'type': 'create', 'sessionId': _sessionId, 'peerId': _myPeerId}); + await createCompleter.future.namedTimeout(const Duration(seconds: 10), operation: 'WatchTogether reconnect create'); + } else { + rethrow; + } } - await completer.future.namedTimeout(const Duration(seconds: 10), operation: 'WatchTogether reconnect'); appLogger.d('WatchTogether: Reconnected successfully'); + onReconnected?.call(); } catch (e) { appLogger.e('WatchTogether: Reconnect failed', error: e); _handleWebSocketClosed(); @@ -384,20 +405,6 @@ class WatchTogetherPeerService with KeepaliveMixin { } } - /// Try to join a session; if it doesn't exist, create it as host. - /// - /// Returns `true` if the user became the host (room was empty). - Future joinOrCreateSession(String sessionId) async { - try { - await joinSession(sessionId); - return false; // joined as guest - } on PeerError catch (e) { - if (e.type != PeerErrorType.serverError) rethrow; - await createSession(sessionId: sessionId); - return true; // created as host - } - } - /// Broadcast a message to all connected peers void broadcast(SyncMessage message) { final payload = message.toJson(); diff --git a/lib/watch_together/services/watch_together_sync_manager.dart b/lib/watch_together/services/watch_together_sync_manager.dart index b04a3b44..ab250cdb 100644 --- a/lib/watch_together/services/watch_together_sync_manager.dart +++ b/lib/watch_together/services/watch_together_sync_manager.dart @@ -228,7 +228,7 @@ class WatchTogetherSyncManager { } finally { _isRemoteAction = false; } - _broadcastPlayPause(true); + // Don't broadcast play — deferred play will broadcast when all peers are ready. return; } @@ -539,20 +539,6 @@ class WatchTogetherSyncManager { return; } - // HOST RELAY: In "anyone" mode, host rebroadcasts control commands from guests - // This is needed because guests only connect to host (star topology), not to each other - if (_session.isHost && _session.controlMode == ControlMode.anyone) { - final isControlMessage = - message.type == SyncMessageType.play || - message.type == SyncMessageType.pause || - message.type == SyncMessageType.seek || - message.type == SyncMessageType.rate; - - if (isControlMessage) { - _peerService.broadcast(message); - } - } - // In hostOnly mode, only process messages from host (unless it's join/leave/sessionConfig) if (_session.controlMode == ControlMode.hostOnly && !_session.isHost) { final isHostMessage = message.peerId == _session.hostPeerId; @@ -669,11 +655,14 @@ class WatchTogetherSyncManager { if (_deferredPlay && isAllReady) { _setDeferredPlay(false); _firstPlayCompleted = true; + final pos = _deferredPlayPosition; + _deferredPlayPosition = null; await _applyRemotePlay( - position: _deferredPlayPosition, + position: pos, expectedAttachmentGeneration: queuedAttachmentGeneration, ); - _deferredPlayPosition = null; + // Broadcast play to all peers now that everyone is ready + _broadcastPlayPause(true); } } break; diff --git a/server/main.go b/server/main.go index 0cc6d992..77316198 100644 --- a/server/main.go +++ b/server/main.go @@ -197,12 +197,62 @@ type serverMsg struct { Payload json.RawMessage `json:"payload,omitempty"` } +// --- Client (serializes writes to a single goroutine) --- + +type Client struct { + conn *websocket.Conn + send chan []byte +} + +func newClient(conn *websocket.Conn) *Client { + c := &Client{conn: conn, send: make(chan []byte, 64)} + go c.writePump() + return c +} + +func (c *Client) writePump() { + ticker := time.NewTicker(pingInterval) + defer ticker.Stop() + for { + select { + case data, ok := <-c.send: + if !ok { + c.conn.WriteMessage(websocket.CloseMessage, nil) + return + } + c.conn.SetWriteDeadline(time.Now().Add(writeWait)) + c.conn.WriteMessage(websocket.TextMessage, data) + case <-ticker.C: + c.conn.SetWriteDeadline(time.Now().Add(writeWait)) + if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil { + return + } + } + } +} + +func (c *Client) sendJSON(msg serverMsg) { + data, err := json.Marshal(msg) + if err != nil { + return + } + select { + case c.send <- data: + default: + // Drop if buffer full (slow client) + } +} + +func (c *Client) close() { + close(c.send) +} + // --- Room --- type Room struct { SessionID string HostPeerID string - Peers map[string]*websocket.Conn + Peers map[string]*Client mu sync.RWMutex CreatedAt time.Time } @@ -220,12 +270,20 @@ func (r *Room) broadcastExcept(senderID string, msg serverMsg) { if err != nil { return } + // Copy peers under lock, then send without holding it r.mu.RLock() - defer r.mu.RUnlock() - for id, conn := range r.Peers { + targets := make([]*Client, 0, len(r.Peers)) + for id, client := range r.Peers { if id != senderID { - conn.SetWriteDeadline(time.Now().Add(writeWait)) - conn.WriteMessage(websocket.TextMessage, data) + targets = append(targets, client) + } + } + r.mu.RUnlock() + + for _, client := range targets { + select { + case client.send <- data: + default: } } } @@ -236,14 +294,17 @@ func (r *Room) sendTo(targetID string, msg serverMsg) bool { return false } r.mu.RLock() - defer r.mu.RUnlock() - conn, ok := r.Peers[targetID] + client, ok := r.Peers[targetID] + r.mu.RUnlock() if !ok { return false } - conn.SetWriteDeadline(time.Now().Add(writeWait)) - conn.WriteMessage(websocket.TextMessage, data) - return true + select { + case client.send <- data: + return true + default: + return false + } } // --- Log store --- @@ -463,18 +524,6 @@ func (s *Server) handleGetLogs(w http.ResponseWriter, r *http.Request) { w.Write(data) } -func (s *Server) sendError(conn *websocket.Conn, code, message string) { - data, _ := json.Marshal(serverMsg{Type: "error", Code: code, Message: message}) - conn.SetWriteDeadline(time.Now().Add(writeWait)) - conn.WriteMessage(websocket.TextMessage, data) -} - -func (s *Server) sendJSON(conn *websocket.Conn, msg serverMsg) { - data, _ := json.Marshal(msg) - conn.SetWriteDeadline(time.Now().Add(writeWait)) - conn.WriteMessage(websocket.TextMessage, data) -} - func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) { ip := clientIP(r) @@ -498,17 +547,9 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) { return nil }) - // Ping ticker - ticker := time.NewTicker(pingInterval) - defer ticker.Stop() - go func() { - for range ticker.C { - conn.SetWriteDeadline(time.Now().Add(writeWait)) - if err := conn.WriteMessage(websocket.PingMessage, nil); err != nil { - return - } - } - }() + // Client wraps the conn with a serialized write channel + ping ticker + client := newClient(conn) + defer client.close() rl := newRateLimiter(rateBurst, rateSustained) var currentRoom *Room @@ -542,24 +583,24 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) { } if !rl.allow() { - s.sendError(conn, "rate_limited", "Too many messages") + client.sendJSON(serverMsg{Type: "error", Code: "rate_limited", Message: "Too many messages"}) continue } var msg clientMsg if err := json.Unmarshal(raw, &msg); err != nil { - s.sendError(conn, "invalid_message", "Invalid JSON") + client.sendJSON(serverMsg{Type: "error", Code: "invalid_message", Message: "Invalid JSON"}) continue } switch msg.Type { case "create": if msg.SessionID == "" || msg.PeerID == "" { - s.sendError(conn, "invalid_message", "sessionId and peerId required") + client.sendJSON(serverMsg{Type: "error", Code: "invalid_message", Message: "sessionId and peerId required"}) continue } if !s.conns.tryCreateRoom(ip) { - s.sendError(conn, "rate_limited", "Too many rooms created") + client.sendJSON(serverMsg{Type: "error", Code: "rate_limited", Message: "Too many rooms created"}) continue } s.mu.Lock() @@ -570,7 +611,7 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) { if !empty { s.mu.Unlock() s.conns.releaseRoom(ip) - s.sendError(conn, "room_exists", "Room already exists") + client.sendJSON(serverMsg{Type: "error", Code: "room_exists", Message: "Room already exists"}) continue } // Empty stale room — reclaim the ID @@ -579,7 +620,7 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) { room := &Room{ SessionID: msg.SessionID, HostPeerID: msg.PeerID, - Peers: map[string]*websocket.Conn{msg.PeerID: conn}, + Peers: map[string]*Client{msg.PeerID: client}, CreatedAt: time.Now(), } s.rooms[msg.SessionID] = room @@ -588,27 +629,27 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) { currentPeerID = msg.PeerID isHost = true log.Printf("room %s created by %s", msg.SessionID, msg.PeerID) - s.sendJSON(conn, serverMsg{Type: "created", SessionID: msg.SessionID}) + client.sendJSON(serverMsg{Type: "created", SessionID: msg.SessionID}) case "join": if msg.SessionID == "" || msg.PeerID == "" { - s.sendError(conn, "invalid_message", "sessionId and peerId required") + client.sendJSON(serverMsg{Type: "error", Code: "invalid_message", Message: "sessionId and peerId required"}) continue } s.mu.RLock() room, exists := s.rooms[msg.SessionID] s.mu.RUnlock() if !exists { - s.sendError(conn, "room_not_found", "Room does not exist") + client.sendJSON(serverMsg{Type: "error", Code: "room_not_found", Message: "Room does not exist"}) continue } room.mu.Lock() if len(room.Peers) >= maxRoomSize { room.mu.Unlock() - s.sendError(conn, "room_full", "Room is full") + client.sendJSON(serverMsg{Type: "error", Code: "room_full", Message: "Room is full"}) continue } - room.Peers[msg.PeerID] = conn + room.Peers[msg.PeerID] = client peers := room.peerIDs() room.mu.Unlock() currentRoom = room @@ -622,12 +663,12 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) { existingPeers = append(existingPeers, p) } } - s.sendJSON(conn, serverMsg{Type: "joined", SessionID: msg.SessionID, Peers: existingPeers}) + client.sendJSON(serverMsg{Type: "joined", SessionID: msg.SessionID, Peers: existingPeers}) room.broadcastExcept(msg.PeerID, serverMsg{Type: "peerJoined", PeerID: msg.PeerID}) case "broadcast": if currentRoom == nil { - s.sendError(conn, "not_in_room", "Not in a room") + client.sendJSON(serverMsg{Type: "error", Code: "not_in_room", Message: "Not in a room"}) continue } currentRoom.broadcastExcept(currentPeerID, serverMsg{ @@ -638,11 +679,11 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) { case "sendTo": if currentRoom == nil { - s.sendError(conn, "not_in_room", "Not in a room") + client.sendJSON(serverMsg{Type: "error", Code: "not_in_room", Message: "Not in a room"}) continue } if msg.To == "" { - s.sendError(conn, "invalid_message", "to field required") + client.sendJSON(serverMsg{Type: "error", Code: "invalid_message", Message: "to field required"}) continue } if !currentRoom.sendTo(msg.To, serverMsg{ @@ -650,14 +691,14 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) { From: currentPeerID, Payload: msg.Payload, }) { - s.sendError(conn, "not_in_room", "Target peer not found") + client.sendJSON(serverMsg{Type: "error", Code: "not_in_room", Message: "Target peer not found"}) } case "ping": - s.sendJSON(conn, serverMsg{Type: "pong"}) + client.sendJSON(serverMsg{Type: "pong"}) default: - s.sendError(conn, "invalid_message", "Unknown message type") + client.sendJSON(serverMsg{Type: "error", Code: "invalid_message", Message: "Unknown message type"}) } } }