yorickr/plezy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: watch together bugs and server write serialization

Browse Source
This commit is contained in:
edde746
2026-04-16 14:00:24 +02:00
parent 3cf1ebc09b
commit 1d3ae3e1c5
8 changed files with 193 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/screens/video_player_screen.dart
+2
View File
@@ -3597,6 +3597,8 @@ class VideoPlayerScreenState extends State<VideoPlayerScreen> with WidgetsBindin
);
},
),
// Watch Together: session indicator and menu
WatchTogetherOverlay(onLeaveSession: _handleBackButton),
// Watch Together: participant join/leave/buffering notifications
const ParticipantNotificationOverlay(),
// Watch Together: waiting for participants to load
lib/services/base_peer_service.dart
+4 -1
View File
@@ -22,7 +22,10 @@ class PeerError {
final String message;
final dynamic originalError;
const PeerError({required this.type, required this.message, this.originalError});
/// Server-provided error code (e.g. 'room_not_found', 'room_full').
final String? serverCode;
const PeerError({required this.type, required this.message, this.originalError, this.serverCode});
@override
String toString() => 'PeerError($type): $message';
lib/watch_together/models/sync_message.dart
+19
View File
@@ -280,6 +280,25 @@ class SyncMessage {
/// Position as Duration (convenience getter)
Duration? get position => positionMs != null ? Duration(milliseconds: positionMs!) : null;
SyncMessage copyWith({String? peerId}) {
return SyncMessage(
type: type,
timestamp: timestamp,
positionMs: positionMs,
bufferingState: bufferingState,
rate: rate,
peerId: peerId ?? this.peerId,
displayName: displayName,
isHost: isHost,
controlMode: controlMode,
pingId: pingId,
ratingKey: ratingKey,
serverId: serverId,
mediaTitle: mediaTitle,
isPlaying: isPlaying,
);
}
/// Serialize to JSON string for sending over data channel
String toJson() {
final map = <String, dynamic>{'t': type.name, 'ts': timestamp};
lib/watch_together/models/watch_session.dart
-11
View File
@@ -79,9 +79,6 @@ class WatchSession {
/// Current connection state
final SessionState state;
/// List of participants in the session
final List<Participant> participants;
/// Error message if state is error
final String? errorMessage;
@@ -102,7 +99,6 @@ class WatchSession {
required this.role,
required this.controlMode,
required this.state,
this.participants = const [],
this.errorMessage,
this.mediaRatingKey,
this.mediaServerId,
@@ -116,15 +112,11 @@ class WatchSession {
/// Whether the session is currently connected
bool get isConnected => state == SessionState.connected;
/// Number of participants (including self)
int get participantCount => participants.length;
WatchSession copyWith({
String? sessionId,
SessionRole? role,
ControlMode? controlMode,
SessionState? state,
List<Participant>? participants,
String? errorMessage,
String? mediaRatingKey,
String? mediaServerId,
@@ -136,7 +128,6 @@ class WatchSession {
role: role ?? this.role,
controlMode: controlMode ?? this.controlMode,
state: state ?? this.state,
participants: participants ?? this.participants,
errorMessage: errorMessage ?? this.errorMessage,
mediaRatingKey: mediaRatingKey ?? this.mediaRatingKey,
mediaServerId: mediaServerId ?? this.mediaServerId,
@@ -163,7 +154,6 @@ class WatchSession {
mediaRatingKey: mediaRatingKey,
mediaServerId: mediaServerId,
mediaTitle: mediaTitle,
participants: [],
);
}
@@ -174,7 +164,6 @@ class WatchSession {
role: SessionRole.guest,
controlMode: ControlMode.hostOnly, // Will be updated when connected
state: SessionState.connecting,
participants: [],
);
}
}
lib/watch_together/providers/watch_together_provider.dart
+41 -30
View File
@@ -36,20 +36,22 @@ class WatchTogetherProvider with ChangeNotifier {
// 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 (_notifyScheduled) return;
if (_disposed || _notifyScheduled) return;
_notifyScheduled = true;
scheduleMicrotask(() {
_notifyScheduled = false;
super.notifyListeners();
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 = {};
@@ -131,7 +133,6 @@ class WatchTogetherProvider with ChangeNotifier {
role: session.role,
controlMode: session.controlMode,
state: session.state,
participants: session.participants,
errorMessage: session.errorMessage,
hostPeerId: session.hostPeerId,
);
@@ -174,6 +175,13 @@ class WatchTogetherProvider with ChangeNotifier {
}
}
/// Wire up reconnection handler to re-announce join after reconnect
void _wireReconnectHandler() {
_peerService!.onReconnected = () {
_syncManager?.announceJoin(_displayName);
};
}
/// Wire up sync manager's state change callback to update provider state
void _wireSyncStateChanges() {
_syncManager!.onSyncStateChanged = (isSyncing) {
@@ -227,6 +235,7 @@ class WatchTogetherProvider with ChangeNotifier {
);
_wireSyncStateChanges();
_wireReconnectHandler();
notifyListeners();
appLogger.d('WatchTogether: Session created: $createdSessionId');
@@ -276,6 +285,7 @@ class WatchTogetherProvider with ChangeNotifier {
};
_wireSyncStateChanges();
_wireReconnectHandler();
// Add self to participants
_participants.add(Participant(peerId: _peerService!.myPeerId!, displayName: _displayName, isHost: false));
@@ -298,25 +308,15 @@ class WatchTogetherProvider with ChangeNotifier {
///
/// Returns `true` if the user became the host.
Future<bool> enterRoom(String sessionId, {ControlMode controlMode = ControlMode.anyone, String? displayName}) async {
// Probe the relay to determine if the room exists, then delegate to
// the existing createSession / joinSession which handle all setup.
final customRelayUrl = SettingsService.instanceOrNull?.getCustomRelayUrl();
final probe = WatchTogetherPeerService(customBaseUrl: customRelayUrl);
try {
final becameHost = await probe.joinOrCreateSession(sessionId);
final shouldBeHost = becameHost || probe.connectedPeers.isEmpty;
await probe.disconnect();
probe.dispose();
if (shouldBeHost) {
await joinSession(sessionId, displayName: displayName);
return false; // joined as guest
} on PeerError catch (e) {
// Room doesn't exist — create it as host
if (e.serverCode == 'room_not_found') {
await createSession(controlMode: controlMode, displayName: displayName, sessionId: sessionId);
} else {
await joinSession(sessionId, displayName: displayName);
return true;
}
return shouldBeHost;
} catch (e) {
await probe.disconnect();
probe.dispose();
rethrow;
}
}
@@ -358,6 +358,7 @@ class WatchTogetherProvider with ChangeNotifier {
_isDeferredPlay = false;
_lastHandledCurrentPlaybackKey = null;
_lastActionEventMs.clear();
_hostIntentionallyLeft = false;
notifyListeners();
appLogger.d('WatchTogether: Session left');
@@ -410,8 +411,9 @@ class WatchTogetherProvider with ChangeNotifier {
_participants.removeWhere((p) => p.peerId == peerId);
// If host disconnected, start grace period for reconnection
if (!isHost && peerId == _session?.hostPeerId) {
// 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(
@@ -459,16 +461,16 @@ class WatchTogetherProvider with ChangeNotifier {
_participantEventController.add(
ParticipantEvent(displayName: message.displayName!, type: ParticipantEventType.joined),
);
}
// Send our join info back so the new peer adds us to their
// participant list. Every peer does this (not just the host)
// so that late joiners learn about all existing participants.
if (_peerService != null) {
_peerService!.sendTo(
message.peerId!,
SyncMessage.join(peerId: _peerService!.myPeerId!, displayName: _displayName, isHost: isHost),
);
// 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();
@@ -487,6 +489,14 @@ class WatchTogetherProvider with ChangeNotifier {
ParticipantEvent(displayName: leavingName, type: ParticipantEventType.left),
);
}
// If the host deliberately left, end the session immediately
// instead of waiting for the 15s reconnect grace period.
if (!isHost && message.peerId == _session?.hostPeerId) {
_hostIntentionallyLeft = true;
_handleHostExitedPlayer(message);
}
notifyListeners();
}
break;
@@ -739,6 +749,7 @@ class WatchTogetherProvider with ChangeNotifier {
@override
void dispose() {
_disposed = true;
_cancelHostReconnectGracePeriod();
_participantEventController.close();
leaveSession();
lib/watch_together/services/watch_together_peer_service.dart
+29 -22
View File
@@ -59,6 +59,9 @@ class WatchTogetherPeerService with KeepaliveMixin {
static const int _maxReconnectAttempts = 3;
Timer? _reconnectTimer;
/// Called after a successful reconnection so the provider can re-announce join.
void Function()? onReconnected;
// Keepalive (via KeepaliveMixin)
@override
Duration get pingInterval => const Duration(seconds: 15);
@@ -192,10 +195,16 @@ class WatchTogetherPeerService with KeepaliveMixin {
case 'message':
final payload = msg['payload'];
final serverFrom = msg['from'] as String?;
if (payload != null) {
try {
final payloadStr = payload is String ? payload : jsonEncode(payload);
final syncMsg = SyncMessage.fromJson(payloadStr);
var syncMsg = SyncMessage.fromJson(payloadStr);
// Use the server-authenticated sender ID instead of the
// self-reported peerId in the payload to prevent spoofing.
if (serverFrom != null && syncMsg.peerId != serverFrom) {
syncMsg = syncMsg.copyWith(peerId: serverFrom);
}
_safeAdd(_messageReceivedController, syncMsg);
} catch (e) {
appLogger.e('WatchTogether: Failed to parse sync message payload', error: e);
@@ -206,7 +215,7 @@ class WatchTogetherPeerService with KeepaliveMixin {
final code = msg['code'] as String? ?? 'unknown';
final message = msg['message'] as String? ?? 'Unknown error';
appLogger.e('WatchTogether: Server error: $code - $message');
final error = PeerError(type: PeerErrorType.serverError, message: '$code: $message');
final error = PeerError(type: PeerErrorType.serverError, message: '$code: $message', serverCode: code);
_safeAdd(_errorController, error);
if (setupCompleter != null && !setupCompleter.isCompleted) {
setupCompleter.completeError(error);
@@ -292,15 +301,27 @@ class WatchTogetherPeerService with KeepaliveMixin {
_listenToChannel(channel, setupCompleter: completer);
startKeepalive();
// Re-send create or join
if (_isHost) {
_sendRaw({'type': 'create', 'sessionId': _sessionId, 'peerId': _myPeerId});
} else {
_sendRaw({'type': 'join', 'sessionId': _sessionId, 'peerId': _myPeerId});
// Always try join first — the room may still have peers (e.g. host
// reconnecting while guests remain). Fall back to create only if
// the room no longer exists and we were the host.
_sendRaw({'type': 'join', 'sessionId': _sessionId, 'peerId': _myPeerId});
try {
await completer.future.namedTimeout(const Duration(seconds: 10), operation: 'WatchTogether reconnect');
} on PeerError catch (e) {
if (_isHost && e.serverCode == 'room_not_found') {
appLogger.d('WatchTogether: Room gone, re-creating as host');
final createCompleter = Completer<void>();
_listenToChannel(channel, setupCompleter: createCompleter);
_sendRaw({'type': 'create', 'sessionId': _sessionId, 'peerId': _myPeerId});
await createCompleter.future.namedTimeout(const Duration(seconds: 10), operation: 'WatchTogether reconnect create');
} else {
rethrow;
}
}
await completer.future.namedTimeout(const Duration(seconds: 10), operation: 'WatchTogether reconnect');
appLogger.d('WatchTogether: Reconnected successfully');
onReconnected?.call();
} catch (e) {
appLogger.e('WatchTogether: Reconnect failed', error: e);
_handleWebSocketClosed();
@@ -384,20 +405,6 @@ class WatchTogetherPeerService with KeepaliveMixin {
}
}
/// Try to join a session; if it doesn't exist, create it as host.
///
/// Returns `true` if the user became the host (room was empty).
Future<bool> joinOrCreateSession(String sessionId) async {
try {
await joinSession(sessionId);
return false; // joined as guest
} on PeerError catch (e) {
if (e.type != PeerErrorType.serverError) rethrow;
await createSession(sessionId: sessionId);
return true; // created as host
}
}
/// Broadcast a message to all connected peers
void broadcast(SyncMessage message) {
final payload = message.toJson();
lib/watch_together/services/watch_together_sync_manager.dart
+6 -17
View File
@@ -228,7 +228,7 @@ class WatchTogetherSyncManager {
} finally {
_isRemoteAction = false;
}
_broadcastPlayPause(true);
// Don't broadcast play — deferred play will broadcast when all peers are ready.
return;
}
@@ -539,20 +539,6 @@ class WatchTogetherSyncManager {
return;
}
// HOST RELAY: In "anyone" mode, host rebroadcasts control commands from guests
// This is needed because guests only connect to host (star topology), not to each other
if (_session.isHost && _session.controlMode == ControlMode.anyone) {
final isControlMessage =
message.type == SyncMessageType.play ||
message.type == SyncMessageType.pause ||
message.type == SyncMessageType.seek ||
message.type == SyncMessageType.rate;
if (isControlMessage) {
_peerService.broadcast(message);
}
}
// In hostOnly mode, only process messages from host (unless it's join/leave/sessionConfig)
if (_session.controlMode == ControlMode.hostOnly && !_session.isHost) {
final isHostMessage = message.peerId == _session.hostPeerId;
@@ -669,11 +655,14 @@ class WatchTogetherSyncManager {
if (_deferredPlay && isAllReady) {
_setDeferredPlay(false);
_firstPlayCompleted = true;
final pos = _deferredPlayPosition;
_deferredPlayPosition = null;
await _applyRemotePlay(
position: _deferredPlayPosition,
position: pos,
expectedAttachmentGeneration: queuedAttachmentGeneration,
);
_deferredPlayPosition = null;
// Broadcast play to all peers now that everyone is ready
_broadcastPlayPause(true);
}
}
break;
server/main.go
+92 -51
View File
@@ -197,12 +197,62 @@ type serverMsg struct {
Payload json.RawMessage `json:"payload,omitempty"`
}
// --- Client (serializes writes to a single goroutine) ---
type Client struct {
conn *websocket.Conn
send chan []byte
}
func newClient(conn *websocket.Conn) *Client {
c := &Client{conn: conn, send: make(chan []byte, 64)}
go c.writePump()
return c
}
func (c *Client) writePump() {
ticker := time.NewTicker(pingInterval)
defer ticker.Stop()
for {
select {
case data, ok := <-c.send:
if !ok {
c.conn.WriteMessage(websocket.CloseMessage, nil)
return
}
c.conn.SetWriteDeadline(time.Now().Add(writeWait))
c.conn.WriteMessage(websocket.TextMessage, data)
case <-ticker.C:
c.conn.SetWriteDeadline(time.Now().Add(writeWait))
if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
return
}
}
}
}
func (c *Client) sendJSON(msg serverMsg) {
data, err := json.Marshal(msg)
if err != nil {
return
}
select {
case c.send <- data:
default:
// Drop if buffer full (slow client)
}
}
func (c *Client) close() {
close(c.send)
}
// --- Room ---
type Room struct {
SessionID string
HostPeerID string
Peers map[string]*websocket.Conn
Peers map[string]*Client
mu sync.RWMutex
CreatedAt time.Time
}
@@ -220,12 +270,20 @@ func (r *Room) broadcastExcept(senderID string, msg serverMsg) {
if err != nil {
return
}
// Copy peers under lock, then send without holding it
r.mu.RLock()
defer r.mu.RUnlock()
for id, conn := range r.Peers {
targets := make([]*Client, 0, len(r.Peers))
for id, client := range r.Peers {
if id != senderID {
conn.SetWriteDeadline(time.Now().Add(writeWait))
conn.WriteMessage(websocket.TextMessage, data)
targets = append(targets, client)
}
}
r.mu.RUnlock()
for _, client := range targets {
select {
case client.send <- data:
default:
}
}
}
@@ -236,14 +294,17 @@ func (r *Room) sendTo(targetID string, msg serverMsg) bool {
return false
}
r.mu.RLock()
defer r.mu.RUnlock()
conn, ok := r.Peers[targetID]
client, ok := r.Peers[targetID]
r.mu.RUnlock()
if !ok {
return false
}
conn.SetWriteDeadline(time.Now().Add(writeWait))
conn.WriteMessage(websocket.TextMessage, data)
return true
select {
case client.send <- data:
return true
default:
return false
}
}
// --- Log store ---
@@ -463,18 +524,6 @@ func (s *Server) handleGetLogs(w http.ResponseWriter, r *http.Request) {
w.Write(data)
}
func (s *Server) sendError(conn *websocket.Conn, code, message string) {
data, _ := json.Marshal(serverMsg{Type: "error", Code: code, Message: message})
conn.SetWriteDeadline(time.Now().Add(writeWait))
conn.WriteMessage(websocket.TextMessage, data)
}
func (s *Server) sendJSON(conn *websocket.Conn, msg serverMsg) {
data, _ := json.Marshal(msg)
conn.SetWriteDeadline(time.Now().Add(writeWait))
conn.WriteMessage(websocket.TextMessage, data)
}
func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {
ip := clientIP(r)
@@ -498,17 +547,9 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {
return nil
})
// Ping ticker
ticker := time.NewTicker(pingInterval)
defer ticker.Stop()
go func() {
for range ticker.C {
conn.SetWriteDeadline(time.Now().Add(writeWait))
if err := conn.WriteMessage(websocket.PingMessage, nil); err != nil {
return
}
}
}()
// Client wraps the conn with a serialized write channel + ping ticker
client := newClient(conn)
defer client.close()
rl := newRateLimiter(rateBurst, rateSustained)
var currentRoom *Room
@@ -542,24 +583,24 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {
}
if !rl.allow() {
s.sendError(conn, "rate_limited", "Too many messages")
client.sendJSON(serverMsg{Type: "error", Code: "rate_limited", Message: "Too many messages"})
continue
}
var msg clientMsg
if err := json.Unmarshal(raw, &msg); err != nil {
s.sendError(conn, "invalid_message", "Invalid JSON")
client.sendJSON(serverMsg{Type: "error", Code: "invalid_message", Message: "Invalid JSON"})
continue
}
switch msg.Type {
case "create":
if msg.SessionID == "" || msg.PeerID == "" {
s.sendError(conn, "invalid_message", "sessionId and peerId required")
client.sendJSON(serverMsg{Type: "error", Code: "invalid_message", Message: "sessionId and peerId required"})
continue
}
if !s.conns.tryCreateRoom(ip) {
s.sendError(conn, "rate_limited", "Too many rooms created")
client.sendJSON(serverMsg{Type: "error", Code: "rate_limited", Message: "Too many rooms created"})
continue
}
s.mu.Lock()
@@ -570,7 +611,7 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {
if !empty {
s.mu.Unlock()
s.conns.releaseRoom(ip)
s.sendError(conn, "room_exists", "Room already exists")
client.sendJSON(serverMsg{Type: "error", Code: "room_exists", Message: "Room already exists"})
continue
}
// Empty stale room — reclaim the ID
@@ -579,7 +620,7 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {
room := &Room{
SessionID: msg.SessionID,
HostPeerID: msg.PeerID,
Peers: map[string]*websocket.Conn{msg.PeerID: conn},
Peers: map[string]*Client{msg.PeerID: client},
CreatedAt: time.Now(),
}
s.rooms[msg.SessionID] = room
@@ -588,27 +629,27 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {
currentPeerID = msg.PeerID
isHost = true
log.Printf("room %s created by %s", msg.SessionID, msg.PeerID)
s.sendJSON(conn, serverMsg{Type: "created", SessionID: msg.SessionID})
client.sendJSON(serverMsg{Type: "created", SessionID: msg.SessionID})
case "join":
if msg.SessionID == "" || msg.PeerID == "" {
s.sendError(conn, "invalid_message", "sessionId and peerId required")
client.sendJSON(serverMsg{Type: "error", Code: "invalid_message", Message: "sessionId and peerId required"})
continue
}
s.mu.RLock()
room, exists := s.rooms[msg.SessionID]
s.mu.RUnlock()
if !exists {
s.sendError(conn, "room_not_found", "Room does not exist")
client.sendJSON(serverMsg{Type: "error", Code: "room_not_found", Message: "Room does not exist"})
continue
}
room.mu.Lock()
if len(room.Peers) >= maxRoomSize {
room.mu.Unlock()
s.sendError(conn, "room_full", "Room is full")
client.sendJSON(serverMsg{Type: "error", Code: "room_full", Message: "Room is full"})
continue
}
room.Peers[msg.PeerID] = conn
room.Peers[msg.PeerID] = client
peers := room.peerIDs()
room.mu.Unlock()
currentRoom = room
@@ -622,12 +663,12 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {
existingPeers = append(existingPeers, p)
}
}
s.sendJSON(conn, serverMsg{Type: "joined", SessionID: msg.SessionID, Peers: existingPeers})
client.sendJSON(serverMsg{Type: "joined", SessionID: msg.SessionID, Peers: existingPeers})
room.broadcastExcept(msg.PeerID, serverMsg{Type: "peerJoined", PeerID: msg.PeerID})
case "broadcast":
if currentRoom == nil {
s.sendError(conn, "not_in_room", "Not in a room")
client.sendJSON(serverMsg{Type: "error", Code: "not_in_room", Message: "Not in a room"})
continue
}
currentRoom.broadcastExcept(currentPeerID, serverMsg{
@@ -638,11 +679,11 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {
case "sendTo":
if currentRoom == nil {
s.sendError(conn, "not_in_room", "Not in a room")
client.sendJSON(serverMsg{Type: "error", Code: "not_in_room", Message: "Not in a room"})
continue
}
if msg.To == "" {
s.sendError(conn, "invalid_message", "to field required")
client.sendJSON(serverMsg{Type: "error", Code: "invalid_message", Message: "to field required"})
continue
}
if !currentRoom.sendTo(msg.To, serverMsg{
@@ -650,14 +691,14 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {
From: currentPeerID,
Payload: msg.Payload,
}) {
s.sendError(conn, "not_in_room", "Target peer not found")
client.sendJSON(serverMsg{Type: "error", Code: "not_in_room", Message: "Target peer not found"})
}
case "ping":
s.sendJSON(conn, serverMsg{Type: "pong"})
client.sendJSON(serverMsg{Type: "pong"})
default:
s.sendError(conn, "invalid_message", "Unknown message type")
client.sendJSON(serverMsg{Type: "error", Code: "invalid_message", Message: "Unknown message type"})
}
}
}