import 'dart:async'; import '../../mpv/mpv.dart'; import '../../utils/app_logger.dart'; import '../models/playback_state.dart'; import '../models/sync_message.dart'; import '../models/watch_session.dart'; import '../primitives.dart'; import 'attached_player.dart'; import 'clock_sync.dart'; import 'guest_playback_reconciler.dart'; import 'host_playback_coordinator.dart'; import 'watch_together_peer_service.dart'; /// Session-scoped playback-sync controller. /// /// Lives for the whole Watch Together session (created at create/join, not /// at player attach), so no sync message is ever dropped during episode /// switches or other attach gaps — the player attachment is just an output /// binding the role engine reconciles against. /// /// Routes the v3 protocol between the relay and the role engine: /// host → [HostPlaybackCoordinator] (single writer of [PlaybackState]), /// guest → [GuestPlaybackReconciler] (+ [ClockSync] against the host). class WatchTogetherController { WatchTogetherController({ required WatchTogetherPeerService peerService, required WatchSession session, int Function()? nowMs, }) : _peerService = peerService, _session = session, _nowMs = nowMs ?? watchTogetherSystemNowMs { if (session.isHost) { _coordinator = HostPlaybackCoordinator( myPeerId: peerService.myPeerId ?? '', controlMode: session.controlMode, sendState: _sendState, callbacks: HostCoordinatorCallbacks( onPhaseChanged: (phase) => onPhaseChanged?.call(phase), onWaitingOnChanged: (peers) => onWaitingOnChanged?.call(peers), onResumedWithout: (peers) => onResumedWithout?.call(peers), onRemoteAction: (peer, hint) => onRemoteAction?.call(peer, hint), ), nowMs: _nowMs, ); } else { _clockSync = ClockSync(sendPing: _sendClockPing, nowMs: _nowMs); _reconciler = GuestPlaybackReconciler( myPeerId: peerService.myPeerId ?? '', sendToHost: _sendToHost, clockSync: _clockSync!, callbacks: GuestReconcilerCallbacks( onMediaSwitchNeeded: (ratingKey, serverId, title) => onMediaStateReceived?.call(ratingKey, serverId, title), onControlModeChanged: (mode) => onControlModeReceived?.call(mode), onPhaseChanged: (phase) => onPhaseChanged?.call(phase), onWaitingOnChanged: (peers) => onWaitingOnChanged?.call(peers), onCorrectingChanged: (correcting) => onCorrectingChanged?.call(correcting), onRemoteAction: (peer, hint) => onRemoteAction?.call(peer, hint), ), nowMs: _nowMs, ); _clockSync!.start(); } _subscriptions.add(peerService.onMessageReceived.listen(_enqueueMessage)); _subscriptions.add(peerService.onPeerDisconnected.listen(_handlePeerDisconnected)); } final WatchTogetherPeerService _peerService; final int Function() _nowMs; WatchSession _session; HostPlaybackCoordinator? _coordinator; GuestPlaybackReconciler? _reconciler; ClockSync? _clockSync; AttachedPlayer? _attachedPlayer; final List> _subscriptions = []; Future _messageQueue = Future.value(); bool _disposed = false; /// Protocol versions learned from join messages (absent ⇒ v1). final Map _peerVersions = {}; final Set _updateToastShown = {}; // Provider-facing callbacks. void Function(PlaybackPhase phase)? onPhaseChanged; void Function(List peerIds)? onWaitingOnChanged; void Function(bool correcting)? onCorrectingChanged; void Function(ControlMode mode)? onControlModeReceived; void Function(String ratingKey, String serverId, String? mediaTitle)? onMediaStateReceived; void Function()? onHostExitedPlayer; void Function(String peerId, PlaybackActionHint hint)? onRemoteAction; void Function(String peerId)? onPeerNeedsUpdate; void Function(List peerIds)? onResumedWithout; bool get hasPlayer => _attachedPlayer != null; PlaybackPhase? get phase => _session.isHost ? _coordinator?.phase : _reconciler?.latestState?.phase; /// Update the session (e.g. when the control mode changes). void updateSession(WatchSession session) { _session = session; _coordinator?.updateControlMode(session.controlMode); } // --------------------------------------------------------------------- // Player attachment // --------------------------------------------------------------------- /// Attach the local player for [ratingKey]/[serverId]. /// /// [hasFirstFrame] is the screen's first-frame snapshot; [startupHold] /// delays readiness until platform startup gates (frame-rate switch) /// release; [remoteSeek] routes sync seeks through the screen's seek path /// (Plex transcode restarts). void attachPlayer( Player player, { required String ratingKey, required String serverId, String? mediaTitle, bool hasFirstFrame = false, Future? startupHold, Future Function(Duration target)? remoteSeek, }) { detachPlayer(); final attached = AttachedPlayer( player: player, onLost: () { appLogger.w('WatchTogether: Player attachment lost, detaching from sync'); detachPlayer(); }, remoteSeek: remoteSeek, nowMs: _nowMs, ); _attachedPlayer = attached; if (_session.isHost) { _coordinator!.attach( attached, ratingKey: ratingKey, serverId: serverId, mediaTitle: mediaTitle, hasFirstFrame: hasFirstFrame, startupHold: startupHold, ); } else { _reconciler!.attach( attached, ratingKey: ratingKey, serverId: serverId, hasFirstFrame: hasFirstFrame, startupHold: startupHold, ); } appLogger.d('WatchTogether: Player attached (host: ${_session.isHost})'); } /// Detach the player. [exiting] means the user left the video player (the /// epoch ends); an episode switch keeps the session and epoch flow. void detachPlayer({bool exiting = false}) { final attached = _attachedPlayer; if (attached == null) return; _attachedPlayer = null; _coordinator?.detachPlayer(exiting: exiting); _reconciler?.detachPlayer(); unawaited( attached.dispose().catchError((Object error, StackTrace stackTrace) { appLogger.e('WatchTogether: Failed to detach player subscriptions', error: error, stackTrace: stackTrace); }), ); appLogger.d('WatchTogether: Player detached (exiting: $exiting)'); } // --------------------------------------------------------------------- // Provider inputs // --------------------------------------------------------------------- /// Host switched media (also called right after attach with the same key, /// which is a no-op). void setCurrentMedia({required String ratingKey, required String serverId, String? mediaTitle}) { _coordinator?.setLocalMedia(ratingKey: ratingKey, serverId: serverId, mediaTitle: mediaTitle); } /// User seek executed locally (screen hook). void onLocalSeek(Duration position) { if (_session.isHost) { _coordinator?.onLocalSeekIntent(position); } else { _reconciler?.onLocalSeekIntent(position); } } void setBackgrounded(bool value) { _coordinator?.setBackgrounded(value); _reconciler?.setBackgrounded(value); } void announceJoin(String displayName) { final peerId = _peerService.myPeerId; if (peerId == null) return; _peerService.broadcast(SyncMessage.join(peerId: peerId, displayName: displayName, isHost: _session.isHost)); } void announceLeave() { final peerId = _peerService.myPeerId; if (peerId == null) return; _peerService.broadcast(SyncMessage.leave(peerId: peerId)); } /// Ask the host to (re-)send its current state. void requestState() { if (_session.isHost) return; final request = SyncMessage.requestState(peerId: _peerService.myPeerId); final hostPeerId = _session.hostPeerId; if (hostPeerId != null) { _peerService.sendTo(hostPeerId, request); } else { _peerService.broadcast(request); } } /// Relay reconnect completed: re-establish mutual state. void onReconnected() { if (_session.isHost) { _coordinator?.onReconnected(); } else { _reconciler?.onReconnected(); requestState(); } } void dispose() { _disposed = true; detachPlayer(exiting: true); for (final subscription in _subscriptions) { unawaited( subscription.cancel().catchError((Object error, StackTrace stackTrace) { appLogger.e('WatchTogether: Failed to cancel controller subscription', error: error, stackTrace: stackTrace); }), ); } _subscriptions.clear(); _clockSync?.stop(); _coordinator?.dispose(); _reconciler?.dispose(); } // --------------------------------------------------------------------- // Transport plumbing // --------------------------------------------------------------------- void _sendState(PlaybackState state, {String? toPeerId}) { final message = SyncMessage.state(state, peerId: _peerService.myPeerId); if (toPeerId != null) { _peerService.sendTo(toPeerId, message); } else { _peerService.broadcast(message); } } void _sendToHost(SyncMessage message) { final hostPeerId = _session.hostPeerId; if (hostPeerId != null) { _peerService.sendTo(hostPeerId, message); } else { _peerService.broadcast(message); } } void _sendClockPing(int pingId) { _sendToHost(SyncMessage.ping(pingId, peerId: _peerService.myPeerId)); } void _enqueueMessage(SyncMessage message) { _messageQueue = _messageQueue.then((_) => _handleMessage(message)).catchError(( Object error, StackTrace stackTrace, ) { appLogger.e('WatchTogether: Failed to handle ${message.type.name} message', error: error, stackTrace: stackTrace); }); } Future _handleMessage(SyncMessage message) async { if (_disposed) return; final senderId = message.peerId; if (senderId == null || senderId == _peerService.myPeerId) return; switch (message.type) { case SyncMessageType.state: // Only the host may author room state. if (_session.isHost || senderId != _session.hostPeerId) return; final state = message.state; if (state != null) _reconciler?.onState(state); break; case SyncMessageType.status: final status = message.status; if (_session.isHost && status != null) { _coordinator?.onPeerStatus(senderId, status); } break; case SyncMessageType.control: if (!_session.isHost) return; // In host-only mode nobody else gets a say. if (_session.controlMode == ControlMode.hostOnly) return; if (_isIncompatible(senderId)) return; final control = message.control; if (control != null) _coordinator?.onControlRequest(senderId, control); break; case SyncMessageType.requestState: if (_session.isHost) _coordinator?.onStateRequested(senderId); break; case SyncMessageType.ping: if (message.pingId != null) { // The pong timestamp is "host clock now" for the guest's offset // math — it must come from the same clock as the state anchors. _peerService.sendTo( senderId, SyncMessage( type: SyncMessageType.pong, timestamp: _nowMs(), pingId: message.pingId, peerId: _peerService.myPeerId, ), ); } break; case SyncMessageType.pong: if (message.pingId != null && !_session.isHost) { _clockSync?.onPong(message.pingId!, message.timestamp); } break; case SyncMessageType.join: _handleJoin(senderId, message); break; case SyncMessageType.leave: _peerVersions.remove(senderId); _coordinator?.onPeerLeft(senderId); break; case SyncMessageType.hostExitedPlayer: // Rides the ordered queue so it can't locally overtake state // messages that preceded it on the wire. Only the host may end the // media epoch. if (!_session.isHost && senderId == _session.hostPeerId) { onHostExitedPlayer?.call(); } break; } } bool _isIncompatible(String peerId) => (_peerVersions[peerId] ?? 1) != SyncMessage.protocolVersion; void _handleJoin(String senderId, SyncMessage message) { final version = message.version ?? 1; final firstSighting = !_peerVersions.containsKey(senderId); _peerVersions[senderId] = version; final compatible = version == SyncMessage.protocolVersion; if (!compatible && _updateToastShown.add(senderId)) { appLogger.w('WatchTogether: Peer $senderId speaks protocol v$version (ours: ${SyncMessage.protocolVersion})'); onPeerNeedsUpdate?.call(senderId); } if (_session.isHost) { _coordinator?.onPeerJoined(senderId, compatible: compatible); } else if (senderId == _session.hostPeerId && firstSighting) { // A fresh host join can mean a restarted host app with a reset // sequence counter — accept its numbering from scratch. _reconciler?.resetSequence(); } } void _handlePeerDisconnected(String peerId) { _peerVersions.remove(peerId); _updateToastShown.remove(peerId); _coordinator?.onPeerLeft(peerId); } }