fix(watch-together): improve sync and buffering handling

2026-01-25 18:47:48 +01:00
parent 27e88207fa
commit e3fbe6d925
3 changed files with 203 additions and 25 deletions
@@ -88,6 +88,9 @@ class SyncMessage {
  /// Title of the media (for mediaSwitch message)
  final String? mediaTitle;

+  /// Whether playback is currently playing (for positionSync heartbeat)
+  final bool? isPlaying;
+
  const SyncMessage({
    required this.type,
    required this.timestamp,
@@ -102,6 +105,7 @@ class SyncMessage {
    this.ratingKey,
    this.serverId,
    this.mediaTitle,
+    this.isPlaying,
  });

  /// Create a PLAY message
@@ -139,13 +143,14 @@ class SyncMessage {
    );
  }

-  /// Create a POSITION_SYNC message
-  factory SyncMessage.positionSync(Duration position, {String? peerId}) {
+  /// Create a POSITION_SYNC message (heartbeat with optional play/pause state)
+  factory SyncMessage.positionSync(Duration position, {String? peerId, bool? isPlaying}) {
    return SyncMessage(
      type: SyncMessageType.positionSync,
      timestamp: DateTime.now().millisecondsSinceEpoch,
      positionMs: position.inMilliseconds,
      peerId: peerId,
+      isPlaying: isPlaying,
    );
  }

@@ -268,6 +273,7 @@ class SyncMessage {
    if (ratingKey != null) map['rk'] = ratingKey;
    if (serverId != null) map['sid'] = serverId;
    if (mediaTitle != null) map['title'] = mediaTitle;
+    if (isPlaying != null) map['pl'] = isPlaying;

    return jsonEncode(map);
  }
@@ -296,6 +302,7 @@ class SyncMessage {
      ratingKey: map['rk'] as String?,
      serverId: map['sid'] as String?,
      mediaTitle: map['title'] as String?,
+      isPlaying: map['pl'] as bool?,
    );
  }

@@ -42,11 +42,19 @@ class WatchTogetherPeerService {
  final _errorController = StreamController<PeerError>.broadcast();
  final _connectionStateController = StreamController<bool>.broadcast();

-  // Reconnection state
+  // Reconnection state (signaling server)
  int _reconnectAttempts = 0;
  static const int _maxReconnectAttempts = 3;
  Timer? _reconnectTimer;

+  // Peer health monitoring (data channel)
+  final Map<String, DateTime> _lastPeerActivity = {};
+  final Map<String, int> _peerReconnectAttempts = {};
+  Timer? _peerHealthCheckTimer;
+  static const Duration _peerTimeout = Duration(seconds: 30);
+  static const Duration _peerHealthCheckInterval = Duration(seconds: 10);
+  static const int _maxPeerReconnectAttempts = 3;
+
  /// Stream of peer IDs when a new peer connects
  Stream<String> get onPeerConnected => _peerConnectedController.stream;

@@ -218,9 +226,15 @@ class WatchTogetherPeerService {
    conn.on('open').listen((_) {
      appLogger.d('WatchTogether: Data channel opened with: $peerId');
      _connections[peerId] = conn;
+      _updatePeerActivity(peerId); // Track initial activity
      _peerConnectedController.add(peerId);
      _connectionStateController.add(true);

+      // Start health monitoring if not already running
+      if (_peerHealthCheckTimer == null) {
+        _startPeerHealthCheck();
+      }
+
      if (completer != null && !completer.isCompleted) {
        completer.complete();
      }
@@ -228,6 +242,7 @@ class WatchTogetherPeerService {

    conn.on('data').listen((data) {
      try {
+        _updatePeerActivity(peerId); // Track activity on each message
        final message = SyncMessage.fromJson(data as String);
        appLogger.d('WatchTogether: Received message: ${message.type} from $peerId');
        _messageReceivedController.add(message);
@@ -239,9 +254,12 @@ class WatchTogetherPeerService {
    conn.on('close').listen((_) {
      appLogger.d('WatchTogether: Connection closed with: $peerId');
      _connections.remove(peerId);
+      _lastPeerActivity.remove(peerId);
+      _peerReconnectAttempts.remove(peerId);
      _peerDisconnectedController.add(peerId);

      if (_connections.isEmpty) {
+        _stopPeerHealthCheck();
        _connectionStateController.add(false);
      }
    });
@@ -255,6 +273,9 @@ class WatchTogetherPeerService {
          originalError: error,
        ),
      );
+
+      // Attempt reconnection on data channel error
+      _attemptPeerReconnect(peerId);
    });
  }

@@ -283,6 +304,84 @@ class WatchTogetherPeerService {
    }
  }

+  /// Start peer health monitoring
+  void _startPeerHealthCheck() {
+    _peerHealthCheckTimer?.cancel();
+    _peerHealthCheckTimer = Timer.periodic(_peerHealthCheckInterval, (_) {
+      _checkPeerHealth();
+    });
+  }
+
+  /// Stop peer health monitoring
+  void _stopPeerHealthCheck() {
+    _peerHealthCheckTimer?.cancel();
+    _peerHealthCheckTimer = null;
+  }
+
+  /// Check health of all peer connections
+  void _checkPeerHealth() {
+    final now = DateTime.now();
+    final peersToReconnect = <String>[];
+
+    for (final peerId in _connections.keys.toList()) {
+      final lastActivity = _lastPeerActivity[peerId];
+      if (lastActivity != null && now.difference(lastActivity) > _peerTimeout) {
+        appLogger.w('WatchTogether: Peer $peerId timed out (no activity for ${_peerTimeout.inSeconds}s)');
+        peersToReconnect.add(peerId);
+      }
+    }
+
+    for (final peerId in peersToReconnect) {
+      _attemptPeerReconnect(peerId);
+    }
+  }
+
+  /// Update peer activity timestamp (called on each message received)
+  void _updatePeerActivity(String peerId) {
+    _lastPeerActivity[peerId] = DateTime.now();
+    // Reset reconnect attempts on successful activity
+    _peerReconnectAttempts[peerId] = 0;
+  }
+
+  /// Attempt to reconnect to a peer with exponential backoff
+  void _attemptPeerReconnect(String peerId) {
+    final attempts = _peerReconnectAttempts[peerId] ?? 0;
+
+    if (attempts >= _maxPeerReconnectAttempts) {
+      appLogger.e('WatchTogether: Max reconnect attempts reached for peer $peerId');
+      // Remove the dead connection and notify
+      _connections.remove(peerId);
+      _lastPeerActivity.remove(peerId);
+      _peerReconnectAttempts.remove(peerId);
+      _peerDisconnectedController.add(peerId);
+
+      if (_connections.isEmpty) {
+        _connectionStateController.add(false);
+      }
+      return;
+    }
+
+    _peerReconnectAttempts[peerId] = attempts + 1;
+    final delay = Duration(seconds: (attempts + 1) * 2); // Exponential backoff
+
+    appLogger.d(
+      'WatchTogether: Attempting peer reconnect to $peerId (${attempts + 1}/$_maxPeerReconnectAttempts) in ${delay.inSeconds}s',
+    );
+
+    // Close existing connection if any
+    _connections[peerId]?.close();
+    _connections.remove(peerId);
+
+    // Schedule reconnection attempt
+    Timer(delay, () {
+      if (_peer != null && !_connections.containsKey(peerId)) {
+        appLogger.d('WatchTogether: Reconnecting to peer $peerId');
+        final conn = _peer!.connect(peerId, options: PeerConnectOption(reliable: true));
+        _handleNewConnection(conn, isOutgoing: true);
+      }
+    });
+  }
+
  /// Broadcast a message to all connected peers
  void broadcast(SyncMessage message) {
    final json = message.toJson();
@@ -318,6 +417,11 @@ class WatchTogetherPeerService {
    _reconnectTimer?.cancel();
    _reconnectTimer = null;

+    // Stop peer health monitoring
+    _stopPeerHealthCheck();
+    _lastPeerActivity.clear();
+    _peerReconnectAttempts.clear();
+
    // Close all data connections
    for (final conn in _connections.values) {
      conn.close();
@@ -39,9 +39,18 @@ class WatchTogetherSyncManager {

  // Drift correction constants
  static const Duration maxAllowedDrift = Duration(seconds: 2);
-  static const Duration positionSyncInterval = Duration(seconds: 5);
+  static const Duration positionSyncInterval = Duration(seconds: 3);
  static const Duration excessiveDrift = Duration(seconds: 10);

+  // Buffering debounce constants
+  static const Duration _bufferingDebounceDelay = Duration(milliseconds: 500);
+
+  // Buffering debounce timer - prevents false pauses from brief buffering events
+  Timer? _bufferingDebounceTimer;
+
+  // Peers with pending (debounced) buffering state
+  final Map<String, bool> _pendingBufferingState = {};
+
  // Track last known state to avoid duplicate broadcasts
  bool _lastKnownPlaying = false;
  double _lastKnownRate = 1.0;
@@ -159,13 +168,16 @@ class WatchTogetherSyncManager {
    _rateSubscription?.cancel();
    _messageSubscription?.cancel();
    _positionSyncTimer?.cancel();
+    _bufferingDebounceTimer?.cancel();

    _playingSubscription = null;
    _bufferingSubscription = null;
    _rateSubscription = null;
    _messageSubscription = null;
    _positionSyncTimer = null;
+    _bufferingDebounceTimer = null;

+    _pendingBufferingState.clear();
    _player = null;
    appLogger.d('WatchTogether: Player detached');
  }
@@ -201,7 +213,7 @@ class WatchTogetherSyncManager {
    });

    // Listen to buffering state changes
-    _bufferingSubscription = _player!.streams.buffering.listen((isBuffering) {
+    _bufferingSubscription = _player!.streams.buffering.listen((isBuffering) async {
      if (_isRemoteAction) return;

      // Announce ready when we stop buffering for the first time (video loaded)
@@ -218,6 +230,13 @@ class WatchTogetherSyncManager {
      }

      _peerService.broadcast(SyncMessage.buffering(isBuffering, peerId: _peerService.myPeerId));
+
+      // Check for auto-resume when LOCAL buffering stops
+      // This fixes the bug where subtitle loading would pause both users but
+      // only resume would trigger from remote buffering messages, not local
+      if (!isBuffering) {
+        await _checkAutoResume();
+      }
    });

    // Listen to rate changes
@@ -239,11 +258,18 @@ class WatchTogetherSyncManager {
  }

  /// Start periodic position sync (host only)
+  /// Includes play/pause state for eventual consistency
  void _startPositionSync() {
    _positionSyncTimer?.cancel();
    _positionSyncTimer = Timer.periodic(positionSyncInterval, (_) {
      if (_player != null && _session.isHost) {
-        _peerService.broadcast(SyncMessage.positionSync(_player!.state.position, peerId: _peerService.myPeerId));
+        _peerService.broadcast(
+          SyncMessage.positionSync(
+            _player!.state.position,
+            peerId: _peerService.myPeerId,
+            isPlaying: _player!.state.playing,
+          ),
+        );
      }
    });
  }
@@ -364,20 +390,36 @@ class WatchTogetherSyncManager {
          break;
        }
        if (message.peerId != null && message.bufferingState != null) {
-          _participantBuffering[message.peerId!] = message.bufferingState!;
+          final peerId = message.peerId!;
+          final isBuffering = message.bufferingState!;

-          // Auto-pause when any peer starts buffering
-          if (isAnyBuffering && _player!.state.playing) {
-            _wasPlayingBeforeBuffering = true;
-            appLogger.d('WatchTogether: Peer buffering, pausing playback');
-            await _applyRemotePause();
-          }
-          // Auto-resume when all peers stop buffering AND all ready (if we were playing before)
-          else if (isAllReady && !isAnyBuffering && !_player!.state.playing && _wasPlayingBeforeBuffering) {
-            _wasPlayingBeforeBuffering = false;
-            appLogger.d('WatchTogether: All peers done buffering, resuming playback');
-            await _applyRemotePlay(position: _pendingPlayPosition);
-            _pendingPlayPosition = null;
+          if (isBuffering) {
+            // Peer started buffering - use debounce to avoid false pauses
+            _pendingBufferingState[peerId] = true;
+
+            // Cancel existing debounce timer if any
+            _bufferingDebounceTimer?.cancel();
+            _bufferingDebounceTimer = Timer(_bufferingDebounceDelay, () async {
+              // Check if still pending after debounce delay
+              if (_pendingBufferingState[peerId] == true) {
+                _participantBuffering[peerId] = true;
+                _pendingBufferingState.remove(peerId);
+
+                // Auto-pause when any peer starts buffering (sustained)
+                if (isAnyBuffering && _player != null && _player!.state.playing) {
+                  _wasPlayingBeforeBuffering = true;
+                  appLogger.d('WatchTogether: Peer buffering (sustained), pausing playback');
+                  await _applyRemotePause();
+                }
+              }
+            });
+          } else {
+            // Peer stopped buffering - cancel pending debounce and update immediately
+            _pendingBufferingState.remove(peerId);
+            _participantBuffering[peerId] = false;
+
+            // Auto-resume when all peers stop buffering AND all ready
+            await _checkAutoResume();
          }
        }
        break;
@@ -386,6 +428,20 @@ class WatchTogetherSyncManager {
        if (message.position != null) {
          _checkAndCorrectDrift(message.position!, message.timestamp);
        }
+        // Reconcile play/pause state if host sent it and we diverged
+        // This provides eventual consistency for play/pause state
+        if (message.isPlaying != null && _player != null && !_session.isHost) {
+          final localPlaying = _player!.state.playing;
+          if (message.isPlaying! && !localPlaying && !isAnyBuffering && isAllReady) {
+            // Host is playing but we're paused - sync up
+            appLogger.d('WatchTogether: Play/pause state diverged, syncing to host (playing)');
+            await _applyRemotePlay(position: message.position);
+          } else if (!message.isPlaying! && localPlaying) {
+            // Host is paused but we're playing - sync up
+            appLogger.d('WatchTogether: Play/pause state diverged, syncing to host (paused)');
+            await _applyRemotePause();
+          }
+        }
        break;

      case SyncMessageType.rate:
@@ -437,12 +493,7 @@ class WatchTogetherSyncManager {
          appLogger.d('WatchTogether: Peer ${message.peerId} player ready: ${message.bufferingState}');

          // If we were waiting to play and all are now ready, start playback
-          if (isAllReady && !isAnyBuffering && _wasPlayingBeforeBuffering) {
-            _wasPlayingBeforeBuffering = false;
-            appLogger.d('WatchTogether: All players ready, starting playback');
-            await _applyRemotePlay(position: _pendingPlayPosition);
-            _pendingPlayPosition = null;
-          }
+          await _checkAutoResume();
        }
        break;
    }
@@ -518,6 +569,21 @@ class WatchTogetherSyncManager {
    }
  }

+  /// Check if conditions are met to auto-resume playback
+  /// Called when local or remote buffering stops, or when a peer becomes ready
+  Future<void> _checkAutoResume() async {
+    if (_player == null) return;
+    if (!isAllReady) return;
+    if (isAnyBuffering) return;
+    if (_player!.state.playing) return;
+    if (!_wasPlayingBeforeBuffering) return;
+
+    _wasPlayingBeforeBuffering = false;
+    appLogger.d('WatchTogether: All conditions met, auto-resuming playback');
+    await _applyRemotePlay(position: _pendingPlayPosition);
+    _pendingPlayPosition = null;
+  }
+
  /// Apply remote rate change
  Future<void> _applyRemoteRate(double rate) async {
    if (_player == null) return;
@@ -691,6 +757,7 @@ class WatchTogetherSyncManager {
    detachPlayer();
    _participantBuffering.clear();
    _participantReady.clear();
+    _pendingBufferingState.clear();
    _hasAnnouncedReady = false;
  }
 }