Ensure live broadcasts stay in sync if ws reconnects

lib/src/model/broadcast/broadcast_game_controller.dart

@@ -36,15 +36,18 @@ class BroadcastGameController extends _$BroadcastGameController
 
   AppLifecycleListener? _appLifecycleListener;
   StreamSubscription<SocketEvent>? _subscription;
+  StreamSubscription<void>? _socketOpenSubscription;
 
   late SocketClient _socketClient;
   late Root _root;
 
   final _engineEvalDebounce = Debouncer(const Duration(milliseconds: 150));
+  final _syncDebouncer = Debouncer(const Duration(milliseconds: 150));
 
-  DateTime? _onPauseAt;
   Timer? _startEngineEvalTimer;
 
+  Object? _key = Object();
+
   @override
   Future<BroadcastGameState> build(
     BroadcastRoundId roundId,
@@ -56,30 +59,37 @@ class BroadcastGameController extends _$BroadcastGameController
 
     _subscription = _socketClient.stream.listen(_handleSocketEvent);
 
+    await _socketClient.firstConnection;
+
+    _socketOpenSubscription = _socketClient.connectedStream.listen((_) {
+      if (state.valueOrNull?.isOngoing == true) {
+        _syncDebouncer(() {
+          _reloadPgn();
+        });
+      }
+    });
+
     final evaluationService = ref.watch(evaluationServiceProvider);
 
     _appLifecycleListener = AppLifecycleListener(
-      onPause: () {
-        _onPauseAt = DateTime.now();
-      },
       onResume: () {
         if (state.valueOrNull?.isOngoing == true) {
-          if (_onPauseAt != null) {
-            final diff = DateTime.now().difference(_onPauseAt!);
-            if (diff >= const Duration(minutes: 5)) {
-              ref.invalidateSelf();
-            }
-          }
+          _syncDebouncer(() {
+            _reloadPgn();
+          });
         }
       },
     );
 
     ref.onDispose(() {
+      _key = null;
       _subscription?.cancel();
+      _socketOpenSubscription?.cancel();
       _startEngineEvalTimer?.cancel();
       _engineEvalDebounce.dispose();
       evaluationService.disposeEngine();
       _appLifecycleListener?.dispose();
+      _syncDebouncer.dispose();
     });
 
     final pgn = await ref.withClient(
@@ -133,6 +143,43 @@ class BroadcastGameController extends _$BroadcastGameController
     return broadcastState;
   }
 
+  Future<void> _reloadPgn() async {
+    if (!state.hasValue) return;
+    final key = _key;
+
+    final pgn = await ref.withClient(
+      (client) => BroadcastRepository(client).getGamePgn(roundId, gameId),
+    );
+
+    // check provider is still mounted
+    if (key == _key) {
+      final game = PgnGame.parsePgn(pgn);
+      final pgnHeaders = IMap(game.headers);
+      final rootComments =
+          IList(game.comments.map((c) => PgnComment.fromPgn(c)));
+
+      final newRoot = Root.fromPgnGame(game);
+
+      final broadcastPath = newRoot.mainlinePath;
+      final lastMove = newRoot.branchAt(newRoot.mainlinePath)?.sanMove.move;
+
+      newRoot.merge(_root);
+
+      _root = newRoot;
+
+      state = AsyncData(
+        state.requireValue.copyWith(
+          pgnHeaders: pgnHeaders,
+          pgnRootComments: rootComments,
+          broadcastPath: broadcastPath,
+          root: _root.view,
+          lastMove: lastMove,
+          clocks: _getClocks(state.requireValue.currentPath),
+        ),
+      );
+    }
+  }
+
   void _handleSocketEvent(SocketEvent event) {
     if (!state.hasValue) return;
 

lib/src/model/broadcast/broadcast_round_controller.dart

@@ -12,6 +12,7 @@ import 'package:lichess_mobile/src/model/common/socket.dart';
 import 'package:lichess_mobile/src/network/http.dart';
 import 'package:lichess_mobile/src/network/socket.dart';
 import 'package:lichess_mobile/src/utils/json.dart';
+import 'package:lichess_mobile/src/utils/rate_limit.dart';
 import 'package:riverpod_annotation/riverpod_annotation.dart';
 
 part 'broadcast_round_controller.g.dart';
@@ -22,12 +23,15 @@ class BroadcastRoundController extends _$BroadcastRoundController {
       Uri(path: 'study/$broadcastRoundId/socket/v6');
 
   StreamSubscription<SocketEvent>? _subscription;
+  StreamSubscription<void>? _socketOpenSubscription;
   AppLifecycleListener? _appLifecycleListener;
 
-  DateTime? _onPauseAt;
-
   late SocketClient _socketClient;
 
+  final _debouncer = Debouncer(const Duration(milliseconds: 150));
+
+  Object? _key = Object();
+
   @override
   Future<BroadcastRoundWithGames> build(
     BroadcastRoundId broadcastRoundId,
@@ -38,32 +42,49 @@ class BroadcastRoundController extends _$BroadcastRoundController {
 
     _subscription = _socketClient.stream.listen(_handleSocketEvent);
 
+    await _socketClient.firstConnection;
+
+    _socketOpenSubscription = _socketClient.connectedStream.listen((_) {
+      if (state.valueOrNull?.round.status == RoundStatus.live) {
+        _debouncer(() {
+          _syncRound();
+        });
+      }
+    });
+
     _appLifecycleListener = AppLifecycleListener(
-      onPause: () {
-        _onPauseAt = DateTime.now();
-      },
       onResume: () {
         if (state.valueOrNull?.round.status == RoundStatus.live) {
-          if (_onPauseAt != null) {
-            final diff = DateTime.now().difference(_onPauseAt!);
-            if (diff >= const Duration(minutes: 5)) {
-              ref.invalidateSelf();
-            }
-          }
+          _debouncer(() {
+            _syncRound();
+          });
         }
       },
     );
 
     ref.onDispose(() {
+      _key = null;
       _subscription?.cancel();
+      _socketOpenSubscription?.cancel();
       _appLifecycleListener?.dispose();
+      _debouncer.dispose();
     });
 
-    final round = await ref.withClient(
+    return ref.withClient(
       (client) => BroadcastRepository(client).getRound(broadcastRoundId),
     );
+  }
 
-    return round;
+  Future<void> _syncRound() async {
+    if (state.hasValue == false) return;
+    final key = _key;
+    final round = await ref.withClient(
+      (client) => BroadcastRepository(client).getRound(broadcastRoundId),
+    );
+    // check provider is still mounted
+    if (key == _key) {
+      state = AsyncData(round);
+    }
   }
 
   void _handleSocketEvent(SocketEvent event) {

lib/src/model/common/eval.dart

@@ -31,6 +31,14 @@ class ExternalEval with _$ExternalEval implements Eval {
     ({String name, String comment})? judgment,
   }) = _ExternalEval;
 
+  factory ExternalEval.fromPgnEval(PgnEvaluation eval) {
+    return ExternalEval(
+      cp: eval.pawns != null ? cpFromPawns(eval.pawns!) : null,
+      mate: eval.mate,
+      depth: eval.depth,
+    );
+  }
+
   factory ExternalEval.fromJson(Map<String, dynamic> json) =>
       _$ExternalEvalFromJson(json);
 

lib/src/model/common/node.dart

@@ -141,6 +141,23 @@ abstract class Node {
     }
   }
 
+  void merge(Node other) {
+    if (other.eval != null) {
+      eval = other.eval;
+    }
+    if (other.opening != null) {
+      opening = other.opening;
+    }
+    for (final otherChild in other.children) {
+      final child = childById(otherChild.id);
+      if (child != null) {
+        child.merge(otherChild);
+      } else {
+        addChild(otherChild);
+      }
+    }
+  }
+
   /// Adds a new node at the given path and returns the new path.
   ///
   /// Returns a tuple of the new path and whether the node was added.
@@ -419,13 +436,69 @@ class Branch extends Node {
   @override
   Branch branchAt(UciPath path) => nodeAt(path) as Branch;
 
-  /// Gets the clock information from the comments.
+  @override
+  void merge(Node other) {
+    if (other is Branch) {
+      other.lichessAnalysisComments?.forEach((c) {
+        if (lichessAnalysisComments == null) {
+          lichessAnalysisComments = [c];
+        } else {
+          final existing = lichessAnalysisComments?.firstWhereOrNull(
+            (e) => e.text == c.text,
+          );
+          if (existing == null) {
+            lichessAnalysisComments?.add(c);
+          }
+        }
+      });
+      other.startingComments?.forEach((c) {
+        if (startingComments == null) {
+          startingComments = [c];
+        } else {
+          final existing = startingComments?.firstWhereOrNull(
+            (e) => e.text == c.text,
+          );
+          if (existing == null) {
+            startingComments?.add(c);
+          }
+        }
+      });
+      other.comments?.forEach((c) {
+        if (comments == null) {
+          comments = [c];
+        } else {
+          final existing = comments?.firstWhereOrNull(
+            (e) => e.text == c.text,
+          );
+          if (existing == null) {
+            comments?.add(c);
+          }
+        }
+      });
+      if (other.nags != null) {
+        nags = other.nags;
+      }
+    }
+    super.merge(other);
+  }
+
+  /// Gets the first available clock from the comments.
   Duration? get clock {
     final clockComment = (lichessAnalysisComments ?? comments)
         ?.firstWhereOrNull((c) => c.clock != null);
     return clockComment?.clock;
   }
 
+  /// Gets the first available external eval from the comments.
+  ExternalEval? get externalEval {
+    final comment = (lichessAnalysisComments ?? comments)?.firstWhereOrNull(
+      (c) => c.eval != null,
+    );
+    return comment?.eval != null
+        ? ExternalEval.fromPgnEval(comment!.eval!)
+        : null;
+  }
+
   @override
   String toString() {
     return 'Branch(id: $id, fen: ${position.fen}, sanMove: $sanMove, eval: $eval, children: $children)';