diff --git a/lib/screens/video_player_screen.dart b/lib/screens/video_player_screen.dart index 8924017c..e6472dbd 100644 --- a/lib/screens/video_player_screen.dart +++ b/lib/screens/video_player_screen.dart @@ -3557,6 +3557,7 @@ class VideoPlayerScreenState extends State with WidgetsBindin // Watch Together overlays (isolated from video surface repaints) RepaintBoundary( child: Stack( + fit: StackFit.expand, children: [ // Watch Together: reconnecting to host overlay Selector( @@ -3597,8 +3598,6 @@ 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/watch_together/providers/watch_together_provider.dart b/lib/watch_together/providers/watch_together_provider.dart index 91f1f82e..a2548335 100644 --- a/lib/watch_together/providers/watch_together_provider.dart +++ b/lib/watch_together/providers/watch_together_provider.dart @@ -175,10 +175,11 @@ class WatchTogetherProvider with ChangeNotifier { } } - /// Wire up reconnection handler to re-announce join after reconnect + /// Wire up reconnection handler to re-announce join and readiness after reconnect void _wireReconnectHandler() { _peerService!.onReconnected = () { _syncManager?.announceJoin(_displayName); + _syncManager?.reannounceReadyIfNeeded(); }; } @@ -308,17 +309,33 @@ 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 with a lightweight peer service to check room occupancy, + // then do a single createSession or joinSession. This avoids the crash-prone + // join→teardown→create cycle on the provider. + final customRelayUrl = SettingsService.instanceOrNull?.getCustomRelayUrl(); + final probe = WatchTogetherPeerService(customBaseUrl: customRelayUrl); + bool shouldBeHost; try { - await joinSession(sessionId, displayName: displayName); - return false; // joined as guest + await probe.joinSession(sessionId); + shouldBeHost = probe.connectedPeers.isEmpty; } 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); - return true; + shouldBeHost = true; + } else { + await probe.disconnect(); + probe.dispose(); + rethrow; } - rethrow; } + await probe.disconnect(); + probe.dispose(); + + if (shouldBeHost) { + await createSession(controlMode: controlMode, displayName: displayName, sessionId: sessionId); + } else { + await joinSession(sessionId, displayName: displayName); + } + return shouldBeHost; } /// Leave the current session @@ -490,11 +507,11 @@ class WatchTogetherProvider with ChangeNotifier { ); } - // If the host deliberately left, end the session immediately - // instead of waiting for the 15s reconnect grace period. + // If the host deliberately left, end the session for everyone. if (!isHost && message.peerId == _session?.hostPeerId) { _hostIntentionallyLeft = true; _handleHostExitedPlayer(message); + leaveSession(); } notifyListeners(); diff --git a/lib/watch_together/services/watch_together_sync_manager.dart b/lib/watch_together/services/watch_together_sync_manager.dart index ab250cdb..a67efe80 100644 --- a/lib/watch_together/services/watch_together_sync_manager.dart +++ b/lib/watch_together/services/watch_together_sync_manager.dart @@ -125,10 +125,19 @@ class WatchTogetherSyncManager { // If host, start broadcasting position periodically if (_session.isHost) { _startPositionSync(); - // Note: sessionConfig is sent after video loads (with correct position) in buffering handler } - // Note: playerReady will be announced when video loads (first buffering: false) + // If the video is already loaded (buffering stream already fired before we + // subscribed), announce ready now so peers aren't stuck waiting. + if (!player.state.buffering && !_hasAnnouncedReady) { + _hasAnnouncedReady = true; + _peerReady[_peerService.myPeerId!] = true; + _peerService.broadcast(SyncMessage.playerReady(peerId: _peerService.myPeerId!, ready: true)); + appLogger.d('WatchTogether: Video already loaded on attach, announcing ready'); + if (_session.isHost) { + _sendSessionConfig(); + } + } // If guest, request current session config from host in case we missed // a mediaSwitch broadcast (e.g., host switched episodes while we were @@ -144,6 +153,8 @@ class WatchTogetherSyncManager { /// Initialize participant tracking from existing session participants /// Call this before attachPlayer() to ensure we know about participants who joined before void initializeParticipants(List peerIds) { + // Clear stale entries (e.g. host's own peerId left over from a previous detachPlayer) + _peerReady.clear(); for (final peerId in peerIds) { if (peerId != _peerService.myPeerId) { if (_session.isHost) { @@ -232,6 +243,7 @@ class WatchTogetherSyncManager { return; } + if (isPlaying && !_firstPlayCompleted) _firstPlayCompleted = true; if (!isPlaying) _setDeferredPlay(false); _broadcastPlayPause(isPlaying); }), @@ -944,6 +956,20 @@ class WatchTogetherSyncManager { } } + /// Re-announce player readiness after reconnect. + /// + /// During reconnect the host resets our _peerReady entry to false via + /// _handlePeerJoin, but our _hasAnnouncedReady flag is still true (never + /// reset because the player stays attached). Re-broadcast so the host + /// doesn't stay stuck in the deferred-play gate. + void reannounceReadyIfNeeded() { + if (_hasAnnouncedReady && _peerService.myPeerId != null) { + _peerReady[_peerService.myPeerId!] = true; + _peerService.broadcast(SyncMessage.playerReady(peerId: _peerService.myPeerId!, ready: true)); + appLogger.d('WatchTogether: Re-announced player ready after reconnect'); + } + } + /// Send join announcement to all peers void announceJoin(String displayName) { _peerService.broadcast( diff --git a/lib/watch_together/widgets/watch_together_overlay.dart b/lib/watch_together/widgets/watch_together_overlay.dart index 7b62be9f..87063f39 100644 --- a/lib/watch_together/widgets/watch_together_overlay.dart +++ b/lib/watch_together/widgets/watch_together_overlay.dart @@ -14,31 +14,24 @@ import '../models/watch_session.dart'; import '../providers/watch_together_provider.dart'; /// Overlay shown on the video player when in a watch together session -class WatchTogetherOverlay extends StatelessWidget { - /// Callback when the user wants to leave the session +/// Session indicator badge for embedding in the video controls header. +/// Shows participant count, host badge, and opens the session menu on tap. +class WatchTogetherSessionIndicator extends StatelessWidget { final VoidCallback? onLeaveSession; - const WatchTogetherOverlay({super.key, this.onLeaveSession}); + const WatchTogetherSessionIndicator({super.key, this.onLeaveSession}); @override Widget build(BuildContext context) { return Consumer( builder: (context, provider, child) { - if (!provider.isInSession) { - return const SizedBox.shrink(); - } - - return Positioned( - top: 16, - right: 16, - child: _SessionIndicator( - participantCount: provider.participantCount, - isHost: provider.isHost, - isSyncing: provider.isSyncing, - controlMode: provider.controlMode, - sessionId: provider.sessionId, - onTap: () => _showSessionMenu(context, provider), - ), + return _SessionIndicator( + participantCount: provider.participantCount, + isHost: provider.isHost, + isSyncing: provider.isSyncing, + controlMode: provider.controlMode, + sessionId: provider.sessionId, + onTap: () => _showSessionMenu(context, provider), ); }, ); @@ -71,8 +64,6 @@ class _SessionIndicator extends StatelessWidget { @override Widget build(BuildContext context) { - final theme = Theme.of(context); - return Material( color: Colors.black54, borderRadius: const BorderRadius.all(Radius.circular(20)), @@ -94,7 +85,7 @@ class _SessionIndicator extends StatelessWidget { : const CircularProgressIndicator(strokeWidth: 2, color: Colors.white), ) else - Icon(Symbols.group, size: 18, color: isHost ? theme.colorScheme.primary : Colors.white), + Icon(Symbols.group, size: 18, color: isHost ? Colors.amber : Colors.white), const SizedBox(width: 6), @@ -109,13 +100,13 @@ class _SessionIndicator extends StatelessWidget { const SizedBox(width: 6), Container( padding: const EdgeInsets.symmetric(horizontal: 6, vertical: 2), - decoration: BoxDecoration( - color: theme.colorScheme.primary, - borderRadius: const BorderRadius.all(Radius.circular(4)), + decoration: const BoxDecoration( + color: Colors.amber, + borderRadius: BorderRadius.all(Radius.circular(4)), ), child: Text( t.watchTogether.hostBadge, - style: const TextStyle(color: Colors.white, fontSize: 10, fontWeight: FontWeight.bold), + style: const TextStyle(color: Colors.black, fontSize: 10, fontWeight: FontWeight.bold), ), ), ], diff --git a/lib/widgets/video_controls/widgets/video_controls_header.dart b/lib/widgets/video_controls/widgets/video_controls_header.dart index 95ddb882..7c74791d 100644 --- a/lib/widgets/video_controls/widgets/video_controls_header.dart +++ b/lib/widgets/video_controls/widgets/video_controls_header.dart @@ -1,8 +1,11 @@ import 'package:flutter/material.dart'; +import 'package:provider/provider.dart'; import 'package:plezy/utils/formatters.dart'; import '../../../models/plex_metadata.dart'; import '../../../i18n/strings.g.dart'; +import '../../../watch_together/widgets/watch_together_overlay.dart'; +import '../../../watch_together/providers/watch_together_provider.dart'; import '../../app_bar_back_button.dart'; /// Header layout style for video controls @@ -47,6 +50,16 @@ class VideoControlsHeader extends StatelessWidget { ), const SizedBox(width: 16), Expanded(child: style == VideoHeaderStyle.singleLine ? _buildSingleLineTitle() : _buildMultiLineTitle()), + Selector( + selector: (_, p) => p.isInSession, + builder: (context, inSession, child) { + if (!inSession) return const SizedBox.shrink(); + return const Padding( + padding: EdgeInsets.only(right: 8), + child: WatchTogetherSessionIndicator(), + ); + }, + ), ?trailing, ], ); diff --git a/server/main.go b/server/main.go index 77316198..de761b77 100644 --- a/server/main.go +++ b/server/main.go @@ -202,10 +202,11 @@ type serverMsg struct { type Client struct { conn *websocket.Conn send chan []byte + done chan struct{} } func newClient(conn *websocket.Conn) *Client { - c := &Client{conn: conn, send: make(chan []byte, 64)} + c := &Client{conn: conn, send: make(chan []byte, 64), done: make(chan struct{})} go c.writePump() return c } @@ -215,13 +216,12 @@ func (c *Client) writePump() { defer ticker.Stop() for { select { - case data, ok := <-c.send: - if !ok { - c.conn.WriteMessage(websocket.CloseMessage, nil) - return - } + case data := <-c.send: c.conn.SetWriteDeadline(time.Now().Add(writeWait)) c.conn.WriteMessage(websocket.TextMessage, data) + case <-c.done: + c.conn.WriteMessage(websocket.CloseMessage, nil) + return case <-ticker.C: c.conn.SetWriteDeadline(time.Now().Add(writeWait)) if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil { @@ -231,20 +231,24 @@ func (c *Client) writePump() { } } +func (c *Client) trySend(data []byte) { + select { + case c.send <- data: + case <-c.done: + default: + } +} + 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) - } + c.trySend(data) } func (c *Client) close() { - close(c.send) + close(c.done) } // --- Room --- @@ -281,10 +285,7 @@ func (r *Room) broadcastExcept(senderID string, msg serverMsg) { r.mu.RUnlock() for _, client := range targets { - select { - case client.send <- data: - default: - } + client.trySend(data) } } @@ -299,12 +300,8 @@ func (r *Room) sendTo(targetID string, msg serverMsg) bool { if !ok { return false } - select { - case client.send <- data: - return true - default: - return false - } + client.trySend(data) + return true } // --- Log store --- @@ -556,20 +553,27 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) { var currentPeerID string var isHost bool - // Cleanup on disconnect + // Cleanup on disconnect — only if our Client is still the one in the room. + // A reconnecting peer reuses the same peerId, so the map entry may have + // been overwritten by a newer Client before this defer runs. defer func() { if currentRoom != nil && currentPeerID != "" { currentRoom.mu.Lock() - delete(currentRoom.Peers, currentPeerID) + stale := currentRoom.Peers[currentPeerID] != client + if !stale { + delete(currentRoom.Peers, currentPeerID) + } currentRoom.mu.Unlock() - currentRoom.broadcastExcept(currentPeerID, serverMsg{ - Type: "peerLeft", - PeerID: currentPeerID, - }) + if !stale { + currentRoom.broadcastExcept(currentPeerID, serverMsg{ + Type: "peerLeft", + PeerID: currentPeerID, + }) + } if isHost { s.conns.releaseRoom(ip) } - log.Printf("peer %s left room %s", currentPeerID, currentRoom.SessionID) + log.Printf("peer %s left room %s (stale=%v)", currentPeerID, currentRoom.SessionID, stale) } }()