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

feat(watch-together): let the host hand the room to another participant

Browse Source 
close #2183

Reusing a room means whoever enters first becomes the host, and moving the
session to someone else's library meant everyone leaving and re-entering in
the right order. The host can now promote any connected guest from the
participant list (Watch Together screen and the in-player session sheet),
with playback, control mode, and reconnect identities carried across.

The relay owns the swap: a new `transferHost` message validates the sender
is the live host and the target a connected modern-protocol guest, then
swaps `HostPeerID` and the reconnect verifiers (each peer keeps its own
token), persists the room, and broadcasts `hostChanged` to every peer.

Clients flip roles in place: the controller swaps its role engine while
keeping the session, message queue, and player attachment. A promoted guest
seeds the coordinator with the room's last known intent (a paused room
stays paused) and the known-peer roster so the fresh epoch re-gates
instead of solo-starting; a demoted host falls back to a reconciler with a
fresh clock sync and asks the new authority for state. Guests re-pin the
host identity, reset their sequence, and re-converge their clocks.

A rejected transfer (`not_host` / `peer_not_found`) surfaces as a toast
instead of tearing the session down. Pre-transfer app builds in the room
ignore `hostChanged` and stop following the room at the next transfer;
targets on an old sync protocol are not offered the action.
This commit is contained in:
edde746
2026-08-31 08:55:45 +02:00
parent 44773fd742
commit 89d54d8f1d
41 changed files with 1383 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/i18n/az.i18n.json
+7
View File
@@ -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",
lib/i18n/bg.i18n.json
+7
View File
@@ -1587,6 +1587,13 @@
"host": "Организатор",
"hostBadge": "ОРГАНИЗАТОР",
"youAreHost": "Вие сте организаторът",
"makeHost": "",
"makeHostQuestion": "",
"makeHostConfirm": "",
"transfer": "",
"hostChangedTo": "",
"youAreNowHost": "",
"hostTransferFailed": "",
"watchingWithOthers": "Гледате с други",
"endSession": "Край на сесията",
"leaveSession": "Напусни сесията",
lib/i18n/da.i18n.json
+7
View File
@@ -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",
lib/i18n/de.i18n.json
+7
View File
@@ -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",
lib/i18n/en.i18n.json
+7
View File
@@ -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",
lib/i18n/es.i18n.json
+7
View File
@@ -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",
lib/i18n/fr.i18n.json
+7
View File
@@ -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",
lib/i18n/hu.i18n.json
+7
View File
@@ -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",
lib/i18n/it.i18n.json
+7
View File
@@ -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",
lib/i18n/ja.i18n.json
+7
View File
@@ -1577,6 +1577,13 @@
"host": "ホスト",
"hostBadge": "ホスト",
"youAreHost": "あなたはホストです",
"makeHost": "",
"makeHostQuestion": "",
"makeHostConfirm": "",
"transfer": "",
"hostChangedTo": "",
"youAreNowHost": "",
"hostTransferFailed": "",
"watchingWithOthers": "他の人と視聴中",
"endSession": "セッションを終了",
"leaveSession": "セッションを退出",
lib/i18n/kk.i18n.json
+7
View File
@@ -1587,6 +1587,13 @@
"host": "Ұйымдастырушы",
"hostBadge": "ҰЙЫМДАСТЫРУШЫ",
"youAreHost": "Сіз ұйымдастырушысыз",
"makeHost": "",
"makeHostQuestion": "",
"makeHostConfirm": "",
"transfer": "",
"hostChangedTo": "",
"youAreNowHost": "",
"hostTransferFailed": "",
"watchingWithOthers": "Басқалармен бірге көрілуде",
"endSession": "Сеансты аяқтау",
"leaveSession": "Сеанстан шығу",
lib/i18n/ko.i18n.json
+7
View File
@@ -1577,6 +1577,13 @@
"host": "호스트",
"hostBadge": "호스트",
"youAreHost": "호스트입니다",
"makeHost": "",
"makeHostQuestion": "",
"makeHostConfirm": "",
"transfer": "",
"hostChangedTo": "",
"youAreNowHost": "",
"hostTransferFailed": "",
"watchingWithOthers": "다른 사람과 함께 시청 중",
"endSession": "세션 종료",
"leaveSession": "세션 나가기",
lib/i18n/nb.i18n.json
+7
View File
@@ -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",
lib/i18n/nl.i18n.json
+7
View File
@@ -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",
lib/i18n/pl.i18n.json
+7
View File
@@ -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ę",
lib/i18n/pt.i18n.json
+7
View File
@@ -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",
lib/i18n/ru.i18n.json
+7
View File
@@ -1607,6 +1607,13 @@
"host": "Организатор",
"hostBadge": "ОРГАНИЗАТОР",
"youAreHost": "Вы — организатор",
"makeHost": "",
"makeHostQuestion": "",
"makeHostConfirm": "",
"transfer": "",
"hostChangedTo": "",
"youAreNowHost": "",
"hostTransferFailed": "",
"watchingWithOthers": "Совместный просмотр",
"endSession": "Завершить сессию",
"leaveSession": "Покинуть сессию",
lib/i18n/strings.g.dart
+1 -1
View File
@@ -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
lib/i18n/strings_en.g.dart
+30 -2
View File
@@ -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',
lib/i18n/sv.i18n.json
+7
View File
@@ -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",
lib/i18n/tr.i18n.json
+7
View File
@@ -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",
lib/i18n/uz.i18n.json
+7
View File
@@ -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",
lib/i18n/zh-Hant.i18n.json
+7
View File
@@ -1577,6 +1577,13 @@
"host": "主持人",
"hostBadge": "主持人",
"youAreHost": "您是主持人",
"makeHost": "",
"makeHostQuestion": "",
"makeHostConfirm": "",
"transfer": "",
"hostChangedTo": "",
"youAreNowHost": "",
"hostTransferFailed": "",
"watchingWithOthers": "與他人一起觀看",
"endSession": "結束工作階段",
"leaveSession": "離開工作階段",
lib/i18n/zh.i18n.json
+7
View File
@@ -1577,6 +1577,13 @@
"host": "主持人",
"hostBadge": "主持人",
"youAreHost": "你是主持人",
"makeHost": "",
"makeHostQuestion": "",
"makeHostConfirm": "",
"transfer": "",
"hostChangedTo": "",
"youAreNowHost": "",
"hostTransferFailed": "",
"watchingWithOthers": "与他人一起观看",
"endSession": "结束会话",
"leaveSession": "离开会话",
lib/watch_together/providers/watch_together_provider.dart
+111 -1
View File
@@ -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<String, int> _lastActionEventMs = {};
@@ -100,6 +104,7 @@ class WatchTogetherProvider with ChangeNotifier {
StreamSubscription<SyncMessage>? _messageSubscription;
StreamSubscription<PeerError>? _errorSubscription;
StreamSubscription<void>? _sessionEndedSubscription;
StreamSubscription<String>? _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 {
lib/watch_together/screens/watch_together_screen.dart
+29 -1
View File
@@ -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<void> _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 {
lib/watch_together/services/clock_sync.dart
+18
View File
@@ -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);
lib/watch_together/services/guest_playback_reconciler.dart
+5
View File
@@ -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
// ---------------------------------------------------------------------
lib/watch_together/services/host_playback_coordinator.dart
+12
View File
@@ -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<void>? 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;
lib/watch_together/services/relay_protocol.g.dart
+4
View File
@@ -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;
lib/watch_together/services/watch_together_controller.dart
+113 -24
View File
@@ -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<StreamSubscription<dynamic>> _subscriptions = [];
Future<void> _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(
lib/watch_together/services/watch_together_peer_service.dart
+33
View File
@@ -82,6 +82,7 @@ class WatchTogetherPeerService with KeepaliveMixin {
final _errorController = StreamController<PeerError>.broadcast();
final _connectionStateController = StreamController<bool>.broadcast();
final _sessionEndedController = StreamController<void>.broadcast();
final _hostChangedController = StreamController<String>.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<void> 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<String> 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();
lib/watch_together/widgets/watch_together_overlay.dart
+55 -24
View File
@@ -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<void> _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<ParticipantNotification
ParticipantEventType.buffering => 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),
relay_protocol.json
+7 -3
View File
@@ -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,
server/main.go
+74
View File
@@ -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"})
server/main_test.go
+265
View File
@@ -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{})
server/relay_protocol_gen.go
+4
View File
@@ -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
test/watch_together/watch_together_controller_test.dart
+77
View File
@@ -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();
});
});
});
}
test/watch_together/watch_together_overlay_test.dart
+64
View File
@@ -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<void> _openLeaveConfirmation(WidgetTester tester, _OverlayHarness harness) async {
@@ -178,8 +228,22 @@ class _FakeWatchTogetherProvider extends WatchTogetherProvider {
@override
List<Participant> get participants => [
Participant(peerId: 'local', displayName: 'Local viewer', isHost: isHostValue),
...extraParticipants,
];
/// Participants appended after the local viewer row.
List<Participant> extraParticipants = const [];
/// Transfer requests forwarded by the sheet.
final List<Participant> 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;
test/watch_together/watch_together_peer_service_test.dart
+149
View File
@@ -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<String>();
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<String>();
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 = <String>[];
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<void>.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 = <PeerError>[];
final errorSubscription = service.onError.listen(errors.add);
addTearDown(errorSubscription.cancel);
final changed = Completer<String>();
final subscription = service.onHostChanged.listen((peerId) {
if (!changed.isCompleted) changed.complete(peerId);
});
addTearDown(subscription.cancel);
final reconnected = Completer<void>();
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);
});
}
test/watch_together/watch_together_provider_test.dart
+178 -2
View File
@@ -32,8 +32,13 @@ class _FakeWatchTogetherPeerService extends WatchTogetherPeerService {
final _messageController = StreamController<SyncMessage>.broadcast();
final _errorController = StreamController<PeerError>.broadcast();
final _sessionEndedController = StreamController<void>.broadcast();
final _hostChangedController = StreamController<String>.broadcast();
final List<SyncMessage> broadcasts = [];
final List<(String, SyncMessage)> directMessages = [];
final List<String> transferRequests = [];
/// Extra peers reported as connected on top of the modeled host link.
final Set<String> connectedGuests = {};
String? _sessionId;
String? _myPeerId;
@@ -59,6 +64,9 @@ class _FakeWatchTogetherPeerService extends WatchTogetherPeerService {
@override
Stream<void> get onSessionEnded => _sessionEndedController.stream;
@override
Stream<String> get onHostChanged => _hostChangedController.stream;
@override
String? get sessionId => _sessionId;
@@ -72,7 +80,10 @@ class _FakeWatchTogetherPeerService extends WatchTogetherPeerService {
bool get isHost => _isHost;
@override
List<String> get connectedPeers => !_isHost && _sessionId != null && _hostConnected ? ['wt-$_sessionId'] : const [];
List<String> get connectedPeers => [
if (!_isHost && _sessionId != null && _hostConnected) 'wt-$_sessionId',
...connectedGuests,
];
@override
Future<String> 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 = <ParticipantEvent>[];
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 = <ParticipantEvent>[];
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 = <ParticipantEvent>[];
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);
});
});
}