sync_replay_project/apps/desktop/test/coord_presence_test.dart

import 'dart:async';
import 'dart:io';

import 'package:flutter_test/flutter_test.dart';
import 'package:media_kit/media_kit.dart';
import 'package:sync_relay_core/core.dart';
import 'package:sync_relay_server/server.dart';
import 'package:hive_ce/hive.dart';
import 'package:watch_together/src/session/session_coordinator.dart';
import 'package:watch_together/src/state/session_state.dart';
import 'package:watch_together/src/sync/sync_client.dart';

Future<void> _poll(
  FutureOr<bool> Function() cond,
  String what, {
  Duration timeout = const Duration(seconds: 15),
}) async {
  final deadline = DateTime.now().add(timeout);
  while (DateTime.now().isBefore(deadline)) {
    if (await cond()) return;
    await Future.delayed(const Duration(milliseconds: 200));
  }
  fail('timed out waiting for $what');
}

void main() {
  test(
    'dead server clears room state and reconnects instead of freezing',
    timeout: const Timeout(Duration(minutes: 2)),
    () async {
      MediaKit.ensureInitialized();
      final server = RoomServer(
          const AppConfig(port: 0, secret: 'presence', name: 'presence'));
      await server.start();
      final coord = SessionCoordinator();
      try {
        await coord.connect(ConnectParams(
          host: '127.0.0.1',
          port: server.boundPort,
          secret: 'presence',
          name: 'watcher',
        ));
        await coord.playback.player.setVolume(0);
        await _poll(() => coord.room.peek() != null, 'room state to arrive');
        expect(coord.reconnecting.peek(), isFalse);

        await server.stop();
        await _poll(
          () => coord.reconnecting.peek() == true,
          'reconnect loop to engage',
        );
        await _poll(
          () => coord.room.peek() == null,
          'stale room (ghost members) to clear',
        );
        coord.cancelReconnect();
        expect(coord.reconnecting.peek(), isFalse);
      } finally {
        await coord.dispose();
      }
    },
  );

  test(
    'cancel then retry within the reconnect delay dials only once',
    timeout: const Timeout(Duration(minutes: 2)),
    () async {
      MediaKit.ensureInitialized();
      final first = RoomServer(
          const AppConfig(port: 0, secret: 'race', name: 'race'));
      await first.start();
      final port = first.boundPort;
      final coord = SessionCoordinator();
      RoomServer? second;
      final observer = SyncClient();
      try {
        await coord.connect(ConnectParams(
            host: '127.0.0.1', port: port, secret: 'race', name: 'watcher'));
        await coord.playback.player.setVolume(0);
        await first.stop();
        await _poll(() => coord.reconnecting.peek(), 'reconnect to engage');
        second = RoomServer(
            AppConfig(port: port, secret: 'race', name: 'race'));
        await second.start();
        var joins = 0;
        final sub = observer.feeds.listen((f) {
          if (f.action == 'join' && f.actor == 'watcher') joins++;
        });
        await observer.connect(
            wsUri: Uri.parse('ws://127.0.0.1:$port/sync'),
            secret: 'race',
            name: 'observer');
        // Within the 1.5 s delay: the first attempt must die with Cancel.
        coord.cancelReconnect();
        coord.retry();
        await _poll(() => !coord.reconnecting.peek() && coord.room.peek() != null,
            'reconnect to land');
        await Future.delayed(const Duration(seconds: 4));
        await sub.cancel();
        expect(joins, 1, reason: 'a stale attempt dialled a second time');
        expect(
            second.hub.engine.members.values
                .where((m) => m.name == 'watcher')
                .length,
            1);
      } finally {
        await observer.dispose();
        await coord.dispose();
        await second?.stop();
      }
    },
  );

  test(
    'reconnect restores the server-side ready flag (force-all included)',
    timeout: const Timeout(Duration(minutes: 2)),
    () async {
      MediaKit.ensureInitialized();
      final first = RoomServer(
          const AppConfig(port: 0, secret: 'force', name: 'force'));
      await first.start();
      final port = first.boundPort;
      final coord = SessionCoordinator();
      final friend = SyncClient();
      RoomServer? second;
      try {
        await coord.connect(ConnectParams(
            host: '127.0.0.1', port: port, secret: 'force', name: 'watcher'));
        await coord.playback.player.setVolume(0);
        await friend.connect(
            wsUri: Uri.parse('ws://127.0.0.1:$port/sync'),
            secret: 'force',
            name: 'friend');
        // Never pressed Space locally; a friend marks everyone ready.
        friend.sendForce('all', true);
        await _poll(() => coord.selfMember.peek()?.ready == true,
            'forced ready to reach watcher');
        await friend.disconnect();
        await first.stop();
        await _poll(() => coord.reconnecting.peek(), 'reconnect to engage');
        second = RoomServer(
            AppConfig(port: port, secret: 'force', name: 'force'));
        await second.start();
        await _poll(
            () => second!.hub.engine.members.values
                .any((m) => m.name == 'watcher' && m.ready),
            'watcher to come back ready');
      } finally {
        await friend.dispose();
        await coord.dispose();
        await second?.stop();
      }
    },
  );

  test(
    'cancel during connect setup aborts without joining',
    timeout: const Timeout(Duration(minutes: 1)),
    () async {
      MediaKit.ensureInitialized();
      final server = RoomServer(
          const AppConfig(port: 0, secret: '', name: 'cancel'));
      await server.start();
      final coord = SessionCoordinator();
      try {
        final pending = coord.connect(ConnectParams(
            host: '127.0.0.1',
            port: server.boundPort,
            secret: '',
            name: 'watcher'));
        await coord.cancelConnect();
        await expectLater(pending, throwsA(isA<ConnectCancelled>()));
        await Future.delayed(const Duration(seconds: 2));
        expect(server.hub.engine.members, isEmpty);
        expect(coord.reconnecting.peek(), isFalse);
      } finally {
        await coord.dispose();
        await server.stop();
      }
    },
  );

  test(
    'cancelling a host setup releases the port',
    timeout: const Timeout(Duration(minutes: 1)),
    () async {
      MediaKit.ensureInitialized();
      Hive.init('/home/foxyz/.claude/jobs/4898998e/tmp/hive_coord_test');
      final probe = await ServerSocket.bind(InternetAddress.anyIPv4, 0);
      final port = probe.port;
      await probe.close();
      final session = SessionState.instance;
      try {
        final pending = session.host(
            ConnectParams(host: '', port: port, secret: '', name: 'host'),
            ConnectForm());
        await session.cancelPending();
        await expectLater(pending, throwsA(isA<ConnectCancelled>()));
        expect(session.route.peek(), isNot(AppRoute.session));
        final rebind = await ServerSocket.bind(InternetAddress.anyIPv4, port);
        await rebind.close();
      } finally {
        await session.coordinator.leave(silent: true);
        await session.stopServer();
      }
    },
  );
}