diff --git a/apps/desktop/lib/src/queue/queue_controller.dart b/apps/desktop/lib/src/queue/queue_controller.dart index 5db2c32..242d1ec 100644 --- a/apps/desktop/lib/src/queue/queue_controller.dart +++ b/apps/desktop/lib/src/queue/queue_controller.dart @@ -43,7 +43,7 @@ class QueueController { QueueController(this.coord); /// Queued items' pictures by hash, from `/i/`. - late final PictureCache thumbs = PictureCache(coord.sync, + late final PictureCache thumbs = PictureCache(coord.connection.client, route: 'i', keep: RelayConstants.maxQueueItems + 8, live: (s) => [for (final q in s.queue) q.thumb]); /// Up next, in order (empty when nothing is queued or not in a server). @@ -77,12 +77,12 @@ class QueueController { String _continuationFor = ''; final _random = Random.secure(); - String get _me => coord.sync.memberId; + String get _me => coord.connection.memberId; void attach() { if (_attached) return; _attached = true; - coord.sync.states.listen(_onState); + coord.connection.states.listen(_onState); effect(() { final next = coord.nextEpisode.value; final plan = coord.jellyfinPlan.value; @@ -121,7 +121,7 @@ class QueueController { final item = built.item; if (item.key.isNotEmpty) _local[item.key] = (pick: pick, at: DateTime.now()); final thumb = built.thumb; - coord.sync.sendQueue(QueueIntent.add(item, next: next, thumb: thumb == null ? '' : base64Encode(thumb))); + coord.connection.sendQueue(QueueIntent.add(item, next: next, thumb: thumb == null ? '' : base64Encode(thumb))); coord.notify(AddedToUpNext(item.title)); return null; } @@ -288,12 +288,12 @@ class QueueController { void move(String id, int to) { coord.markInput(); - coord.sync.sendQueue(QueueIntent.move(id, to)); + coord.connection.sendQueue(QueueIntent.move(id, to)); } void remove(String id) { coord.markInput(); - coord.sync.sendQueue(QueueIntent.remove(id)); + coord.connection.sendQueue(QueueIntent.remove(id)); } /// Plays [id] now: this app puts it up when it is the one that can, @@ -306,7 +306,7 @@ class QueueController { if (queuePerformer(item, r.members).id == _me) { _perform(item); } else { - coord.sync.sendQueue(QueueIntent.play(id)); + coord.connection.sendQueue(QueueIntent.play(id)); } } @@ -388,7 +388,7 @@ class QueueController { if (perf.id == null) { final why = perf.why; if (why != null && queueJanitor(r.members) == _me && _skipped.add(item.id)) { - coord.sync.sendQueue(QueueIntent.remove(item.id, why: why)); + coord.connection.sendQueue(QueueIntent.remove(item.id, why: why)); } return; } @@ -461,7 +461,7 @@ class QueueController { } if (error != null) { debugPrint('queue: ${item.id} cannot go up: $error'); - coord.sync.sendQueue(QueueIntent.remove(item.id, why: "couldn't play it ($error)")); + coord.connection.sendQueue(QueueIntent.remove(item.id, why: "couldn't play it ($error)")); } }()); } @@ -615,7 +615,7 @@ class QueueController { final head = r?.queue.firstOrNull; final relay = head?.media.relay; if (r == null || head == null || relay == null || relay.fetch) return; - final key = '${head.id}|${coord.sync.token}'; + final key = '${head.id}|${coord.connection.token}'; if (key == _fedWarm) return; switch (head.how) { case QueueHow.through: diff --git a/apps/desktop/lib/src/queue/queue_ui.dart b/apps/desktop/lib/src/queue/queue_ui.dart index 70f3cce..c5482cb 100644 --- a/apps/desktop/lib/src/queue/queue_ui.dart +++ b/apps/desktop/lib/src/queue/queue_ui.dart @@ -344,7 +344,7 @@ class UpNextPreview extends StatelessWidget { const picture = _width - 2 * _pad; if (item != null) { final r = queue.coord.serverState.value; - final selfId = queue.coord.sync.memberId; + final selfId = queue.coord.connection.memberId; final more = queue.items.value.length - 1; final ready = warm?.id == item.id ? warm!.state : null; return Column( @@ -449,7 +449,7 @@ class _KeyCap extends StatelessWidget { String? performerNote(QueueController queue, QueueItem item, StateBroadcast? r) { if (r == null) return null; final id = queuePerformer(item, r.members).id; - if (id == null || id == queue.coord.sync.memberId) return null; + if (id == null || id == queue.coord.connection.memberId) return null; final name = r.members.where((m) => m.id == id).firstOrNull?.name; return name == null ? null : '$name puts it up'; } @@ -841,7 +841,7 @@ class _QueuePanelState extends State { final cont = _q.shownContinuation.value; final warm = _q.warm.value; final r = _q.coord.serverState.value; - final selfId = _q.coord.sync.memberId; + final selfId = _q.coord.connection.memberId; return Column( mainAxisSize: MainAxisSize.min, crossAxisAlignment: CrossAxisAlignment.stretch, @@ -1558,7 +1558,7 @@ class UpNextCard extends StatelessWidget { final item = c.item; final left = c.secondsLeft; final r = queue.coord.serverState.value; - final selfId = queue.coord.sync.memberId; + final selfId = queue.coord.connection.memberId; final waiting = left == null ? performerNote(queue, item, r) : null; return AppGlass( key: const ValueKey('up-next-card'), diff --git a/apps/desktop/lib/src/session/connection.dart b/apps/desktop/lib/src/session/connection.dart new file mode 100644 index 0000000..b4cbeb0 --- /dev/null +++ b/apps/desktop/lib/src/session/connection.dart @@ -0,0 +1,532 @@ +import 'dart:async'; +import 'dart:io'; + +import 'package:flutter/foundation.dart' show debugPrint; +import 'package:signals/signals_core.dart'; +import 'package:sync_relay_core/core.dart'; + +import '../screens/home/ui.dart' show appName, displayVersion; +import '../state/settings.dart'; +import '../sync/sync_client.dart'; +import 'app_events.dart'; + +/// Thrown when a connect is superseded by cancel/leave before it finished. +class ConnectCancelled implements Exception { + const ConnectCancelled(); + @override + String toString() => 'connection cancelled'; +} + +class ConnectParams { + final String host; + final int port; + final String secret; + final String name; + final String avatar; + final String mediaPath; + final String linkUrl; + /// Sent in hello by the hosting app's own client (see AppConfig.hostKey). + final String hostKey; + /// Put up with a `swap` right after joining (a Jellyfin ref or a link); + /// wins over [linkUrl]. + final HelloMedia? media; + const ConnectParams({ + required this.host, + required this.port, + required this.secret, + required this.name, + this.avatar = '', + this.mediaPath = '', + this.linkUrl = '', + this.hostKey = '', + this.media, + }); + + ConnectParams withMedia(HelloMedia? media) => ConnectParams( + host: host, + port: port, + secret: secret, + name: name, + avatar: avatar, + mediaPath: mediaPath, + linkUrl: media == null ? linkUrl : '', + hostKey: hostKey, + media: media, + ); + + ConnectParams withProfile({String? name, String? avatar}) => ConnectParams( + host: host, + port: port, + secret: secret, + name: name ?? this.name, + avatar: avatar ?? this.avatar, + mediaPath: mediaPath, + linkUrl: linkUrl, + hostKey: hostKey, + media: media, + ); +} + +/// Where the [Connection] is. Each change is a new value (a repeat is +/// notified too). +sealed class ConnState { + const ConnState(); + + /// The reconnect loop runs ([ConnLost] or [ConnReconnecting]). + bool get retrying => false; +} + +/// Not connected and not trying: before the first connect, after leave, +/// cancel or "stop reconnecting". +final class ConnClosed extends ConnState { + const ConnClosed(); + @override + String toString() => 'closed'; +} + +/// The first dial of a session (hello sent, waiting for welcome). +final class ConnConnecting extends ConnState { + const ConnConnecting(); + @override + String toString() => 'connecting'; +} + +/// Welcome received. +final class ConnOnline extends ConnState { + const ConnOnline(); + @override + String toString() => 'online'; +} + +/// Dropped (the socket closed, or no state for [Connection.silenceLimit]); +/// the next attempt is scheduled. +final class ConnLost extends ConnState { + const ConnLost(); + @override + bool get retrying => true; + @override + String toString() => 'lost'; +} + +/// Reconnect attempt [attempt] (1 is the first after the drop) is +/// dialling; after it failed it stays until the next one, unless the +/// socket's close said [ConnLost] meanwhile. +final class ConnReconnecting extends ConnState { + final int attempt; + const ConnReconnecting(this.attempt); + @override + bool get retrying => true; + @override + String toString() => 'reconnecting($attempt)'; +} + +/// The first dial failed; nothing retries. [code]: `version` or `secret` +/// when the server said so, else empty (offline, timeout, anything else); +/// [message]: in plain words ([connectErrorText]). +final class ConnRefused extends ConnState { + final String code; + final String message; + const ConnRefused(this.code, this.message); + @override + String toString() => 'refused($code: $message)'; +} + +/// Another socket took this member over. Not produced yet (protocol 16 +/// adds the `replaced` refusal). +final class ConnReplaced extends ConnState { + const ConnReplaced(); + @override + String toString() => 'replaced'; +} + +/// A failed dial in plain words. +String connectErrorText(Object e) { + final msg = '$e'; + if (msg.contains('proto mismatch')) return 'host runs a different version — update the app'; + if (msg.contains('wrong secret')) return 'wrong server password (401)'; + if (msg.contains('SocketException') || msg.contains('Connection refused')) { + return 'host offline — check address and port'; + } + return 'connect failed: $e'; +} + +ConnRefused _refused(Object e) { + final msg = '$e'; + final code = msg.contains('proto mismatch') + ? 'version' + : msg.contains('wrong secret') + ? 'secret' + : ''; + return ConnRefused(code, connectErrorText(e)); +} + +/// Builds the `hello` for [params]. Injectable so later protocol versions +/// add fields without changing [Connection]. +typedef HelloBuilder = HelloMessage Function(ConnectParams params); + +/// Today's hello. +HelloMessage defaultHello(ConnectParams params) => HelloMessage( + proto: RelayConstants.protoVersion, + secret: params.secret, + user: HelloUser(name: params.name, avatar: params.avatar), + media: const HelloMedia(name: '', size: 0, duration: 0), + hostKey: params.hostKey, + ); + +/// The session's part of a reconnect: called right before an attempt +/// dials (read what to restore then; the welcome's state replaces it). The +/// function it returns runs once that attempt is online; [stale] turns +/// true when a cancel, retry, leave or new connect superseded it. An error +/// there counts as a failed attempt. +typedef ReconnectHandler = Future Function(WelcomeMessage welcome, bool Function() stale) Function(); + +/// The link to the server: the [SyncClient] socket, the first connect, +/// reconnects ([reconnectDelay] between attempts, the previous token +/// offered as `resume`, epochs that make a cancelled attempt stale), the +/// "no state for [silenceLimit] = lost" policy, the 1 s heartbeat (clock +/// ping, diagnostics), the connection lost/back events and [state]. What +/// a reconnect restores in the session is the [onReconnect] handler's. +class Connection { + final SyncClient client; + /// Where [ConnectionLost] / [ConnectionBack] go (the Notifier). + final void Function(AppEvent event) _notify; + /// Builds each dial's hello. + final HelloBuilder hello; + /// Wait before each reconnect attempt. + final Duration reconnectDelay; + /// No state broadcast for this long means the connection is lost. + final Duration silenceLimit; + + Connection( + {void Function(AppEvent event)? notify, + SyncClient? client, + this.hello = defaultHello, + this.reconnectDelay = const Duration(milliseconds: 1500), + this.silenceLimit = const Duration(seconds: 5)}) + : _notify = notify ?? ((_) {}), + client = client ?? SyncClient() { + _subs = [ + this.client.status.listen(_onStatus), + this.client.states.listen((_) => _lastStateAt = DateTime.now().toUtc()), + ]; + } + + late final List _subs; + + final _state = signal(const ConnClosed()); + /// Every change is notified, a repeat too. + ReadonlySignal get state => _state; + + /// Set by the session (see [ReconnectHandler]). + ReconnectHandler? onReconnect; + + /// This app's log lines to the server's session log (Settings: "Share + /// diagnostics with the server"), and "Mark a problem". Runs from + /// [begin] to [close]. + late final DiagUploader diag = DiagUploader( + target: _diagTarget, version: '$appName ${displayVersion()} on ${Platform.operatingSystem}'); + + DiagTarget? _diagTarget() { + final uri = client.wsUri; + if (client.token.isEmpty || uri == null || !client.clockSync.hasSample) return null; + return (base: DiagUploader.baseOf(uri), token: client.token, offsetMs: client.clockSync.offsetMs); + } + + /// No state broadcast for [silenceLimit] (checked every second while not + /// closed): what the server said is stale. Synchronous, before [state] + /// goes lost. + Stream get silence => _silence.stream; + final _silence = StreamController.broadcast(sync: true); + + ConnectParams? _params; + /// The last connect's parameters (with [updateProfile]'s changes); a + /// reconnect dials with them. + ConnectParams? get params => _params; + + // Bumped by every connect/leave/cancel/retry: stale attempts no-op. + int _epoch = 0; + Timer? _reconnectTimer; + Timer? _heartbeat; + // Set by close/cancel: no reconnect, no lost detection. + bool _closed = false; + bool _everOnline = false; + /// The connection-lost sound played; the reconnected one answers it once. + bool _lostSounded = false; + // Token of the current connection, offered as `resume` on reconnect. + String _resumeToken = ''; + DateTime? _lastStateAt; + int _attempt = 0; + + // The socket and what the server said, read through. + Stream get states => client.states; + /// Feed entries, chat included. + Stream get feeds => client.feeds; + Stream get nudges => client.nudges; + /// `err` frames after the welcome. + Stream get errors => client.errors; + String get token => client.token; + String get memberId => client.memberId; + /// This member's `/j/...` URL from `welcome`. + String? get relayUrl => client.relayUrl; + StateBroadcast? get lastState => client.lastState; + /// Now on the server's clock (the ping/pong clock offset). + DateTime serverNow() => client.serverNow(); + + void sendReady(bool ready, {String source = 'manual'}) => client.sendReady(ready, source: source); + void sendForce(String target, bool ready) => client.sendForce(target, ready); + void sendSeek(double position, {int seq = 0}) => client.sendSeek(position, seq: seq); + void sendSwap(HelloMedia media, {String? queueId}) => client.sendSwap(media, queueId: queueId); + void sendQueue(QueueIntent intent) => client.sendQueue(intent); + void sendReport({ + required double position, + required double buffered, + required int eventId, + double duration = 0, + int mediaId = 0, + Map? jf, + String error = '', + }) => + client.sendReport( + position: position, + buffered: buffered, + eventId: eventId, + duration: duration, + mediaId: mediaId, + jf: jf, + error: error); + void sendChat(String text) => client.sendChat(text); + void sendArt(int mediaId, {String? poster, String? banner}) => + client.sendArt(mediaId, poster: poster, banner: banner); + void sendOut(String target, bool out) => client.sendOut(target, out); + void sendNudge(String target) => client.sendNudge(target); + + /// Name and/or picture change for everyone now; reconnects send them too. + void updateProfile({String? name, String? avatar}) { + _params = _params?.withProfile(name: name, avatar: avatar); + client.sendProfile(name: name, avatar: avatar); + } + + /// A new epoch: attempts in flight become stale, a scheduled reconnect + /// is dropped. Returns it. + int invalidate() { + _epoch++; + _reconnectTimer?.cancel(); + _reconnectTimer = null; + return _epoch; + } + + /// False once a cancel, retry, leave or new connect followed [epoch]. + bool isCurrent(int epoch) => epoch == _epoch; + + /// A new session with [params]: forgets the last one's resume token and + /// starts the diagnostics. [dial] connects. + void begin(ConnectParams params) { + _params = params; + _closed = false; + _everOnline = false; + _lostSounded = false; + _resumeToken = ''; + _attempt = 0; + diag + ..share = AppSettings.instance.shareDiagnostics.peek() + ..start(); + } + + /// The first dial of the session [begin] started. Throws + /// [ConnectCancelled] when [epoch] went stale meanwhile (whoever bumped + /// it also closed the socket); any other failure leaves [state] refused. + Future dial(int epoch) async { + final params = _params!; + _set(const ConnConnecting()); + try { + final welcome = await client.connectWith(wsUri: _wsUri(params), hello: hello(params)); + if (epoch != _epoch) throw const ConnectCancelled(); + _online(); + return welcome; + } catch (e) { + if (epoch != _epoch) throw const ConnectCancelled(); + debugPrint('connect failed: $e'); + _set(_refused(e)); + rethrow; + } + } + + static Uri _wsUri(ConnectParams p) => Uri.parse('ws://${p.host}:${p.port}/sync'); + + /// [hello] offering the previous connection's token. + HelloMessage _resumeHello(ConnectParams params) { + final h = hello(params); + return HelloMessage( + proto: h.proto, secret: h.secret, user: h.user, media: h.media, hostKey: h.hostKey, resume: _resumeToken); + } + + void _online() { + _attempt = 0; + _everOnline = true; + if (_lostSounded) { + _lostSounded = false; + _notify(const ConnectionBack()); + } + _resumeToken = client.token; + _lastStateAt = DateTime.now().toUtc(); + _set(const ConnOnline()); + _startHeartbeat(); + } + + void _set(ConnState s) => _state.set(s, force: true); + + void _onStatus(SyncStatus s) { + // Before the first welcome a close is just the dial failing. + if (s != SyncStatus.closed || _closed || !_everOnline) return; + _dropped(); + } + + /// Lost (said again while already retrying), the lost sound once per + /// drop, and a reconnect unless one is already scheduled or dialling. + void _dropped() { + if (_state.peek().retrying) { + _set(const ConnLost()); + } else { + scheduleReconnect(); + } + _soundLost(); + } + + void _startHeartbeat() { + _heartbeat?.cancel(); + _heartbeat = Timer.periodic(const Duration(seconds: 1), (_) => _beat()); + } + + /// Every second while connected: a clock sample, the diagnostics, and + /// the silence check. + void _beat() { + final now = DateTime.now().toUtc(); + if (client.token.isNotEmpty) client.sendPing(); + diag + ..share = AppSettings.instance.shareDiagnostics.peek() + ..tick(); + final last = _lastStateAt; + if (!_closed && last != null && now.difference(last) >= silenceLimit) { + _silence.add(null); + if (!_state.peek().retrying) _dropped(); + } + } + + void _soundLost() { + if (!_everOnline || _lostSounded) return; + _lostSounded = true; + _notify(const ConnectionLost()); + } + + /// Reconnect after [reconnectDelay] ([state] lost), unless closed, + /// never online (unless [manual]) or already retrying. + void scheduleReconnect({bool manual = false}) { + if (_closed || _params == null) return; + if (!manual && !_everOnline) return; + if (_state.peek().retrying) return; + _set(const ConnLost()); + _armReconnect(); + } + + void _armReconnect() { + final epoch = _epoch; + _reconnectTimer?.cancel(); + _reconnectTimer = Timer(reconnectDelay, () => _reconnect(epoch)); + } + + /// One reconnect attempt. An attempt from an older epoch must not touch + /// state (it would kill the newer connection and leave a ghost member). + Future _reconnect(int epoch) async { + _reconnectTimer = null; + bool stale() => epoch != _epoch; + final params = _params; + if (stale() || params == null || _closed) return; + final after = onReconnect?.call(); + try { + _set(ConnReconnecting(++_attempt)); + final welcome = await client.connectWith(wsUri: _wsUri(params), hello: _resumeHello(params)); + // Whoever bumped the epoch also closed (or replaced) the socket. + if (stale()) return; + _online(); + if (after != null) await after(welcome, stale); + } catch (e) { + if (stale()) return; + debugPrint('reconnect failed: $e'); + _retryAfterFailure(e); + } + } + + void _retryAfterFailure(Object e) { + if (_closed) { + _set(const ConnClosed()); + } else if (!_everOnline) { + // Only a manual retry before any welcome gets here. + _set(_refused(e)); + } else { + if (!_state.peek().retrying) _set(const ConnLost()); + _armReconnect(); + } + } + + /// "Stop reconnecting": no more attempts; one dialling is aborted. + void cancelReconnect() { + _halt(); + unawaited(client.disconnect().catchError((_) {})); + } + + /// Cancel during setup: aborts the dial. + Future cancelConnect() async { + _halt(); + try { + await client.disconnect(); + } catch (_) {} + } + + void _halt() { + _closed = true; + invalidate(); + _heartbeat?.cancel(); + _heartbeat = null; + _set(const ConnClosed()); + } + + /// "Retry": a reconnect after [reconnectDelay] whatever the state. + /// False without a session to retry. + bool retry() { + if (_params == null) return false; + invalidate(); + _closed = false; + _set(const ConnClosed()); + scheduleReconnect(manual: true); + return true; + } + + /// No more reconnects or silence checks, [state] closed; the socket + /// stays until [close] (the session's last words still go out). + void stop() { + _closed = true; + _reconnectTimer?.cancel(); + _reconnectTimer = null; + _heartbeat?.cancel(); + _heartbeat = null; + _set(const ConnClosed()); + } + + /// [stop], the last diagnostics (before the token goes), the socket. + Future close() async { + stop(); + await diag.stop(); + try { + await client.disconnect(); + } catch (_) {} + } + + Future dispose() async { + stop(); + for (final s in _subs) { + await s.cancel(); + } + await _silence.close(); + await client.dispose(); + } +} diff --git a/apps/desktop/lib/src/session/session_coordinator.dart b/apps/desktop/lib/src/session/session_coordinator.dart index ac2a515..2481a5a 100644 --- a/apps/desktop/lib/src/session/session_coordinator.dart +++ b/apps/desktop/lib/src/session/session_coordinator.dart @@ -17,15 +17,14 @@ import '../playback/playback_controller.dart'; import '../platform/ports/link_resolver.dart'; import '../playback/yt_dlp_app.dart' show AppYtDlp; import '../schedule/anilist.dart'; -import '../screens/home/ui.dart' show appName, displayVersion; import '../screens/home/picked_media.dart'; import '../sfx/sfx.dart' show NoopSfxService, SfxService; import '../state/settings.dart'; import '../sync/avatar_cache.dart'; import '../sync/state_diff.dart'; -import '../sync/sync_client.dart'; import '../sync/user_seek.dart'; import '../queue/queue_controller.dart'; +import 'connection.dart'; import 'delivery.dart'; import 'health.dart'; import 'idle.dart'; @@ -35,59 +34,15 @@ import 'now_playing.dart'; import 'now_playing_info.dart'; import 'switch_trace.dart'; +export 'connection.dart' show ConnectCancelled, ConnectParams; export 'notifier.dart' show ToastItem; final _log = createLogger('session'); -/// Thrown when a connect is superseded by cancel/leave before it finished. -class ConnectCancelled implements Exception { - const ConnectCancelled(); - @override - String toString() => 'connection cancelled'; -} - /// Display name for a link: last path segment, decoded, never the query /// (Jellyfin-style links carry api_key there and names are broadcast). String linkDisplayName(String url) => linkName(url); -class ConnectParams { - final String host; - final int port; - final String secret; - final String name; - final String avatar; - final String mediaPath; - final String linkUrl; - /// Sent in hello by the hosting app's own client (see AppConfig.hostKey). - final String hostKey; - /// Put up with a `swap` right after joining (a Jellyfin ref or a link); - /// wins over [linkUrl]. - final HelloMedia? media; - const ConnectParams({ - required this.host, - required this.port, - required this.secret, - required this.name, - this.avatar = '', - this.mediaPath = '', - this.linkUrl = '', - this.hostKey = '', - this.media, - }); - - ConnectParams withMedia(HelloMedia? media) => ConnectParams( - host: host, - port: port, - secret: secret, - name: name, - avatar: avatar, - mediaPath: mediaPath, - linkUrl: media == null ? linkUrl : '', - hostKey: hostKey, - media: media, - ); -} - /// The end-of-episode countdown (auto-advance). Every client with the /// setting on shows it; only [swapper] puts the next episode up. class AutoAdvance { @@ -146,6 +101,17 @@ final class _JellyfinOrigin extends _RelayOrigin { const _JellyfinOrigin(this.link); } +/// The health line a connection change writes; null writes none (closed: +/// whoever closed says why). +HealthStatus? connectionHealth(ConnState s) => switch (s) { + ConnConnecting() => const HealthStatus(HealthSeverity.info, 'connecting…'), + ConnOnline() => const HealthStatus(HealthSeverity.ok, 'connected'), + ConnLost() => const HealthStatus(HealthSeverity.warn, 'connection lost — retrying…'), + ConnReconnecting() => const HealthStatus(HealthSeverity.info, 'reconnecting…'), + ConnRefused(:final message) => HealthStatus(HealthSeverity.error, message), + ConnReplaced() || ConnClosed() => null, + }; + String? stateMediaUrl(StateBroadcast state) { final media = state.media; if (media == null) return null; @@ -167,17 +133,11 @@ class SessionCoordinator { Future<({String name, int size})?> Function(String path)? hostFile; /// The app's yt-dlp and link probes (tests pass their own). final LinkResolver ytDlp; - final SyncClient sync = SyncClient(); - /// This app's log lines to the server's session log (Settings: "Share - /// diagnostics with the server"), and "Mark a problem". - late final DiagUploader diag = DiagUploader( - target: _diagTarget, version: '$appName ${displayVersion()} on ${Platform.operatingSystem}'); - - DiagTarget? _diagTarget() { - final uri = sync.wsUri; - if (sync.token.isEmpty || uri == null || !sync.clockSync.hasSample) return null; - return (base: DiagUploader.baseOf(uri), token: sync.token, offsetMs: sync.clockSync.offsetMs); - } + /// The link to the server (socket, reconnects, clock offset, its + /// [ConnState]); made in the composition root. + final Connection connection; + /// This app's log lines to the server's session log ([Connection.diag]). + DiagUploader get diag => connection.diag; /// "Mark a problem": a marker with [note] in the server's session log /// (and this app's log). False when not in a session: then only this @@ -197,11 +157,11 @@ class SessionCoordinator { /// Up next: the server's queue, adding to it, and putting items up. late final QueueController queue = QueueController(this); /// The server has a password (it fetches links only then). - bool get serverHasPassword => (_params?.secret ?? '').isNotEmpty; + bool get serverHasPassword => (connection.params?.secret ?? '').isNotEmpty; /// Member pictures by `avatarHash`, fetched from the server. - late final AvatarCache avatars = AvatarCache(sync); + late final AvatarCache avatars = AvatarCache(connection.client); /// The current media's poster and banner by `state.media.art` hash. - late final ArtCache art = ArtCache(sync); + late final ArtCache art = ArtCache(connection.client); final PlaybackController playback = PlaybackController(); /// Timings of each media change here (see [SwitchTrace]). final SwitchTrace switchTrace = SwitchTrace(); @@ -214,7 +174,8 @@ class SessionCoordinator { Signal> get feedLog => notifier.feedLog; /// Toasts on screen ([Notifier.toasts]). Signal> get toasts => notifier.toasts; - final reconnecting = signal(false); + /// The reconnect loop runs ([ConnState.retrying]). + late final Computed reconnecting = computed(() => connection.state.value.retrying); final starving = signal(null); /// Why this player cannot open the current media, in plain words /// ([playErrorText]); empty while it can. It rides in `report.error`, so @@ -264,9 +225,9 @@ class SessionCoordinator { late final Computed selfMember = computed(() { final r = serverState.value; - if (r == null || sync.memberId.isEmpty) return null; + if (r == null || connection.memberId.isEmpty) return null; for (final m in r.members) { - if (m.id == sync.memberId) return m; + if (m.id == connection.memberId) return m; } return null; }); @@ -279,9 +240,9 @@ class SessionCoordinator { StreamSubscription? _feedSub; StreamSubscription? _nudgeSub; StreamSubscription? _errSub; - StreamSubscription? _statusSub; + StreamSubscription? _silenceSub; + void Function()? _connectionHealth; - bool _manualClose = false; /// A solo session (set by SessionState): no idle unready / sit out, /// there is nobody to hold up. bool solo = false; @@ -303,9 +264,6 @@ class SessionCoordinator { // Last mediaId seen in a state; unlike [_mediaId] it survives a // reconnect, so only a real swap counts as a new video. int _seenMediaId = -1; - // Bumped by every connect/leave/cancel/retry: stale attempts no-op. - int _epoch = 0; - Timer? _reconnectTimer; int _mediaId = -1; // The mediaId whose media the player holds (or is opening); -1 for none. // Unlike [_mediaId] it survives a lost connection, so a reconnect to the @@ -350,14 +308,7 @@ class SessionCoordinator { // True while a media open is in flight: the player's duration may still // belong to the previous media then. bool _opening = false; - // Token of the current connection, offered as `resume` on reconnect. - String _resumeToken = ''; - DateTime? _lastStateAt; final IdleTracker _idle = IdleTracker(DateTime.now().toUtc()); - ConnectParams? _params; - bool _everConnected = false; - /// The connection-lost sound played; the reconnected one answers it once. - bool _lostSounded = false; Map? _openHeaders; // Jellyfin: the resolved ref, the mediaId it belongs to, a generation @@ -379,17 +330,33 @@ class SessionCoordinator { // Members' last `error`, by id, so a new one toasts once. Map _memberErrors = const {}; - /// [notifier] defaults to one playing [sfx] (tests). - SessionCoordinator( + /// [notifier] defaults to one playing [sfx], [connection] to one + /// telling that notifier (tests). + factory SessionCoordinator( {SfxService sfx = const NoopSfxService(), Notifier? notifier, + Connection? connection, JellyfinAccounts? jellyfin, TrackPrefs? trackPrefs, - LinkResolver? ytDlp}) - : notifier = notifier ?? Notifier(sfx: sfx), - ytDlp = ytDlp ?? AppYtDlp.instance, - jellyfin = jellyfin ?? JellyfinAccounts.instance, - trackPrefs = trackPrefs ?? TrackPrefs(); + LinkResolver? ytDlp}) { + final n = notifier ?? Notifier(sfx: sfx); + return SessionCoordinator._( + notifier: n, + connection: connection ?? Connection(notify: n.notify), + ytDlp: ytDlp ?? AppYtDlp.instance, + jellyfin: jellyfin ?? JellyfinAccounts.instance, + trackPrefs: trackPrefs ?? TrackPrefs(), + ); + } + + SessionCoordinator._( + {required this.notifier, + required this.connection, + required this.ytDlp, + required this.jellyfin, + required this.trackPrefs}) { + connection.onReconnect = _onReconnect; + } /// The session's first player: created here on connect, disposed by /// [leave]; each new media gets its own ([PlaybackController.openUrl]). @@ -434,7 +401,7 @@ class SessionCoordinator { final now = {for (final m in r.members) if (m.error.isNotEmpty) m.id: m.error}; for (final m in r.members) { final e = now[m.id]; - if (e != null && m.id != sync.memberId && _memberErrors[m.id] != e) { + if (e != null && m.id != connection.memberId && _memberErrors[m.id] != e) { notify(MemberCantPlay(m.name, e)); } } @@ -446,7 +413,7 @@ class SessionCoordinator { void _warnSlowSource(StateBroadcast r) { memberNames.observe(r); if (_slowWarnedFor == r.mediaId) return; - final warning = slowSourceWarning(r, sync.memberId); + final warning = slowSourceWarning(r, connection.memberId); if (warning == null) return; _slowWarnedFor = r.mediaId; notify(SlowSource(warning)); @@ -462,7 +429,7 @@ class SessionCoordinator { /// edge). The ready flag is never touched. Future reloadVideo() async { // Also allowed after a failed reload (player gone, session still on). - if (_params == null || _reloadingVideo) return; + if (connection.params == null || _reloadingVideo) return; _reloadingVideo = true; _playerReady = false; try { @@ -486,11 +453,10 @@ class SessionCoordinator { Future connect(ConnectParams params) async { // Own epoch first: a cancel during any await below makes this stale. - final epoch = ++_epoch; + final epoch = connection.invalidate(); await _teardown(silent: true); - if (epoch != _epoch) throw const ConnectCancelled(); - _params = params; - _manualClose = false; + if (!connection.isCurrent(epoch)) throw const ConnectCancelled(); + connection.begin(params); _wsHost = params.host; _wsPort = params.port; diff.reset(); @@ -507,9 +473,6 @@ class SessionCoordinator { _joinBackAt = null; _seenMediaId = -1; _wasPlayAllowed = false; - _everConnected = false; - _lostSounded = false; - _resumeToken = ''; _seekGate.reset(); _clearPendingSeek(); _lastEventId = -1; @@ -522,30 +485,13 @@ class SessionCoordinator { } markInput(); _subscribe(); - diag - ..share = AppSettings.instance.shareDiagnostics.peek() - ..start(); notifier.clear(); await initPlayer(); - if (epoch != _epoch) throw const ConnectCancelled(); - health.value = - const HealthStatus(HealthSeverity.info, 'connecting…'); - final wsUri = Uri.parse('ws://${params.host}:${params.port}/sync'); + if (!connection.isCurrent(epoch)) throw const ConnectCancelled(); + // Health follows the connection's state from here (connecting, + // connected; a failed dial leaves its reason). + final welcome = await connection.dial(epoch); try { - final welcome = await sync.connect( - wsUri: wsUri, - secret: params.secret, - name: params.name, - avatar: params.avatar, - hostKey: params.hostKey, - ); - // Whoever bumped the epoch (cancel/leave) also closed the socket. - if (epoch != _epoch) throw const ConnectCancelled(); - reconnecting.value = false; - _everConnected = true; - _resumeToken = sync.token; - health.value = const HealthStatus(HealthSeverity.ok, 'connected'); - _lastStateAt = DateTime.now().toUtc(); final url = welcome.mediaUrl; if (url != null && url.isNotEmpty) { await _openMedia(url); @@ -555,37 +501,46 @@ class SessionCoordinator { } final media = params.media ?? (params.linkUrl.isEmpty ? null : await mediaForLink(jellyfin, params.linkUrl)); - if (epoch != _epoch) throw const ConnectCancelled(); + if (!connection.isCurrent(epoch)) throw const ConnectCancelled(); if (media != null) _swap(media); _startTicker(); } catch (e) { - if (epoch != _epoch) throw const ConnectCancelled(); + if (!connection.isCurrent(epoch)) throw const ConnectCancelled(); debugPrint('connect failed: $e'); + connection.scheduleReconnect(); health.value = - HealthStatus(HealthSeverity.error, _friendlyError(e)); - _scheduleReconnect(); + HealthStatus(HealthSeverity.error, connectErrorText(e)); rethrow; } } void _subscribe() { queue.attach(); - _stateSub ??= sync.states.listen(_onState); - _feedSub ??= sync.feeds.listen(_onFeed); - _nudgeSub ??= sync.nudges.listen(_onNudge); - _errSub ??= sync.errors.listen((msg) { + _stateSub ??= connection.states.listen(_onState); + _feedSub ??= connection.feeds.listen(_onFeed); + _nudgeSub ??= connection.nudges.listen(_onNudge); + _errSub ??= connection.errors.listen((msg) { health.value = HealthStatus(HealthSeverity.warn, msg); }); - _statusSub ??= sync.status.listen((s) { - if (s == SyncStatus.closed && !_manualClose) { - health.value = const HealthStatus( - HealthSeverity.warn, 'connection lost — retrying…'); - _soundLost(); - _scheduleReconnect(); - } + _silenceSub ??= connection.silence.listen((_) => _onConnectionSilent()); + _connectionHealth ??= connection.state.subscribe((s) { + final h = connectionHealth(s); + if (h != null) health.value = h; }); } + /// The drop handling: no state for [Connection.silenceLimit], so what + /// the server said is stale. (A closed socket keeps it until then.) + void _onConnectionSilent() { + if (serverState.peek() == null) return; + serverState.value = null; + starving.value = null; + _wasPlayAllowed = false; + _mediaId = -1; + _clearPendingSeek(); + diff.reset(); + } + /// Opens [url] (with [headers] on every request); false on failure. A /// new media gets a fresh player (or the preload holding it); the media /// the player held before (a reconnect) keeps it. @@ -620,7 +575,7 @@ class SessionCoordinator { } // Tell the server at once (a preloaded media already has its buffer). final r = serverState.peek(); - if (ok && r != null && r.mediaId == mediaId && url == _openUrl && sync.token.isNotEmpty && _playerReady) { + if (ok && r != null && r.mediaId == mediaId && url == _openUrl && connection.token.isNotEmpty && _playerReady) { _sendReport(r); } return ok; @@ -635,7 +590,7 @@ class SessionCoordinator { final r = serverState.peek(); if (r == null || r.transport != 'held' || _bufferReportedFor == r.mediaId) return; if (_lastOut || _opening || _openUrl == null || _playerMediaId != r.mediaId || !_playerReady) return; - if (sync.token.isEmpty || playback.bufferedSecsAhead < r.bufferNeed) return; + if (connection.token.isEmpty || playback.bufferedSecsAhead < r.bufferNeed) return; _bufferReportedFor = r.mediaId; _sendReport(r); } @@ -715,7 +670,7 @@ class SessionCoordinator { final relay = stateMediaRelay(r); final waiting = relay != null && relay.source.isNotEmpty && - relay.source != sync.memberId && + relay.source != connection.memberId && !relay.live && _openUrl != null && jellyfinPlan.peek() is! OwnPlay; @@ -737,7 +692,7 @@ class SessionCoordinator { /// queue id in the URL, so the preload of its warm relay matches. String _relayMediaUrlFor(String name, int mediaId, [List tracks = const [], String queueId = '']) => - relayViewerUrl('http://$_wsHost:$_wsPort/r/default/${sync.token}/${queueId.isEmpty ? mediaId : queueId}', name, + relayViewerUrl('http://$_wsHost:$_wsPort/r/default/${connection.token}/${queueId.isEmpty ? mediaId : queueId}', name, tracks: tracks); /// This viewer's URL of the warm relay of Up next's first [item] @@ -762,7 +717,7 @@ class SessionCoordinator { } stopWarmSource(); final src = RelaySource( - Uri.parse('ws://$_wsHost:$_wsPort/r/default/${sync.token}/source?q=${Uri.encodeQueryComponent(queueId)}'), + Uri.parse('ws://$_wsHost:$_wsPort/r/default/${connection.token}/source?q=${Uri.encodeQueryComponent(queueId)}'), media); _warmSource = (queueId: queueId, src: src); unawaited(() async { @@ -813,7 +768,7 @@ class SessionCoordinator { return; } final src = RelaySource( - Uri.parse('ws://$_wsHost:$_wsPort/r/default/${sync.token}/source'), media); + Uri.parse('ws://$_wsHost:$_wsPort/r/default/${connection.token}/source'), media); _relaySource = src; _relaySourceMediaId = mediaId; relaying.value = true; @@ -847,7 +802,7 @@ class SessionCoordinator { } String _fileUrlFor(String name) { - return 'http://$_wsHost:$_wsPort/m/default/${sync.token}/${Uri.encodeComponent(name)}'; + return 'http://$_wsHost:$_wsPort/m/default/${connection.token}/${Uri.encodeComponent(name)}'; } StateBroadcast? _queuedState; @@ -858,7 +813,6 @@ class SessionCoordinator { /// three viewers readying at once). A state that arrives meanwhile /// replaces any still waiting. void _onState(StateBroadcast r) { - _lastStateAt = DateTime.now().toUtc(); serverState.value = r; if (r.transport == 'playing') switchTrace.mark(r.mediaId, SwitchTrace.playing); _toastMemberErrors(r); @@ -893,7 +847,7 @@ class SessionCoordinator { await _unloadForOut(); return; } - _seekGate.observe(r, sync.memberId); + _seekGate.observe(r, connection.memberId); final newEvent = r.eventId != _lastEventId; _lastEventId = r.eventId; if (r.mediaId != _mediaId) { @@ -907,7 +861,7 @@ class SessionCoordinator { sameJellyfinItem(ref, _jfRef) && jellyfinPlan.peek() is OwnPlay; final relay = stateMediaRelay(r); - final mine = relay != null && relay.source.isNotEmpty && relay.source == sync.memberId; + final mine = relay != null && relay.source.isNotEmpty && relay.source == connection.memberId; _RelayOrigin? origin; if (mine) { final pending = _pendingOrigin; @@ -956,7 +910,7 @@ class SessionCoordinator { } } else { var url = stateMediaUrl(r); - if ((url == null || url.isEmpty) && sync.token.isNotEmpty) { + if ((url == null || url.isEmpty) && connection.token.isNotEmpty) { final name = (r.media?['name'] as String?) ?? ''; if (name.isNotEmpty) url = _fileUrlFor(name); } @@ -984,8 +938,8 @@ class SessionCoordinator { return; } // Server clock: serverTime in the state is the server's. - final now = sync.serverNow(); - final allowed = localPlayAllowed(r, sync.memberId); + final now = connection.serverNow(); + final allowed = localPlayAllowed(r, connection.memberId); var actions = diff.diff(r, playback.positionSecs, now); if (r.transport == 'playing' && !allowed) { actions = [ @@ -1006,7 +960,7 @@ class SessionCoordinator { // echo of our own seek arriving: the player is where the user put it. // Later drift corrections within that event still apply. if (_seekGate.holdsPlayer(DateTime.now().toUtc()) || - (newEvent && _seekGate.isOwnEcho(r, sync.memberId))) { + (newEvent && _seekGate.isOwnEcho(r, connection.memberId))) { actions = [for (final a in actions) if (a is! SeekAction) a]; } if (actions.any((a) => a is SeekAction)) { @@ -1044,7 +998,7 @@ class SessionCoordinator { ready: self.ready || _restoringReady)) { _lastReady = false; _restoringReady = false; - sync.sendReady(false, source: 'auto'); + connection.sendReady(false, source: 'auto'); } } @@ -1163,7 +1117,7 @@ class SessionCoordinator { final error = playerError.peek(); final buffered = holds ? playback.bufferedSecsAhead : 0.0; if (holds && buffered >= r.bufferNeed) switchTrace.mark(r.mediaId, SwitchTrace.buffered); - sync.sendReport( + connection.sendReport( position: _lastOut ? 0 : playback.positionSecs, buffered: buffered, eventId: r.eventId, @@ -1178,35 +1132,13 @@ class SessionCoordinator { final now = DateTime.now().toUtc(); notifier.expire(now); final r = serverState.peek(); - if (sync.token.isNotEmpty) sync.sendPing(); - diag - ..share = AppSettings.instance.shareDiagnostics.peek() - ..tick(); - if (sync.token.isNotEmpty && r != null && _playerReady) { + if (connection.token.isNotEmpty && r != null && _playerReady) { _sendReport(r); if (!_lastOut) { _checkSeekLanded(r); unawaited(_tickJellyfin(r)); } } - if (!_manualClose && - _lastStateAt != null && - now.difference(_lastStateAt!).inSeconds >= 5) { - if (serverState.peek() != null) { - serverState.value = null; - starving.value = null; - _wasPlayAllowed = false; - _mediaId = -1; - _clearPendingSeek(); - diff.reset(); - } - if (!reconnecting.peek()) { - health.value = const HealthStatus( - HealthSeverity.warn, 'connection lost — retrying…'); - _soundLost(); - _scheduleReconnect(); - } - } // No warning when there is nothing to lose (unready), and none while // PLAYING: the ready flag gates nothing then, and watching counts as // activity so a pause after a long watch starts a fresh idle clock. @@ -1225,7 +1157,7 @@ class SessionCoordinator { if (sitOutWhenIdle) { sitOut(); } else { - sync.sendReady(false, source: 'auto'); + connection.sendReady(false, source: 'auto'); } case IdleEvent.none: break; @@ -1249,7 +1181,7 @@ class SessionCoordinator { _joinBackAt = null; if (!self.ready) { _lastReady = true; - sync.sendReady(true, source: 'manual'); + connection.sendReady(true, source: 'manual'); } } } @@ -1273,7 +1205,7 @@ class SessionCoordinator { return; } if (now.difference(at).inMilliseconds < 1500) return; - final target = serverPositionAt(r, sync.serverNow()); + final target = serverPositionAt(r, connection.serverNow()); if (seekLanded(ownPosition: playback.positionSecs, target: target)) { _clearPendingSeek(); return; @@ -1315,7 +1247,7 @@ class SessionCoordinator { (own - position).abs() < 2.0) { position = own; } - sync.sendSeek(position, seq: settled.seq); + connection.sendSeek(position, seq: settled.seq); } void toggleReady() { @@ -1329,17 +1261,17 @@ class SessionCoordinator { final next = !((self?.ready ?? false) && (self?.inSet ?? false)); markInput(); _lastReady = next; - sync.sendReady(next, source: 'manual'); + connection.sendReady(next, source: 'manual'); } void forceTarget(String memberId, bool ready) { markInput(); - sync.sendForce(memberId, ready); + connection.sendForce(memberId, ready); } void forcePlayAll() { markInput(); - sync.sendForce('all', true); + connection.sendForce('all', true); } void seekTo(double position) => _userSeek(position); @@ -1349,7 +1281,7 @@ class SessionCoordinator { void nudge(double deltaSecs) { final r = serverState.peek(); if (r == null || _openUrl == null) return; - final base = _seekGate.target ?? serverPositionAt(r, sync.serverNow()); + final base = _seekGate.target ?? serverPositionAt(r, connection.serverNow()); final target = base + deltaSecs; if (target.isNaN) return; _userSeek(target < 0 ? 0 : target); @@ -1388,36 +1320,22 @@ class SessionCoordinator { // second arrives empty. Drop blanks so at most one message ships. if (text.trim().isEmpty) return; markInput(); - sync.sendChat(text); + connection.sendChat(text); } /// Name and/or picture (base64 PNG, empty removes it) change for /// everyone now; a later fresh join sends them too. void updateProfile({String? name, String? avatar}) { - final p = _params; - if (p != null) { - _params = ConnectParams( - host: p.host, - port: p.port, - secret: p.secret, - name: name ?? p.name, - avatar: avatar ?? p.avatar, - mediaPath: p.mediaPath, - linkUrl: p.linkUrl, - hostKey: p.hostKey, - media: p.media, - ); - } - sync.sendProfile(name: name, avatar: avatar); + connection.updateProfile(name: name, avatar: avatar); } /// Sits [memberId] out (anyone, the host included), self when null. void sitOut([String? memberId]) { - final target = memberId ?? sync.memberId; + final target = memberId ?? connection.memberId; if (target.isEmpty) return; markInput(); - if (target == sync.memberId) _joinBackAt = null; - sync.sendOut(target, true); + if (target == connection.memberId) _joinBackAt = null; + connection.sendOut(target, true); } /// Back from sitting out: the media opens at the server position and @@ -1427,21 +1345,21 @@ class SessionCoordinator { final self = selfMember.peek(); if (self == null || !self.out) return; markInput(); - sync.sendOut(self.id, false); + connection.sendOut(self.id, false); if (serverState.peek()?.transport == 'playing') { _joinBackAt = DateTime.now().toUtc(); } else { _joinBackAt = null; _lastReady = true; - sync.sendReady(true, source: 'manual'); + connection.sendReady(true, source: 'manual'); } } /// Pokes [memberId]: only they hear it (sitting out or not). void nudgeMember(String memberId) { - if (memberId.isEmpty || memberId == sync.memberId) return; + if (memberId.isEmpty || memberId == connection.memberId) return; markInput(); - sync.sendNudge(memberId); + connection.sendNudge(memberId); } /// [queueId]: the Up next item this puts up (the server takes it off @@ -1450,9 +1368,9 @@ class SessionCoordinator { // Putting something up means watching it: back first. Not for Up next // coming up by itself: whoever puts it up may be sitting out. final self = selfMember.peek(); - if (self != null && self.out && queueId == null) sync.sendOut(self.id, false); + if (self != null && self.out && queueId == null) connection.sendOut(self.id, false); _pendingArt = (media: media, after: serverState.peek()?.mediaId ?? 0, link: link ?? media.url, art: art); - sync.sendSwap(media, queueId: queueId); + connection.sendSwap(media, queueId: queueId); } /// An Up next item put up as queued, for one this app cannot build @@ -1466,7 +1384,7 @@ class SessionCoordinator { void _maybeSendArt(StateBroadcast r) { final pending = _pendingArt; if (pending == null || r.mediaId <= pending.after) return; - if (sync.memberId.isEmpty || r.media?['swapBy'] != sync.memberId) return; + if (connection.memberId.isEmpty || r.media?['swapBy'] != connection.memberId) return; _pendingArt = null; final link = pending.link; _ownLink = link == null || link.isEmpty ? null : (r.mediaId, link); @@ -1514,7 +1432,7 @@ class SessionCoordinator { final poster = bytes?.poster, banner = bytes?.banner; if (poster == null && banner == null) return; if (serverState.peek()?.mediaId != mediaId) return; - sync.sendArt(mediaId, + connection.sendArt(mediaId, poster: poster == null ? null : base64Encode(poster), banner: banner == null ? null : base64Encode(banner)); } catch (e) { @@ -1749,14 +1667,14 @@ class SessionCoordinator { String? jellyfinRelayUrl(JellyfinRef ref) => _relayUrlFor(ref); String? _relayUrlFor(JellyfinRef ref) { - if (sync.token.isEmpty) return null; - final welcome = sync.relayUrl; + if (connection.token.isEmpty) return null; + final welcome = connection.relayUrl; if (welcome != null && - welcome.contains('/${sync.token}/') && + welcome.contains('/${connection.token}/') && welcome.endsWith('/${Uri.encodeComponent(ref.itemId)}')) { return welcome; } - return relayUrlFor(_wsHost, _wsPort, sync.token, ref.itemId); + return relayUrlFor(_wsHost, _wsPort, connection.token, ref.itemId); } /// `report.jf`: every linked account that works (so a queued Jellyfin @@ -1990,7 +1908,7 @@ class SessionCoordinator { } bool _isSwapper(StateBroadcast r, JellyfinRef ref) => - autoAdvanceSwapper(r.members, ref.serverId) == sync.memberId; + autoAdvanceSwapper(r.members, ref.serverId) == connection.memberId; void _advanceTick() { final a = autoAdvance.peek(); @@ -2039,55 +1957,13 @@ class SessionCoordinator { unawaited(_playJellyfin(ref, r)); } - void _scheduleReconnect({bool manual = false}) { - if (_manualClose || _params == null) return; - if (!manual && !_everConnected) return; - if (reconnecting.peek()) return; - reconnecting.value = true; - final epoch = _epoch; - _reconnectTimer?.cancel(); - _reconnectTimer = Timer( - const Duration(milliseconds: 1500), () => _reconnect(epoch)); - } - - /// One reconnect attempt. Every cancel/retry/leave/connect bumps - /// [_epoch]; an attempt from an older epoch must not touch state (it - /// would kill the newer connection and leave a ghost member). - Future _reconnect(int epoch) async { - _reconnectTimer = null; - bool stale() => epoch != _epoch; - final params = _params; - if (stale() || params == null) return; - if (_manualClose) { - reconnecting.value = false; - return; - } + /// A reconnect's part here (see [ReconnectHandler]): what to restore is + /// read before the attempt dials. + Future Function(WelcomeMessage, bool Function()) _onReconnect() { final wasOut = _lastOut; var wantReady = _lastReady && !wasOut; final seenBefore = _seenMediaId; - try { - health.value = const HealthStatus( - HealthSeverity.info, 'reconnecting…'); - final wsUri = Uri.parse('ws://${params.host}:${params.port}/sync'); - final welcome = await sync.connect( - wsUri: wsUri, - secret: params.secret, - name: params.name, - avatar: params.avatar, - hostKey: params.hostKey, - resume: _resumeToken, - ); - // Whoever bumped the epoch also closed (or replaced) the socket. - if (stale()) return; - reconnecting.value = false; - _everConnected = true; - if (_lostSounded) { - _lostSounded = false; - notify(const ConnectionBack()); - } - _resumeToken = sync.token; - _lastStateAt = DateTime.now().toUtc(); - health.value = const HealthStatus(HealthSeverity.ok, 'connected'); + return (welcome, stale) async { final url = welcome.mediaUrl; if (url != null && url.isNotEmpty && url != _openUrl) { await _openMedia(url); @@ -2099,77 +1975,36 @@ class SessionCoordinator { if (unreadyForNewMedia( enabled: AppSettings.instance.unreadyOnSwap.peek(), lastMediaId: seenBefore, - mediaId: sync.lastState?.mediaId ?? seenBefore, + mediaId: connection.lastState?.mediaId ?? seenBefore, ready: wantReady)) { wantReady = false; _lastReady = false; } // A new member (the server dropped the old one) sits out again; a // resumed one still does. - final selfNow = sync.lastState?.members.where((m) => m.id == sync.memberId).firstOrNull; - if (wasOut && !(selfNow?.out ?? false)) sync.sendOut(sync.memberId, true); + final selfNow = connection.lastState?.members.where((m) => m.id == connection.memberId).firstOrNull; + if (wasOut && !(selfNow?.out ?? false)) connection.sendOut(connection.memberId, true); if (wantReady) { _lastReady = true; _restoringReady = true; - sync.sendReady(true, source: 'manual'); + connection.sendReady(true, source: 'manual'); } _startTicker(); - } catch (e) { - if (stale()) return; - debugPrint('reconnect failed: $e'); - reconnecting.value = false; - _scheduleReconnect(); - } + }; } - void _soundLost() { - if (!_everConnected || _lostSounded) return; - _lostSounded = true; - notify(const ConnectionLost()); - } - - void _invalidateAttempts() { - _epoch++; - _reconnectTimer?.cancel(); - _reconnectTimer = null; - } - - void cancelReconnect() { - _manualClose = true; - _invalidateAttempts(); - reconnecting.value = false; - // Abort an attempt already dialling (no live session while retrying). - unawaited(sync.disconnect().catchError((_) {})); - } + /// "Stop reconnecting". + void cancelReconnect() => connection.cancelReconnect(); Future cancelConnect() async { - _manualClose = true; - _invalidateAttempts(); - reconnecting.value = false; - try { - await sync.disconnect(); - } catch (_) {} + await connection.cancelConnect(); health.value = const HealthStatus(HealthSeverity.info, 'cancelled'); } void retry() { - if (_params == null) return; - _invalidateAttempts(); - _manualClose = false; - reconnecting.value = false; + if (!connection.retry()) return; health.value = const HealthStatus(HealthSeverity.info, 'retrying…'); - _scheduleReconnect(manual: true); - } - - String _friendlyError(Object e) { - final msg = '$e'; - if (msg.contains('proto mismatch')) return 'host runs a different version — update the app'; - if (msg.contains('wrong secret')) return 'wrong server password (401)'; - if (msg.contains('SocketException') || msg.contains('Connection refused')) { - return 'host offline — check address and port'; - } - return 'connect failed: $e'; } /// App close, before [leave]: the players go first (no mpv may outlive @@ -2184,20 +2019,17 @@ class SessionCoordinator { } Future leave({bool silent = false}) async { - _invalidateAttempts(); + connection.invalidate(); await _teardown(silent: silent); } Future _teardown({bool silent = false}) async { - _manualClose = true; - _reconnectTimer?.cancel(); - _reconnectTimer = null; + connection.stop(); _ticker?.cancel(); _ticker = null; _seekTimer?.cancel(); _seekTimer = null; _seekGate.reset(); - reconnecting.value = false; await _stopJellyfin(); _stopRelaySource(); stopWarmSource(); @@ -2210,10 +2042,7 @@ class SessionCoordinator { await playback.dispose(); } catch (_) {} // The last lines go before the token does. - await diag.stop(); - try { - await sync.disconnect(); - } catch (_) {} + await connection.close(); if (!silent) { serverState.value = null; starving.value = null; @@ -2231,10 +2060,11 @@ class SessionCoordinator { await _feedSub?.cancel(); await _nudgeSub?.cancel(); await _errSub?.cancel(); - await _statusSub?.cancel(); + await _silenceSub?.cancel(); + _connectionHealth?.call(); _bufferSub?.call(); _aniList?.close(); - await sync.dispose(); + await connection.dispose(); await playback.dispose(); } } diff --git a/apps/desktop/lib/src/shell/avatar.dart b/apps/desktop/lib/src/shell/avatar.dart index 7ec56dc..c4bf95b 100644 --- a/apps/desktop/lib/src/shell/avatar.dart +++ b/apps/desktop/lib/src/shell/avatar.dart @@ -78,7 +78,7 @@ MemoryImage _memo(Object key, Uint8List bytes) { /// Callers rebuild on [SessionCoordinator.avatars] and [Profile.picture] /// (see [PartyWatch]). ImageProvider? memberAvatarImage(MemberView m, SessionCoordinator coord) { - if (m.id.isNotEmpty && m.id == coord.sync.memberId) { + if (m.id.isNotEmpty && m.id == coord.connection.memberId) { final own = Profile.instance.picture.value; if (own != null) return _memo('self', own); } @@ -105,10 +105,10 @@ AvatarInfo selfAvatarInfo({bool ready = false, bool host = false, bool out = fal /// [m] as the party strip, toasts and the chat show it. AvatarInfo memberAvatarInfo(MemberView m, StateBroadcast? r, SessionCoordinator coord) { - final self = m.id.isNotEmpty && m.id == coord.sync.memberId; + final self = m.id.isNotEmpty && m.id == coord.connection.memberId; final relay = r == null ? null : stateMediaRelay(r); final playing = r?.media != null; - final delivery = r == null ? null : deliveryOf(r, selfId: coord.sync.memberId, names: coord.memberNames); + final delivery = r == null ? null : deliveryOf(r, selfId: coord.connection.memberId, names: coord.memberNames); return AvatarInfo( id: m.id, name: m.name, diff --git a/apps/desktop/lib/src/shell/party_strip.dart b/apps/desktop/lib/src/shell/party_strip.dart index 60f6264..6390ab2 100644 --- a/apps/desktop/lib/src/shell/party_strip.dart +++ b/apps/desktop/lib/src/shell/party_strip.dart @@ -516,7 +516,7 @@ class MembersCard extends StatelessWidget { : '${session.lastForm.host}:${session.lastForm.port}'; final delivery = r == null ? null - : deliveryOf(r, selfId: coord.sync.memberId, names: coord.memberNames); + : deliveryOf(r, selfId: coord.connection.memberId, names: coord.memberNames); return Column( mainAxisSize: MainAxisSize.min, crossAxisAlignment: CrossAxisAlignment.stretch, diff --git a/apps/desktop/lib/src/shell/sit_out_card.dart b/apps/desktop/lib/src/shell/sit_out_card.dart index 0cee068..74d56dc 100644 --- a/apps/desktop/lib/src/shell/sit_out_card.dart +++ b/apps/desktop/lib/src/shell/sit_out_card.dart @@ -50,7 +50,7 @@ class _SitOutCardState extends State { final now = coord.nowPlaying.value; final playing = r?.transport == 'playing'; final duration = (r?.media?['duration'] as num?)?.toDouble() ?? 0; - final at = r == null ? 0.0 : serverPositionAt(r, coord.sync.serverNow()); + final at = r == null ? 0.0 : serverPositionAt(r, coord.connection.serverNow()); Duration secs(double s) => Duration(milliseconds: (s * 1000).round()); final where = r?.media == null ? 'Nothing is up' diff --git a/apps/desktop/lib/src/state/session_state.dart b/apps/desktop/lib/src/state/session_state.dart index b831539..06e90ce 100644 --- a/apps/desktop/lib/src/state/session_state.dart +++ b/apps/desktop/lib/src/state/session_state.dart @@ -18,6 +18,7 @@ import '../sfx/sound_bank.dart'; import '../sfx/synth_sfx.dart'; import '../session/health.dart'; import '../session/notifier.dart'; +import '../session/connection.dart' show Connection; import '../session/session_coordinator.dart'; import '../widgets/avatar_pipeline.dart'; import 'connect_prefs.dart'; @@ -173,7 +174,10 @@ class SessionState { final Ports ports = desktopPorts(); /// Every toast, sound and activity-log line goes through it. late final Notifier notifier = Notifier(sfx: sfx, attention: ports.shell.requestAttention); - late final SessionCoordinator coordinator = SessionCoordinator(notifier: notifier, ytDlp: ports.links); + /// The link to the server; lost/back go to the notifier. + late final Connection connection = Connection(notify: notifier.notify); + late final SessionCoordinator coordinator = + SessionCoordinator(notifier: notifier, connection: connection, ytDlp: ports.links); final lastForm = ConnectForm(); JellyfinAccounts get jellyfin => JellyfinAccounts.instance; /// Set when the hosted server died mid-session; Home shows it on the @@ -348,7 +352,7 @@ class SessionState { }); try { await landed.future.timeout(const Duration(seconds: 20)); - coordinator.sync.sendSeek(carry.position); + coordinator.connection.sendSeek(carry.position); } on TimeoutException { debugPrint('resume: the media did not come up; not seeking'); } finally { diff --git a/apps/desktop/lib/src/sync/sync_client.dart b/apps/desktop/lib/src/sync/sync_client.dart index bfe87b6..92b9a85 100644 --- a/apps/desktop/lib/src/sync/sync_client.dart +++ b/apps/desktop/lib/src/sync/sync_client.dart @@ -53,7 +53,21 @@ class SyncClient { HelloMedia? media, String hostKey = '', String resume = '', - }) async { + }) => + connectWith( + wsUri: wsUri, + hello: HelloMessage( + proto: RelayConstants.protoVersion, + secret: secret, + user: HelloUser(name: name, avatar: avatar), + media: media ?? const HelloMedia(name: '', size: 0, duration: 0), + hostKey: hostKey, + resume: resume, + ), + ); + + /// Dials [wsUri], sends [hello] and waits for the welcome (10 s). + Future connectWith({required Uri wsUri, required HelloMessage hello}) async { await disconnect(); clockSync.reset(); this.wsUri = wsUri; @@ -78,14 +92,6 @@ class SyncClient { }, cancelOnError: false, ); - final hello = HelloMessage( - proto: RelayConstants.protoVersion, - secret: secret, - user: HelloUser(name: name, avatar: avatar), - media: media ?? const HelloMedia(name: '', size: 0, duration: 0), - hostKey: hostKey, - resume: resume, - ); ch.sink.add(jsonEncode(hello.toJson())); _pendingWelcome = welcomeCompleter; try { diff --git a/apps/desktop/test/chapters_mpv_e2e_test.dart b/apps/desktop/test/chapters_mpv_e2e_test.dart index 8d50cfb..aa4b551 100644 --- a/apps/desktop/test/chapters_mpv_e2e_test.dart +++ b/apps/desktop/test/chapters_mpv_e2e_test.dart @@ -114,7 +114,7 @@ void main() { a.seekTo(40); await _poll(() { final r = a.serverState.peek(); - return r != null && r.seekBy == a.sync.memberId && (r.position - 40).abs() < 2; + return r != null && r.seekBy == a.connection.memberId && (r.position - 40).abs() < 2; }, 'server at 0:40'); await _poll(() => a.skipOffer.peek()?.kind == SkipKind.opening, 'Skip OP offered'); expect(a.skipOffer.peek()!.target, const Duration(minutes: 2)); @@ -125,9 +125,9 @@ void main() { await _poll(() { final r = a.serverState.peek(); return r != null && - r.seekBy == a.sync.memberId && - serverPositionAt(r, a.sync.serverNow()) >= 120 && - serverPositionAt(r, a.sync.serverNow()) < 125; + r.seekBy == a.connection.memberId && + serverPositionAt(r, a.connection.serverNow()) >= 120 && + serverPositionAt(r, a.connection.serverNow()) < 125; }, 'server at the OP end'); await _poll(() => b.playback.positionSecs >= 119.5 && b.playback.positionSecs < 126, 'b follows'); await _poll(() => seeks(b) == seeksBefore + 1, 'b saw the Skip OP seek'); diff --git a/apps/desktop/test/relay_e2e_test.dart b/apps/desktop/test/relay_e2e_test.dart index 38728a7..ab421e5 100644 --- a/apps/desktop/test/relay_e2e_test.dart +++ b/apps/desktop/test/relay_e2e_test.dart @@ -131,14 +131,14 @@ void main() { final alice = await member('alice'); final bob = await member('bob'); expect(await bob.swapPicked(FileMedia(file)), isNull); - await _poll(() => stateMediaRelay(alice.serverState.peek()!)?.source == bob.sync.memberId, 'relay state'); + await _poll(() => stateMediaRelay(alice.serverState.peek()!)?.source == bob.connection.memberId, 'relay state'); final state = alice.serverState.peek()!; expect(state.media!.containsKey('url'), isFalse); expect(bob.relaying.peek(), isTrue); await playTogether([alice, bob]); final relay = server.hub.relay!; expect(relay.buffers.single.bytesToViewers, greaterThan(0), reason: 'alice played through the relay'); - expect(relay.sourceToken, bob.sync.token); + expect(relay.sourceToken, bob.connection.token); // The next swap retires bob's source and the buffer. Nothing answers // at that link, so its probe refuses it unless put up anyway. @@ -156,7 +156,7 @@ void main() { final link = 'http://127.0.0.1:${origin.port}/private/ep.mkv?api_key=SECRET'; expect(await bob.swapPicked(LinkMedia(link, relay: LinkRelay.me)), isNull); - await _poll(() => stateMediaRelay(alice.serverState.peek()!)?.source == bob.sync.memberId, 'me state'); + await _poll(() => stateMediaRelay(alice.serverState.peek()!)?.source == bob.connection.memberId, 'me state'); expect(alice.serverState.peek()!.toJson().toString(), isNot(contains('SECRET'))); await playTogether([alice, bob]); expect(server.hub.relay!.buffers.single.bytesToViewers, greaterThan(0)); @@ -240,7 +240,7 @@ void main() { expect(await bob.swapPicked(FileMedia(file)), isNull); await _poll(() => server.hub.relay?.live == true, 'source live'); await _poll(() => alice.playback.durationSecs > 19 && bob.playback.durationSecs > 19, 'both opened'); - final token = bob.sync.token; + final token = bob.connection.token; // Bob's network goes silent (no FIN): states stop, he reconnects // with `resume`, the server still has him. @@ -248,7 +248,7 @@ void main() { await _poll(() => bob.reconnecting.peek(), 'reconnect engaged', timeout: const Duration(seconds: 15)); await _poll(() => !bob.reconnecting.peek() && bob.serverState.peek() != null, 'back', timeout: const Duration(seconds: 20)); - expect(bob.sync.token, token, reason: 'resumed as the same member'); + expect(bob.connection.token, token, reason: 'resumed as the same member'); await _poll(() => server.hub.relay?.live == true, 'source redialled', timeout: const Duration(seconds: 15)); expect(bob.relaying.peek(), isTrue); expect(bob.playback.durationSecs, greaterThan(19), reason: 'still his own copy'); diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index d6e9b2d..114b90c 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -9,7 +9,7 @@ Plain overview first: [HOW-IT-WORKS.md](HOW-IT-WORKS.md). Wire format: |---|---|---| | `packages/core` | protocol messages, sync engine (`SyncEngine`), constants, config, Range parsing, logging, Jellyfin client, updater, watched rule (`WatchSession`) | pure Dart, no Flutter | | `packages/server` | Relic HTTP + WebSocket server (`SyncServer`, `ServerHub`, `Session`, media endpoint, relay buffer, Jellyfin relay), the `serve()` entrypoint, `bin/server.dart` (dedicated server) | pure Dart, no Flutter | -| `apps/desktop` | Flutter app: UI, `SyncClient`, `SessionCoordinator`, `PlaybackController` (media_kit/mpv), `ServerIsolate`, update UI (`UpdateController`) | ships with `watch-together-server` | +| `apps/desktop` | Flutter app: UI, `SyncClient`, `Connection`, `SessionCoordinator`, `PlaybackController` (media_kit/mpv), `ServerIsolate`, update UI (`UpdateController`) | ships with `watch-together-server` | | `tool` | harnesses: `range_logger`, `shaper`, `fake_viewer`, `e2e_check` | dev only | | `deploy` | systemd unit for the dedicated server | | @@ -93,8 +93,9 @@ Loosely modeled on Syncplay, with a different readiness design. a `ping` with its clock every second and estimates the offset NTP style (`ClockSync`: `offset = s - (t0 + t3) / 2` of the sample with the smallest round trip among the last 16). All extrapolation uses - `SyncClient.serverNow()` (local clock plus offset), so a client whose clock - is off still lands on the server position. + `Connection.serverNow()` (= `SyncClient.serverNow()`: local clock plus + offset), so a client whose clock is off still lands on the server + position. - **Reports**: clients send `position`, `buffered`, `eventId` every 1 s, plus the media `duration` and its `mediaId` once the player knows it. The server stores `buffered` and liveness, and takes the first reported @@ -191,9 +192,11 @@ Loosely modeled on Syncplay, with a different readiness design. as a new member. - **Client policies**: no state broadcast for 5 s means connection lost, clear the server state and reconnect every 1.5 s until cancelled (an epoch counter makes - a cancelled attempt stale). After a reconnect the ready flag is + a cancelled attempt stale). A closed socket means lost at once (the + server state stays until the 5 s pass). After a reconnect the ready flag is restored from the server's state. Cancel during setup aborts the join. Closing the window leaves gracefully (see "App close" under Desktop UI). + Where it lives: see "Connection" below. - **Idle** (`IdleTracker`): measured from the later of the last input and the moment playback stopped. While HELD and self ready: warning at 8 min, auto-unready at 10 min, and never less than 2 min after the warning. @@ -213,6 +216,45 @@ Loosely modeled on Syncplay, with a different readiness design. Additive wire changes only; bump `proto` and PROTOCOL.md in the same commit. +## Connection + +`Connection` (`session/connection.dart`) is the link to the server, made +in the composition root (`SessionState.connection`) and passed to the +`SessionCoordinator`. It owns the `SyncClient` socket, the first connect +(`begin` + `dial`), reconnects (`reconnectDelay`, 1.5 s; the previous +token as `hello.resume`), the epoch (`invalidate`, `isCurrent`) that makes +a cancelled connect or attempt stale, the 1 s heartbeat (clock `ping`, +the diagnostics tick, the silence check: `silenceLimit`, 5 s), the +diagnostics uploader (`diag`, started with the session, stopped by +`close` before the socket), "Stop reconnecting" (`cancelReconnect`), +"Retry" (`retry`) and setup cancel (`cancelConnect`). The `hello` is +built by an injectable `HelloBuilder` (`defaultHello`). The incoming +streams (`states`, `feeds`, `nudges`, `errors`), the send methods, the +`token`, `memberId` and `serverNow()` are read through it. + +`Connection.state` (a signal of the sealed `ConnState`): `ConnClosed` +(before a connect, after leave or a cancel), `ConnConnecting` (the first +dial), `ConnOnline` (welcome received), `ConnLost` (dropped: the socket +closed or the silence limit passed; the next attempt is scheduled), +`ConnReconnecting(attempt)` (an attempt dialling, and after it failed +until the next one), `ConnRefused(code, message)` (the first dial failed: +`version`, `secret` or empty, and the plain-words reason; nothing +retries) and `ConnReplaced` (not produced yet). Every change notifies, +a repeat too. The coordinator maps it to `health` (`connectionHealth`: +"connecting…", "connected", "connection lost — retrying…", +"reconnecting…", the reason; closed writes nothing, a cancel writes +"cancelled", "Retry" "retrying…") and to `reconnecting` (lost or +reconnecting). The first drop of a connection raises `ConnectionLost`, +the reconnect that follows `ConnectionBack` (each once per drop, through +the Notifier). + +The session's part of a reconnect is the coordinator's +`onReconnect` handler: before an attempt dials it reads what to restore +(ready, sat out, the media seen), once online it opens the media again +if its URL changed, sends `out`/`ready` back and restarts the session's +ticker. The silence (`Connection.silence`) clears the coordinator's +server state (`_onConnectionSilent`). + ## Media Three sources, same sync. Any of them can also go through the relay @@ -1109,7 +1151,7 @@ tab's card errors are still its own snack bars. **Ports.** UI and session code reach the platform through interfaces (`platform/ports/`), made once in the composition root (`SessionState.ports`, `desktopPorts()` in `platform/desktop/`) and -handed on from there (the coordinator gets its `Notifier` and +handed on from there (the coordinator gets its `Notifier`, `Connection` and `LinkResolver` as constructor arguments): - `ShellPort` (`DesktopShell`): window start (remembered bounds, the close request routed to `shutDownApp`), fullscreen, minimize/maximize @@ -1697,10 +1739,10 @@ dedicated server (`package:logging` loggers named by area: `relay`, - **The session log** (`packages/server/lib/src/session_log.dart`, `packages/core/lib/src/diag_share.dart`): one shared log per server so a problem can be read afterwards from every side. While in a session - the app (`DiagUploader`, owned by the coordinator) keeps its own + the app (`DiagUploader`, owned by the `Connection`) keeps its own records (not its hosted server's `server.*`, which the server logs - itself), redacted, and sends them every 10 s from the session's 1 s - ticker and when it leaves (before its token goes, at most 1.5 s), as + itself), redacted, and sends them every 10 s from the connection's 1 s + heartbeat and when it leaves (before its token goes, at most 1.5 s), as gzip batches to `POST /l//` (PROTOCOL "Diagnostics"), timestamps moved to the server's clock with the ping offset (`ClockSync`), the app version in every batch. Until it can diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 374e433..450652b 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -85,15 +85,13 @@ the automatic continuation and countdown, relay state), the device (client id, library and folders, match corrections, watch history, settings, notification prefs, what this app is sourcing) and the integrations (Jellyfin played/resume, AniList list). -0. Safety net and seams, no visible change. Left: Connection extracted - from the coordinator (one sealed connection state, hello built by a - factory, reconnect delay and silence threshold as parameters); a - `noticeFor` table test; a flaky-link test harness (pure-Dart TCP proxy: - pass, blackhole, delay, reset, late frame); grace and silence through +0. Safety net and seams, no visible change. Left: a `noticeFor` table + test; a flaky-link test harness (pure-Dart TCP proxy: pass, blackhole, + delay, reset, late frame); grace and silence through the engine clock; an env override for client id and data folder (two instances on one machine); the scenario list in DEVELOPMENT "Tests". (Done: tests audited and tagged, P27, AppEvent + Notifier, ports, - protocol 16 planned.) + protocol 16 planned, Connection extracted from the coordinator.) R. Rename (before step 1, so dev builds install side by side with 1.2.x): app id `dev.foxyz.starsync`, binary, Windows company `foxyz`; Hive out of `~/Documents` into the app's support folder, copied not moved;