diff --git a/lib/media/media_server_client.dart b/lib/media/media_server_client.dart index 6167e9fd..59804107 100644 --- a/lib/media/media_server_client.dart +++ b/lib/media/media_server_client.dart @@ -735,20 +735,41 @@ extension MediaServerClientScope on MediaServerClient { _ => serverId, }; - /// Mark [item] watched because it crossed [watchedThreshold] during playback, - /// when a playback-stopped report is/was also sent for the same playback. + /// Mark [item] watched because playback crossed [watchedThreshold], for paths + /// where the backend cannot mark it from the playback reports themselves: + /// queued offline replay, external players, Plex same-file siblings, and + /// in-player sessions whose crossing the backend never observed. + /// /// Backends that mark played from the stop report /// ([marksWatchedOnPlaybackStopped]) skip the server call — issuing /// [markWatched] too would double-scrobble via the Jellyfin Trakt plugin /// (#1287). The single local event emitted here keeps the UI and Plezy's /// own Trakt sync (which key on `watched` events, not progress) in sync; /// the stop report syncs the server. + /// + /// In-player sessions that *did* give the backend an observable crossing use + /// [notifyWatchedFromPlaybackSession] instead. Future markWatchedFromPlaybackStop(MediaItem item) async { if (!marksWatchedOnPlaybackStopped) { await markWatched(item); } WatchStateNotifier().notifyWatched(item: item, isNowWatched: true, cacheServerId: cacheServerId); } + + /// Emit the local watched event for [item] without touching the server. + /// + /// Used when a live playback-reporting session has already given the backend + /// everything it needs to mark the item itself — a report below + /// [watchedThreshold] followed by one at or above it. Both backends act on + /// that crossing: Jellyfin through `/Sessions/Playing/Stopped` + /// (`MaxResumePct`), Plex through `/:/timeline` past + /// `LibraryVideoPlayedThreshold`. Adding an explicit [markWatched] on top + /// records the same watch twice — a second Trakt-plugin scrobble on Jellyfin + /// (#1287), a second Play History row and an inflated `viewCount` on Plex + /// (#1740). + void notifyWatchedFromPlaybackSession(MediaItem item) { + WatchStateNotifier().notifyWatched(item: item, isNowWatched: true, cacheServerId: cacheServerId); + } } extension MediaServerClientLiveTv on MediaServerClient { diff --git a/lib/services/playback_progress_tracker.dart b/lib/services/playback_progress_tracker.dart index e35e2b8d..4748cc2a 100644 --- a/lib/services/playback_progress_tracker.dart +++ b/lib/services/playback_progress_tracker.dart @@ -20,9 +20,14 @@ import '../utils/watch_state_notifier.dart'; /// Both Plex and Jellyfin go through the unified /// [MediaServerClient.reportPlayback*] surface — Plex maps the three signals /// onto `/:/timeline` updates with appropriate `state`, Jellyfin uses the -/// three `/Sessions/Playing*` endpoints. Scrobble fires once the position -/// crosses the client's [watchedThreshold] (per-server pref on Plex, fixed -/// 90% on Jellyfin). +/// three `/Sessions/Playing*` endpoints. +/// +/// Local watched state flips as soon as the position crosses the client's +/// [MediaServerClient.watchedThreshold] (per-server pref on Plex, fixed 90% on +/// Jellyfin). The *server-side* mark is a separate decision: both backends +/// already mark an item played from a threshold crossing they observe in the +/// reports this tracker sends, so an explicit mark is issued only for sessions +/// that gave them no such crossing (#1287, #1740). class PlaybackProgressTracker { /// Server client for online progress updates (null when offline). Pinned /// for the tracker's lifetime — one playback session against the server @@ -96,9 +101,44 @@ class PlaybackProgressTracker { /// Timer ticks to skip before retrying after failures (exponential backoff). int _ticksToSkip = 0; - /// Whether we've already scrobbled (marked as watched) for this playback session. + /// Whether this playback session considers the item watched locally. Latched + /// on the first observed threshold crossing, delivered to the server or not. bool _scrobbled = false; + /// The backend has received a report from this session at a position that is + /// both strictly positive and below [MediaServerClient.watchedThreshold], so + /// a later at-or-above report reads as a crossing. + /// + /// Position zero does not count. Verified against PMS 1.43: a session + /// reporting `time=0` and then the full duration is not marked played, while + /// the same session starting at `time=1000` is. Plex treats a zero position + /// as session initialisation rather than progress, so it has nothing to + /// cross from. (Retention of a resume point is a separate, higher bar — + /// reports at 5s and 30s arm the crossing without persisting an offset.) + bool _deliveredBelow = false; + + /// The backend has received an at-or-above-threshold report while already + /// holding a sub-threshold offset — it observed the crossing and marked the + /// item itself, so an explicit mark would record the same watch twice + /// (#1287 Jellyfin, #1740 Plex). + bool _serverObservedCrossing = false; + + /// The terminal stopped report has been delivered; no further report can + /// change what the backend saw. + bool _sessionEnded = false; + + /// The in-flight [_settleServerMark], so the terminal stopped report can wait + /// for the explicit mark it triggers instead of leaving it racing teardown. + Future? _pendingSettle; + + /// The server-side mark is resolved: either the backend marked the item from + /// its own crossing, or we issued the explicit mark. Reset on failure so the + /// next delivered report retries. + bool _serverMarkSettled = false; + + /// The post-watch hook has run; it fires at most once per tracker. + bool _scrobbledHookRan = false; + /// Whether the final stopped progress event was already emitted locally. bool _stopProgressNotified = false; @@ -109,7 +149,8 @@ class PlaybackProgressTracker { static const Duration _progressNotifyDelta = Duration(seconds: 30); - final PlaybackReportSession? _reportSession; + /// Built in the constructor body so the delivery callback can bind `this`. + late final PlaybackReportSession? _reportSession; PlaybackProgressTracker({ required this.client, @@ -127,15 +168,18 @@ class PlaybackProgressTracker { this.hasRenderedPlayback, this.updateInterval = const Duration(seconds: 10), }) : assert(!isOffline || offlineWatchService != null, 'offlineWatchService is required when isOffline is true'), - assert(isOffline || client != null, 'client is required when isOffline is false'), - _reportSession = isOffline || client == null - ? null - : PlaybackReportSession( - client: client, - itemId: metadata.id, - playSessionId: playSessionId, - playMethod: playMethod, - ); + assert(isOffline || client != null, 'client is required when isOffline is false') { + final reportingClient = client; + _reportSession = isOffline || reportingClient == null + ? null + : PlaybackReportSession( + client: reportingClient, + itemId: metadata.id, + playSessionId: playSessionId, + playMethod: playMethod, + onDelivered: _onReportDelivered, + ); + } void startTracking() { if (_progressTimer != null) { @@ -203,6 +247,12 @@ class PlaybackProgressTracker { void resumeAfterStoppedReport() { _stoppedProgressFuture = null; _reportSession?.resetAfterStop(); + // A re-armed session is a new server-side session: backends only act on a + // threshold crossing observed within one, so it must earn its own + // below-threshold report before we can rely on it again. + _deliveredBelow = false; + _serverObservedCrossing = false; + _sessionEnded = false; } Future _sendProgress(String state, {Duration? positionOverride}) async { @@ -240,6 +290,10 @@ class PlaybackProgressTracker { // by a fatal error, use the last position captured while output was // healthy rather than the still-advancing native media clock. final accepted = await _sendOnlineProgress(state, position, duration, allowScrobble: canCommitStoppedProgress); + // The explicit mark is resolved at session end, so it has to ride the + // terminal report's future — callers that await the stop before tearing + // the player down would otherwise drop it. + await _pendingSettle; _resetBackoff(); if (accepted && canCommitStoppedProgress) { _notifyProgressIfNeeded(position, duration, force: true); @@ -363,44 +417,128 @@ class PlaybackProgressTracker { return info == null ? PlaybackStreamSelection.none : PlaybackStreamSelection(mediaSourceId: info.mediaSourceId); } + /// Records what the backend actually received, then re-evaluates whether the + /// explicit mark is still needed. + /// + /// Both backends mark an item played from a watched-threshold *crossing* + /// observed inside a single reporting session — a report below the threshold + /// followed by one at or above it. Absolute position is not enough: a session + /// whose every report sits above the threshold, or one resuming past it, is + /// never marked server-side. + void _onReportDelivered(PlaybackReportSnapshot snapshot) { + final threshold = client?.watchedThreshold; + // isWatchedProgress reports false for an unknown duration; treating that as + // a below-threshold report would wrongly arm the crossing. + if (threshold == null || snapshot.duration.inMilliseconds <= 0) return; + if (snapshot.isStopped) _sessionEnded = true; + if (isWatchedProgress( + positionMs: snapshot.position.inMilliseconds, + durationMs: snapshot.duration.inMilliseconds, + threshold: threshold, + )) { + if (_deliveredBelow) _serverObservedCrossing = true; + } else if (snapshot.position > Duration.zero) { + // Zero is session initialisation, not progress: the backend has nothing + // to cross from, so it will never mark the item off such a session. + _deliveredBelow = true; + } + // _settleServerMark swallows its own failures, so this never escapes. + final settle = _settleServerMark(client); + _pendingSettle = settle; + unawaited(settle); + } + + /// Issues the explicit server-side mark, but only once it is clear the + /// backend will not record the watch itself. + /// + /// Called after the local crossing latches and again on every delivered + /// report — between them those cover every transition that can change the + /// answer. + Future _settleServerMark(MediaServerClient? c) async { + if (c == null || !_scrobbled || _serverMarkSettled) return; + // Backends that mark played from the playback-stopped report do it there, + // and the terminal stop is always sent. An explicit mark on top would + // double-scrobble through the Jellyfin Trakt plugin (#1287). + if (c.marksWatchedOnPlaybackStopped) { + _serverMarkSettled = true; + await _runScrobbledHook(); + return; + } + // The backend observed the crossing and marked the item itself. Marking + // again records the same watch twice (#1740). + if (_serverObservedCrossing) { + _serverMarkSettled = true; + await _runScrobbledHook(); + return; + } + // Never mark while the session is still live. A crossing can appear at any + // point until the stop: even a session that began past the threshold can + // seek back below it and cross again, and the backend records that crossing + // itself. No eager decision can know a future rewind won't create one. + // Marking eagerly and then hitting that path reproduces the very + // double-count this guards against — verified against PMS 1.43, where an + // explicit mark followed by an in-session crossing leaves viewCount at 2 + // with a Play History row (#1740). + // + // A session that only ever sent its stop reaches this already ended, so the + // common crossing-less case is still resolved immediately. + if (!_sessionEnded) return; + // The session is over and the backend never saw a crossing: a resume that + // stayed past the threshold, one that only sent its stop, or one whose + // crossing was coalesced away and never re-delivered. It will not mark this + // itself. + _serverMarkSettled = true; + try { + await c.markWatched(metadata); + } catch (e) { + appLogger.w('Failed to mark ${metadata.id} watched', error: e); + _serverMarkSettled = false; // Retry on the next delivered report. + return; + } + await _runScrobbledHook(); + } + + /// Runs the post-watch hook once, after the item's own watched state is + /// accounted for server-side. + /// + /// Its production caller marks same-file sibling episodes (#1500) with real + /// server writes, so it must never run ahead of the primary: a resumed + /// session whose explicit mark is still pending — or has just failed — would + /// otherwise leave the siblings watched and the episode actually played + /// unwatched. Hook failures are logged and never un-settle the mark. + Future _runScrobbledHook() async { + final hook = onScrobbled; + if (hook == null || _scrobbledHookRan) return; + _scrobbledHookRan = true; + try { + await hook(); + } catch (e) { + appLogger.w('Post-scrobble hook failed for ${metadata.id}', error: e); + } + } + Future _maybeScrobble(MediaServerClient c, Duration position, Duration duration) async { - // Explicitly scrobble once progress crosses the watched threshold. - // Some servers (Plex with no active play session, Jellyfin always) - // don't auto-mark from progress updates alone. - if (!_scrobbled && - isWatchedProgress( + if (_scrobbled || + !isWatchedProgress( positionMs: position.inMilliseconds, durationMs: duration.inMilliseconds, threshold: c.watchedThreshold, )) { - final percent = position.inMilliseconds / duration.inMilliseconds; - final threshold = c.watchedThreshold; - _scrobbled = true; - try { - // Backends that mark the item played from the playback-stopped report - // (Jellyfin) only emit the local watch event here — an explicit - // markWatched would double-scrobble via the Trakt plugin (#1287). - // Plex still issues the server call. Either path emits the watched - // event through WatchStateNotifier, so no extra notify is needed. - await c.markWatchedFromPlaybackStop(metadata); - appLogger.d( - 'Scrobbled ${metadata.id} (${(percent * 100).toStringAsFixed(0)}% >= ${(threshold * 100).toStringAsFixed(0)}%)', - ); - } catch (e) { - appLogger.w('Failed to scrobble ${metadata.id}', error: e); - _scrobbled = false; // Retry on next tick - } - // After (and only after) the primary mark succeeded. A failure here - // must not reset _scrobbled — that would re-scrobble the primary - // item and inflate its view count. - if (_scrobbled && onScrobbled != null) { - try { - await onScrobbled!(); - } catch (e) { - appLogger.w('Post-scrobble hook failed for ${metadata.id}', error: e); - } - } + return; } + final percent = position.inMilliseconds / duration.inMilliseconds; + final threshold = c.watchedThreshold; + _scrobbled = true; + // Local state flips on the observed crossing, whether or not the backend + // received that particular report. The server-side mark is a separate + // question, answered by _settleServerMark once delivery is known. + c.notifyWatchedFromPlaybackSession(metadata); + appLogger.d( + 'Watched ${metadata.id} (${(percent * 100).toStringAsFixed(0)}% >= ${(threshold * 100).toStringAsFixed(0)}%)', + ); + // The #1500 sibling hook runs from _settleServerMark, once this item's own + // watched state is accounted for server-side. + await _settleServerMark(c); } Future _currentStreamSelectionForProgress() async { diff --git a/lib/services/playback_report_session.dart b/lib/services/playback_report_session.dart index a1ba94d0..05db4e0a 100644 --- a/lib/services/playback_report_session.dart +++ b/lib/services/playback_report_session.dart @@ -64,6 +64,7 @@ class PlaybackReportSession { this.playSessionId, this.playMethod, this.liveStreamId, + this.onDelivered, }); final MediaServerClient client; @@ -72,6 +73,16 @@ class PlaybackReportSession { final String? playMethod; final String? liveStreamId; + /// Invoked with each snapshot the backend actually received, right after its + /// report call returns. + /// + /// Coalesced and superseded snapshots never reach this: [report] resolving + /// `true` is not delivery. A same-state heartbeat arriving while the start + /// report is in flight is dropped by [_reportProgress] yet still completes + /// its future, so callers that need to know what the server saw — such as + /// watched-threshold crossing detection — must key on this instead. + final void Function(PlaybackReportSnapshot snapshot)? onDelivered; + _PlaybackReportState _state = _PlaybackReportState.idle; PlaybackReportSnapshot? _startSnapshot; _PendingProgressReport? _pendingProgress; @@ -256,6 +267,7 @@ class PlaybackReportSession { audioStreamIndex: selection.audioStreamIndex, subtitleStreamIndex: selection.subtitleStreamIndex, ); + onDelivered?.call(snapshot); } Future _sendProgress(PlaybackReportSnapshot snapshot) async { @@ -273,6 +285,7 @@ class PlaybackReportSession { audioStreamIndex: selection.audioStreamIndex, subtitleStreamIndex: selection.subtitleStreamIndex, ); + onDelivered?.call(snapshot); return true; } @@ -287,5 +300,6 @@ class PlaybackReportSession { mediaSourceId: selection.mediaSourceId, report: snapshot.report, ); + onDelivered?.call(snapshot); } } diff --git a/lib/services/plex_client.dart b/lib/services/plex_client.dart index 11bde7e4..76c88d48 100644 --- a/lib/services/plex_client.dart +++ b/lib/services/plex_client.dart @@ -3839,9 +3839,18 @@ class PlexClient @override double get watchedThreshold => watchedThresholdPercent / 100.0; - /// Plex's `/:/timeline?state=stopped` doesn't reliably mark watched without - /// an active play session, so the in-player auto-scrobble still issues the - /// explicit `markWatched` (`/:/scrobble`). See [marksWatchedOnPlaybackStopped]. + /// A single `/:/timeline?state=stopped` does not mark watched: PMS only acts + /// on a threshold crossing it observes inside one session — a report below + /// `LibraryVideoPlayedThreshold` followed by one at or above it. Verified + /// against PMS 1.43: consecutive above-threshold reports mark nothing (it + /// won't even store an above-threshold `viewOffset`), and a resume point left + /// by an earlier session does not arm a new one. + /// + /// So paths with no observable crossing — queued offline replay, external + /// players, same-file siblings — still need the explicit `markWatched` + /// (`/:/scrobble`). In-player sessions that did produce a crossing must not + /// send it: PMS has already recorded the watch, and the extra call inflates + /// `viewCount` and (before PMS 1.40) adds a second Play History row (#1740). @override bool get marksWatchedOnPlaybackStopped => false; diff --git a/test/services/playback_progress_tracker_test.dart b/test/services/playback_progress_tracker_test.dart index 3573bb4a..d5a1b8a0 100644 --- a/test/services/playback_progress_tracker_test.dart +++ b/test/services/playback_progress_tracker_test.dart @@ -759,8 +759,9 @@ void main() { expect(client.markWatchedCalls, isEmpty); }); - test('scrobbles when percent >= watchedThresholdPercent', () async { - // 95% >= 90% threshold. + test('a session with no observable crossing still issues the explicit mark', () async { + // A lone stopped report at 95%: Plex never held a sub-threshold offset + // for this session, so it will not mark the item itself (#1740). final client = _FakePlexClient(thresholdPercent: 90); final player = _FakePlayer(position: const Duration(seconds: 95), duration: const Duration(seconds: 100)); final tracker = PlaybackProgressTracker( @@ -835,39 +836,28 @@ void main() { expect(client.markWatchedCalls, hasLength(1)); }); - test('a failed scrobble is retried on the next call (resets _scrobbled)', () async { - final client = _FakePlexClient(thresholdPercent: 90); + test('a failed explicit mark is retried on the next session end', () async { + // The mark is only attempted once a session ends with no crossing the + // backend could see. A failure must leave it unsettled so the next + // terminal report tries again. final player = _FakePlayer(position: const Duration(seconds: 95), duration: const Duration(seconds: 100)); - final tracker = PlaybackProgressTracker(client: client, metadata: _meta(), player: player, isOffline: false); - addTearDown(tracker.dispose); - - // First call: updateProgress succeeds, then markAsWatched throws. - // To make the *second* method (markAsWatched) throw, we need a flag that - // only triggers on the 2nd call. The fake's `throwOnNextCall` consumes - // on the first call, which is updateProgress. Workaround: arm the throw - // immediately before sendProgress, so updateProgress fails. The catch - // branch in PlaybackProgressTracker still bumps the failure counter for - // online stopped calls (and skips scrobble). Then arm again — updateProgress - // succeeds (because the throw was consumed) — and assert markAsWatched - // succeeds and scrobbles. - // - // To target ONLY markAsWatched, we instead use a custom client. final precise = _ScrobblePreciseClient(thresholdPercent: 90, failScrobbleFirstTime: true); - final tracker2 = PlaybackProgressTracker( + final tracker = PlaybackProgressTracker( client: precise, metadata: _meta(ratingKey: '42'), player: player, isOffline: false, ); - addTearDown(tracker2.dispose); + addTearDown(tracker.dispose); - await tracker2.sendProgress('playing'); - await Future.delayed(Duration.zero); + await tracker.sendProgress('stopped'); + await pumpEventQueue(); expect(precise.markWatchedAttempts, 1); + expect(precise.markWatchedSuccesses, 0); - // Retry — markAsWatched now succeeds. - await tracker2.sendProgress('playing'); - await Future.delayed(Duration.zero); + tracker.resumeAfterStoppedReport(); + await tracker.sendProgress('stopped'); + await pumpEventQueue(); expect(precise.markWatchedAttempts, 2); expect(precise.markWatchedSuccesses, 1); }); @@ -909,8 +899,12 @@ void main() { expect(hookCalls, 0); }); - test('onScrobbled waits for a successful scrobble when the first attempt fails', () async { - final precise = _ScrobblePreciseClient(thresholdPercent: 90, failScrobbleFirstTime: true); + test('onScrobbled waits for the explicit mark and never runs ahead of it (#1500)', () async { + // The hook marks same-file siblings with real server writes, so it must + // not run while this item's own mark is still pending: a hard kill in + // between would leave the siblings watched and the episode actually + // played unwatched. + final precise = _ScrobblePreciseClient(thresholdPercent: 90); final player = _FakePlayer(position: const Duration(seconds: 95), duration: const Duration(seconds: 100)); var hookCalls = 0; final tracker = PlaybackProgressTracker( @@ -923,12 +917,40 @@ void main() { addTearDown(tracker.dispose); await tracker.sendProgress('playing'); - await Future.delayed(Duration.zero); - expect(precise.markWatchedAttempts, 1); - expect(hookCalls, 0); + await pumpEventQueue(); + expect(precise.markWatchedAttempts, 0, reason: 'the mark waits for the session to end'); + expect(hookCalls, 0, reason: 'and the siblings wait for the mark'); - await tracker.sendProgress('playing'); - await Future.delayed(Duration.zero); + await tracker.sendProgress('stopped'); + await pumpEventQueue(); + expect(precise.markWatchedSuccesses, 1); + expect(hookCalls, 1); + }); + + test('onScrobbled does not run when the explicit mark fails, and runs on the retry', () async { + // The exact inconsistency to avoid: siblings marked watched while the + // primary episode is still unmarked on the server. + final precise = _ScrobblePreciseClient(thresholdPercent: 90, failScrobbleFirstTime: true); + final player = _FakePlayer(position: const Duration(seconds: 95), duration: const Duration(seconds: 100)); + var hookCalls = 0; + final tracker = PlaybackProgressTracker( + client: precise, + metadata: _meta(ratingKey: '42'), + player: player, + isOffline: false, + onScrobbled: () async => hookCalls++, + ); + addTearDown(tracker.dispose); + + await tracker.sendProgress('stopped'); + await pumpEventQueue(); + expect(precise.markWatchedAttempts, 1); + expect(precise.markWatchedSuccesses, 0); + expect(hookCalls, 0, reason: 'siblings must not be marked while the primary is not'); + + tracker.resumeAfterStoppedReport(); + await tracker.sendProgress('stopped'); + await pumpEventQueue(); expect(precise.markWatchedSuccesses, 1); expect(hookCalls, 1); }); @@ -952,13 +974,378 @@ void main() { addTearDown(tracker.dispose); await tracker.sendProgress('playing'); - await Future.delayed(Duration.zero); - await tracker.sendProgress('playing'); - await Future.delayed(Duration.zero); + await pumpEventQueue(); + await tracker.sendProgress('stopped'); + await pumpEventQueue(); expect(client.markWatchedCalls, hasLength(1)); expect(hookCalls, 1); }); + + // ---------------------------------------------------------- + // Server-side crossing detection (#1740) + // + // Both backends mark an item played from a watched-threshold crossing they + // observe inside one reporting session: a report below the threshold + // followed by one at or above it. Verified against PMS 1.43 — consecutive + // above-threshold reports mark nothing, and a resume point from an earlier + // session does not arm a new one. The explicit mark must therefore be sent + // only when the backend had no crossing to observe. + // ---------------------------------------------------------- + + test('a delivered crossing marks watched locally and never issues the explicit mark (#1740)', () async { + final client = _FakePlexClient(thresholdPercent: 90); + final player = _FakePlayer(position: const Duration(seconds: 50), duration: const Duration(seconds: 100)); + final tracker = PlaybackProgressTracker( + client: client, + metadata: _meta(ratingKey: '42'), + player: player, + isOffline: false, + ); + addTearDown(tracker.dispose); + + final watched = []; + final sub = WatchStateNotifier() + .forItem('42') + .where((e) => e.changeType == WatchStateChangeType.watched) + .listen(watched.add); + addTearDown(sub.cancel); + + await tracker.sendProgress('playing'); + await pumpEventQueue(); + player.position = const Duration(seconds: 95); + await tracker.sendProgress('playing'); + await pumpEventQueue(); + + expect(client.markWatchedCalls, isEmpty, reason: 'the server saw the crossing and marked it itself'); + expect(watched, hasLength(1), reason: 'local watched state still flips exactly once'); + }); + + test('a session that begins past the threshold marks explicitly, but only once it ends', () async { + // Resume at 95%: the server never holds a sub-threshold offset for this + // session, so no crossing is observable and it will not mark the item. + // The mark still waits for the stop — until then the session could seek + // back and create a crossing of its own. + final client = _FakePlexClient(thresholdPercent: 90); + final player = _FakePlayer(position: const Duration(seconds: 95), duration: const Duration(seconds: 100)); + final tracker = PlaybackProgressTracker( + client: client, + metadata: _meta(ratingKey: '42'), + player: player, + isOffline: false, + ); + addTearDown(tracker.dispose); + + await tracker.sendProgress('playing'); + await pumpEventQueue(); + expect(client.markWatchedCalls, isEmpty, reason: 'still live — a rewind could yet create a crossing'); + + player.position = const Duration(seconds: 100); + await tracker.sendProgress('stopped'); + await pumpEventQueue(); + + expect(client.markWatchedCalls, ['42']); + }); + + test('the explicit mark completes with the stopped report future', () async { + // The mark is resolved at session end, so it must ride the terminal + // report rather than race teardown: callers await the stop and then + // dispose the tracker. + final client = _ScrobblePreciseClient(thresholdPercent: 90)..markGate = Completer(); + final player = _FakePlayer(position: const Duration(seconds: 95), duration: const Duration(seconds: 100)); + final tracker = PlaybackProgressTracker( + client: client, + metadata: _meta(ratingKey: '42'), + player: player, + isOffline: false, + ); + addTearDown(tracker.dispose); + + await tracker.sendProgress('playing'); + await pumpEventQueue(); + + var stopDone = false; + final stop = tracker.sendStoppedProgressOnce().then((_) => stopDone = true); + await pumpEventQueue(); + expect(client.markWatchedAttempts, 1); + expect(stopDone, isFalse, reason: 'the stopped future must wait for the mark it triggered'); + + client.markGate!.complete(); + await stop; + expect(client.markWatchedSuccesses, 1); + }); + + test('a session that begins past the threshold then rewinds and re-crosses never marks explicitly', () async { + // Verified against PMS 1.43: an explicit mark followed by an in-session + // crossing leaves viewCount at 2 with a Play History row. Marking eagerly + // at the start of a resumed-past-threshold session would recreate #1740 + // for anyone who rewinds and watches the end again. + final client = _FakePlexClient(thresholdPercent: 90); + final player = _FakePlayer(position: const Duration(seconds: 95), duration: const Duration(seconds: 100)); + final tracker = PlaybackProgressTracker( + client: client, + metadata: _meta(ratingKey: '42'), + player: player, + isOffline: false, + ); + addTearDown(tracker.dispose); + + await tracker.sendProgress('playing'); + await pumpEventQueue(); + player.position = const Duration(seconds: 50); + await tracker.sendProgress('playing'); + await pumpEventQueue(); + player.position = const Duration(seconds: 92); + await tracker.sendProgress('playing'); + await pumpEventQueue(); + player.position = const Duration(seconds: 100); + await tracker.sendProgress('stopped'); + await pumpEventQueue(); + + expect(client.markWatchedCalls, isEmpty, reason: 'the server observed the 50% -> 92% crossing itself'); + }); + + test('a crossing landing on the stopped report suppresses the explicit mark', () async { + final client = _FakePlexClient(thresholdPercent: 90); + final player = _FakePlayer(position: const Duration(seconds: 50), duration: const Duration(seconds: 100)); + final tracker = PlaybackProgressTracker( + client: client, + metadata: _meta(ratingKey: '42'), + player: player, + isOffline: false, + ); + addTearDown(tracker.dispose); + + await tracker.sendProgress('playing'); + await pumpEventQueue(); + player.position = const Duration(seconds: 100); + await tracker.sendProgress('stopped'); + await pumpEventQueue(); + + expect(client.markWatchedCalls, isEmpty); + }); + + test('a crossing landing on a paused report suppresses the explicit mark', () async { + // The shape in the issue log: the threshold is crossed on a paused + // heartbeat, ~2 minutes before playback actually ends. + final client = _FakePlexClient(thresholdPercent: 90); + final player = _FakePlayer(position: const Duration(seconds: 50), duration: const Duration(seconds: 100)); + final tracker = PlaybackProgressTracker( + client: client, + metadata: _meta(ratingKey: '42'), + player: player, + isOffline: false, + ); + addTearDown(tracker.dispose); + + await tracker.sendProgress('playing'); + await pumpEventQueue(); + player.position = const Duration(seconds: 92); + await tracker.sendProgress('paused'); + await pumpEventQueue(); + + expect(client.markWatchedCalls, isEmpty); + }); + + test('onScrobbled still fires when the explicit mark is suppressed (#1500)', () async { + final client = _FakePlexClient(thresholdPercent: 90); + final player = _FakePlayer(position: const Duration(seconds: 50), duration: const Duration(seconds: 100)); + var hookCalls = 0; + final tracker = PlaybackProgressTracker( + client: client, + metadata: _meta(ratingKey: '42'), + player: player, + isOffline: false, + onScrobbled: () async => hookCalls++, + ); + addTearDown(tracker.dispose); + + await tracker.sendProgress('playing'); + await pumpEventQueue(); + player.position = const Duration(seconds: 95); + await tracker.sendProgress('playing'); + await pumpEventQueue(); + + expect(client.markWatchedCalls, isEmpty); + expect(hookCalls, 1, reason: 'same-file siblings are still marked'); + }); + + test('resumeAfterStoppedReport clears the crossing state for the next session', () async { + // Session 1 only ever reports below the threshold. If its sub-threshold + // offset leaked into session 2, the first above-threshold report there + // would look like a crossing and the explicit mark would be skipped — + // but the server treats them as separate sessions and marks nothing. + final client = _FakePlexClient(thresholdPercent: 90); + final player = _FakePlayer(position: const Duration(seconds: 40), duration: const Duration(seconds: 100)); + final tracker = PlaybackProgressTracker( + client: client, + metadata: _meta(ratingKey: '42'), + player: player, + isOffline: false, + ); + addTearDown(tracker.dispose); + + await tracker.sendProgress('playing'); + await pumpEventQueue(); + await tracker.sendProgress('stopped'); + await pumpEventQueue(); + expect(client.markWatchedCalls, isEmpty, reason: 'never crossed the threshold'); + + tracker.resumeAfterStoppedReport(); + player.position = const Duration(seconds: 95); + await tracker.sendProgress('playing'); + await pumpEventQueue(); + await tracker.sendProgress('stopped'); + await pumpEventQueue(); + + expect(client.markWatchedCalls, ['42']); + }); + + test('a below-threshold report dropped during startup does not arm the crossing', () async { + // The start report (95%) is in flight; a same-state 50% heartbeat arriving + // meanwhile is coalesced away by PlaybackReportSession even though its + // future resolves true. The server only ever saw 95%, so there is no + // crossing and the explicit mark must still go out. + final client = _DelayedStartClient(); + final player = _FakePlayer(position: const Duration(seconds: 95), duration: const Duration(seconds: 100)); + final tracker = PlaybackProgressTracker( + client: client, + metadata: _meta(ratingKey: '42'), + player: player, + isOffline: false, + ); + addTearDown(tracker.dispose); + + final crossing = tracker.sendProgress('playing'); + await pumpEventQueue(); + player.position = const Duration(seconds: 50); + final dropped = tracker.sendProgress('playing'); + await pumpEventQueue(); + + client.startCompleter.complete(); + await Future.wait([crossing, dropped]); + await pumpEventQueue(); + + expect(client.updateProgressCalls.map((c) => c.time), [95000], reason: 'the 50% snapshot was never sent'); + + await tracker.sendProgress('stopped'); + await pumpEventQueue(); + expect(client.markWatchedCalls, ['42']); + }); + + test('a dropped crossing followed by a seek back below the threshold marks explicitly', () async { + // The crossing snapshot is coalesced away, so the server never sees an + // at-or-above report; playback then seeks back and the terminal stop + // carries a sub-threshold position. Deciding at the crossing would have + // lost this watch entirely. + final client = _DelayedStartClient(); + final player = _FakePlayer(position: const Duration(seconds: 50), duration: const Duration(seconds: 100)); + final tracker = PlaybackProgressTracker( + client: client, + metadata: _meta(ratingKey: '42'), + player: player, + isOffline: false, + ); + addTearDown(tracker.dispose); + + final below = tracker.sendProgress('playing'); + await pumpEventQueue(); + player.position = const Duration(seconds: 95); + final dropped = tracker.sendProgress('playing'); + await pumpEventQueue(); + + client.startCompleter.complete(); + await Future.wait([below, dropped]); + await pumpEventQueue(); + + expect(client.updateProgressCalls.map((c) => c.time), [50000], reason: 'the crossing snapshot was dropped'); + expect(client.markWatchedCalls, isEmpty, reason: 'the decision is deferred, not made'); + + player.position = const Duration(seconds: 50); + await tracker.sendProgress('stopped'); + await pumpEventQueue(); + + expect(client.markWatchedCalls, ['42'], reason: 'the session ended with no crossing the server could see'); + }); + + test('a dropped crossing settles as server-marked once a later report is delivered', () async { + final client = _DelayedStartClient(); + final player = _FakePlayer(position: const Duration(seconds: 50), duration: const Duration(seconds: 100)); + final tracker = PlaybackProgressTracker( + client: client, + metadata: _meta(ratingKey: '42'), + player: player, + isOffline: false, + ); + addTearDown(tracker.dispose); + + final below = tracker.sendProgress('playing'); + await pumpEventQueue(); + player.position = const Duration(seconds: 95); + final dropped = tracker.sendProgress('playing'); + await pumpEventQueue(); + + client.startCompleter.complete(); + await Future.wait([below, dropped]); + await pumpEventQueue(); + expect(client.markWatchedCalls, isEmpty); + + player.position = const Duration(seconds: 96); + await tracker.sendProgress('playing'); + await pumpEventQueue(); + player.position = const Duration(seconds: 100); + await tracker.sendProgress('stopped'); + await pumpEventQueue(); + + expect(client.markWatchedCalls, isEmpty, reason: 'the later delivery completed the crossing server-side'); + }); + + test('a report at position zero does not arm the crossing', () async { + // Verified against PMS 1.43: a session reporting time=0 and then the full + // duration is NOT marked played, while the same session starting at + // time=1000 is. Plex treats zero as session initialisation, so it has + // nothing to cross from. Short music tracks hit this — the initial report + // fires at 0 and the next one is the terminal stop at duration. + final client = _FakePlexClient(thresholdPercent: 90); + final player = _FakePlayer(position: Duration.zero, duration: const Duration(seconds: 100)); + final tracker = PlaybackProgressTracker( + client: client, + metadata: _meta(ratingKey: '42'), + player: player, + isOffline: false, + ); + addTearDown(tracker.dispose); + + await tracker.sendProgress('playing'); + await pumpEventQueue(); + player.position = const Duration(seconds: 100); + await tracker.sendProgress('stopped'); + await pumpEventQueue(); + + expect(client.markWatchedCalls, ['42']); + }); + + test('a report one second in does arm the crossing', () async { + // The boundary is strictly positive, not some larger minimum: PMS marks + // a 1s -> 100% session played even though it persists no resume point. + final client = _FakePlexClient(thresholdPercent: 90); + final player = _FakePlayer(position: const Duration(seconds: 1), duration: const Duration(seconds: 100)); + final tracker = PlaybackProgressTracker( + client: client, + metadata: _meta(ratingKey: '42'), + player: player, + isOffline: false, + ); + addTearDown(tracker.dispose); + + await tracker.sendProgress('playing'); + await pumpEventQueue(); + player.position = const Duration(seconds: 100); + await tracker.sendProgress('stopped'); + await pumpEventQueue(); + + expect(client.markWatchedCalls, isEmpty); + }); }); // ============================================================ @@ -1453,6 +1840,9 @@ class _ScrobblePreciseClient with PlaybackReportRecorder implements PlexClient { int markWatchedAttempts = 0; int markWatchedSuccesses = 0; + /// When set, [markWatched] blocks on this so a test can observe whether the + /// caller awaits the mark. + Completer? markGate; @override Future updateProgress( String ratingKey, { @@ -1469,6 +1859,8 @@ class _ScrobblePreciseClient with PlaybackReportRecorder implements PlexClient { @override Future markWatched(MediaItem item) async { markWatchedAttempts++; + final gate = markGate; + if (gate != null) await gate.future; if (failScrobbleFirstTime) { failScrobbleFirstTime = false; throw StateError('simulated scrobble failure'); diff --git a/test/services/playback_report_session_test.dart b/test/services/playback_report_session_test.dart index f538a28d..fa9d80e9 100644 --- a/test/services/playback_report_session_test.dart +++ b/test/services/playback_report_session_test.dart @@ -170,4 +170,44 @@ void main() { expect(await session.report(_snapshot('playing', positionMs: 4000)), isTrue); expect(client.calls, ['stopped-attempt:3000:null', 'stopped:3000:null', 'started:4000:null:null:null']); }); + + test('onDelivered fires only for snapshots the backend actually received', () async { + // Callers that must know what the server saw — watched-threshold crossing + // detection (#1740) — cannot use report()'s bool: a same-state heartbeat + // arriving while the start report is in flight resolves true but is + // coalesced away. + final client = _RecordingClient()..startGate = Completer(); + final delivered = []; + final session = PlaybackReportSession( + client: client, + itemId: 'item-1', + onDelivered: (snapshot) => delivered.add(snapshot.position.inMilliseconds), + ); + + final first = session.report(_snapshot('playing', positionMs: 1000)); + final second = session.report(_snapshot('playing', positionMs: 2000)); + await Future.delayed(Duration.zero); + + client.startGate!.complete(); + expect(await Future.wait([first, second]), [true, true]); + + expect(client.calls, ['started:1000:null:null:null']); + expect(delivered, [1000], reason: 'the 2000ms snapshot was dropped, so it was never delivered'); + }); + + test('onDelivered reports progress and stopped snapshots as they are sent', () async { + final client = _RecordingClient(); + final delivered = []; + final session = PlaybackReportSession( + client: client, + itemId: 'item-1', + onDelivered: (snapshot) => delivered.add('${snapshot.state}:${snapshot.position.inMilliseconds}'), + ); + + await session.report(_snapshot('playing', positionMs: 1000)); + await session.report(_snapshot('paused', positionMs: 2000)); + await session.report(_snapshot('stopped', positionMs: 3000)); + + expect(delivered, ['playing:1000', 'paused:2000', 'stopped:3000']); + }); }