Files
plezy/lib/watch_together/providers/watch_together_provider.dart
T

792 lines
27 KiB
Dart

import 'dart:async';
import 'dart:math';
import 'package:flutter/foundation.dart';
import '../../mpv/mpv.dart';
import '../../services/settings_service.dart';
import '../../utils/app_logger.dart';
import '../models/sync_message.dart';
import '../models/watch_session.dart';
import '../services/watch_together_peer_service.dart';
import '../services/watch_together_sync_manager.dart';
/// Callback type for when media switches (for guest navigation)
typedef MediaSwitchCallback = void Function(String ratingKey, String serverId, String mediaTitle);
/// Provider for Watch Together functionality
///
/// This provider manages:
/// - Session creation/joining
/// - Peer connections
/// - Playback synchronization
/// - Participant list
/// - Media switching across the session
class WatchTogetherProvider with ChangeNotifier {
WatchSession? _session;
WatchTogetherPeerService? _peerService;
WatchTogetherSyncManager? _syncManager;
final List<Participant> _participants = [];
bool _isSyncing = false;
bool _isDeferredPlay = false;
String _displayName = 'User';
String? _lastHandledCurrentPlaybackKey;
// Coalesce rapid-fire notifyListeners() calls into a single rebuild per frame.
// During Watch Together join, 4-5 notifications fire within milliseconds;
// this batches them into one rebuild to avoid overwhelming low-end devices.
bool _notifyScheduled = false;
bool _disposed = false;
@override
void notifyListeners() {
if (_disposed || _notifyScheduled) return;
_notifyScheduled = true;
scheduleMicrotask(() {
_notifyScheduled = false;
if (!_disposed) super.notifyListeners();
});
}
// Host reconnect grace period
Timer? _hostReconnectTimer;
bool _isWaitingForHostReconnect = false;
bool _hostIntentionallyLeft = false;
// Debounce map for action events (peerId+type → last emission timestamp)
final Map<String, int> _lastActionEventMs = {};
/// Generate a random display name for this session
static String _generateDisplayName() {
const adjectives = ['Happy', 'Sleepy', 'Sunny', 'Cozy', 'Chill', 'Swift', 'Brave', 'Calm', 'Jolly', 'Lucky'];
const nouns = ['Panda', 'Koala', 'Fox', 'Owl', 'Cat', 'Dog', 'Bear', 'Bunny', 'Duck', 'Penguin'];
final random = Random();
return '${adjectives[random.nextInt(adjectives.length)]} ${nouns[random.nextInt(nouns.length)]}';
}
/// Callback for when host switches media (guests should navigate)
/// Used by MainScreen when VideoPlayerScreen is not active
MediaSwitchCallback? onMediaSwitched;
/// Callback for VideoPlayerScreen to handle media switch internally (guest only)
/// When set, takes priority over onMediaSwitched for proper navigation context
MediaSwitchCallback? onPlayerMediaSwitched;
/// Callback for when host exits the video player (guests should exit too)
VoidCallback? onHostExitedPlayer;
// Stream subscriptions
StreamSubscription<String>? _peerConnectedSubscription;
StreamSubscription<String>? _peerDisconnectedSubscription;
StreamSubscription<SyncMessage>? _messageSubscription;
StreamSubscription<PeerError>? _errorSubscription;
// Getters
bool get isInSession => _session != null && _session!.state != SessionState.disconnected;
bool get isHost => _session?.isHost ?? false;
bool get isConnected => _session?.isConnected ?? false;
bool get isSyncing => _isSyncing;
bool get isDeferredPlay => _isDeferredPlay;
WatchSession? get session => _session;
List<Participant> get participants => List.unmodifiable(_participants);
int get participantCount => _participants.length;
ControlMode get controlMode => _session?.controlMode ?? ControlMode.hostOnly;
String? get sessionId => _session?.sessionId;
WatchTogetherSyncManager? get syncManager => _syncManager;
bool get isWaitingForHostReconnect => _isWaitingForHostReconnect;
// Participant join/leave event stream
final StreamController<ParticipantEvent> _participantEventController = StreamController<ParticipantEvent>.broadcast();
Stream<ParticipantEvent> get participantEvents => _participantEventController.stream;
// Current media getters
String? get currentMediaRatingKey => _session?.mediaRatingKey;
String? get currentMediaServerId => _session?.mediaServerId;
String? get currentMediaTitle => _session?.mediaTitle;
bool get hasCurrentPlayback =>
currentMediaRatingKey != null && currentMediaServerId != null && currentMediaTitle != null;
/// Set the display name for this user
void setDisplayName(String name) {
_displayName = name;
}
String? _buildPlaybackKey(String? ratingKey, String? serverId) {
if (ratingKey == null || serverId == null) return null;
return '$serverId:$ratingKey';
}
void _updateCurrentPlaybackSnapshot({
required String ratingKey,
required String serverId,
required String mediaTitle,
}) {
_session = _session?.copyWith(mediaRatingKey: ratingKey, mediaServerId: serverId, mediaTitle: mediaTitle);
}
void _clearCurrentPlaybackSnapshot() {
final session = _session;
if (session == null) return;
_session = WatchSession(
sessionId: session.sessionId,
role: session.role,
controlMode: session.controlMode,
state: session.state,
errorMessage: session.errorMessage,
hostPeerId: session.hostPeerId,
);
_lastHandledCurrentPlaybackKey = null;
}
void _dispatchCurrentPlayback({
required String ratingKey,
required String serverId,
required String mediaTitle,
required String source,
}) {
final callback = onPlayerMediaSwitched ?? onMediaSwitched;
if (callback == null) {
appLogger.d('WatchTogether: No media switch callback set, keeping snapshot from $source only');
return;
}
_lastHandledCurrentPlaybackKey = _buildPlaybackKey(ratingKey, serverId);
appLogger.d('WatchTogether: Dispatching current playback from $source: $mediaTitle');
callback(ratingKey, serverId, mediaTitle);
}
void markCurrentPlaybackHandled({required String ratingKey, required String serverId}) {
_lastHandledCurrentPlaybackKey = _buildPlaybackKey(ratingKey, serverId);
}
void requestCurrentPlaybackSnapshot() {
if (isHost || _peerService == null || _session == null || _peerService!.myPeerId == null) {
return;
}
final request = SyncMessage.requestSessionConfig(peerId: _peerService!.myPeerId);
if (_session!.hostPeerId != null) {
appLogger.d('WatchTogether: Requesting current playback snapshot from host');
_peerService!.sendTo(_session!.hostPeerId!, request);
} else {
appLogger.d('WatchTogether: Host peer unknown, broadcasting current playback snapshot request');
_peerService!.broadcast(request);
}
}
/// Wire up reconnection handler to re-announce join and readiness after reconnect
void _wireReconnectHandler() {
_peerService!.onReconnected = () {
_syncManager?.announceJoin(_displayName);
_syncManager?.reannounceReadyIfNeeded();
};
}
/// Wire up sync manager's state change callback to update provider state
void _wireSyncStateChanges() {
_syncManager!.onSyncStateChanged = (isSyncing) {
_isSyncing = isSyncing;
notifyListeners();
};
_syncManager!.onDeferredPlayChanged = (isDeferredPlay) {
_isDeferredPlay = isDeferredPlay;
notifyListeners();
};
}
/// Create a new watch together session as host
Future<String> createSession({
required ControlMode controlMode,
String? displayName,
String? sessionId,
String? mediaRatingKey,
String? mediaServerId,
String? mediaTitle,
}) async {
// Clean up any existing session
await leaveSession();
_lastHandledCurrentPlaybackKey = null;
appLogger.d('WatchTogether: Creating session with control mode: $controlMode');
final customRelayUrl = SettingsService.instanceOrNull?.getCustomRelayUrl();
_peerService = WatchTogetherPeerService(customBaseUrl: customRelayUrl);
_setupPeerServiceListeners();
try {
final createdSessionId = await _peerService!.createSession(sessionId: sessionId);
_session = WatchSession.createAsHost(
sessionId: createdSessionId,
hostPeerId: _peerService!.myPeerId!,
controlMode: controlMode,
mediaRatingKey: mediaRatingKey,
mediaServerId: mediaServerId,
mediaTitle: mediaTitle,
).copyWith(state: SessionState.connected);
_displayName = displayName ?? _generateDisplayName();
_participants.add(Participant(peerId: _peerService!.myPeerId!, displayName: _displayName, isHost: true));
_syncManager = WatchTogetherSyncManager(
peerService: _peerService!,
session: _session!,
displayName: _displayName,
);
_wireSyncStateChanges();
_wireReconnectHandler();
notifyListeners();
appLogger.d('WatchTogether: Session created: $createdSessionId');
return createdSessionId;
} catch (e) {
appLogger.e('WatchTogether: Failed to create session', error: e);
_session = _session?.copyWith(state: SessionState.error, errorMessage: e.toString());
notifyListeners();
rethrow;
}
}
/// Join an existing session as guest
Future<void> joinSession(String sessionId, {String? displayName}) async {
// Clean up any existing session
await leaveSession();
_lastHandledCurrentPlaybackKey = null;
appLogger.d('WatchTogether: Joining session: $sessionId');
final customRelayUrl = SettingsService.instanceOrNull?.getCustomRelayUrl();
_peerService = WatchTogetherPeerService(customBaseUrl: customRelayUrl);
_setupPeerServiceListeners();
_session = WatchSession.joinAsGuest(sessionId: sessionId);
notifyListeners();
try {
await _peerService!.joinSession(sessionId);
// Session will be fully configured when we receive sessionConfig from host
_session = _session!.copyWith(state: SessionState.connected, hostPeerId: 'wt-${sessionId.toUpperCase()}');
_displayName = displayName ?? _generateDisplayName();
_syncManager = WatchTogetherSyncManager(
peerService: _peerService!,
session: _session!,
displayName: _displayName,
);
_syncManager!.onSessionConfigReceived = (controlMode) {
_session = _session!.copyWith(controlMode: controlMode);
_syncManager!.updateSession(_session!);
notifyListeners();
};
_wireSyncStateChanges();
_wireReconnectHandler();
// Add self to participants
_participants.add(Participant(peerId: _peerService!.myPeerId!, displayName: _displayName, isHost: false));
// Announce join to other participants
_syncManager!.announceJoin(_displayName);
requestCurrentPlaybackSnapshot();
notifyListeners();
appLogger.d('WatchTogether: Joined session successfully');
} catch (e) {
appLogger.e('WatchTogether: Failed to join session', error: e);
_session = _session?.copyWith(state: SessionState.error, errorMessage: e.toString());
notifyListeners();
rethrow;
}
}
/// Enter a room by code — joins if it exists, creates if empty.
///
/// Returns `true` if the user became the host.
Future<bool> 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 probe.joinSession(sessionId);
shouldBeHost = probe.connectedPeers.isEmpty;
} on PeerError catch (e) {
if (e.serverCode == 'room_not_found') {
shouldBeHost = true;
} else {
await probe.disconnect();
probe.dispose();
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
Future<void> leaveSession() async {
if (_session == null) return;
appLogger.d('WatchTogether: Leaving session');
// Announce leave if connected
_syncManager?.announceLeave();
// Clean up subscriptions
_peerConnectedSubscription?.cancel();
_peerDisconnectedSubscription?.cancel();
_messageSubscription?.cancel();
_errorSubscription?.cancel();
_peerConnectedSubscription = null;
_peerDisconnectedSubscription = null;
_messageSubscription = null;
_errorSubscription = null;
// Cancel host reconnect grace period
_cancelHostReconnectGracePeriod();
// Clean up services
_syncManager?.dispose();
_syncManager = null;
await _peerService?.disconnect();
_peerService?.dispose();
_peerService = null;
_session = null;
_participants.clear();
_isSyncing = false;
_isDeferredPlay = false;
_lastHandledCurrentPlaybackKey = null;
_lastActionEventMs.clear();
_hostIntentionallyLeft = false;
notifyListeners();
appLogger.d('WatchTogether: Session left');
}
/// Attach a player to the sync manager
void attachPlayer(Player player) {
if (_syncManager == null) {
appLogger.w('WatchTogether: Cannot attach player - no sync manager');
return;
}
// Initialize sync manager with existing participants (may have joined before player attached)
final peerIds = _participants.map((p) => p.peerId).toList();
_syncManager!.initializeParticipants(peerIds);
_syncManager!.attachPlayer(player);
appLogger.d('WatchTogether: Player attached to sync manager');
}
/// Detach the player from the sync manager
void detachPlayer() {
_syncManager?.detachPlayer();
appLogger.d('WatchTogether: Player detached from sync manager');
}
/// Suppress position sync while the app is backgrounded.
void setBackgrounded(bool value) {
_syncManager?.setBackgrounded(value);
}
/// Set up listeners for peer service events
void _setupPeerServiceListeners() {
_peerConnectedSubscription = _peerService!.onPeerConnected.listen((peerId) {
appLogger.d('WatchTogether: Peer connected: $peerId');
// If host reconnected during grace period, cancel the timer
if (!isHost && peerId == _session?.hostPeerId && _isWaitingForHostReconnect) {
_cancelHostReconnectGracePeriod();
}
if (!isHost && peerId == _session?.hostPeerId) {
requestCurrentPlaybackSnapshot();
}
// Peer will announce themselves with a join message
notifyListeners();
});
_peerDisconnectedSubscription = _peerService!.onPeerDisconnected.listen((peerId) {
appLogger.d('WatchTogether: Peer disconnected: $peerId');
// Capture display name before removal for notification
final disconnectedName = _participants.where((p) => p.peerId == peerId).map((p) => p.displayName).firstOrNull;
_participants.removeWhere((p) => p.peerId == peerId);
// If host disconnected unexpectedly, start grace period for reconnection.
// Skip if the host already sent a deliberate leave message.
if (!isHost && peerId == _session?.hostPeerId && !_hostIntentionallyLeft) {
_startHostReconnectGracePeriod();
} else if (disconnectedName != null) {
_participantEventController.add(
ParticipantEvent(displayName: disconnectedName, type: ParticipantEventType.left),
);
}
notifyListeners();
});
_messageSubscription = _peerService!.onMessageReceived.listen((message) {
_handleSyncMessage(message);
});
_errorSubscription = _peerService!.onError.listen((error) {
appLogger.e('WatchTogether: Peer error: ${error.message}');
// Update session state on error
if (_session != null && _session!.state == SessionState.connected) {
_session = _session!.copyWith(state: SessionState.error, errorMessage: error.message);
notifyListeners();
}
});
}
/// Handle incoming sync messages for participant management
void _handleSyncMessage(SyncMessage message) {
switch (message.type) {
case SyncMessageType.join:
if (message.peerId != null && message.displayName != null) {
// Check if participant already exists
final existingIndex = _participants.indexWhere((p) => p.peerId == message.peerId);
if (existingIndex >= 0) {
// Update existing participant
_participants[existingIndex] = Participant(
peerId: message.peerId!,
displayName: message.displayName!,
isHost: message.isHost ?? false,
);
} else {
// Add new participant
_participants.add(
Participant(peerId: message.peerId!, displayName: message.displayName!, isHost: message.isHost ?? false),
);
_participantEventController.add(
ParticipantEvent(displayName: message.displayName!, type: ParticipantEventType.joined),
);
// Send our join info back so the new peer adds us to their
// participant list. Only reply to NEW peers to avoid an
// infinite join ping-pong (A→join→B→join→A→...).
if (_peerService != null) {
_peerService!.sendTo(
message.peerId!,
SyncMessage.join(peerId: _peerService!.myPeerId!, displayName: _displayName, isHost: isHost),
);
}
}
notifyListeners();
}
break;
case SyncMessageType.leave:
if (message.peerId != null) {
final leavingName = _participants
.where((p) => p.peerId == message.peerId)
.map((p) => p.displayName)
.firstOrNull;
_participants.removeWhere((p) => p.peerId == message.peerId);
if (leavingName != null) {
_participantEventController.add(
ParticipantEvent(displayName: leavingName, type: ParticipantEventType.left),
);
}
// If the host deliberately left, end the session for everyone.
if (!isHost && message.peerId == _session?.hostPeerId) {
_hostIntentionallyLeft = true;
_handleHostExitedPlayer(message);
leaveSession();
}
notifyListeners();
}
break;
case SyncMessageType.buffering:
if (message.peerId != null) {
final index = _participants.indexWhere((p) => p.peerId == message.peerId);
if (index >= 0) {
final newState = message.bufferingState ?? false;
if (_participants[index].isBuffering != newState) {
_participants[index] = _participants[index].copyWith(isBuffering: newState);
if (newState) {
_emitActionEvent(message.peerId, ParticipantEventType.buffering);
}
notifyListeners();
}
}
}
break;
case SyncMessageType.positionSync:
if (message.peerId != null && message.position != null) {
final index = _participants.indexWhere((p) => p.peerId == message.peerId);
if (index >= 0) {
_participants[index] = _participants[index].copyWith(lastKnownPosition: message.position);
// Don't notify for position updates - too frequent
}
}
break;
case SyncMessageType.mediaSwitch:
_handleMediaSwitch(message);
break;
case SyncMessageType.hostExitedPlayer:
_handleHostExitedPlayer(message);
break;
case SyncMessageType.sessionConfig:
_handleSessionConfig(message);
break;
case SyncMessageType.requestSessionConfig:
// Handled at sync manager level (host responds with config)
break;
case SyncMessageType.play:
_emitActionEvent(message.peerId, ParticipantEventType.resumed);
break;
case SyncMessageType.pause:
_emitActionEvent(message.peerId, ParticipantEventType.paused);
break;
case SyncMessageType.seek:
_emitActionEvent(message.peerId, ParticipantEventType.seeked);
break;
default:
break;
}
}
/// Emit an action event for a remote peer (with 1s debounce per peer+type)
void _emitActionEvent(String? peerId, ParticipantEventType type) {
if (peerId == null || peerId == _peerService?.myPeerId) return;
final key = '$peerId:${type.name}';
final now = DateTime.now().millisecondsSinceEpoch;
final last = _lastActionEventMs[key] ?? 0;
if (now - last < 1000) return;
_lastActionEventMs[key] = now;
final name = _participants.where((p) => p.peerId == peerId).map((p) => p.displayName).firstOrNull;
if (name != null) {
_participantEventController.add(ParticipantEvent(displayName: name, type: type));
}
}
/// Handle session config from host (guest only)
/// This is handled at provider level so it's processed even before player is attached
void _handleSessionConfig(SyncMessage message) {
if (isHost) return; // Host doesn't need to process config
if (message.controlMode != null) {
appLogger.d('WatchTogether: Received session config, controlMode: ${message.controlMode}');
_session = _session!.copyWith(controlMode: message.controlMode);
_syncManager?.updateSession(_session!); // Update sync manager if it exists
notifyListeners();
}
if (message.ratingKey != null && message.serverId != null && message.mediaTitle != null) {
final playbackKey = _buildPlaybackKey(message.ratingKey, message.serverId);
final shouldDispatch = playbackKey != _lastHandledCurrentPlaybackKey;
_updateCurrentPlaybackSnapshot(
ratingKey: message.ratingKey!,
serverId: message.serverId!,
mediaTitle: message.mediaTitle!,
);
notifyListeners();
if (shouldDispatch) {
_dispatchCurrentPlayback(
ratingKey: message.ratingKey!,
serverId: message.serverId!,
mediaTitle: message.mediaTitle!,
source: 'session config',
);
}
}
}
/// Called when user seeks locally (to broadcast to peers)
void onLocalSeek(Duration position) {
_syncManager?.onLocalSeek(position);
}
/// Whether the current user can control playback
bool canControl() {
if (_session == null) return true; // Not in session, can control
if (_session!.controlMode == ControlMode.anyone) return true;
return isHost;
}
/// Set the current media (host only) and broadcast to guests
///
/// Call this when the host starts playing new content.
/// Guests will receive a media switch notification and should navigate.
void setCurrentMedia({required String ratingKey, required String serverId, required String mediaTitle}) {
if (!isHost || _session == null || _peerService == null) {
appLogger.w('WatchTogether: Cannot set media - not host or not in session');
return;
}
appLogger.d('WatchTogether: Host setting current media: $mediaTitle (ratingKey: $ratingKey)');
// Update session with new media info
_session = _session!.copyWith(mediaRatingKey: ratingKey, mediaServerId: serverId, mediaTitle: mediaTitle);
// Broadcast media switch to all guests
_peerService!.broadcast(
SyncMessage.mediaSwitch(
ratingKey: ratingKey,
serverId: serverId,
mediaTitle: mediaTitle,
peerId: _peerService!.myPeerId,
),
);
notifyListeners();
}
/// Handle media switch message from host (guest only)
void _handleMediaSwitch(SyncMessage message) {
if (isHost) return; // Host doesn't need to handle their own switch
if (message.ratingKey == null || message.serverId == null || message.mediaTitle == null) {
appLogger.w('WatchTogether: Received incomplete media switch message');
return;
}
final playbackKey = _buildPlaybackKey(message.ratingKey, message.serverId);
final shouldDispatch = playbackKey != _lastHandledCurrentPlaybackKey;
_updateCurrentPlaybackSnapshot(
ratingKey: message.ratingKey!,
serverId: message.serverId!,
mediaTitle: message.mediaTitle!,
);
notifyListeners();
if (!shouldDispatch) {
appLogger.d('WatchTogether: Ignoring duplicate media switch for ${message.ratingKey}');
return;
}
appLogger.d('WatchTogether: Received media switch: ${message.mediaTitle}');
_dispatchCurrentPlayback(
ratingKey: message.ratingKey!,
serverId: message.serverId!,
mediaTitle: message.mediaTitle!,
source: 'media switch',
);
}
/// Notify guests that host is exiting the video player
///
/// Call this from video player dispose when host exits.
void notifyHostExitedPlayer() {
if (!isHost || _session == null || _peerService == null) {
return;
}
appLogger.d('WatchTogether: Host exiting player, notifying guests');
_peerService!.broadcast(SyncMessage.hostExitedPlayer(peerId: _peerService!.myPeerId));
}
/// Handle host exited player message (guest only)
void _handleHostExitedPlayer(SyncMessage _) {
if (isHost) return; // Host doesn't need to handle their own exit
appLogger.d('WatchTogether: Host exited player, callback set: ${onHostExitedPlayer != null}');
_clearCurrentPlaybackSnapshot();
// Clear the player callback BEFORE popping so that any mediaSwitch message
// arriving during the pop animation routes to MainScreen's handler instead
// of the dying VideoPlayerScreen.
onPlayerMediaSwitched = null;
notifyListeners();
// Trigger callback for the app to navigate guest out of player
if (onHostExitedPlayer != null) {
onHostExitedPlayer!.call();
} else {
appLogger.w('WatchTogether: onHostExitedPlayer callback not set!');
}
}
/// Start a grace period for host reconnection (guest only)
void _startHostReconnectGracePeriod() {
_cancelHostReconnectGracePeriod();
_isWaitingForHostReconnect = true;
appLogger.d('WatchTogether: Host disconnected, waiting 15s for reconnection');
notifyListeners();
_hostReconnectTimer = Timer(const Duration(seconds: 15), () {
if (_isWaitingForHostReconnect) {
appLogger.d('WatchTogether: Host reconnect grace period expired');
_isWaitingForHostReconnect = false;
_session = _session?.copyWith(state: SessionState.error, errorMessage: 'Host left the session');
onHostExitedPlayer?.call();
notifyListeners();
}
});
}
/// Cancel host reconnect grace period
void _cancelHostReconnectGracePeriod() {
_hostReconnectTimer?.cancel();
_hostReconnectTimer = null;
if (_isWaitingForHostReconnect) {
_isWaitingForHostReconnect = false;
appLogger.d('WatchTogether: Host reconnected, grace period cancelled');
notifyListeners();
}
}
@override
void dispose() {
_disposed = true;
_cancelHostReconnectGracePeriod();
_participantEventController.close();
leaveSession();
super.dispose();
}
}
/// Type of participant event
enum ParticipantEventType { joined, left, paused, resumed, seeked, buffering }
/// Event emitted when a participant joins or leaves
class ParticipantEvent {
final String displayName;
final ParticipantEventType type;
const ParticipantEvent({required this.displayName, required this.type});
}