diff --git a/lib/i18n/az.i18n.json b/lib/i18n/az.i18n.json index 68488d591..29c16d732 100644 --- a/lib/i18n/az.i18n.json +++ b/lib/i18n/az.i18n.json @@ -1587,6 +1587,13 @@ "host": "Təşkilatçı", "hostBadge": "TƏŞKİLATÇI", "youAreHost": "Təşkilatçı sizsiniz", + "makeHost": "", + "makeHostQuestion": "", + "makeHostConfirm": "", + "transfer": "", + "hostChangedTo": "", + "youAreNowHost": "", + "hostTransferFailed": "", "watchingWithOthers": "Başqaları ilə izlənilir", "endSession": "Seansı bitir", "leaveSession": "Seansdan çıx", diff --git a/lib/i18n/bg.i18n.json b/lib/i18n/bg.i18n.json index bfb9080e6..03ca1b4e7 100644 --- a/lib/i18n/bg.i18n.json +++ b/lib/i18n/bg.i18n.json @@ -1587,6 +1587,13 @@ "host": "Организатор", "hostBadge": "ОРГАНИЗАТОР", "youAreHost": "Вие сте организаторът", + "makeHost": "", + "makeHostQuestion": "", + "makeHostConfirm": "", + "transfer": "", + "hostChangedTo": "", + "youAreNowHost": "", + "hostTransferFailed": "", "watchingWithOthers": "Гледате с други", "endSession": "Край на сесията", "leaveSession": "Напусни сесията", diff --git a/lib/i18n/da.i18n.json b/lib/i18n/da.i18n.json index 11c489597..b36da5ae0 100644 --- a/lib/i18n/da.i18n.json +++ b/lib/i18n/da.i18n.json @@ -1587,6 +1587,13 @@ "host": "Vært", "hostBadge": "VÆRT", "youAreHost": "Du er vært", + "makeHost": "", + "makeHostQuestion": "", + "makeHostConfirm": "", + "transfer": "", + "hostChangedTo": "", + "youAreNowHost": "", + "hostTransferFailed": "", "watchingWithOthers": "Ser med andre", "endSession": "Afslut session", "leaveSession": "Forlad session", diff --git a/lib/i18n/de.i18n.json b/lib/i18n/de.i18n.json index 33d63eea6..5277459b7 100644 --- a/lib/i18n/de.i18n.json +++ b/lib/i18n/de.i18n.json @@ -1587,6 +1587,13 @@ "host": "Host", "hostBadge": "HOST", "youAreHost": "Du bist der Host", + "makeHost": "", + "makeHostQuestion": "", + "makeHostConfirm": "", + "transfer": "", + "hostChangedTo": "", + "youAreNowHost": "", + "hostTransferFailed": "", "watchingWithOthers": "Mit anderen schauen", "endSession": "Sitzung beenden", "leaveSession": "Sitzung verlassen", diff --git a/lib/i18n/en.i18n.json b/lib/i18n/en.i18n.json index 49e44f2fe..95b33fc9e 100644 --- a/lib/i18n/en.i18n.json +++ b/lib/i18n/en.i18n.json @@ -1587,6 +1587,13 @@ "host": "Host", "hostBadge": "HOST", "youAreHost": "You are the host", + "makeHost": "Make host", + "makeHostQuestion": "Transfer host?", + "makeHostConfirm": "${name} will control playback and drive the session for everyone.", + "transfer": "Transfer", + "hostChangedTo": "${name} is now the host", + "youAreNowHost": "You are now the host", + "hostTransferFailed": "Couldn't make ${name} the host", "watchingWithOthers": "Watching with others", "endSession": "End Session", "leaveSession": "Leave Session", diff --git a/lib/i18n/es.i18n.json b/lib/i18n/es.i18n.json index 38cee003a..159faad5e 100644 --- a/lib/i18n/es.i18n.json +++ b/lib/i18n/es.i18n.json @@ -1587,6 +1587,13 @@ "host": "Anfitrión", "hostBadge": "ANFITRIÓN", "youAreHost": "Eres el anfitrión", + "makeHost": "", + "makeHostQuestion": "", + "makeHostConfirm": "", + "transfer": "", + "hostChangedTo": "", + "youAreNowHost": "", + "hostTransferFailed": "", "watchingWithOthers": "Viendo contenido con otras personas", "endSession": "Finalizar sesión", "leaveSession": "Salir de la sesión", diff --git a/lib/i18n/fr.i18n.json b/lib/i18n/fr.i18n.json index 8bdf68481..933225ce9 100644 --- a/lib/i18n/fr.i18n.json +++ b/lib/i18n/fr.i18n.json @@ -1587,6 +1587,13 @@ "host": "Hôte", "hostBadge": "HÔTE", "youAreHost": "Vous êtes l'hôte", + "makeHost": "", + "makeHostQuestion": "", + "makeHostConfirm": "", + "transfer": "", + "hostChangedTo": "", + "youAreNowHost": "", + "hostTransferFailed": "", "watchingWithOthers": "Regarder avec d'autres personnes", "endSession": "Terminer la session", "leaveSession": "Quitter la session", diff --git a/lib/i18n/hu.i18n.json b/lib/i18n/hu.i18n.json index 149daff17..a8caccb65 100644 --- a/lib/i18n/hu.i18n.json +++ b/lib/i18n/hu.i18n.json @@ -1587,6 +1587,13 @@ "host": "Házigazda", "hostBadge": "HÁZIGAZDA", "youAreHost": "Te vagy a házigazda", + "makeHost": "", + "makeHostQuestion": "", + "makeHostConfirm": "", + "transfer": "", + "hostChangedTo": "", + "youAreNowHost": "", + "hostTransferFailed": "", "watchingWithOthers": "Nézés másokkal", "endSession": "Munkamenet befejezése", "leaveSession": "Munkamenet elhagyása", diff --git a/lib/i18n/it.i18n.json b/lib/i18n/it.i18n.json index 3887db920..6d409f377 100644 --- a/lib/i18n/it.i18n.json +++ b/lib/i18n/it.i18n.json @@ -1587,6 +1587,13 @@ "host": "Host", "hostBadge": "HOST", "youAreHost": "Sei l'host", + "makeHost": "", + "makeHostQuestion": "", + "makeHostConfirm": "", + "transfer": "", + "hostChangedTo": "", + "youAreNowHost": "", + "hostTransferFailed": "", "watchingWithOthers": "In visione con altri partecipanti", "endSession": "Termina la sessione", "leaveSession": "Abbandona la sessione", diff --git a/lib/i18n/ja.i18n.json b/lib/i18n/ja.i18n.json index 783913f9a..f04c6c3bd 100644 --- a/lib/i18n/ja.i18n.json +++ b/lib/i18n/ja.i18n.json @@ -1577,6 +1577,13 @@ "host": "ホスト", "hostBadge": "ホスト", "youAreHost": "あなたはホストです", + "makeHost": "", + "makeHostQuestion": "", + "makeHostConfirm": "", + "transfer": "", + "hostChangedTo": "", + "youAreNowHost": "", + "hostTransferFailed": "", "watchingWithOthers": "他の人と視聴中", "endSession": "セッションを終了", "leaveSession": "セッションを退出", diff --git a/lib/i18n/kk.i18n.json b/lib/i18n/kk.i18n.json index b90430010..14d50795a 100644 --- a/lib/i18n/kk.i18n.json +++ b/lib/i18n/kk.i18n.json @@ -1587,6 +1587,13 @@ "host": "Ұйымдастырушы", "hostBadge": "ҰЙЫМДАСТЫРУШЫ", "youAreHost": "Сіз ұйымдастырушысыз", + "makeHost": "", + "makeHostQuestion": "", + "makeHostConfirm": "", + "transfer": "", + "hostChangedTo": "", + "youAreNowHost": "", + "hostTransferFailed": "", "watchingWithOthers": "Басқалармен бірге көрілуде", "endSession": "Сеансты аяқтау", "leaveSession": "Сеанстан шығу", diff --git a/lib/i18n/ko.i18n.json b/lib/i18n/ko.i18n.json index fb29c5aae..9fb30ae76 100644 --- a/lib/i18n/ko.i18n.json +++ b/lib/i18n/ko.i18n.json @@ -1577,6 +1577,13 @@ "host": "호스트", "hostBadge": "호스트", "youAreHost": "호스트입니다", + "makeHost": "", + "makeHostQuestion": "", + "makeHostConfirm": "", + "transfer": "", + "hostChangedTo": "", + "youAreNowHost": "", + "hostTransferFailed": "", "watchingWithOthers": "다른 사람과 함께 시청 중", "endSession": "세션 종료", "leaveSession": "세션 나가기", diff --git a/lib/i18n/nb.i18n.json b/lib/i18n/nb.i18n.json index c0613c170..ce3fb2aef 100644 --- a/lib/i18n/nb.i18n.json +++ b/lib/i18n/nb.i18n.json @@ -1587,6 +1587,13 @@ "host": "Vert", "hostBadge": "VERT", "youAreHost": "Du er verten", + "makeHost": "", + "makeHostQuestion": "", + "makeHostConfirm": "", + "transfer": "", + "hostChangedTo": "", + "youAreNowHost": "", + "hostTransferFailed": "", "watchingWithOthers": "Ser med andre", "endSession": "Avslutt økt", "leaveSession": "Forlat økt", diff --git a/lib/i18n/nl.i18n.json b/lib/i18n/nl.i18n.json index c4ffd2b34..3bc99a443 100644 --- a/lib/i18n/nl.i18n.json +++ b/lib/i18n/nl.i18n.json @@ -1587,6 +1587,13 @@ "host": "Host", "hostBadge": "HOST", "youAreHost": "Jij bent de host", + "makeHost": "", + "makeHostQuestion": "", + "makeHostConfirm": "", + "transfer": "", + "hostChangedTo": "", + "youAreNowHost": "", + "hostTransferFailed": "", "watchingWithOthers": "Kijken met anderen", "endSession": "Sessie beëindigen", "leaveSession": "Sessie verlaten", diff --git a/lib/i18n/pl.i18n.json b/lib/i18n/pl.i18n.json index 88f72d5aa..d4a2ea486 100644 --- a/lib/i18n/pl.i18n.json +++ b/lib/i18n/pl.i18n.json @@ -1607,6 +1607,13 @@ "host": "Gospodarz", "hostBadge": "GOSPODARZ", "youAreHost": "Jesteś gospodarzem", + "makeHost": "", + "makeHostQuestion": "", + "makeHostConfirm": "", + "transfer": "", + "hostChangedTo": "", + "youAreNowHost": "", + "hostTransferFailed": "", "watchingWithOthers": "Oglądasz z innymi", "endSession": "Zakończ sesję", "leaveSession": "Opuść sesję", diff --git a/lib/i18n/pt.i18n.json b/lib/i18n/pt.i18n.json index 4d7d243d7..da2d6c84e 100644 --- a/lib/i18n/pt.i18n.json +++ b/lib/i18n/pt.i18n.json @@ -1587,6 +1587,13 @@ "host": "Anfitrião", "hostBadge": "ANFITRIÃO", "youAreHost": "Você é o anfitrião", + "makeHost": "", + "makeHostQuestion": "", + "makeHostConfirm": "", + "transfer": "", + "hostChangedTo": "", + "youAreNowHost": "", + "hostTransferFailed": "", "watchingWithOthers": "Assistindo com outras pessoas", "endSession": "Encerrar sessão", "leaveSession": "Sair da sessão", diff --git a/lib/i18n/ru.i18n.json b/lib/i18n/ru.i18n.json index d75d13954..cd4ab82bd 100644 --- a/lib/i18n/ru.i18n.json +++ b/lib/i18n/ru.i18n.json @@ -1607,6 +1607,13 @@ "host": "Организатор", "hostBadge": "ОРГАНИЗАТОР", "youAreHost": "Вы — организатор", + "makeHost": "", + "makeHostQuestion": "", + "makeHostConfirm": "", + "transfer": "", + "hostChangedTo": "", + "youAreNowHost": "", + "hostTransferFailed": "", "watchingWithOthers": "Совместный просмотр", "endSession": "Завершить сессию", "leaveSession": "Покинуть сессию", diff --git a/lib/i18n/strings.g.dart b/lib/i18n/strings.g.dart index 58220153b..506b64a3e 100644 --- a/lib/i18n/strings.g.dart +++ b/lib/i18n/strings.g.dart @@ -4,7 +4,7 @@ /// To regenerate, run: `dart run slang` /// /// Locales: 22 -/// Strings: 43738 (1988 per locale) +/// Strings: 43745 (1988 per locale) // coverage:ignore-file // ignore_for_file: type=lint, unused_import diff --git a/lib/i18n/strings_en.g.dart b/lib/i18n/strings_en.g.dart index dfa4dd41e..5c2b1a3a7 100644 --- a/lib/i18n/strings_en.g.dart +++ b/lib/i18n/strings_en.g.dart @@ -4022,6 +4022,27 @@ class Translations$watchTogether$en { /// en: 'You are the host' String get youAreHost => 'You are the host'; + /// en: 'Make host' + String get makeHost => 'Make host'; + + /// en: 'Transfer host?' + String get makeHostQuestion => 'Transfer host?'; + + /// en: '${name} will control playback and drive the session for everyone.' + String makeHostConfirm({required Object name}) => '${name} will control playback and drive the session for everyone.'; + + /// en: 'Transfer' + String get transfer => 'Transfer'; + + /// en: '${name} is now the host' + String hostChangedTo({required Object name}) => '${name} is now the host'; + + /// en: 'You are now the host' + String get youAreNowHost => 'You are now the host'; + + /// en: 'Couldn't make ${name} the host' + String hostTransferFailed({required Object name}) => 'Couldn\'t make ${name} the host'; + /// en: 'Watching with others' String get watchingWithOthers => 'Watching with others'; @@ -8354,6 +8375,13 @@ extension on Translations { 'watchTogether.host' => 'Host', 'watchTogether.hostBadge' => 'HOST', 'watchTogether.youAreHost' => 'You are the host', + 'watchTogether.makeHost' => 'Make host', + 'watchTogether.makeHostQuestion' => 'Transfer host?', + 'watchTogether.makeHostConfirm' => ({required Object name}) => '${name} will control playback and drive the session for everyone.', + 'watchTogether.transfer' => 'Transfer', + 'watchTogether.hostChangedTo' => ({required Object name}) => '${name} is now the host', + 'watchTogether.youAreNowHost' => 'You are now the host', + 'watchTogether.hostTransferFailed' => ({required Object name}) => 'Couldn\'t make ${name} the host', 'watchTogether.watchingWithOthers' => 'Watching with others', 'watchTogether.endSession' => 'End Session', 'watchTogether.leaveSession' => 'Leave Session', @@ -8453,6 +8481,8 @@ extension on Translations { 'downloads.removeSyncRuleConfirm' => ({required Object title}) => 'Stop syncing "${title}"? Downloaded episodes will be kept.', 'downloads.removeListSyncRuleConfirm' => ({required Object title}) => 'Stop syncing "${title}"?', 'downloads.deleteSyncRuleDownloads' => 'Also delete associated downloads', + _ => null, + } ?? switch (path) { 'downloads.deleteSyncRuleDownloadsDescription' => 'Downloads used by another sync rule or profile will be kept.', 'downloads.syncRuleCreated' => ({required Object count}) => 'Sync rule created — keeping ${count} unwatched episodes', 'downloads.syncRuleUpdated' => 'Sync rule updated', @@ -8460,8 +8490,6 @@ extension on Translations { 'downloads.syncRuleAndDownloadsRemoved' => 'Sync rule and associated downloads removed', 'downloads.syncRuleCleanupBusy' => 'Sync rules are currently updating. Try again in a moment.', 'downloads.syncRuleCleanupUnavailable' => 'Associated downloads could not be identified safely. Reconnect the server and try again, or remove the rule without deleting downloads.', - _ => null, - } ?? switch (path) { 'downloads.syncedNewEpisodes' => ({required Object count, required Object title}) => 'Synced ${count} new episodes for ${title}', 'downloads.activeSyncRules' => 'Sync rules', 'downloads.noSyncRules' => 'No sync rules', diff --git a/lib/i18n/sv.i18n.json b/lib/i18n/sv.i18n.json index 4866b11f2..2e1f3197e 100644 --- a/lib/i18n/sv.i18n.json +++ b/lib/i18n/sv.i18n.json @@ -1587,6 +1587,13 @@ "host": "Värd", "hostBadge": "VÄRD", "youAreHost": "Du är värden", + "makeHost": "", + "makeHostQuestion": "", + "makeHostConfirm": "", + "transfer": "", + "hostChangedTo": "", + "youAreNowHost": "", + "hostTransferFailed": "", "watchingWithOthers": "Tittar med andra", "endSession": "Avsluta session", "leaveSession": "Lämna session", diff --git a/lib/i18n/tr.i18n.json b/lib/i18n/tr.i18n.json index 26ca69a4f..0cfb34523 100644 --- a/lib/i18n/tr.i18n.json +++ b/lib/i18n/tr.i18n.json @@ -1587,6 +1587,13 @@ "host": "Kurucu", "hostBadge": "KURUCU", "youAreHost": "Kurucu sizsiniz", + "makeHost": "", + "makeHostQuestion": "", + "makeHostConfirm": "", + "transfer": "", + "hostChangedTo": "", + "youAreNowHost": "", + "hostTransferFailed": "", "watchingWithOthers": "Başkalarıyla birlikte izleniyor", "endSession": "Oturumu Bitir", "leaveSession": "Oturumdan Ayrıl", diff --git a/lib/i18n/uz.i18n.json b/lib/i18n/uz.i18n.json index af1a4bdce..c10a67b02 100644 --- a/lib/i18n/uz.i18n.json +++ b/lib/i18n/uz.i18n.json @@ -1587,6 +1587,13 @@ "host": "Tashkilotchi", "hostBadge": "TASHKILOTCHI", "youAreHost": "Siz tashkilotchisiz", + "makeHost": "", + "makeHostQuestion": "", + "makeHostConfirm": "", + "transfer": "", + "hostChangedTo": "", + "youAreNowHost": "", + "hostTransferFailed": "", "watchingWithOthers": "Boshqalar bilan tomosha qilinmoqda", "endSession": "Seansni yakunlash", "leaveSession": "Seansdan chiqish", diff --git a/lib/i18n/zh-Hant.i18n.json b/lib/i18n/zh-Hant.i18n.json index 6c3072d20..15808e62c 100644 --- a/lib/i18n/zh-Hant.i18n.json +++ b/lib/i18n/zh-Hant.i18n.json @@ -1577,6 +1577,13 @@ "host": "主持人", "hostBadge": "主持人", "youAreHost": "您是主持人", + "makeHost": "", + "makeHostQuestion": "", + "makeHostConfirm": "", + "transfer": "", + "hostChangedTo": "", + "youAreNowHost": "", + "hostTransferFailed": "", "watchingWithOthers": "與他人一起觀看", "endSession": "結束工作階段", "leaveSession": "離開工作階段", diff --git a/lib/i18n/zh.i18n.json b/lib/i18n/zh.i18n.json index 3eb78e6af..eed3bc8c5 100644 --- a/lib/i18n/zh.i18n.json +++ b/lib/i18n/zh.i18n.json @@ -1577,6 +1577,13 @@ "host": "主持人", "hostBadge": "主持人", "youAreHost": "你是主持人", + "makeHost": "", + "makeHostQuestion": "", + "makeHostConfirm": "", + "transfer": "", + "hostChangedTo": "", + "youAreNowHost": "", + "hostTransferFailed": "", "watchingWithOthers": "与他人一起观看", "endSession": "结束会话", "leaveSession": "离开会话", diff --git a/lib/watch_together/providers/watch_together_provider.dart b/lib/watch_together/providers/watch_together_provider.dart index daf107a76..4e9e06e5b 100644 --- a/lib/watch_together/providers/watch_together_provider.dart +++ b/lib/watch_together/providers/watch_together_provider.dart @@ -72,6 +72,10 @@ class WatchTogetherProvider with ChangeNotifier { bool _isWaitingForHostReconnect = false; bool _hostIntentionallyLeft = false; + // Display name of the guest a host transfer was requested for; names the + // failure toast if the relay rejects the request. + String? _pendingTransferTargetName; + // Debounce map for action events (peerId+type → last emission timestamp) final Map _lastActionEventMs = {}; @@ -100,6 +104,7 @@ class WatchTogetherProvider with ChangeNotifier { StreamSubscription? _messageSubscription; StreamSubscription? _errorSubscription; StreamSubscription? _sessionEndedSubscription; + StreamSubscription? _hostChangedSubscription; // Getters bool get isInSession => _session != null; @@ -508,11 +513,13 @@ class WatchTogetherProvider with ChangeNotifier { _observeSubscriptionCancellation(_messageSubscription?.cancel()); _observeSubscriptionCancellation(_errorSubscription?.cancel()); _observeSubscriptionCancellation(_sessionEndedSubscription?.cancel()); + _observeSubscriptionCancellation(_hostChangedSubscription?.cancel()); _peerConnectedSubscription = null; _peerDisconnectedSubscription = null; _messageSubscription = null; _errorSubscription = null; _sessionEndedSubscription = null; + _hostChangedSubscription = null; _hostReconnectTimer?.cancel(); _hostReconnectTimer = null; @@ -529,6 +536,7 @@ class WatchTogetherProvider with ChangeNotifier { _playbackPhase = null; _playbackDispatcher.reset(); _lastActionEventMs.clear(); + _pendingTransferTargetName = null; _hostIntentionallyLeft = false; if (!_disposed) notifyListeners(); @@ -689,6 +697,19 @@ class WatchTogetherProvider with ChangeNotifier { appLogger.d('WatchTogether: Declared host is not connected yet; keeping the retained-room join pending'); return; } + if (error.serverCode == RelayProtocol.notHostCode || error.serverCode == RelayProtocol.peerNotFoundCode) { + // A rejected host transfer is not a session failure — surface a toast + // and keep the room connected. + appLogger.w('WatchTogether: Host transfer rejected: ${error.message}'); + final targetName = _pendingTransferTargetName; + _pendingTransferTargetName = null; + if (targetName != null) { + _participantEventController.add( + ParticipantEvent(displayName: targetName, type: ParticipantEventType.hostTransferFailed), + ); + } + return; + } appLogger.e('WatchTogether: Peer error: ${error.message}'); // Only established transport stream errors are recoverable. Relay and @@ -713,6 +734,10 @@ class WatchTogetherProvider with ChangeNotifier { ); } }); + _hostChangedSubscription = peerService.onHostChanged.listen((newHostPeerId) { + if (_disposed || !identical(_peerService, peerService)) return; + _handleHostChanged(newHostPeerId); + }); } /// Handle incoming sync messages for participant management @@ -878,6 +903,79 @@ class WatchTogetherProvider with ChangeNotifier { notifyListeners(); } + /// Whether the current user (as host) may hand host authority to + /// [participant]: connected guest speaking the current sync protocol. + bool canTransferHostTo(Participant participant) { + final peerService = _peerService; + final controller = _controller; + if (peerService == null || controller == null) return false; + if (!isHost || !isConnected) return false; + if (participant.isHost || participant.peerId == peerService.myPeerId) return false; + if (!peerService.connectedPeers.contains(participant.peerId)) return false; + return controller.isPeerCompatible(participant.peerId); + } + + /// Ask the relay to make [participant] the host (host only). Roles flip + /// when the relay's `hostChanged` broadcast arrives; a rejection surfaces + /// as a [ParticipantEventType.hostTransferFailed] event. + void transferHost(Participant participant) { + if (!canTransferHostTo(participant)) { + appLogger.w('WatchTogether: Ignoring host transfer to ineligible peer ${participant.peerId}'); + return; + } + appLogger.d('WatchTogether: Requesting host transfer to ${participant.peerId}'); + _pendingTransferTargetName = participant.displayName; + _peerService!.transferHost(participant.peerId); + } + + /// The relay reassigned host authority ([peerService] already flipped its + /// own role state): rebuild session/participants and swap the controller's + /// role engine in place. + void _handleHostChanged(String newHostPeerId) { + final session = _session; + final peerService = _peerService; + if (session == null || peerService == null) return; + + final wasHost = session.isHost; + final amHost = newHostPeerId == peerService.myPeerId; + _pendingTransferTargetName = null; + + // Any host-departure bookkeeping referred to the previous host. + _cancelHostReconnectGracePeriod(); + _hostIntentionallyLeft = false; + + final updated = session.copyWith(role: amHost ? SessionRole.host : SessionRole.guest, hostPeerId: newHostPeerId); + _session = updated; + + for (var i = 0; i < _participants.length; i++) { + final isHostNow = _participants[i].peerId == newHostPeerId; + if (_participants[i].isHost != isHostNow) { + _participants[i] = _participants[i].copyWith(isHost: isHostNow); + } + } + + _controller?.applyHostChange(updated); + + if (amHost && !wasHost) { + // Re-announce as host — the lobby-safe carrier that teaches guests the + // room's control mode now comes from this peer. + _controller?.announceJoin(_displayName); + _participantEventController.add( + ParticipantEvent(displayName: _displayName, type: ParticipantEventType.becameHost), + ); + } else if (!amHost) { + _participantEventController.add( + ParticipantEvent( + displayName: _displayNameForPeer(newHostPeerId) ?? '?', + type: ParticipantEventType.hostChanged, + ), + ); + } + + appLogger.d('WatchTogether: Host changed to $newHostPeerId (self: $amHost)'); + notifyListeners(); + } + /// Notify guests that host is exiting the video player /// /// Call this from video player dispose when host exits. @@ -961,7 +1059,19 @@ class WatchTogetherProvider with ChangeNotifier { } /// Type of participant event -enum ParticipantEventType { joined, left, paused, resumed, seeked, buffering, needsUpdate, resumedWithout } +enum ParticipantEventType { + joined, + left, + paused, + resumed, + seeked, + buffering, + needsUpdate, + resumedWithout, + hostChanged, + becameHost, + hostTransferFailed, +} /// Event emitted when a participant joins or leaves class ParticipantEvent { diff --git a/lib/watch_together/screens/watch_together_screen.dart b/lib/watch_together/screens/watch_together_screen.dart index 8138e9145..a3b2baa72 100644 --- a/lib/watch_together/screens/watch_together_screen.dart +++ b/lib/watch_together/screens/watch_together_screen.dart @@ -606,7 +606,13 @@ class _ActiveSessionContent extends StatelessWidget { color: participant.isHost ? Colors.amber : theme.colorScheme.onSurfaceVariant, ), const SizedBox(width: 12), - Text(participant.displayName, style: theme.textTheme.bodyMedium), + Flexible( + child: Text( + participant.displayName, + style: theme.textTheme.bodyMedium, + overflow: TextOverflow.ellipsis, + ), + ), if (participant.isHost) ...[ const SizedBox(width: 8), Container( @@ -629,6 +635,17 @@ class _ActiveSessionContent extends StatelessWidget { child: CircularProgressIndicator(strokeWidth: 2, color: theme.colorScheme.primary), ), ], + if (watchTogether.canTransferHostTo(participant)) ...[ + const Spacer(), + FocusableButton( + useBackgroundFocus: true, + onPressed: () => _confirmTransferHost(context, participant), + child: TextButton( + onPressed: () => _confirmTransferHost(context, participant), + child: Text(t.watchTogether.makeHost), + ), + ), + ], ], ), ), @@ -682,6 +699,17 @@ class _ActiveSessionContent extends StatelessWidget { appLogger.e('WatchTogether: Session leave failed', error: error, stackTrace: stackTrace); } } + + Future _confirmTransferHost(BuildContext context, Participant participant) async { + final confirmed = await showConfirmDialog( + context, + title: t.watchTogether.makeHostQuestion, + message: t.watchTogether.makeHostConfirm(name: participant.displayName), + confirmText: t.watchTogether.transfer, + ); + if (!confirmed) return; + watchTogether.transferHost(participant); + } } class _JoinCurrentPlaybackCard extends StatefulWidget { diff --git a/lib/watch_together/services/clock_sync.dart b/lib/watch_together/services/clock_sync.dart index fca552fd4..f08f4bc64 100644 --- a/lib/watch_together/services/clock_sync.dart +++ b/lib/watch_together/services/clock_sync.dart @@ -83,6 +83,24 @@ class ClockSync { _pending.clear(); } + /// Discard every sample and in-flight ping (the peer the offset was + /// measured against changed — e.g. a host transfer) and re-converge + /// against the current host with a fresh burst. + void reset() { + _samples.clear(); + if (_started) { + _started = false; + _timer?.cancel(); + _timer = null; + _burstTimer?.cancel(); + _burstTimer = null; + _pending.clear(); + start(); + } else { + _pending.clear(); + } + } + void _ping() { final now = _nowMs(); _pending.removeWhere((_, sentAt) => now - sentAt > _pendingExpiryMs); diff --git a/lib/watch_together/services/guest_playback_reconciler.dart b/lib/watch_together/services/guest_playback_reconciler.dart index b134e3b24..a7661395c 100644 --- a/lib/watch_together/services/guest_playback_reconciler.dart +++ b/lib/watch_together/services/guest_playback_reconciler.dart @@ -113,6 +113,11 @@ class GuestPlaybackReconciler { PlaybackState? get latestState => _latestState; + /// Whether the attached player has rendered a frame — carried across a + /// host-transfer engine swap so the promoted coordinator doesn't wait for + /// a first frame that already happened. + bool get firstFrameSeen => _firstFrameSeen; + // --------------------------------------------------------------------- // Public inputs // --------------------------------------------------------------------- diff --git a/lib/watch_together/services/host_playback_coordinator.dart b/lib/watch_together/services/host_playback_coordinator.dart index 7674b5bef..1e2cea66c 100644 --- a/lib/watch_together/services/host_playback_coordinator.dart +++ b/lib/watch_together/services/host_playback_coordinator.dart @@ -111,6 +111,11 @@ class HostPlaybackCoordinator { PlaybackPhase get phase => _phase; + /// Whether the attached player has reported readiness — carried across a + /// host-transfer engine swap so the demoted reconciler doesn't wait for a + /// first frame that already happened. + bool get localPlayerReady => _localReady; + // --------------------------------------------------------------------- // Public inputs // --------------------------------------------------------------------- @@ -146,18 +151,25 @@ class HostPlaybackCoordinator { /// screen's first-frame snapshot (covers attaching to an already-rendering /// player); [startupHold] delays readiness past platform startup gates /// (e.g. the Android frame-rate switch). + /// + /// [intendPlaying] seeds the room's play intent for the new epoch. The + /// default (true) is the normal "opening media implies play" flow; a host + /// promoted mid-session passes the room's last known intent so a paused + /// room doesn't start playing just because its host changed. void attach( AttachedPlayer player, { required String ratingKey, required String serverId, String? mediaTitle, bool hasFirstFrame = false, + bool intendPlaying = true, Future? startupHold, }) { detachPlayer(); final sameEpoch = hasActiveEpoch && PlaybackState.mediaKeyFor(ratingKey: ratingKey, serverId: serverId) == _mediaKey; setLocalMedia(ratingKey: ratingKey, serverId: serverId, mediaTitle: mediaTitle); + _intendedPlaying = intendPlaying; _player = player; _rate = player.rate; diff --git a/lib/watch_together/services/relay_protocol.g.dart b/lib/watch_together/services/relay_protocol.g.dart index afce560fb..5537c21ac 100644 --- a/lib/watch_together/services/relay_protocol.g.dart +++ b/lib/watch_together/services/relay_protocol.g.dart @@ -11,6 +11,7 @@ abstract final class RelayProtocol { static const String ping = 'ping'; static const String leave = 'leave'; static const String endSession = 'endSession'; + static const String transferHost = 'transferHost'; static const String created = 'created'; static const String joined = 'joined'; static const String peerJoined = 'peerJoined'; @@ -20,6 +21,7 @@ abstract final class RelayProtocol { static const String pong = 'pong'; static const String left = 'left'; static const String ended = 'ended'; + static const String hostChanged = 'hostChanged'; static const String rateLimitedCode = 'rate_limited'; static const String invalidMessageCode = 'invalid_message'; static const String roomExistsCode = 'room_exists'; @@ -29,6 +31,8 @@ abstract final class RelayProtocol { static const String alreadyInRoomCode = 'already_in_room'; static const String peerIdUnavailableCode = 'peer_id_unavailable'; static const String protocolMismatchCode = 'protocol_mismatch'; + static const String notHostCode = 'not_host'; + static const String peerNotFoundCode = 'peer_not_found'; static const int maxRoomSize = 8; static const int maxMessageSize = 65536; diff --git a/lib/watch_together/services/watch_together_controller.dart b/lib/watch_together/services/watch_together_controller.dart index f40782459..b358b9cf9 100644 --- a/lib/watch_together/services/watch_together_controller.dart +++ b/lib/watch_together/services/watch_together_controller.dart @@ -31,37 +31,45 @@ class WatchTogetherController { _session = session, _nowMs = nowMs ?? watchTogetherSystemNowMs { if (session.isHost) { - _coordinator = HostPlaybackCoordinator( - myPeerId: peerService.myPeerId ?? '', - controlMode: session.controlMode, - sendState: _sendState, - onPhaseChanged: (phase) => onPhaseChanged?.call(phase), - onWaitingOnChanged: (peers) => onWaitingOnChanged?.call(peers), - onResumedWithout: (peers) => onResumedWithout?.call(peers), - onRemoteAction: (peer, hint) => onRemoteAction?.call(peer, hint), - nowMs: _nowMs, - ); + _createCoordinator(); } else { - _clockSync = ClockSync(sendPing: _sendClockPing, nowMs: _nowMs); - _reconciler = GuestPlaybackReconciler( - myPeerId: peerService.myPeerId ?? '', - sendToHost: _sendToHost, - clockSync: _clockSync!, - 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(); + _createReconciler(); } _subscriptions.add(peerService.onMessageReceived.listen(_enqueueMessage)); _subscriptions.add(peerService.onPeerDisconnected.listen(_handlePeerDisconnected)); } + void _createCoordinator() { + _coordinator = HostPlaybackCoordinator( + myPeerId: _peerService.myPeerId ?? '', + controlMode: _session.controlMode, + sendState: _sendState, + onPhaseChanged: (phase) => onPhaseChanged?.call(phase), + onWaitingOnChanged: (peers) => onWaitingOnChanged?.call(peers), + onResumedWithout: (peers) => onResumedWithout?.call(peers), + onRemoteAction: (peer, hint) => onRemoteAction?.call(peer, hint), + nowMs: _nowMs, + ); + } + + void _createReconciler() { + _clockSync = ClockSync(sendPing: _sendClockPing, nowMs: _nowMs); + _reconciler = GuestPlaybackReconciler( + myPeerId: _peerService.myPeerId ?? '', + sendToHost: _sendToHost, + clockSync: _clockSync!, + 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(); + } + final WatchTogetherPeerService _peerService; final int Function() _nowMs; WatchSession _session; @@ -71,6 +79,9 @@ class WatchTogetherController { ClockSync? _clockSync; AttachedPlayer? _attachedPlayer; + String? _attachedRatingKey; + String? _attachedServerId; + String? _attachedMediaTitle; final List> _subscriptions = []; Future _messageQueue = Future.value(); bool _disposed = false; @@ -100,6 +111,78 @@ class WatchTogetherController { _coordinator?.updateControlMode(session.controlMode); } + /// Whether [peerId] speaks the current sync protocol (drives host-transfer + /// eligibility — a peer on an old protocol can't take over the room). + bool isPeerCompatible(String peerId) => (_peerVersions[peerId] ?? 1) == SyncMessage.protocolVersion; + + /// The relay reassigned host authority: swap the role engine while keeping + /// the session, message queue, peer knowledge, and player attachment. + /// + /// [session] already carries the new role and host peer ID. + void applyHostChange(WatchSession session) { + final wasHost = _session.isHost; + _session = session; + + if (wasHost == session.isHost) { + if (!session.isHost) { + // Still a guest, but the authority (and its clock) moved: discard + // offsets and sequence numbering learned from the old host and + // re-converge against the new one. + _clockSync?.reset(); + _reconciler?.resetSequence(); + requestState(); + } + return; + } + + final attached = _attachedPlayer; + if (session.isHost) { + // Promotion: adopt the room where the old host left it. + final lastState = _reconciler?.latestState; + final firstFrameSeen = _reconciler?.firstFrameSeen ?? false; + _clockSync?.stop(); + _clockSync = null; + _reconciler?.dispose(); + _reconciler = null; + _createCoordinator(); + if (attached != null && _attachedRatingKey != null && _attachedServerId != null) { + _coordinator!.attach( + attached, + ratingKey: _attachedRatingKey!, + serverId: _attachedServerId!, + mediaTitle: _attachedMediaTitle, + hasFirstFrame: firstFrameSeen, + // A paused room must not start playing just because its host + // changed; anything else (playing, a stall, mid-load) carries the + // intent to (re)start. + intendPlaying: lastState == null || lastState.phase != PlaybackPhase.paused, + ); + } + // Seed the roster so the fresh epoch gates on the peers we already + // know instead of solo-starting before their first status report. + for (final entry in _peerVersions.entries) { + if (entry.key == _peerService.myPeerId) continue; + _coordinator!.onPeerJoined(entry.key, compatible: entry.value == SyncMessage.protocolVersion); + } + } else { + // Demotion: hand the room to the new host and fall in line. + final localReady = _coordinator?.localPlayerReady ?? false; + _coordinator?.dispose(); + _coordinator = null; + _createReconciler(); + if (attached != null && _attachedRatingKey != null && _attachedServerId != null) { + _reconciler!.attach( + attached, + ratingKey: _attachedRatingKey!, + serverId: _attachedServerId!, + hasFirstFrame: localReady, + ); + } + requestState(); + } + appLogger.d('WatchTogether: Host change applied (host: ${session.isHost})'); + } + // --------------------------------------------------------------------- // Player attachment // --------------------------------------------------------------------- @@ -131,6 +214,9 @@ class WatchTogetherController { nowMs: _nowMs, ); _attachedPlayer = attached; + _attachedRatingKey = ratingKey; + _attachedServerId = serverId; + _attachedMediaTitle = mediaTitle; if (_session.isHost) { _coordinator!.attach( @@ -159,6 +245,9 @@ class WatchTogetherController { final attached = _attachedPlayer; if (attached == null) return; _attachedPlayer = null; + _attachedRatingKey = null; + _attachedServerId = null; + _attachedMediaTitle = null; _coordinator?.detachPlayer(exiting: exiting); _reconciler?.detachPlayer(); unawaited( diff --git a/lib/watch_together/services/watch_together_peer_service.dart b/lib/watch_together/services/watch_together_peer_service.dart index e6b00ce0f..e45002261 100644 --- a/lib/watch_together/services/watch_together_peer_service.dart +++ b/lib/watch_together/services/watch_together_peer_service.dart @@ -82,6 +82,7 @@ class WatchTogetherPeerService with KeepaliveMixin { final _errorController = StreamController.broadcast(); final _connectionStateController = StreamController.broadcast(); final _sessionEndedController = StreamController.broadcast(); + final _hostChangedController = StreamController.broadcast(); // Reconnection state int _reconnectAttempts = 0; @@ -124,6 +125,11 @@ class WatchTogetherPeerService with KeepaliveMixin { /// Emitted when the host has durably ended the relay room. Stream get onSessionEnded => _sessionEndedController.stream; + /// Emitted with the new host's peer ID when the relay reassigns host + /// authority ([transferHost]). [hostPeerId] and [isHost] are already + /// updated when this fires. + Stream get onHostChanged => _hostChangedController.stream; + /// Current session ID (null if not in a session) String? get sessionId => _sessionId; @@ -489,6 +495,20 @@ class WatchTogetherPeerService with KeepaliveMixin { _failSetup(_invalidSetupResponse(RelayProtocol.ended)); } + case RelayProtocol.hostChanged: + final newHostPeerId = msg['hostPeerId']; + if (msg['sessionId'] != _sessionId || + newHostPeerId is! String || + !RelayProtocol.isValidPeerId(newHostPeerId)) { + appLogger.w('WatchTogether: Relay returned an invalid hostChanged message'); + break; + } + if (newHostPeerId == _hostPeerId) break; // Duplicate delivery. + appLogger.d('WatchTogether: Host authority moved to $newHostPeerId'); + _hostPeerId = newHostPeerId; + _isHost = newHostPeerId == _myPeerId; + _safeAdd(_hostChangedController, newHostPeerId); + case RelayProtocol.error: final code = msg['code'] as String? ?? 'unknown'; final message = msg['message'] as String? ?? t.common.unknown; @@ -773,6 +793,18 @@ class WatchTogetherPeerService with KeepaliveMixin { _sendRaw({'type': RelayProtocol.sendTo, 'to': peerId, 'payload': payload}); } + /// Ask the relay to reassign host authority to [peerId] (host only). + /// + /// The relay answers with a `hostChanged` broadcast on success or a + /// `not_host`/`peer_not_found` error on the error stream; local role state + /// only flips when the broadcast arrives. + void transferHost(String peerId) { + if (!RelayProtocol.isValidPeerId(peerId)) { + throw ArgumentError.value(peerId, 'peerId', 'Must be 1–${RelayProtocol.maxPeerIdLength} letters, digits, _ or -'); + } + _sendRaw({'type': RelayProtocol.transferHost, 'to': peerId, 'protocolVersion': _relayProtocolVersion}); + } + /// Explicitly release this peer's relay ownership. Hosts destroy the room; /// guests release their reserved reconnect identity. If transport was lost, /// authenticate a fresh connection first so an intentional exit is not @@ -881,6 +913,7 @@ class WatchTogetherPeerService with KeepaliveMixin { _disposed = true; unawaited(disconnect()); + _hostChangedController.close(); _peerConnectedController.close(); _peerDisconnectedController.close(); _messageReceivedController.close(); diff --git a/lib/watch_together/widgets/watch_together_overlay.dart b/lib/watch_together/widgets/watch_together_overlay.dart index 564f5cabc..ff0acd59b 100644 --- a/lib/watch_together/widgets/watch_together_overlay.dart +++ b/lib/watch_together/widgets/watch_together_overlay.dart @@ -215,31 +215,25 @@ class _SessionMenuSheet extends StatelessWidget { Text(t.watchTogether.participants, style: theme.textTheme.titleSmall), const SizedBox(height: 8), for (final participant in provider.participants) - ListTile( - leading: CircleAvatar( - backgroundColor: participant.isHost - ? theme.colorScheme.primary - : theme.colorScheme.surfaceContainerHighest, - child: AppIcon( - participant.isHost ? Symbols.star_rounded : Symbols.person_rounded, - color: participant.isHost ? Colors.white : theme.colorScheme.onSurfaceVariant, - size: 20, - ), + if (provider.canTransferHostTo(participant)) + FocusableListTile( + leading: _participantAvatar(theme, participant), + title: Text(participant.displayName), + trailing: + _bufferingIndicator(participant) ?? + AppIcon(Symbols.star_rounded, size: 20, color: theme.colorScheme.onSurfaceVariant), + onTap: () => unawaited(_confirmTransferHost(context, participant)), + contentPadding: .zero, + ) + else + ListTile( + leading: _participantAvatar(theme, participant), + title: Text(participant.displayName), + subtitle: participant.isHost ? Text(t.watchTogether.host) : null, + trailing: _bufferingIndicator(participant), + dense: true, + contentPadding: .zero, ), - title: Text(participant.displayName), - subtitle: participant.isHost ? Text(t.watchTogether.host) : null, - trailing: participant.isBuffering - ? SizedBox( - width: 16, - height: 16, - child: PlatformDetector.isTV() - ? const AppIcon(Symbols.hourglass_empty_rounded, size: 16) - : const CircularProgressIndicator(strokeWidth: 2), - ) - : null, - dense: true, - contentPadding: .zero, - ), const SizedBox(height: 16), const Divider(), const SizedBox(height: 8), @@ -259,6 +253,40 @@ class _SessionMenuSheet extends StatelessWidget { ); } + Widget _participantAvatar(ThemeData theme, Participant participant) { + return CircleAvatar( + backgroundColor: participant.isHost ? theme.colorScheme.primary : theme.colorScheme.surfaceContainerHighest, + child: AppIcon( + participant.isHost ? Symbols.star_rounded : Symbols.person_rounded, + color: participant.isHost ? Colors.white : theme.colorScheme.onSurfaceVariant, + size: 20, + ), + ); + } + + Widget? _bufferingIndicator(Participant participant) { + if (!participant.isBuffering) return null; + return SizedBox( + width: 16, + height: 16, + child: PlatformDetector.isTV() + ? const AppIcon(Symbols.hourglass_empty_rounded, size: 16) + : const CircularProgressIndicator(strokeWidth: 2), + ); + } + + Future _confirmTransferHost(BuildContext context, Participant participant) async { + final confirmed = await showConfirmDialog( + context, + title: t.watchTogether.makeHostQuestion, + message: t.watchTogether.makeHostConfirm(name: participant.displayName), + confirmText: t.watchTogether.transfer, + ); + if (!confirmed || !context.mounted) return; + provider.transferHost(participant); + OverlaySheetController.closeAdaptive(context); + } + void _copySessionCode(BuildContext context, String sessionId) { Clipboard.setData(ClipboardData(text: sessionId)); showSuccessSnackBar(context, t.watchTogether.sessionCodeCopied); @@ -351,6 +379,9 @@ class _ParticipantNotificationOverlayState extends State t.watchTogether.participantBuffering(name: n.event.displayName), ParticipantEventType.needsUpdate => t.watchTogether.participantNeedsUpdate(name: n.event.displayName), ParticipantEventType.resumedWithout => t.watchTogether.resumingWithout(name: n.event.displayName), + ParticipantEventType.hostChanged => t.watchTogether.hostChangedTo(name: n.event.displayName), + ParticipantEventType.becameHost => t.watchTogether.youAreNowHost, + ParticipantEventType.hostTransferFailed => t.watchTogether.hostTransferFailed(name: n.event.displayName), }; return Container( key: ValueKey(n.id), diff --git a/relay_protocol.json b/relay_protocol.json index 47a79c2e8..3b66cebc9 100644 --- a/relay_protocol.json +++ b/relay_protocol.json @@ -8,7 +8,8 @@ "sendTo": "sendTo", "ping": "ping", "leave": "leave", - "endSession": "endSession" + "endSession": "endSession", + "transferHost": "transferHost" }, "serverMessageTypes": { "created": "created", @@ -19,7 +20,8 @@ "error": "error", "pong": "pong", "left": "left", - "ended": "ended" + "ended": "ended", + "hostChanged": "hostChanged" }, "errorCodes": { "rateLimited": "rate_limited", @@ -30,7 +32,9 @@ "notInRoom": "not_in_room", "alreadyInRoom": "already_in_room", "peerIdUnavailable": "peer_id_unavailable", - "protocolMismatch": "protocol_mismatch" + "protocolMismatch": "protocol_mismatch", + "notHost": "not_host", + "peerNotFound": "peer_not_found" }, "limits": { "maxRoomSize": 8, diff --git a/server/main.go b/server/main.go index 4939f72c3..9f87dbfdb 100644 --- a/server/main.go +++ b/server/main.go @@ -2278,6 +2278,80 @@ func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) { guest.close() } + case relayTypeTransferHost: + if currentRoom == nil { + client.sendJSON(serverMsg{Type: relayTypeError, Code: relayErrorNotInRoom, Message: "Not in a room"}) + continue + } + room := currentRoom + room.mu.Lock() + authorized := + !room.closing && + room.Peers[currentPeerID] == client && + msg.ProtocolVersion == room.ProtocolVersion + if !authorized { + room.mu.Unlock() + client.sendJSON(serverMsg{Type: relayTypeError, Code: relayErrorPeerIdUnavailable, Message: "Unable to transfer host authority"}) + continue + } + if currentPeerID != room.HostPeerID { + room.mu.Unlock() + client.sendJSON(serverMsg{Type: relayTypeError, Code: relayErrorNotHost, Message: "Only the host can transfer host authority"}) + continue + } + if room.ProtocolVersion != relayProtocolVersion { + room.mu.Unlock() + client.sendJSON(serverMsg{Type: relayTypeError, Code: relayErrorInvalidMessage, Message: "Host transfer requires current protocol"}) + continue + } + targetPeerID := msg.To + targetConnected := false + if validRelayID(targetPeerID, maxPeerIDLength) && targetPeerID != currentPeerID { + _, targetConnected = room.Peers[targetPeerID] + } + if !targetConnected { + room.mu.Unlock() + client.sendJSON(serverMsg{Type: relayTypeError, Code: relayErrorPeerNotFound, Message: "Target peer not found"}) + continue + } + targetReservation, targetReserved := room.peerReservations[targetPeerID] + if !targetReserved || targetReservation.releasePending { + room.mu.Unlock() + if !targetReserved { + // Connected modern-room guests always hold a reservation. + log.Printf("transferHost: connected guest missing reservation") + } + client.sendJSON(serverMsg{Type: relayTypeError, Code: relayErrorPeerNotFound, Message: "Target peer not found"}) + continue + } + + // Swap authority. The host never holds a peer reservation (snapshot + // loading rejects that), so the target's reservation becomes the + // host verifier and the old host gains a connected reservation. + // Reconnect tokens are untouched: both peers keep their own. + oldHostPeerID := currentPeerID + delete(room.peerReservations, targetPeerID) + room.peerReservations[oldHostPeerID] = peerReservation{verifier: room.hostVerifier} + room.hostVerifier = targetReservation.verifier + room.HostPeerID = targetPeerID + room.LastActivityAt = time.Now() + recipients := make([]*Client, 0, len(room.Peers)) + for _, peerClient := range room.Peers { + recipients = append(recipients, peerClient) + } + hostChanged := serverMsg{ + Type: relayTypeHostChanged, + SessionID: room.SessionID, + HostPeerID: targetPeerID, + From: oldHostPeerID, + } + s.snap.recordMutation() + room.mu.Unlock() + + for _, peerClient := range recipients { + peerClient.sendJSON(hostChanged) + } + case relayTypeBroadcast: if currentRoom == nil { client.sendJSON(serverMsg{Type: relayTypeError, Code: relayErrorNotInRoom, Message: "Not in a room"}) diff --git a/server/main_test.go b/server/main_test.go index 823322710..c917511ff 100644 --- a/server/main_test.go +++ b/server/main_test.go @@ -5294,6 +5294,271 @@ func TestAuthenticatedModernHostEndDeletesRoomAndIsRetrySafe(t *testing.T) { retry.expectError(relayErrorRoomNotFound) } +func TestTransferHostSwapsAuthorityAndBroadcastsHostChanged(t *testing.T) { + h := newRelayHarness(t) + hostToken, _ := mustReconnectToken(t) + host := h.dial(t, "6.3.0.1") + host.send(clientMsg{ + Type: relayTypeCreate, + SessionID: "XFER_HAPPY", + PeerID: "H", + ReconnectToken: hostToken, + ProtocolVersion: relayProtocolVersion, + }) + host.expectAuthority(relayTypeCreated, "H") + + guestAToken, _ := mustReconnectToken(t) + guestA := h.dial(t, "6.3.0.2") + guestA.send(clientMsg{ + Type: relayTypeJoin, + SessionID: "XFER_HAPPY", + PeerID: "A", + ReconnectToken: guestAToken, + ProtocolVersion: relayProtocolVersion, + }) + guestA.expectAuthority(relayTypeJoined, "H") + host.expect(relayTypePeerJoined) + + guestBToken, _ := mustReconnectToken(t) + guestB := h.dial(t, "6.3.0.3") + guestB.send(clientMsg{ + Type: relayTypeJoin, + SessionID: "XFER_HAPPY", + PeerID: "B", + ReconnectToken: guestBToken, + ProtocolVersion: relayProtocolVersion, + }) + guestB.expectAuthority(relayTypeJoined, "H") + host.expect(relayTypePeerJoined) + guestA.expect(relayTypePeerJoined) + + host.send(clientMsg{Type: relayTypeTransferHost, To: "A", ProtocolVersion: relayProtocolVersion}) + for _, peer := range []*testConn{host, guestA, guestB} { + changed := peer.expect(relayTypeHostChanged) + if changed.SessionID != "XFER_HAPPY" || changed.HostPeerID != "A" || changed.From != "H" { + t.Fatalf("hostChanged=%+v, want sessionId=XFER_HAPPY hostPeerId=A from=H", changed) + } + } + + // The old host lost end-session authority even with its valid token. + host.send(clientMsg{ + Type: relayTypeEndSession, + ReconnectToken: hostToken, + ProtocolVersion: relayProtocolVersion, + }) + host.expectError(relayErrorPeerIdUnavailable) + + // The new host ends the room with its own unchanged reconnect token. + guestA.send(clientMsg{ + Type: relayTypeEndSession, + ReconnectToken: guestAToken, + ProtocolVersion: relayProtocolVersion, + }) + ended := guestA.expect(relayTypeEnded) + if ended.SessionID != "XFER_HAPPY" || ended.ProtocolVersion != relayProtocolVersion { + t.Fatalf("ended acknowledgement=%+v", ended) + } + for _, peer := range []*testConn{host, guestB} { + messages, err := peer.recvUntilClosed(2 * time.Second) + if err != nil { + t.Fatalf("peer stayed connected after transfer-then-end: %v (frames=%v)", err, messages) + } + if len(messages) != 1 || + messages[0].Type != relayTypeEnded || + messages[0].SessionID != "XFER_HAPPY" { + t.Fatalf("terminal frames=%+v, want one ended notification", messages) + } + } +} + +func TestTransferHostRejectsNonHostSender(t *testing.T) { + h := newRelayHarness(t) + host, guest, hostToken, _ := createModernRoomWithGuest(t, h, "XFER_NOT_HOST", "6.3.1.1", "6.3.1.2") + guest.send(clientMsg{Type: relayTypeTransferHost, To: "H", ProtocolVersion: relayProtocolVersion}) + guest.expectError(relayErrorNotHost) + host.recvNothing(200 * time.Millisecond) + + h.srv.mu.RLock() + room := h.srv.rooms["XFER_NOT_HOST"] + h.srv.mu.RUnlock() + if room == nil { + t.Fatal("room missing after rejected transfer") + } + presented, ok := reconnectVerifierFromToken(hostToken) + if !ok { + t.Fatal("host token has invalid shape") + } + room.mu.RLock() + hostPeerID := room.HostPeerID + hostAuthorityIntact := reconnectVerifierMatches(room.hostVerifier, presented) + _, guestReserved := room.peerReservations["G"] + room.mu.RUnlock() + if hostPeerID != "H" { + t.Fatalf("hostPeerId=%q after rejected transfer, want H", hostPeerID) + } + if !hostAuthorityIntact { + t.Fatal("host verifier changed after rejected transfer") + } + if !guestReserved { + t.Fatal("guest reservation lost after rejected transfer") + } +} + +func TestTransferHostRejectsInvalidTargets(t *testing.T) { + h := newRelayHarness(t) + outsider := h.dial(t, "6.3.2.9") + outsider.send(clientMsg{Type: relayTypeTransferHost, To: "G", ProtocolVersion: relayProtocolVersion}) + outsider.expectError(relayErrorNotInRoom) + + host, guest, _, _ := createModernRoomWithGuest(t, h, "XFER_TARGETS", "6.3.2.1", "6.3.2.2") + for _, target := range []string{"", "H", "UNKNOWN", "bad peer!"} { + host.send(clientMsg{Type: relayTypeTransferHost, To: target, ProtocolVersion: relayProtocolVersion}) + host.expectError(relayErrorPeerNotFound) + } + guest.recvNothing(200 * time.Millisecond) + + // A reserved-but-disconnected guest is not a valid transfer target. + if err := guest.conn.Close(); err != nil { + t.Fatalf("close guest: %v", err) + } + left := host.expect(relayTypePeerLeft) + if left.PeerID != "G" { + t.Fatalf("peerLeft peerId=%q, want G", left.PeerID) + } + host.send(clientMsg{Type: relayTypeTransferHost, To: "G", ProtocolVersion: relayProtocolVersion}) + host.expectError(relayErrorPeerNotFound) +} + +func TestTransferHostRejectsLegacyRoom(t *testing.T) { + h := newRelayHarness(t) + host := h.dial(t, "6.3.3.1") + host.send(clientMsg{Type: relayTypeCreate, SessionID: "XFER_LEGACY", PeerID: "H"}) + host.expect(relayTypeCreated) + guest := h.dial(t, "6.3.3.2") + guest.send(clientMsg{Type: relayTypeJoin, SessionID: "XFER_LEGACY", PeerID: "G"}) + guest.expect(relayTypeJoined) + host.expect(relayTypePeerJoined) + + host.send(clientMsg{Type: relayTypeTransferHost, To: "G"}) + host.expectError(relayErrorInvalidMessage) + guest.recvNothing(200 * time.Millisecond) +} + +func TestTransferHostPreservesReconnectAuthority(t *testing.T) { + h := newRelayHarness(t) + host, guest, hostToken, guestToken := createModernRoomWithGuest(t, h, "XFER_RECONNECT", "6.3.4.1", "6.3.4.2") + host.send(clientMsg{Type: relayTypeTransferHost, To: "G", ProtocolVersion: relayProtocolVersion}) + host.expect(relayTypeHostChanged) + guest.expect(relayTypeHostChanged) + + // The old host drops and rejoins with its original token as a guest. + if err := host.conn.Close(); err != nil { + t.Fatalf("close old host: %v", err) + } + left := guest.expect(relayTypePeerLeft) + if left.PeerID != "H" { + t.Fatalf("peerLeft peerId=%q, want H", left.PeerID) + } + oldHost := h.dial(t, "6.3.4.3") + oldHost.send(clientMsg{ + Type: relayTypeJoin, + SessionID: "XFER_RECONNECT", + PeerID: "H", + ReconnectToken: hostToken, + ProtocolVersion: relayProtocolVersion, + }) + rejoinedGuest := oldHost.expectAuthority(relayTypeJoined, "G") + if rejoinedGuest.ReconnectToken != hostToken { + t.Fatalf("old host reconnect token changed: %+v", rejoinedGuest) + } + guest.expect(relayTypePeerJoined) + + // The new host drops and rejoins with its original token as the host. + if err := guest.conn.Close(); err != nil { + t.Fatalf("close new host: %v", err) + } + left = oldHost.expect(relayTypePeerLeft) + if left.PeerID != "G" { + t.Fatalf("peerLeft peerId=%q, want G", left.PeerID) + } + newHost := h.dial(t, "6.3.4.4") + newHost.send(clientMsg{ + Type: relayTypeJoin, + SessionID: "XFER_RECONNECT", + PeerID: "G", + ReconnectToken: guestToken, + ProtocolVersion: relayProtocolVersion, + }) + rejoinedHost := newHost.expectAuthority(relayTypeJoined, "G") + if rejoinedHost.ReconnectToken != guestToken { + t.Fatalf("new host reconnect token changed: %+v", rejoinedHost) + } + oldHost.expect(relayTypePeerJoined) +} + +func TestTransferHostSurvivesSnapshotRestart(t *testing.T) { + stateFile := filepath.Join(t.TempDir(), "rooms.json") + hA := newRelayHarnessAt(t, t.TempDir(), stateFile) + host, guest, hostToken, guestToken := createModernRoomWithGuest(t, hA, "XFER_RESTART", "6.3.5.1", "6.3.5.2") + host.send(clientMsg{Type: relayTypeTransferHost, To: "G", ProtocolVersion: relayProtocolVersion}) + host.expect(relayTypeHostChanged) + guest.expect(relayTypeHostChanged) + + if err := hA.srv.snap.flushAndStop(2 * time.Second); err != nil { + t.Fatalf("flush snapshot after transfer: %v", err) + } + + hB := newRelayHarnessAt(t, t.TempDir(), stateFile) + hB.srv.mu.RLock() + room := hB.srv.rooms["XFER_RESTART"] + hB.srv.mu.RUnlock() + if room == nil { + t.Fatal("restored room missing") + } + newHostVerifier, ok := reconnectVerifierFromToken(guestToken) + if !ok { + t.Fatal("new host token has invalid shape") + } + room.mu.RLock() + hostPeerID := room.HostPeerID + hostAuthority := reconnectVerifierMatches(room.hostVerifier, newHostVerifier) + _, oldHostReserved := room.peerReservations["H"] + _, newHostReserved := room.peerReservations["G"] + room.mu.RUnlock() + if hostPeerID != "G" { + t.Fatalf("restored hostPeerId=%q, want G", hostPeerID) + } + if !hostAuthority { + t.Fatal("restored host verifier does not validate the new host's token") + } + if !oldHostReserved { + t.Fatal("restored room lost the old host's guest reservation") + } + if newHostReserved { + t.Fatal("restored room kept a guest reservation for the new host") + } + + newHost := hB.dial(t, "6.3.5.3") + newHost.send(clientMsg{ + Type: relayTypeJoin, + SessionID: "XFER_RESTART", + PeerID: "G", + ReconnectToken: guestToken, + ProtocolVersion: relayProtocolVersion, + }) + newHost.expectAuthority(relayTypeJoined, "G") + oldHost := hB.dial(t, "6.3.5.4") + oldHost.send(clientMsg{ + Type: relayTypeJoin, + SessionID: "XFER_RESTART", + PeerID: "H", + ReconnectToken: hostToken, + ProtocolVersion: relayProtocolVersion, + }) + oldHost.expectAuthority(relayTypeJoined, "G") + newHost.expect(relayTypePeerJoined) +} + func TestHostEndDeliversEndedAfterConcurrentGuestTraffic(t *testing.T) { h := newRelayHarness(t) endDeliveryReady := make(chan struct{}) diff --git a/server/relay_protocol_gen.go b/server/relay_protocol_gen.go index c4814c690..144ca1ed7 100644 --- a/server/relay_protocol_gen.go +++ b/server/relay_protocol_gen.go @@ -13,6 +13,7 @@ const ( relayTypePing = "ping" relayTypeLeave = "leave" relayTypeEndSession = "endSession" + relayTypeTransferHost = "transferHost" relayTypeCreated = "created" relayTypeJoined = "joined" relayTypePeerJoined = "peerJoined" @@ -22,6 +23,7 @@ const ( relayTypePong = "pong" relayTypeLeft = "left" relayTypeEnded = "ended" + relayTypeHostChanged = "hostChanged" relayErrorRateLimited = "rate_limited" relayErrorInvalidMessage = "invalid_message" relayErrorRoomExists = "room_exists" @@ -31,6 +33,8 @@ const ( relayErrorAlreadyInRoom = "already_in_room" relayErrorPeerIdUnavailable = "peer_id_unavailable" relayErrorProtocolMismatch = "protocol_mismatch" + relayErrorNotHost = "not_host" + relayErrorPeerNotFound = "peer_not_found" maxRoomSize = 8 maxMessageSize = 65536 diff --git a/test/watch_together/watch_together_controller_test.dart b/test/watch_together/watch_together_controller_test.dart index 56b3b2c91..ceb35bebc 100644 --- a/test/watch_together/watch_together_controller_test.dart +++ b/test/watch_together/watch_together_controller_test.dart @@ -579,4 +579,81 @@ void main() { }); }); }); + + group('host transfer', () { + WatchSession sessionAs(SessionRole role) => WatchSession( + sessionId: 'ROOM1', + role: role, + controlMode: ControlMode.hostOnly, + state: SessionState.connected, + hostPeerId: 'guest', + ); + + test('mid-playback: the promoted guest re-anchors the room and the demoted host follows', () { + fakeAsync((async) { + final room = _Room(async); + room.hostStartsMedia(); + room.guestJoinsMedia(); + room.bothBecomeReady(); + async.elapse(const Duration(seconds: 3)); + expect(room.hostPlayer.state.playing, isTrue); + expect(room.guestPlayer.state.playing, isTrue); + final originalKey = room.lastHostState().mediaKey; + final statesBefore = room.hostService.outgoingLog.where((m) => m.type == SyncMessageType.state).length; + + // The relay reassigned authority: each side applies its own hostChanged. + room.host.applyHostChange(sessionAs(SessionRole.guest)); + room.guest.applyHostChange(sessionAs(SessionRole.host)); + async.flushMicrotasks(); + + // The demoted host falls in line: it asks the new authority for state. + expect(room.hostService.outgoingLog.any((m) => m.type == SyncMessageType.requestState), isTrue); + + // The promoted guest re-runs the epoch gate and both players resume. + async.elapse(const Duration(seconds: 5)); + final adopted = room.guestService.outgoingLog.lastWhere((m) => m.type == SyncMessageType.state).state!; + expect(adopted.phase, PlaybackPhase.playing); + expect(adopted.mediaKey, originalKey); + expect(room.hostPlayer.state.playing, isTrue); + expect(room.guestPlayer.state.playing, isTrue); + + // The old host's coordinator is gone — it authors no further states. + expect(room.hostService.outgoingLog.where((m) => m.type == SyncMessageType.state).length, statesBefore); + room.dispose(); + }); + }); + + test('a paused room stays paused across the handoff', () { + fakeAsync((async) { + final room = _Room(async); + room.hostStartsMedia(); + room.guestJoinsMedia(); + room.bothBecomeReady(); + async.elapse(const Duration(seconds: 3)); + expect(room.hostPlayer.state.playing, isTrue); + + // The host pauses the room before handing it over. + room.hostPlayer.emitPlaying(false); + async.flushMicrotasks(); + async.elapse(const Duration(seconds: 1)); + expect(room.lastHostState().phase, PlaybackPhase.paused); + expect(room.guestPlayer.state.playing, isFalse); + final guestPlays = room.guestPlayer.commandLog.where((c) => c == 'play').length; + final hostPlays = room.hostPlayer.commandLog.where((c) => c == 'play').length; + + room.host.applyHostChange(sessionAs(SessionRole.guest)); + room.guest.applyHostChange(sessionAs(SessionRole.host)); + async.flushMicrotasks(); + async.elapse(const Duration(seconds: 5)); + + final adopted = room.guestService.outgoingLog.lastWhere((m) => m.type == SyncMessageType.state).state!; + expect(adopted.phase, PlaybackPhase.paused, reason: 'a host change is not a play intent'); + expect(room.hostPlayer.state.playing, isFalse); + expect(room.guestPlayer.state.playing, isFalse); + expect(room.guestPlayer.commandLog.where((c) => c == 'play').length, guestPlays); + expect(room.hostPlayer.commandLog.where((c) => c == 'play').length, hostPlays); + room.dispose(); + }); + }); + }); } diff --git a/test/watch_together/watch_together_overlay_test.dart b/test/watch_together/watch_together_overlay_test.dart index dfd509e01..12012c3ad 100644 --- a/test/watch_together/watch_together_overlay_test.dart +++ b/test/watch_together/watch_together_overlay_test.dart @@ -111,6 +111,56 @@ void main() { await tester.pump(); semantics.dispose(); }); + + testWidgets('host promotes a guest from the session sheet', (tester) async { + final harness = _OverlayHarness(isHost: true); + harness.provider.extraParticipants = [const Participant(peerId: 'g1', displayName: 'Guest One', isHost: false)]; + addTearDown(harness.dispose); + await tester.pumpWidget(harness.build()); + + await tester.tap(find.byKey(_OverlayHarness.indicatorKey)); + await tester.pumpAndSettle(); + await tester.tap(find.text('Guest One')); + await tester.pumpAndSettle(); + + expect(find.text(t.watchTogether.makeHostQuestion), findsOneWidget); + await tester.tap(find.text(t.watchTogether.transfer)); + await tester.pumpAndSettle(); + + expect(harness.provider.transferredTo.single.peerId, 'g1'); + expect(harness.sheetController.isOpen, isFalse, reason: 'the sheet closes once the transfer is requested'); + }); + + testWidgets('cancelling the promote confirmation transfers nothing', (tester) async { + final harness = _OverlayHarness(isHost: true); + harness.provider.extraParticipants = [const Participant(peerId: 'g1', displayName: 'Guest One', isHost: false)]; + addTearDown(harness.dispose); + await tester.pumpWidget(harness.build()); + + await tester.tap(find.byKey(_OverlayHarness.indicatorKey)); + await tester.pumpAndSettle(); + await tester.tap(find.text('Guest One')); + await tester.pumpAndSettle(); + await tester.tap(find.text(t.common.cancel)); + await tester.pumpAndSettle(); + + expect(harness.provider.transferredTo, isEmpty); + }); + + testWidgets('guests see fellow participants without a promote affordance', (tester) async { + final harness = _OverlayHarness(isHost: false); + harness.provider.extraParticipants = [const Participant(peerId: 'g1', displayName: 'Guest One', isHost: false)]; + addTearDown(harness.dispose); + await tester.pumpWidget(harness.build()); + + await tester.tap(find.byKey(_OverlayHarness.indicatorKey)); + await tester.pumpAndSettle(); + await tester.tap(find.text('Guest One'), warnIfMissed: false); + await tester.pumpAndSettle(); + + expect(find.text(t.watchTogether.makeHostQuestion), findsNothing); + expect(harness.provider.transferredTo, isEmpty); + }); } Future _openLeaveConfirmation(WidgetTester tester, _OverlayHarness harness) async { @@ -178,8 +228,22 @@ class _FakeWatchTogetherProvider extends WatchTogetherProvider { @override List get participants => [ Participant(peerId: 'local', displayName: 'Local viewer', isHost: isHostValue), + ...extraParticipants, ]; + /// Participants appended after the local viewer row. + List extraParticipants = const []; + + /// Transfer requests forwarded by the sheet. + final List transferredTo = []; + + @override + bool canTransferHostTo(Participant participant) => + isHostValue && !participant.isHost && participant.peerId != 'local'; + + @override + void transferHost(Participant participant) => transferredTo.add(participant); + @override int get participantCount => participants.length; diff --git a/test/watch_together/watch_together_peer_service_test.dart b/test/watch_together/watch_together_peer_service_test.dart index 50ce2b01a..8df178d24 100644 --- a/test/watch_together/watch_together_peer_service_test.dart +++ b/test/watch_together/watch_together_peer_service_test.dart @@ -236,6 +236,7 @@ void main() { () => service.sendTo('bad peer', const SyncMessage(type: SyncMessageType.requestState, timestamp: 0)), throwsArgumentError, ); + expect(() => service.transferHost('bad peer'), throwsArgumentError); }); test('host stores relay authority and uses a random routing ID', () async { @@ -1232,4 +1233,152 @@ void main() { expect(reconnectCallbacks, 0); }); + + test('host transfer request reaches the relay and the broadcast demotes the sender', () async { + late final _RelayServer relay; + relay = await relayWith((_, socket, message) { + if (message['type'] == 'create') { + relay.send(socket, { + 'type': 'created', + 'sessionId': message['sessionId'], + 'hostPeerId': message['peerId'], + 'reconnectToken': message['reconnectToken'], + 'protocolVersion': 2, + }); + } else if (message['type'] == 'transferHost') { + relay.send(socket, { + 'type': 'hostChanged', + 'sessionId': 'XFER1', + 'hostPeerId': message['to'], + 'from': message['peerId'], + }); + } + }); + final service = serviceFor(relay); + final changed = Completer(); + final subscription = service.onHostChanged.listen((peerId) { + if (!changed.isCompleted) changed.complete(peerId); + }); + addTearDown(subscription.cancel); + + await service.createSession(sessionId: 'xfer1'); + service.transferHost('guest-1'); + + expect(await changed.future.timeout(const Duration(seconds: 5)), 'guest-1'); + expect(service.isHost, isFalse); + expect(service.hostPeerId, 'guest-1'); + expect(relay.messages.single.last, {'type': 'transferHost', 'to': 'guest-1', 'protocolVersion': 2}); + }); + + test('a guest named in hostChanged adopts host authority', () async { + late final _RelayServer relay; + relay = await relayWith((_, socket, message) { + if (message['type'] == 'join') { + relay.send(socket, { + 'type': 'joined', + 'sessionId': message['sessionId'], + 'hostPeerId': _relayHostId, + 'reconnectToken': message['reconnectToken'], + 'protocolVersion': 2, + 'peers': [_relayHostId], + }); + } + }); + final service = serviceFor(relay); + final changed = Completer(); + final subscription = service.onHostChanged.listen((peerId) { + if (!changed.isCompleted) changed.complete(peerId); + }); + addTearDown(subscription.cancel); + + await service.joinSession('xfer2'); + relay.send(relay.sockets.single, { + 'type': 'hostChanged', + 'sessionId': 'XFER2', + 'hostPeerId': service.myPeerId, + 'from': _relayHostId, + }); + + expect(await changed.future.timeout(const Duration(seconds: 5)), service.myPeerId); + expect(service.isHost, isTrue); + expect(service.hostPeerId, service.myPeerId); + }); + + test('invalid and duplicate hostChanged messages are ignored', () async { + late final _RelayServer relay; + relay = await relayWith((_, socket, message) { + if (message['type'] == 'join') { + relay.send(socket, { + 'type': 'joined', + 'sessionId': message['sessionId'], + 'hostPeerId': _relayHostId, + 'reconnectToken': message['reconnectToken'], + 'protocolVersion': 2, + 'peers': [_relayHostId], + }); + } + }); + final service = serviceFor(relay); + final observed = []; + final subscription = service.onHostChanged.listen(observed.add); + addTearDown(subscription.cancel); + + await service.joinSession('xfer3'); + final socket = relay.sockets.single; + // Wrong room, malformed peer, and a no-op "change" to the current host + // must all be dropped; the valid change afterwards proves ordering. + relay.send(socket, {'type': 'hostChanged', 'sessionId': 'OTHER', 'hostPeerId': 'guest-9'}); + relay.send(socket, {'type': 'hostChanged', 'sessionId': 'XFER3', 'hostPeerId': 'bad peer'}); + relay.send(socket, {'type': 'hostChanged', 'sessionId': 'XFER3', 'hostPeerId': _relayHostId}); + relay.send(socket, {'type': 'hostChanged', 'sessionId': 'XFER3', 'hostPeerId': 'guest-2'}); + + while (observed.isEmpty) { + await Future.delayed(const Duration(milliseconds: 10)); + } + expect(observed, ['guest-2']); + expect(service.hostPeerId, 'guest-2'); + expect(service.isHost, isFalse); + }); + + test('guest reconnect accepts the host identity pinned by a transfer', () async { + late final _RelayServer relay; + relay = await relayWith((connection, socket, message) { + if (message['type'] == 'join') { + relay.send(socket, { + 'type': 'joined', + 'sessionId': message['sessionId'], + 'hostPeerId': connection == 0 ? _relayHostId : 'guest-2', + 'reconnectToken': message['reconnectToken'], + 'protocolVersion': 2, + 'peers': ['guest-2'], + }); + } + }); + final service = serviceFor(relay); + final errors = []; + final errorSubscription = service.onError.listen(errors.add); + addTearDown(errorSubscription.cancel); + final changed = Completer(); + final subscription = service.onHostChanged.listen((peerId) { + if (!changed.isCompleted) changed.complete(peerId); + }); + addTearDown(subscription.cancel); + final reconnected = Completer(); + service.onReconnected = reconnected.complete; + + await _withShortenedTimer( + original: const Duration(seconds: 2), + replacement: const Duration(milliseconds: 10), + body: () => service.joinSession('xfer4'), + ); + relay.send(relay.sockets.single, {'type': 'hostChanged', 'sessionId': 'XFER4', 'hostPeerId': 'guest-2'}); + await changed.future.timeout(const Duration(seconds: 5)); + + await relay.sockets.single.close(); + await reconnected.future.timeout(const Duration(seconds: 6)); + + expect(service.hostPeerId, 'guest-2'); + expect(service.isHost, isFalse); + expect(errors, isEmpty); + }); } diff --git a/test/watch_together/watch_together_provider_test.dart b/test/watch_together/watch_together_provider_test.dart index 8f238a996..35dfa92ef 100644 --- a/test/watch_together/watch_together_provider_test.dart +++ b/test/watch_together/watch_together_provider_test.dart @@ -32,8 +32,13 @@ class _FakeWatchTogetherPeerService extends WatchTogetherPeerService { final _messageController = StreamController.broadcast(); final _errorController = StreamController.broadcast(); final _sessionEndedController = StreamController.broadcast(); + final _hostChangedController = StreamController.broadcast(); final List broadcasts = []; final List<(String, SyncMessage)> directMessages = []; + final List transferRequests = []; + + /// Extra peers reported as connected on top of the modeled host link. + final Set connectedGuests = {}; String? _sessionId; String? _myPeerId; @@ -59,6 +64,9 @@ class _FakeWatchTogetherPeerService extends WatchTogetherPeerService { @override Stream get onSessionEnded => _sessionEndedController.stream; + @override + Stream get onHostChanged => _hostChangedController.stream; + @override String? get sessionId => _sessionId; @@ -72,7 +80,10 @@ class _FakeWatchTogetherPeerService extends WatchTogetherPeerService { bool get isHost => _isHost; @override - List get connectedPeers => !_isHost && _sessionId != null && _hostConnected ? ['wt-$_sessionId'] : const []; + List get connectedPeers => [ + if (!_isHost && _sessionId != null && _hostConnected) 'wt-$_sessionId', + ...connectedGuests, + ]; @override Future createSession({String? sessionId}) { @@ -108,6 +119,15 @@ class _FakeWatchTogetherPeerService extends WatchTogetherPeerService { directMessages.add((peerId, message)); } + @override + void transferHost(String peerId) => transferRequests.add(peerId); + + void emitHostChanged(String newHostPeerId) { + _hostPeerId = newHostPeerId; + _isHost = newHostPeerId == _myPeerId; + _hostChangedController.add(newHostPeerId); + } + void emitError(PeerError error) => _errorController.add(error); void emitPeerConnected(String peerId) { if (peerId == _hostPeerId) _hostConnected = true; @@ -127,7 +147,8 @@ class _FakeWatchTogetherPeerService extends WatchTogetherPeerService { _peerDisconnectedController.hasListener || _messageController.hasListener || _errorController.hasListener || - _sessionEndedController.hasListener; + _sessionEndedController.hasListener || + _hostChangedController.hasListener; bool get isDisposed => _disposed; bool get didDisconnect => _didDisconnect; @@ -162,6 +183,7 @@ class _FakeWatchTogetherPeerService extends WatchTogetherPeerService { _messageController.close(); _errorController.close(); _sessionEndedController.close(); + _hostChangedController.close(); super.dispose(); } } @@ -997,4 +1019,158 @@ void main() { expect(service.isDisposed, isTrue); }); }); + + group('WatchTogetherProvider — host transfer', () { + test('hostChanged demotes the host, updates participants, and toasts the new host', () async { + final factory = _FakePeerServiceFactory(); + final provider = WatchTogetherProvider(peerServiceFactory: factory.call); + addTearDown(provider.dispose); + final events = []; + final subscription = provider.participantEvents.listen(events.add); + addTearDown(subscription.cancel); + + await provider.createSession( + controlMode: ControlMode.anyone, + relayEndpoint: WatchTogetherRelayEndpoint.defaultEndpoint, + displayName: 'Host', + sessionId: 'xfer1', + ); + final service = factory.services.single; + service.connectedGuests.add('guest-a'); + service.emitMessage(SyncMessage.join(peerId: 'guest-a', displayName: 'Guest A', isHost: false)); + await _flushProviderEvents(); + + final guest = provider.participants.singleWhere((p) => p.peerId == 'guest-a'); + expect(provider.canTransferHostTo(guest), isTrue); + provider.transferHost(guest); + expect(service.transferRequests, ['guest-a']); + + // Roles only flip when the relay's broadcast lands. + expect(provider.isHost, isTrue); + service.emitHostChanged('guest-a'); + await _flushProviderEvents(); + + expect(provider.isHost, isFalse); + expect(provider.session?.hostPeerId, 'guest-a'); + expect(provider.session?.state, SessionState.connected); + expect(provider.participants.singleWhere((p) => p.peerId == 'guest-a').isHost, isTrue); + expect(provider.participants.singleWhere((p) => p.peerId != 'guest-a').isHost, isFalse); + final event = events.singleWhere((e) => e.type == ParticipantEventType.hostChanged); + expect(event.displayName, 'Guest A'); + }); + + test('hostChanged promotion re-announces the room control mode as host', () async { + final factory = _FakePeerServiceFactory(); + final provider = WatchTogetherProvider(peerServiceFactory: factory.call); + addTearDown(provider.dispose); + final events = []; + final subscription = provider.participantEvents.listen(events.add); + addTearDown(subscription.cancel); + + await provider.joinSession( + 'xfer2', + relayEndpoint: WatchTogetherRelayEndpoint.defaultEndpoint, + displayName: 'Guest', + ); + final service = factory.services.single; + // The host's join teaches guests the room's real control mode. + service.emitMessage( + SyncMessage.join(peerId: 'wt-XFER2', displayName: 'Host', isHost: true, controlMode: ControlMode.anyone), + ); + await _flushProviderEvents(); + expect(provider.controlMode, ControlMode.anyone); + + service.emitHostChanged(service.myPeerId!); + await _flushProviderEvents(); + + expect(provider.isHost, isTrue); + expect(provider.session?.role, SessionRole.host); + expect(provider.session?.hostPeerId, service.myPeerId); + expect(provider.participants.singleWhere((p) => p.peerId == service.myPeerId).isHost, isTrue); + expect(provider.participants.singleWhere((p) => p.peerId == 'wt-XFER2').isHost, isFalse); + expect(events.map((e) => e.type), contains(ParticipantEventType.becameHost)); + // The promoted host re-broadcasts a join carrying the inherited mode. + final announced = service.broadcasts.lastWhere((m) => m.type == SyncMessageType.join); + expect(announced.isHost, isTrue); + expect(announced.controlMode, ControlMode.anyone); + // Reverse state: the demoted-side wiring is exercised by the demotion + // test above; a promoted host can end the session (host-only path). + expect(provider.canControl(), isTrue); + }); + + test('a rejected transfer keeps the session connected and toasts the failure', () async { + final factory = _FakePeerServiceFactory(); + final provider = WatchTogetherProvider(peerServiceFactory: factory.call); + addTearDown(provider.dispose); + final events = []; + final subscription = provider.participantEvents.listen(events.add); + addTearDown(subscription.cancel); + + await provider.createSession( + controlMode: ControlMode.hostOnly, + relayEndpoint: WatchTogetherRelayEndpoint.defaultEndpoint, + displayName: 'Host', + sessionId: 'xfer3', + ); + final service = factory.services.single; + service.connectedGuests.add('guest-b'); + service.emitMessage(SyncMessage.join(peerId: 'guest-b', displayName: 'Guest B', isHost: false)); + await _flushProviderEvents(); + + provider.transferHost(provider.participants.singleWhere((p) => p.peerId == 'guest-b')); + service.emitError( + const PeerError( + type: PeerErrorType.serverError, + message: 'peer_not_found: Peer is not in the room', + serverCode: 'peer_not_found', + ), + ); + await _flushProviderEvents(); + + expect(provider.session?.state, SessionState.connected); + expect(provider.isHost, isTrue); + final event = events.singleWhere((e) => e.type == ParticipantEventType.hostTransferFailed); + expect(event.displayName, 'Guest B'); + }); + + test('canTransferHostTo requires host role, a connected target, and protocol compatibility', () async { + final factory = _FakePeerServiceFactory(); + final provider = WatchTogetherProvider(peerServiceFactory: factory.call); + addTearDown(provider.dispose); + + await provider.createSession( + controlMode: ControlMode.anyone, + relayEndpoint: WatchTogetherRelayEndpoint.defaultEndpoint, + displayName: 'Host', + sessionId: 'xfer4', + ); + final service = factory.services.single; + service.connectedGuests.add('guest-new'); + service.emitMessage(SyncMessage.join(peerId: 'guest-new', displayName: 'New', isHost: false)); + // An old-protocol guest (version below the current sync protocol). + service.connectedGuests.add('guest-old'); + service.emitMessage( + SyncMessage.fromJson('{"t":"join","ts":0,"pid":"guest-old","name":"Old","host":false,"v":2}'), + ); + // A guest the relay no longer reports as connected. + service.emitMessage(SyncMessage.join(peerId: 'guest-gone', displayName: 'Gone', isHost: false)); + await _flushProviderEvents(); + + Participant byId(String id) => provider.participants.singleWhere((p) => p.peerId == id); + expect(provider.canTransferHostTo(byId('guest-new')), isTrue); + expect(provider.canTransferHostTo(byId('guest-old')), isFalse); + expect(provider.canTransferHostTo(byId('guest-gone')), isFalse); + expect(provider.canTransferHostTo(byId(service.myPeerId!)), isFalse); + + // Ineligible targets never reach the relay. + provider.transferHost(byId('guest-old')); + expect(service.transferRequests, isEmpty); + + // A guest can never initiate a transfer. + service.emitHostChanged('guest-new'); + await _flushProviderEvents(); + expect(provider.isHost, isFalse); + expect(provider.canTransferHostTo(byId('guest-old')), isFalse); + }); + }); }