diff --git a/lib/database/app_database.dart b/lib/database/app_database.dart index b4d48f4f..bf1a5539 100644 --- a/lib/database/app_database.dart +++ b/lib/database/app_database.dart @@ -882,7 +882,7 @@ class AppDatabase extends _$AppDatabase { } /// Insert or update a progress action (merges with existing). - Future upsertProgressAction({ + Future<({int rowId, int revision})> upsertProgressAction({ String? profileId, required ServerId serverId, String? clientScopeId, @@ -895,7 +895,7 @@ class AppDatabase extends _$AppDatabase { final globalKey = buildGlobalKey(ServerId(serverId), ratingKey); final now = DateTime.now().millisecondsSinceEpoch; - await transaction(() async { + return transaction(() async { final existing = await (select(offlineWatchProgress) ..where( @@ -910,6 +910,12 @@ class AppDatabase extends _$AppDatabase { final keep = existing.isEmpty ? null : existing.first; if (keep != null) { + // The row id survives a merge, so its timestamp must advance even + // when multiple playback updates land in one clock millisecond. + final nextRevision = keep.updatedAt + 1; + final revision = now > nextRevision ? now : nextRevision; + // A merge is a new logical action; retry history belongs only to + // the revision whose server write failed. await (update(offlineWatchProgress)..where((t) => t.id.equals(keep.id))).write( OfflineWatchProgressCompanion( viewOffset: Value(viewOffset), @@ -917,37 +923,41 @@ class AppDatabase extends _$AppDatabase { shouldMarkWatched: Value(shouldMarkWatched), profileId: Value(profileId), clientScopeId: Value(clientScopeId), - updatedAt: Value(now), + updatedAt: Value(revision), + syncAttempts: const Value(0), + lastError: const Value(null), ), ); final duplicateIds = existing.skip(1).map((row) => row.id).toList(growable: false); if (duplicateIds.isNotEmpty) { await (delete(offlineWatchProgress)..where((t) => t.id.isIn(duplicateIds))).go(); } - } else { - await into(offlineWatchProgress).insert( - OfflineWatchProgressCompanion.insert( - serverId: serverId, - profileId: Value(profileId), - clientScopeId: Value(clientScopeId), - ratingKey: ratingKey, - globalKey: globalKey, - actionType: OfflineActionType.progress.id, - viewOffset: Value(viewOffset), - duration: Value(duration), - shouldMarkWatched: Value(shouldMarkWatched), - createdAt: now, - updatedAt: now, - ), - ); + return (rowId: keep.id, revision: revision); } + + final rowId = await into(offlineWatchProgress).insert( + OfflineWatchProgressCompanion.insert( + serverId: serverId, + profileId: Value(profileId), + clientScopeId: Value(clientScopeId), + ratingKey: ratingKey, + globalKey: globalKey, + actionType: OfflineActionType.progress.id, + viewOffset: Value(viewOffset), + duration: Value(duration), + shouldMarkWatched: Value(shouldMarkWatched), + createdAt: now, + updatedAt: now, + ), + ); + return (rowId: rowId, revision: now); }); }); } /// Insert a manual watch action (watched or unwatched). /// Removes conflicting actions for the same item. - Future insertWatchAction({ + Future<({int rowId, int revision})> insertWatchAction({ String? profileId, required ServerId serverId, String? clientScopeId, @@ -958,7 +968,7 @@ class AppDatabase extends _$AppDatabase { final globalKey = buildGlobalKey(ServerId(serverId), ratingKey); final now = DateTime.now().millisecondsSinceEpoch; - await transaction(() async { + return transaction(() async { // Remove conflicting actions (opposite action type and progress). await (delete(offlineWatchProgress)..where( (t) => @@ -968,7 +978,7 @@ class AppDatabase extends _$AppDatabase { )) .go(); - await into(offlineWatchProgress).insert( + final rowId = await into(offlineWatchProgress).insert( OfflineWatchProgressCompanion.insert( serverId: serverId, profileId: Value(profileId), @@ -980,6 +990,7 @@ class AppDatabase extends _$AppDatabase { updatedAt: now, ), ); + return (rowId: rowId, revision: now); }); }); } @@ -994,13 +1005,15 @@ class AppDatabase extends _$AppDatabase { /// orders by `createdAt` — and replaying it rewrites the resume position the /// mark just cleared, pinning the item to Continue Watching (#1812). /// - /// Progress queued *after* a mark is a genuine rewatch and is not affected: - /// this only runs at the moment the mark lands. + /// When [beforeRevision] is present, the notifier is settling a persisted + /// offline mark. Its listener is asynchronous, so only older revisions are + /// stale; an equal or newer progress revision is a genuine concurrent rewatch. Future deleteQueuedProgressForItem({ String? profileId, required ServerId serverId, String? clientScopeId, required String ratingKey, + int? beforeRevision, }) { return _runPendingMutation(() async { final globalKey = buildGlobalKey(ServerId(serverId), ratingKey); @@ -1009,29 +1022,58 @@ class AppDatabase extends _$AppDatabase { t.globalKey.equals(globalKey) & _nullableTextPredicate(t.profileId, profileId) & _nullableTextPredicate(t.clientScopeId, clientScopeId) & - t.actionType.equals(OfflineActionType.progress.id), + t.actionType.equals(OfflineActionType.progress.id) & + (beforeRevision == null ? const Constant(true) : t.updatedAt.isSmallerThanValue(beforeRevision)), )) .go(); }); } - /// Delete a specific watch action after successful sync + /// Delete a watch action only if it is still the snapshotted revision. + Future deleteWatchActionIfUnchanged(int id, int revision) { + return _runPendingMutation(() async { + final deleted = await (delete( + offlineWatchProgress, + )..where((t) => t.id.equals(id) & t.updatedAt.equals(revision))).go(); + return deleted != 0; + }); + } + + /// Update the retry state only if the action is still the snapshotted revision. + Future updateSyncAttemptIfUnchanged(int id, int revision, String? errorMessage) { + return _runPendingMutation(() async { + final existing = await (select( + offlineWatchProgress, + )..where((t) => t.id.equals(id) & t.updatedAt.equals(revision))).getSingleOrNull(); + if (existing == null) return false; + + final updated = await (update(offlineWatchProgress)..where((t) => t.id.equals(id) & t.updatedAt.equals(revision))) + .write( + OfflineWatchProgressCompanion( + syncAttempts: Value(existing.syncAttempts + 1), + lastError: Value(errorMessage), + ), + ); + return updated != 0; + }); + } + + /// Delete a specific watch action outside a snapshotted replay. Future deleteWatchAction(int id) { return _runPendingMutation(() async { await (delete(offlineWatchProgress)..where((t) => t.id.equals(id))).go(); }); } - /// Update sync attempt count and error message - Future updateSyncAttempt(int id, String? errorMessage) async { + /// Update retry state outside a snapshotted replay. + Future updateSyncAttempt(int id, String? errorMessage) { return _runPendingMutation(() async { final existing = await (select(offlineWatchProgress)..where((t) => t.id.equals(id))).getSingleOrNull(); + if (existing == null) return; - if (existing != null) { - await (update(offlineWatchProgress)..where((t) => t.id.equals(id))).write( - OfflineWatchProgressCompanion(syncAttempts: Value(existing.syncAttempts + 1), lastError: Value(errorMessage)), - ); - } + await (update(offlineWatchProgress)..where((t) => t.id.equals(id))).write( + OfflineWatchProgressCompanion(syncAttempts: Value(existing.syncAttempts + 1), lastError: Value(errorMessage)), + ); }); } diff --git a/lib/media/media_server_client.dart b/lib/media/media_server_client.dart index 8e808abe..dff5384f 100644 --- a/lib/media/media_server_client.dart +++ b/lib/media/media_server_client.dart @@ -70,6 +70,42 @@ abstract interface class GracefullyCloseable { Future closeGracefully({Duration drainTimeout}); } +/// Per-leg outcome sink for hub fetches. +/// +/// Home rows are best-effort by design: a hub leg that fails degrades to an +/// empty list rather than sinking the screen. That makes "this server has no +/// rows" and "every row request failed" indistinguishable at the aggregation +/// boundary, so a totally failed refresh was reported to the user as a +/// success (#1829). +/// +/// Callers that need to tell them apart pass a sink and the backend records +/// each leg it degraded. A sink rather than a widened return type keeps every +/// existing caller untouched, and lets a response carry *partial* rows +/// alongside a failure — which throwing cannot. +/// +/// Deliberately not recorded: legs whose degradation is intentional rather +/// than a fault, i.e. Emby's series-reconstruction deadline and its +/// per-series probes. +class HubFetchDiagnostics { + var _failed = false; + var _cancelled = false; + + /// A leg failed for a reason that is not a client-side abort. + bool get failed => _failed; + + /// A leg was aborted client-side. Cancellation is not failure: a disrupted + /// pass says nothing about the server's actual content. + bool get cancelled => _cancelled; + + void recordFailure(Object error) { + if (error is MediaServerHttpException && error.isCancellation) { + _cancelled = true; + } else { + _failed = true; + } + } +} + abstract class MediaServerClient { ServerId get serverId; String? get serverName; @@ -302,7 +338,14 @@ abstract class MediaServerClient { /// Curated home-screen hubs across all libraries (Plex Discover; Jellyfin /// synthesizes `Latest` plus optional `Resume` + `NextUp`). - Future> fetchGlobalHubs({int limit = defaultHubPreviewLimit, bool includePlaybackHubs = true}); + /// + /// [diagnostics], when supplied, receives every leg this call degraded to + /// empty, so the caller can tell an empty home from a failed one (#1829). + Future> fetchGlobalHubs({ + int limit = defaultHubPreviewLimit, + bool includePlaybackHubs = true, + HubFetchDiagnostics? diagnostics, + }); /// Hubs scoped to a single library section. [libraryName] is baked into /// the title of synthetic hubs (Jellyfin) so per-library "Recently Added" @@ -310,12 +353,14 @@ abstract class MediaServerClient { /// [includePlaybackHubs] lets surfaces that already render Continue /// Watching skip duplicate playback rows. [libraryKind] lets backends avoid /// irrelevant expensive probes, e.g. Jellyfin `NextUp` for movie libraries. + /// [diagnostics] carries degraded legs as in [fetchGlobalHubs]. Future> fetchLibraryHubs( String libraryId, { required String libraryName, int limit = defaultHubPreviewLimit, bool includePlaybackHubs = true, MediaKind? libraryKind, + HubFetchDiagnostics? diagnostics, }); /// "More like this" recommendations for [id]. @@ -769,10 +814,18 @@ extension MediaServerClientScope on MediaServerClient { /// In-player sessions that *did* give the backend an observable crossing use /// [notifyWatchedFromPlaybackSession] instead. Future markWatchedFromPlaybackStop(MediaItem item) async { - if (!marksWatchedOnPlaybackStopped) { + final performedExplicitMark = !marksWatchedOnPlaybackStopped; + if (performedExplicitMark) { await markWatched(item); } - WatchStateNotifier().notifyWatched(item: item, isNowWatched: true, cacheServerId: cacheServerId); + WatchStateNotifier().notifyWatched( + item: item, + isNowWatched: true, + cacheServerId: cacheServerId, + // A successful report cannot prove that the backend marked the item + // played; only the explicit mutation settles this patch. + serverAcknowledged: performedExplicitMark, + ); } /// Emit the local watched event for [item] without touching the server. @@ -786,8 +839,17 @@ extension MediaServerClientScope on MediaServerClient { /// 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); + /// + /// The event remains unacknowledged because reporting success proves only + /// receipt, not that the server classified the item as played. The caller + /// keeps the returned patch id so a later explicit mark can promote it. + WatchPatchId? notifyWatchedFromPlaybackSession(MediaItem item) { + return WatchStateNotifier().notifyWatched( + item: item, + isNowWatched: true, + cacheServerId: cacheServerId, + serverAcknowledged: false, + ); } } diff --git a/lib/navigation/profile_session_screen.dart b/lib/navigation/profile_session_screen.dart index 2a1a493c..18bd4734 100644 --- a/lib/navigation/profile_session_screen.dart +++ b/lib/navigation/profile_session_screen.dart @@ -225,6 +225,10 @@ class _ProfileSessionScreenState extends State { context.read(), context.read(), context.read(), + // Created above in this same subtree, so its lifetime + // matches; the proxy reuses `previous`, so the reference + // stays valid for as long as this provider does. + watchStateStore: context.read(), isProfileBinding: () => activeProfile.isBinding, profileId: activeId, ); diff --git a/lib/providers/discover_provider.dart b/lib/providers/discover_provider.dart index f82181dc..bb7b14e6 100644 --- a/lib/providers/discover_provider.dart +++ b/lib/providers/discover_provider.dart @@ -15,14 +15,46 @@ import '../utils/app_logger.dart'; import '../utils/coalesced_load_coordinator.dart'; import '../utils/deletion_notifier.dart'; import '../utils/media_event_keys.dart'; +import '../utils/global_key_utils.dart'; import '../utils/media_hub_ordering.dart'; import '../utils/watch_state_notifier.dart'; import 'hidden_libraries_provider.dart'; import 'libraries_provider.dart'; import 'multi_server_provider.dart'; +import 'watch_state_store.dart'; enum DiscoverLoadState { initial, loading, loaded, error } +enum DiscoverRefreshOutcome { + /// Both surfaces completed without a failed or cancelled server leg. + refreshed, + + /// Some content refreshed, but at least one server leg failed or was cancelled. + degraded, + + /// At least one server leg failed and none succeeded. + failed, + + /// The pass was interrupted or had no attempted server legs. + cancelled, +} + +DiscoverRefreshOutcome _refreshOutcome({ + required Set succeededServerIds, + required Set failedServerIds, + required Set cancelledServerIds, + required bool cancelled, +}) { + if (cancelled) return DiscoverRefreshOutcome.cancelled; + if (failedServerIds.isNotEmpty) { + return succeededServerIds.isEmpty ? DiscoverRefreshOutcome.failed : DiscoverRefreshOutcome.degraded; + } + if (cancelledServerIds.isNotEmpty) { + return succeededServerIds.isEmpty ? DiscoverRefreshOutcome.cancelled : DiscoverRefreshOutcome.degraded; + } + return succeededServerIds.isEmpty ? DiscoverRefreshOutcome.cancelled : DiscoverRefreshOutcome.refreshed; +} + /// Owns the Discover tab's data: the Continue Watching row and the home hub /// list, including the refresh policy that used to live in the screen — /// durable watch events refresh only Continue Watching (one on-deck call, @@ -48,8 +80,12 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin this._libraries, { required this.profileId, required this.isProfileBinding, + WatchStateStore? watchStateStore, Future Function(String profileId, List)? syncSystemShelf, - }) : _syncSystemShelfOverride = syncSystemShelf { + // A private field cannot be a named initializing formal callers can pass. + // ignore: prefer_initializing_formals + }) : _watchStateStore = watchStateStore, + _syncSystemShelfOverride = syncSystemShelf { _loadCoordinator = CoalescedLoadCoordinator(onFull: _loadOnce, onDelta: _loadDeltaOnce); // Late server connects (reconnect after outage, slow wave) refresh // discover the same way they refresh libraries. Removed in [dispose] so a @@ -79,8 +115,46 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin final MultiServerProvider _multiServer; final HiddenLibrariesProvider _hiddenLibraries; final LibrariesProvider _libraries; + + /// Authoritative fetches tell the store which items the server re-observed, + /// so a stale local watch patch stops overriding fresh server state (#1829). + /// Optional: tests and isolated subtrees may have no store. + final WatchStateStore? _watchStateStore; final String? profileId; + /// Watermark plus store identity captured before an authoritative request. + /// Null when there is no store to reconcile against. + ({int watermark, Object epoch})? _beginObservation() { + final store = _watchStateStore; + if (store == null) return null; + return (watermark: store.observationWatermark, epoch: store.observationEpoch); + } + + /// Tell the store which items a *successful and committed* pass re-observed, + /// so their stale local watch patches stop overriding the fresh server rows. + /// + /// Only ever called once the same disposed / generation / exception checks + /// that authorise committing those rows have passed. Recording earlier would + /// suppress a patch whose fresh row was then discarded or rolled back, + /// leaving an older snapshot on screen with nothing to correct it. A + /// zero-success pass observed nothing and records nothing. + void _recordObservations( + ({int watermark, Object epoch})? observation, + List<({MediaItem item, String? clientScope})> rows, + Set succeededServerIds, + ) { + final store = _watchStateStore; + if (store == null || observation == null || succeededServerIds.isEmpty || rows.isEmpty) return; + store.recordObservations( + [ + for (final row in rows) + if (succeededServerIds.contains(row.item.serverId)) row, + ], + watermark: observation.watermark, + epoch: observation.epoch, + ); + } + /// Whether the profile binder is still wiring servers — a no-servers load /// during binding stays in the loading state instead of flashing an error, /// and a zero-success pass during binding stays in the loading state @@ -100,19 +174,21 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin String? _errorMessage; int _loadGeneration = 0; int _contentRevision = 0; + int _commitRevision = 0; + DiscoverRefreshOutcome _lastOutcome = DiscoverRefreshOutcome.cancelled; Future? _continueWatchingRefreshFuture; bool _continueWatchingRefreshQueued = false; Set _lastSeenHiddenKeys = {}; List _lastSeenLibraryOrderKeys = const []; - /// Online servers whose Continue Watching fetch succeeded in the current - /// on-deck list. Tracked separately from hubs so a transient failure in one - /// surface does not cache the other as loaded forever or force unnecessary - /// refetches. + /// Online servers whose Continue Watching legs succeeded without a failure + /// or cancellation in the current on-deck list. Tracked separately from hubs + /// so a transient failure in one surface does not cache the other as loaded + /// forever or force unnecessary refetches. Set _loadedOnDeckServerIds = {}; - /// Online servers whose home-hub fetch succeeded in the current hub list. + /// Online servers whose home-hub legs all succeeded in the current hub list. Set _loadedHubServerIds = {}; Set get _fullyLoadedServerIds => _loadedOnDeckServerIds.intersection(_loadedHubServerIds); @@ -167,25 +243,60 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin return _loadCoordinator.requestFull(); } + Future refreshNow() async { + if (isDisposed) return DiscoverRefreshOutcome.cancelled; + await _loadCoordinator.requestFull(); + if (isDisposed) return DiscoverRefreshOutcome.cancelled; + return _lastOutcome; + } + /// Whether a [load] pass is already running. The startup online-entry hook /// uses this to skip a prime that would only duplicate the load the screen /// started in `initState`. bool get isLoadInFlight => _loadCoordinator.isBusy; Future _loadOnce() async { - // Yield to the microtask queue before the first notify so a load() - // kicked off during build (the screen's initState) doesn't mark - // listening widgets dirty mid-build. - await null; - if (isDisposed) return; - ++_contentRevision; - appLogger.d('DiscoverProvider: loading content from all servers'); - _onDeckState = DiscoverLoadState.loading; - _hubsState = DiscoverLoadState.loading; - _errorMessage = null; - safeNotifyListeners(); + var outcome = DiscoverRefreshOutcome.cancelled; + var passClearedExceptionBoundary = false; + List? systemShelfPassToken; + // Observations are staged with the pass, not recorded as each leg lands: + // suppressing a patch whose fresh row is then discarded or rolled back + // would leave an older snapshot on screen with nothing to correct it. + final pendingObservations = <({MediaItem item, String? clientScope})>[]; + final observedServerIds = {}; + // Assigned inside the try, after the preparatory awaits, so the watermark + // brackets the network calls rather than the whole method. + ({int watermark, Object epoch})? observation; + final succeededServerIds = {}; + final failedServerIds = {}; + final cancelledServerIds = {}; + final previousOnDeck = _onDeck; + final previousHubs = _hubs; + final previousHasMoreContinueWatching = _hasMoreContinueWatching; + final previousLoadedOnDeckServerIds = _loadedOnDeckServerIds; + final previousLoadedHubServerIds = _loadedHubServerIds; + final previousOnDeckState = _onDeckState; + final previousHubsState = _hubsState; + final previousLoadGeneration = _loadGeneration; + final previousCommitRevision = _commitRevision; + var expectedCommitRevision = previousCommitRevision; + var onDeckFetchCompleted = false; + var hubFetchCompleted = false; + var replacedOnDeck = false; try { + // Yield to the microtask queue before the first notify so a load() + // kicked off during build (the screen's initState) doesn't mark + // listening widgets dirty mid-build. + await null; + if (isDisposed) return; + ++_contentRevision; + appLogger.d('DiscoverProvider: loading content from all servers'); + _onDeckState = DiscoverLoadState.loading; + _hubsState = DiscoverLoadState.loading; + _errorMessage = null; + safeNotifyListeners(); + if (!_multiServer.hasConnectedServers) { if (isProfileBinding()) return; throw Exception('No servers available'); @@ -202,6 +313,11 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin // On-deck and hubs fetch in parallel; on-deck is published as soon as // it lands so the hero renders while hubs are still loading. + // + // The watermark is captured immediately before the requests, after + // every preparatory await: a patch recorded later has a higher sequence + // and must survive this pass's observations. + observation = _beginObservation(); final onDeckFuture = aggregation.getOnDeckFromAllServers( limit: _continueWatchingProbeLimit, hiddenLibraryKeys: _hiddenLibraries.hiddenLibraryKeys, @@ -219,13 +335,23 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin // follow-up load is guaranteed (binding-settle prime, or // syncToOnlineServers falling through to load() while not loaded). final fetchedOnDeck = await onDeckFuture; + onDeckFetchCompleted = true; + succeededServerIds.addAll(fetchedOnDeck.succeededServerIds); + failedServerIds.addAll(fetchedOnDeck.failedServerIds); + cancelledServerIds.addAll(fetchedOnDeck.cancelledServerIds); + pendingObservations.addAll(fetchedOnDeck.observedItems); + observedServerIds.addAll(fetchedOnDeck.succeededServerIds); if (isDisposed) return; if (fetchedOnDeck.succeededServerIds.isEmpty && _onDeck.isNotEmpty) { // Keep the stale rows; the empty succeeded set makes the next status // emission refetch every server. appLogger.w('DiscoverProvider: on-deck pass failed on all servers; keeping previous items'); _onDeckState = DiscoverLoadState.loaded; - _loadedOnDeckServerIds = fetchedOnDeck.succeededServerIds; + _loadedOnDeckServerIds = _authoritativeSucceededServerIds( + fetchedOnDeck.succeededServerIds, + fetchedOnDeck.failedServerIds, + fetchedOnDeck.cancelledServerIds, + ); safeNotifyListeners(); } else if (fetchedOnDeck.succeededServerIds.isEmpty && (fetchedOnDeck.cancelledServerIds.isNotEmpty || isProfileBinding())) { @@ -235,25 +361,60 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin appLogger.d('DiscoverProvider: on-deck pass disrupted with no prior content; keeping loading state'); } else { _applyOnDeck(fetchedOnDeck.items); + ++expectedCommitRevision; + replacedOnDeck = true; _onDeckState = DiscoverLoadState.loaded; - _loadedOnDeckServerIds = fetchedOnDeck.succeededServerIds; - _loadGeneration++; + _loadedOnDeckServerIds = _authoritativeSucceededServerIds( + fetchedOnDeck.succeededServerIds, + fetchedOnDeck.failedServerIds, + fetchedOnDeck.cancelledServerIds, + ); + systemShelfPassToken = List.unmodifiable(_onDeck); safeNotifyListeners(); - unawaited(_syncSystemShelf(_onDeck)); } final fetchedHubs = await hubsFuture; + hubFetchCompleted = true; + succeededServerIds.addAll(fetchedHubs.succeededServerIds); + failedServerIds.addAll(fetchedHubs.failedServerIds); + pendingObservations.addAll(fetchedHubs.observedItems); + observedServerIds.addAll(fetchedHubs.succeededServerIds); + cancelledServerIds.addAll(fetchedHubs.cancelledServerIds); if (isDisposed) return; if (fetchedHubs.succeededServerIds.isEmpty && _hubs.isNotEmpty) { appLogger.w('DiscoverProvider: hub pass failed on all servers; keeping previous hubs'); _hubsState = DiscoverLoadState.loaded; - _loadedHubServerIds = fetchedHubs.succeededServerIds; + _loadedHubServerIds = _authoritativeSucceededServerIds( + fetchedHubs.succeededServerIds, + fetchedHubs.failedServerIds, + fetchedHubs.cancelledServerIds, + ); safeNotifyListeners(); + outcome = _refreshOutcome( + succeededServerIds: succeededServerIds, + failedServerIds: failedServerIds, + cancelledServerIds: cancelledServerIds, + cancelled: isProfileBinding(), + ); + if (replacedOnDeck && outcome != DiscoverRefreshOutcome.failed && outcome != DiscoverRefreshOutcome.cancelled) { + ++_loadGeneration; + } + passClearedExceptionBoundary = true; return; } if (fetchedHubs.succeededServerIds.isEmpty && (fetchedHubs.cancelledServerIds.isNotEmpty || isProfileBinding())) { appLogger.d('DiscoverProvider: hub pass disrupted with no prior content; keeping loading state'); + outcome = _refreshOutcome( + succeededServerIds: succeededServerIds, + failedServerIds: failedServerIds, + cancelledServerIds: cancelledServerIds, + cancelled: isProfileBinding(), + ); + if (replacedOnDeck && outcome != DiscoverRefreshOutcome.failed && outcome != DiscoverRefreshOutcome.cancelled) { + ++_loadGeneration; + } + passClearedExceptionBoundary = true; return; } @@ -261,17 +422,80 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin sortMediaHubsByLibraryOrder(filteredHubs, _libraries.libraries); appLogger.d('DiscoverProvider: ${_onDeck.length} on-deck items, ${filteredHubs.length} hubs'); - _hubs = filteredHubs; + _replaceHubs(filteredHubs); + ++expectedCommitRevision; _hubsState = DiscoverLoadState.loaded; - _loadedHubServerIds = fetchedHubs.succeededServerIds; + _loadedHubServerIds = _authoritativeSucceededServerIds( + fetchedHubs.succeededServerIds, + fetchedHubs.failedServerIds, + fetchedHubs.cancelledServerIds, + ); + outcome = _refreshOutcome( + succeededServerIds: succeededServerIds, + failedServerIds: failedServerIds, + cancelledServerIds: cancelledServerIds, + cancelled: isProfileBinding(), + ); + if (replacedOnDeck && outcome != DiscoverRefreshOutcome.failed && outcome != DiscoverRefreshOutcome.cancelled) { + ++_loadGeneration; + } + passClearedExceptionBoundary = true; safeNotifyListeners(); } catch (e) { if (isDisposed) return; + outcome = isProfileBinding() ? DiscoverRefreshOutcome.cancelled : DiscoverRefreshOutcome.failed; appLogger.e('Failed to load discover content', error: e); - _errorMessage = e.toString(); - _onDeckState = DiscoverLoadState.error; - _hubsState = DiscoverLoadState.error; + final hadPriorContent = previousOnDeck.isNotEmpty || previousHubs.isNotEmpty; + if (!hadPriorContent) { + _errorMessage = e.toString(); + _onDeckState = DiscoverLoadState.error; + _hubsState = DiscoverLoadState.error; + safeNotifyListeners(); + return; + } + + if (_commitRevision == expectedCommitRevision) { + _replaceOnDeck( + _withoutHiddenLibraries(previousOnDeck, _hiddenLibraries.hiddenLibraryKeys), + hasMore: previousHasMoreContinueWatching, + ); + _replaceHubs(_hubsWithoutHiddenLibraries(previousHubs, _hiddenLibraries.hiddenLibraryKeys)); + _loadedOnDeckServerIds = Set.of(previousLoadedOnDeckServerIds); + _loadedHubServerIds = Set.of(previousLoadedHubServerIds); + _onDeckState = previousOnDeckState; + _hubsState = previousHubsState; + _loadGeneration = previousLoadGeneration; + } else { + _filterCurrentContentForHiddenLibraries(); + } + + final hiddenServerIds = _serverIdsForLibraryKeys(_hiddenLibraries.hiddenLibraryKeys); + if (!onDeckFetchCompleted) { + _loadedOnDeckServerIds = {}; + } else { + _loadedOnDeckServerIds = Set.of(_loadedOnDeckServerIds) + ..removeAll(failedServerIds) + ..removeAll(cancelledServerIds); + } + if (!hubFetchCompleted) { + _loadedHubServerIds = {}; + } else { + _loadedHubServerIds = Set.of(_loadedHubServerIds) + ..removeAll(failedServerIds) + ..removeAll(cancelledServerIds); + } + _loadedOnDeckServerIds = Set.of(_loadedOnDeckServerIds)..removeAll(hiddenServerIds); + _loadedHubServerIds = Set.of(_loadedHubServerIds)..removeAll(hiddenServerIds); + _errorMessage = null; + _onDeckState = DiscoverLoadState.loaded; + _hubsState = DiscoverLoadState.loaded; safeNotifyListeners(); + } finally { + _lastOutcome = outcome; + if (passClearedExceptionBoundary && _commitRevision == expectedCommitRevision && !isDisposed) { + _recordObservations(observation, pendingObservations, observedServerIds); + if (systemShelfPassToken != null) unawaited(_syncSystemShelf(systemShelfPassToken)); + } } } @@ -297,6 +521,7 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin final useGlobalHubs = settings.read(SettingsService.useGlobalHubs); final aggregation = _multiServer.aggregationService; + final observation = _beginObservation(); final Future onDeckFuture = onDeckIds.isEmpty ? Future.value() : aggregation.getOnDeckFromAllServers( @@ -316,6 +541,11 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin final freshOnDeck = await onDeckFuture; final freshHubs = await hubsFuture; if (isDisposed) return; + // Staged, not recorded here: the merge below can still throw, and the + // catch keeps the previous rows. Suppressing patches against rows that + // were then discarded would strand an older snapshot on screen. + final pendingObservations = <({MediaItem item, String? clientScope})>[]; + final observedServerIds = {}; if (freshOnDeck != null) { final hadMore = _hasMoreContinueWatching; @@ -329,7 +559,9 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin // The stored list is already trimmed, so the merge can't see old items // past the cap — a previously-true "more" affordance stays true. if (hadMore) _hasMoreContinueWatching = true; - _loadedOnDeckServerIds = {..._loadedOnDeckServerIds, ...freshOnDeck.succeededServerIds}; + _loadedOnDeckServerIds = {..._loadedOnDeckServerIds, ...freshOnDeck.succeededServerIds} + ..removeAll(freshOnDeck.failedServerIds) + ..removeAll(freshOnDeck.cancelledServerIds); // No _loadGeneration bump: a delta behaves like the background Continue // Watching refresh (the hero clamps instead of resetting). } @@ -341,10 +573,22 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin ..._filterDiscoverHubs(freshHubs.hubs), ]; sortMediaHubsByLibraryOrder(mergedHubs, _libraries.libraries); - _hubs = mergedHubs; - _loadedHubServerIds = {..._loadedHubServerIds, ...succeededHubIds}; + _replaceHubs(mergedHubs); + _loadedHubServerIds = {..._loadedHubServerIds, ...succeededHubIds} + ..removeAll(freshHubs.failedServerIds) + ..removeAll(freshHubs.cancelledServerIds); } + if (freshOnDeck != null) { + pendingObservations.addAll(freshOnDeck.observedItems); + observedServerIds.addAll(freshOnDeck.succeededServerIds); + } + if (freshHubs != null) { + pendingObservations.addAll(freshHubs.observedItems); + observedServerIds.addAll(freshHubs.succeededServerIds); + } + _recordObservations(observation, pendingObservations, observedServerIds); + appLogger.d('DiscoverProvider: ${_onDeck.length} on-deck items, ${_hubs.length} hubs after merging $ids'); safeNotifyListeners(); unawaited(_syncSystemShelf(_onDeck)); @@ -369,6 +613,69 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin }).toList(); } + Set _authoritativeSucceededServerIds( + Set succeededServerIds, + Set failedServerIds, + Set cancelledServerIds, + ) { + return Set.of(succeededServerIds) + ..removeAll(failedServerIds) + ..removeAll(cancelledServerIds); + } + + List _withoutHiddenLibraries(List items, Set hiddenLibraryKeys) { + if (hiddenLibraryKeys.isEmpty) return items; + return items.where((item) { + final libraryKey = item.libraryGlobalKey; + return libraryKey == null || !hiddenLibraryKeys.contains(libraryKey); + }).toList(); + } + + List _hubsWithoutHiddenLibraries(List hubs, Set hiddenLibraryKeys) { + if (hiddenLibraryKeys.isEmpty) return hubs; + final filteredHubs = []; + for (final hub in hubs) { + final filteredItems = hub.items.where((item) { + var libraryKey = item.libraryGlobalKey; + final libraryId = item.libraryId; + final serverId = item.serverId ?? hub.serverId; + if (libraryKey == null && libraryId != null && serverId != null) { + libraryKey = buildGlobalKey(ServerId(serverId), libraryId); + } + return libraryKey == null || !hiddenLibraryKeys.contains(libraryKey); + }).toList(); + if (filteredItems.isNotEmpty) { + filteredHubs.add( + filteredItems.length == hub.items.length + ? hub + : hub.copyWith(items: filteredItems, size: filteredItems.length), + ); + } + } + return filteredHubs; + } + + Set _serverIdsForLibraryKeys(Set libraryKeys) { + final serverIds = {}; + for (final key in libraryKeys) { + final parsed = parseGlobalKey(key); + if (parsed != null) serverIds.add(parsed.serverId); + } + return serverIds; + } + + void _filterCurrentContentForHiddenLibraries() { + final hiddenLibraryKeys = _hiddenLibraries.hiddenLibraryKeys; + final filteredOnDeck = _withoutHiddenLibraries(_onDeck, hiddenLibraryKeys); + if (filteredOnDeck.length != _onDeck.length) { + _replaceOnDeck(filteredOnDeck, hasMore: _hasMoreContinueWatching); + } + final filteredHubs = _hubsWithoutHiddenLibraries(_hubs, hiddenLibraryKeys); + if (!listEquals(filteredHubs, _hubs)) { + _replaceHubs(filteredHubs); + } + } + /// Background refresh of Continue Watching only. Concurrent events coalesce /// into the active request plus at most one trailing fresh request. Future refreshContinueWatching() { @@ -399,17 +706,30 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin if (!_multiServer.hasConnectedServers) return; final revision = _contentRevision; final hiddenKeys = Set.of(_hiddenLibraries.hiddenLibraryKeys); + final observation = _beginObservation(); final fetched = await _multiServer.aggregationService.getOnDeckFromAllServers( limit: _continueWatchingProbeLimit, hiddenLibraryKeys: hiddenKeys, ); if (isDisposed) return; if (revision != _contentRevision) { + // A newer mutation landed while this was in flight, so these rows are + // discarded — recording them would suppress patches against data the + // user never sees. _continueWatchingRefreshQueued = true; return; } + _loadedOnDeckServerIds = _authoritativeSucceededServerIds( + fetched.succeededServerIds, + fetched.failedServerIds, + fetched.cancelledServerIds, + ); + if (fetched.succeededServerIds.isEmpty) { + safeNotifyListeners(); + return; + } _applyOnDeck(fetched.items); - _loadedOnDeckServerIds = fetched.succeededServerIds; + _recordObservations(observation, fetched.observedItems, fetched.succeededServerIds); safeNotifyListeners(); unawaited(_syncSystemShelf(_onDeck)); } catch (e) { @@ -422,9 +742,13 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin if (!_multiServer.hasConnectedServers) return const []; await _hiddenLibraries.ensureInitialized(); if (isDisposed) return const []; + final observation = _beginObservation(); final fetched = await _multiServer.aggregationService.getOnDeckFromAllServers( hiddenLibraryKeys: _hiddenLibraries.hiddenLibraryKeys, ); + // "View All" renders through the same overlay as the row it expands, so + // it has to reconcile too or the stale patch simply reappears there. + _recordObservations(observation, fetched.observedItems, fetched.succeededServerIds); return fetched.items; } @@ -448,7 +772,7 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin void _updateItemInLists(String sourceGlobalKey, MediaItem updatedItem) { final onDeckIndex = _onDeck.indexWhere((item) => item.globalKey == sourceGlobalKey); if (onDeckIndex != -1) { - _onDeck = List.of(_onDeck)..[onDeckIndex] = updatedItem; + _replaceOnDeck(List.of(_onDeck)..[onDeckIndex] = updatedItem, hasMore: _hasMoreContinueWatching); } for (var i = 0; i < _hubs.length; i++) { @@ -457,15 +781,25 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin if (itemIndex != -1) { final newItems = List.from(hub.items); newItems[itemIndex] = updatedItem; - _hubs = List.of(_hubs)..[i] = hub.copyWith(items: newItems); + _replaceHubs(List.of(_hubs)..[i] = hub.copyWith(items: newItems)); } } } void _applyOnDeck(List fetched) { final hasMore = fetched.length > continueWatchingPreviewLimit; - _onDeck = hasMore ? fetched.take(continueWatchingPreviewLimit).toList() : fetched; + _replaceOnDeck(hasMore ? fetched.take(continueWatchingPreviewLimit).toList() : fetched, hasMore: hasMore); + } + + void _replaceOnDeck(List onDeck, {required bool hasMore}) { + _onDeck = onDeck; _hasMoreContinueWatching = hasMore; + ++_commitRevision; + } + + void _replaceHubs(List hubs) { + _hubs = hubs; + ++_commitRevision; } // --- Event reactions ----------------------------------------------------- @@ -481,7 +815,10 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin final viewOffset = event.viewOffset; final index = _onDeck.indexWhere((item) => item.globalKey == event.globalKey); if (viewOffset != null && index != -1 && _onDeck[index].viewOffsetMs != viewOffset) { - _onDeck = List.of(_onDeck)..[index] = _onDeck[index].copyWith(viewOffsetMs: viewOffset); + _replaceOnDeck( + List.of(_onDeck)..[index] = _onDeck[index].copyWith(viewOffsetMs: viewOffset), + hasMore: _hasMoreContinueWatching, + ); safeNotifyListeners(); unawaited(_syncSystemShelf(_onDeck)); } @@ -507,7 +844,7 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin void _evictFromOnDeck(bool Function(MediaItem item) matches) { final remaining = _onDeck.where((item) => !matches(item)).toList(); if (remaining.length == _onDeck.length) return; - _onDeck = remaining; + _replaceOnDeck(remaining, hasMore: _hasMoreContinueWatching); safeNotifyListeners(); unawaited(_syncSystemShelf(_onDeck)); } @@ -533,14 +870,14 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin var changed = false; final remainingOnDeck = _onDeck.where((item) => !affected(item)).toList(); if (remainingOnDeck.length != _onDeck.length) { - _onDeck = remainingOnDeck; + _replaceOnDeck(remainingOnDeck, hasMore: _hasMoreContinueWatching); changed = true; } for (var i = 0; i < _hubs.length; i++) { final hub = _hubs[i]; final newItems = hub.items.where((item) => !affected(item)).toList(); if (newItems.length != hub.items.length) { - _hubs = List.of(_hubs)..[i] = hub.copyWith(items: newItems); + _replaceHubs(List.of(_hubs)..[i] = hub.copyWith(items: newItems)); changed = true; } } @@ -565,7 +902,7 @@ class DiscoverProvider extends ChangeNotifier with DisposableChangeNotifierMixin final sortedHubs = List.from(_hubs); if (!sortMediaHubsByLibraryOrder(sortedHubs, _libraries.libraries)) return; - _hubs = sortedHubs; + _replaceHubs(sortedHubs); safeNotifyListeners(); } diff --git a/lib/providers/offline_watch_provider.dart b/lib/providers/offline_watch_provider.dart index c4abd3f5..0290d1ec 100644 --- a/lib/providers/offline_watch_provider.dart +++ b/lib/providers/offline_watch_provider.dart @@ -144,12 +144,18 @@ class OfflineWatchProvider extends ChangeNotifier with DisposableChangeNotifierM required String itemId, required bool isNowWatched, required WatchStateChangeType changeType, + required WatchPatchId patchId, String? cacheServerId, }) { final globalKey = buildGlobalKey(ServerId(serverId), itemId); final metadata = _downloadProvider.getMetadata(globalKey); if (metadata != null) { - WatchStateNotifier().notifyWatched(item: metadata, isNowWatched: isNowWatched, cacheServerId: cacheServerId); + WatchStateNotifier().notifyWatched( + item: metadata, + isNowWatched: isNowWatched, + cacheServerId: cacheServerId, + patchId: patchId, + ); } else { // Fallback: emit minimal event without parent chain. WatchStateNotifier().notify( @@ -161,6 +167,7 @@ class OfflineWatchProvider extends ChangeNotifier with DisposableChangeNotifierM parentChain: [], mediaType: 'unknown', isNowWatched: isNowWatched, + patchId: patchId, ), ); } @@ -170,13 +177,14 @@ class OfflineWatchProvider extends ChangeNotifier with DisposableChangeNotifierM /// /// This queues the action for sync when online and emits a [WatchStateEvent]. Future markAsWatched({required ServerId serverId, required String itemId}) async { - final cacheServerId = await _syncService.queueMarkWatched(serverId: serverId, itemId: itemId); + final queued = await _syncService.queueMarkWatched(serverId: serverId, itemId: itemId); _emitWatchStateChange( serverId: serverId, itemId: itemId, isNowWatched: true, changeType: WatchStateChangeType.watched, - cacheServerId: cacheServerId, + cacheServerId: queued.clientScopeId, + patchId: WatchPatchId.offlineAction(profileId: queued.profileId, rowId: queued.rowId, revision: queued.revision), ); safeNotifyListeners(); _autoDeleteIfWatched(serverId, itemId); @@ -212,13 +220,14 @@ class OfflineWatchProvider extends ChangeNotifier with DisposableChangeNotifierM /// /// This queues the action for sync when online and emits a [WatchStateEvent]. Future markAsUnwatched({required ServerId serverId, required String itemId}) async { - final cacheServerId = await _syncService.queueMarkUnwatched(serverId: serverId, itemId: itemId); + final queued = await _syncService.queueMarkUnwatched(serverId: serverId, itemId: itemId); _emitWatchStateChange( serverId: serverId, itemId: itemId, isNowWatched: false, changeType: WatchStateChangeType.unwatched, - cacheServerId: cacheServerId, + cacheServerId: queued.clientScopeId, + patchId: WatchPatchId.offlineAction(profileId: queued.profileId, rowId: queued.rowId, revision: queued.revision), ); safeNotifyListeners(); } diff --git a/lib/providers/watch_state_store.dart b/lib/providers/watch_state_store.dart index f4cac153..309f8b72 100644 --- a/lib/providers/watch_state_store.dart +++ b/lib/providers/watch_state_store.dart @@ -26,19 +26,57 @@ class HydratedWatchStatePatch { }); } +/// Key for [WatchStateStore]'s observation map: a global key already resolved +/// through the client scope its request was issued under. A plain `String` +/// would work, but the wrapper keeps the two key spaces from being confused +/// with the patch maps' keys at a glance. +class _ObservationKey { + final String value; + + const _ObservationKey(this.value); + + @override + bool operator ==(Object other) => other is _ObservationKey && other.value == value; + + @override + int get hashCode => value.hashCode; +} + class _WatchStatePatchEntry { final WatchStateSnapshot patch; final int updatedAt; final int sequence; final bool isSessionEvent; + /// Whether the server had accepted this exact state when the entry was + /// recorded. Only an acknowledged entry may be superseded by a later + /// authoritative read; an unacknowledged one is a write still owed to the + /// server and must outlive any read (#1829). + final bool serverAcknowledged; + + /// Identity used to promote this exact entry once its write settles. + /// Promotion never matches by global key: a same-item rewatch may already + /// have replaced the entry, and it must not be promoted in its place. + final WatchPatchId? patchId; + const _WatchStatePatchEntry( this.patch, { required this.updatedAt, required this.sequence, required this.isSessionEvent, + this.serverAcknowledged = false, + this.patchId, }); + _WatchStatePatchEntry acknowledgedAt(int sequence) => _WatchStatePatchEntry( + patch, + updatedAt: updatedAt, + sequence: sequence, + isSessionEvent: true, + serverAcknowledged: true, + patchId: patchId, + ); + bool isNewerThan(_WatchStatePatchEntry other) { if (updatedAt != other.updatedAt) return updatedAt > other.updatedAt; if (isSessionEvent != other.isSessionEvent) return isSessionEvent; @@ -51,10 +89,12 @@ class _WatchStatePatchEntry { other.patch == patch && other.updatedAt == updatedAt && other.sequence == sequence && - other.isSessionEvent == isSessionEvent; + other.isSessionEvent == isSessionEvent && + other.serverAcknowledged == serverAcknowledged && + other.patchId == patchId; @override - int get hashCode => Object.hash(patch, updatedAt, sequence, isSessionEvent); + int get hashCode => Object.hash(patch, updatedAt, sequence, isSessionEvent, serverAcknowledged, patchId); } /// The single session-local layer for watch-state freshness. @@ -69,50 +109,175 @@ class _WatchStatePatchEntry { class WatchStateStore extends ChangeNotifier with DisposableChangeNotifierMixin { WatchStateStore() { _subscription = WatchStateNotifier().stream.listen(_onWatchStateEvent); + // A second, separately owned subscription: promotions are deliberately + // not WatchStateEvents, and BaseNotifier carries exactly one typed + // stream, so they cannot arrive on the channel above. + _promotionSubscription = WatchPatchPromotionNotifier().stream.listen(_onPromotion); } StreamSubscription? _subscription; + StreamSubscription? _promotionSubscription; final Map _patches = {}; final Map _hydratedPatches = {}; String? _activeProfileId; Map _activeClientScopesByServer = const {}; int _sequence = 0; - _WatchStatePatchEntry? _exactEntryFor(String globalKey) { - final session = _patches[globalKey]; + /// Highest watermark at which a qualifying authoritative response returned a + /// key, under the client scope its request was issued for. + /// + /// This records *that* the server re-observed an item, never *what* it + /// observed. Storing the observed state instead would need full fidelity + /// ([WatchStateSnapshot] cannot hold a container's leaf counts) and would + /// make two responses sharing a watermark first-response-wins. A watermark + /// has neither problem: it is a single int, and [recordObservations] keeps + /// the max, so response order cannot matter. + final Map<_ObservationKey, int> _observedAt = {}; + + /// Snapshot the current sequence *before* issuing an authoritative request; + /// pass it back to [recordObservations] when that request succeeds. + int get observationWatermark => _sequence; + + /// Identifies this store instance. Watermarks are store-local, but clients — + /// and their in-flight requests — outlive a profile switch, so a response + /// minted against the previous store must not be applied to this one. + final Object _epoch = Object(); + + Object get observationEpoch => _epoch; + + _WatchStatePatchEntry? _exactEntryFor(String globalKey, int? observedAt) { + final session = _live(_patches[globalKey], observedAt); final hydrated = _hydratedPatches[globalKey]; if (session == null) return hydrated; if (hydrated == null) return session; return session.isNewerThan(hydrated) ? session : hydrated; } - _WatchStatePatchEntry? _entryFor(String globalKey) { + /// Drops an acknowledged session entry the server has re-observed since. + /// + /// Only acknowledged entries are eligible: an unacknowledged one is a write + /// still owed to the server, and a read must never retire it. Hydrated + /// entries are the persisted owed-write layer and are likewise untouched — + /// filtering happens per layer, before [_exactEntryFor] chooses, so a + /// suppressed session entry falls back to an older hydrated one instead of + /// suppressing both. + static _WatchStatePatchEntry? _live(_WatchStatePatchEntry? entry, int? observedAt) { + if (entry == null) return null; + if (!entry.serverAcknowledged || observedAt == null) return entry; + return observedAt >= entry.sequence ? null : entry; + } + + /// Candidate keys for [globalKey], in the order [_entryFor] consults them: + /// the active client scope first, then the public fallback. + List _candidateKeys(String globalKey) { final parsed = parseGlobalKey(globalKey); if (parsed != null) { final scoped = _activeClientScopesByServer[parsed.serverId]; if (scoped != null && scoped.isNotEmpty) { - return _exactEntryFor(buildGlobalKey(ServerId(scoped), parsed.ratingKey)) ?? _exactEntryFor(globalKey); + return [buildGlobalKey(ServerId(scoped), parsed.ratingKey), globalKey]; } } - return _exactEntryFor(globalKey); + return [globalKey]; } + _WatchStatePatchEntry? _entryFor(String globalKey, [int? observedAt]) { + for (final key in _candidateKeys(globalKey)) { + final entry = _exactEntryFor(key, observedAt); + if (entry != null) return entry; + } + return null; + } + + /// The watermark at which [globalKey] was last authoritatively observed. + /// + /// Resolved through the *primary* candidate only — the active client scope + /// when there is one, else the public key. Deliberately not a max over both: + /// on a user-scoped backend the public server id is shared between users, so + /// letting a public observation satisfy a scoped patch would suppress + /// another user's watch state. + int? _observationFor(String globalKey) => _observedAt[_ObservationKey(_candidateKeys(globalKey).first)]; + @visibleForTesting - WatchStateSnapshot? patchForGlobalKey(String globalKey) => _entryFor(globalKey)?.patch; + WatchStateSnapshot? patchForGlobalKey(String globalKey) => _entryFor(globalKey, _observationFor(globalKey))?.patch; WatchStateSnapshot? patchForItem(MediaItem item) { - var best = _entryFor(item.globalKey); + if (_patches.isEmpty && _hydratedPatches.isEmpty) return null; + // One barrier for the whole resolution: an authoritative read of *this* + // item already incorporates any container mark that preceded it, so the + // ancestor candidates are judged against the item's own observation too. + // Suppressing only the item's own entry would let an older season patch + // win and render something worse than either value (#1829 review). + final observedAt = _observationFor(item.globalKey); + var best = _entryFor(item.globalKey, observedAt); if (item.parentChain.isNotEmpty) { final serverId = serverIdOrNull(item.serverId); for (final parentId in item.parentChain) { // Mirror MediaItem.globalKey's bare-id fallback when serverId is missing. - final entry = _entryFor(serverId != null ? buildGlobalKey(serverId, parentId) : parentId); + final entry = _entryFor(serverId != null ? buildGlobalKey(serverId, parentId) : parentId, observedAt); if (entry != null && (best == null || entry.isNewerThan(best))) best = entry; } } return best?.patch; } + /// Record that a successful authoritative response returned [rows]. + /// + /// [watermark] must be [observationWatermark] captured immediately before + /// the network call, and [epoch] the [observationEpoch] of the store that + /// captured it. Entries recorded *during* the request keep a higher + /// sequence and survive, so a local action taken mid-flight always wins. + /// + /// Each row carries the immutable cache scope of the client that fetched + /// it: on a user-scoped backend the public server id is shared, so a row + /// must only ever reconcile the scope it actually came from. + void recordObservations( + Iterable<({MediaItem item, String? clientScope})> rows, { + required int watermark, + required Object epoch, + }) { + // Watermarks are store-local, but clients — and their in-flight requests + // — outlive a profile switch, so a response minted against the previous + // store carries a sequence that means nothing here. + if (!identical(epoch, _epoch)) return; + if (_patches.isEmpty) return; + var changed = false; + for (final row in rows) { + final item = row.item; + // An observation is only ever consulted to suppress a patch, so keep + // one only while some patch could still apply to this item — its own, + // or an ancestor's, which the barrier in [patchForItem] also judges + // against this key. That bounds the map by the items the user acted + // on rather than by everything ever fetched. + if (!_hasSuppressibleEntry(item)) continue; + final scope = row.clientScope; + final key = _ObservationKey( + scope != null && scope.isNotEmpty && scope != item.serverId + ? buildGlobalKey(ServerId(scope), item.id) + : item.globalKey, + ); + final existing = _observedAt[key]; + if (existing != null && existing >= watermark) continue; + _observedAt[key] = watermark; + changed = true; + } + if (changed) safeNotifyListeners(); + } + + bool _hasSuppressibleEntry(MediaItem item) { + for (final key in _candidateKeys(item.globalKey)) { + if (_patches[key]?.serverAcknowledged ?? false) return true; + } + if (item.parentChain.isEmpty) return false; + final serverId = serverIdOrNull(item.serverId); + for (final parentId in item.parentChain) { + final parentKey = serverId != null ? buildGlobalKey(serverId, parentId) : parentId; + for (final key in _candidateKeys(parentKey)) { + if (_patches[key]?.serverAcknowledged ?? false) return true; + } + } + return false; + } + MediaItem apply(MediaItem item) { return applyPatch(item, patchForItem(item)); } @@ -127,9 +292,11 @@ class WatchStateStore extends ChangeNotifier with DisposableChangeNotifierMixin void setActiveProfileId(String? profileId) { if (_activeProfileId == profileId) return; _activeProfileId = profileId; - if (_patches.isEmpty && _hydratedPatches.isEmpty) return; + if (_patches.isEmpty && _hydratedPatches.isEmpty && _observedAt.isEmpty) return; _patches.clear(); _hydratedPatches.clear(); + // Observations are only meaningful against the patches they suppress. + _observedAt.clear(); safeNotifyListeners(); } @@ -188,14 +355,56 @@ class WatchStateStore extends ChangeNotifier with DisposableChangeNotifierMixin updatedAt: DateTime.now().millisecondsSinceEpoch, sequence: ++_sequence, isSessionEvent: true, + serverAcknowledged: event.serverAcknowledged, + patchId: event.patchId, ); safeNotifyListeners(); } + /// The write behind [promotion] has settled, so its entry may now be + /// superseded by a later authoritative read. + /// + /// Matching is by patch identity, never by key: an unknown id — because a + /// newer action replaced the entry, or because the process restarted — is a + /// deliberate no-op. The fresh sequence matters as much as the flag: an + /// observation captured *before* the write landed must not suppress the + /// entry it predates. + void _onPromotion(WatchPatchPromotion promotion) { + final sessionKey = _keyForPatchId(_patches, promotion.patchId); + if (sessionKey != null) { + _patches[sessionKey] = _patches[sessionKey]!.acknowledgedAt(++_sequence); + safeNotifyListeners(); + return; + } + + // A hydrated entry is the persisted owed-write layer. Once the write has + // landed it no longer belongs there, and leaving it would make it + // immortal, since suppression only ever considers session entries. Move + // it — but never over a newer session entry on the same key, which would + // defeat the exact-identity rule promotion exists for. + final hydratedKey = _keyForPatchId(_hydratedPatches, promotion.patchId); + if (hydratedKey == null) return; + final promoted = _hydratedPatches.remove(hydratedKey)!; + final existing = _patches[hydratedKey]; + if (existing == null || existing.patchId == promotion.patchId) { + _patches[hydratedKey] = promoted.acknowledgedAt(++_sequence); + } + safeNotifyListeners(); + } + + static String? _keyForPatchId(Map entries, WatchPatchId patchId) { + for (final entry in entries.entries) { + if (entry.value.patchId == patchId) return entry.key; + } + return null; + } + @override void dispose() { _subscription?.cancel(); _subscription = null; + _promotionSubscription?.cancel(); + _promotionSubscription = null; super.dispose(); } } diff --git a/lib/screens/discover_screen.dart b/lib/screens/discover_screen.dart index 7de2d875..b3904a91 100644 --- a/lib/screens/discover_screen.dart +++ b/lib/screens/discover_screen.dart @@ -50,6 +50,7 @@ import '../utils/formatters.dart'; import '../utils/hub_icons.dart'; import '../utils/media_navigation_helper.dart'; import '../utils/provider_extensions.dart'; +import '../utils/snackbar_helper.dart'; import '../utils/video_player_navigation.dart'; import '../utils/layout_constants.dart'; import '../utils/platform_detector.dart'; @@ -756,7 +757,23 @@ class _DiscoverScreenState extends State onNavigateLeft: _navigateToSidebar, onNavigateDown: _focusContentFromAppBar, actions: [ - FocusableAction(icon: Symbols.refresh_rounded, iconColor: foregroundColor, onPressed: _discover.load), + FocusableAction( + icon: Symbols.refresh_rounded, + iconColor: foregroundColor, + onPressed: () async { + final outcome = await _discover.refreshNow(); + if (!context.mounted) return; + switch (outcome) { + case DiscoverRefreshOutcome.failed: + showErrorSnackBar(context, t.errors.unableToLoad(context: t.discover.title)); + case DiscoverRefreshOutcome.degraded: + appLogger.w('Discover refresh completed with partial server failures'); + case DiscoverRefreshOutcome.cancelled: + case DiscoverRefreshOutcome.refreshed: + break; + } + }, + ), // Watch Together FocusableAction( onPressed: () => diff --git a/lib/services/data_aggregation_service.dart b/lib/services/data_aggregation_service.dart index 7988a84d..c116a102 100644 --- a/lib/services/data_aggregation_service.dart +++ b/lib/services/data_aggregation_service.dart @@ -16,16 +16,38 @@ import '../utils/media_server_http_client.dart'; import 'local_playback_history.dart'; import 'multi_server_manager.dart'; +/// A row a successful response actually returned, paired with the immutable +/// cache scope of the client that fetched it. +/// +/// The scope cannot be recovered later: flattening and dedup lose which +/// client produced a row, and a user-scoped backend's public server id is +/// shared between users — so reconciling by public key could suppress a +/// different user's watch state. +typedef ObservedRow = ({MediaItem item, String? clientScope}); + +/// [observedItems] are the rows the successful responses actually returned, +/// *before* dedup or limiting. Continue Watching dedup can drop one server's +/// copy in favour of another's, and a dropped copy would otherwise leave that +/// key unreconciled even though its server did return it (#1829). typedef OnDeckAggregationResult = ({ List items, + List observedItems, Set succeededServerIds, Set cancelledServerIds, + Set failedServerIds, +}); +typedef HubAggregationResult = ({ + List hubs, + List observedItems, + Set succeededServerIds, + Set cancelledServerIds, + Set failedServerIds, }); -typedef HubAggregationResult = ({List hubs, Set succeededServerIds, Set cancelledServerIds}); typedef LibraryAggregationResult = ({ List libraries, Set succeededServerIds, Set cancelledServerIds, + Set failedServerIds, }); typedef SearchAggregationResult = ({ List items, @@ -165,6 +187,7 @@ class DataAggregationService { libraries: const [], succeededServerIds: const {}, cancelledServerIds: const {}, + failedServerIds: const {}, ); } final fetched = await _fanOut( @@ -176,6 +199,7 @@ class DataAggregationService { libraries: fetched.items, succeededServerIds: fetched.succeededServerIds, cancelledServerIds: fetched.cancelledServerIds, + failedServerIds: fetched.failedServerIds, ); } @@ -190,14 +214,27 @@ class DataAggregationService { }) async { final clients = _clientsFor(serverIds); if (clients.isEmpty) { - appLogger.w('No online servers available for fetching on deck'); - return (items: const [], succeededServerIds: const {}, cancelledServerIds: const {}); + return ( + items: const [], + observedItems: const [], + succeededServerIds: const {}, + cancelledServerIds: const {}, + failedServerIds: const {}, + ); } + final observedRows = []; final fetched = await _fanOut( clients, failureMessage: (serverId) => 'Failed on-deck fetch from $serverId', - fetch: (_, client) => client.fetchContinueWatching(count: limit), + fetch: (_, client) async { + final rows = await client.fetchContinueWatching(count: limit); + // Capture the scope here: after the fan-out flattens and dedup runs, + // there is no way back to the client that produced a row. + final scope = client.cacheServerId; + observedRows.addAll([for (final row in rows) (item: row, clientScope: scope)]); + return rows; + }, ); // Filter out items from hidden libraries var filteredOnDeck = _withoutHiddenLibraries(fetched.items, hiddenLibraryKeys); @@ -216,8 +253,12 @@ class DataAggregationService { return ( items: items, + // Pre-dedup, pre-limit: a copy dropped by dedup or trimmed by the limit + // was still authoritatively returned by its server. + observedItems: observedRows, succeededServerIds: fetched.succeededServerIds, cancelledServerIds: fetched.cancelledServerIds, + failedServerIds: fetched.failedServerIds, ); } @@ -436,7 +477,13 @@ class DataAggregationService { final clients = _clientsFor(serverIds); if (clients.isEmpty) { appLogger.w('No online servers available for fetching hubs'); - return (hubs: const [], succeededServerIds: const {}, cancelledServerIds: const {}); + return ( + hubs: const [], + observedItems: const [], + succeededServerIds: const {}, + cancelledServerIds: const {}, + failedServerIds: const {}, + ); } // Home layout needs the library list for every client: fallback backends @@ -444,9 +491,13 @@ class DataAggregationService { // (Plex) need it to detect visible music libraries, whose hubs the // global-hub endpoint excludes. One `fetchLibraries` per server, served // from the per-backend API cache when warm. - final libraries = useGlobalHubs - ? _groupLibrariesByServer((await getMediaLibrariesFromAllServers(serverIds: serverIds)).libraries) - : null; + final libraryFetch = useGlobalHubs ? await getMediaLibrariesFromAllServers(serverIds: serverIds) : null; + final libraries = libraryFetch == null ? null : _groupLibrariesByServer(libraryFetch.libraries); + final legSucceededServerIds = {}; + final legFailedServerIds = {if (libraryFetch != null) ...libraryFetch.failedServerIds}; + final legCancelledServerIds = {if (libraryFetch != null) ...libraryFetch.cancelledServerIds}; + final globalDiagnosticsByServer = {}; + final observedRows = []; final fetched = await _fanOut( clients, @@ -454,6 +505,7 @@ class DataAggregationService { fetch: (serverId, client) async { final serverLibraries = libraries?[serverId]; final shouldUseGlobalHubs = useGlobalHubs && client.capabilities.richHubs; + final prefetchDegraded = legFailedServerIds.contains(serverId) || legCancelledServerIds.contains(serverId); final hubItemLimit = limit ?? defaultHubPreviewLimit; List hubs; if (shouldUseGlobalHubs) { @@ -461,36 +513,91 @@ class DataAggregationService { // Spreading `...await a, ...await b` into one list literal evaluates // them in order, which serialised the music rows behind the global // hub round trip. - final globalFuture = client.fetchGlobalHubs(limit: hubItemLimit, includePlaybackHubs: includePlaybackHubs); + final globalDiagnostics = HubFetchDiagnostics(); + globalDiagnosticsByServer[serverId] = globalDiagnostics; + final globalFuture = client.fetchGlobalHubs( + limit: hubItemLimit, + includePlaybackHubs: includePlaybackHubs, + diagnostics: globalDiagnostics, + ); // Plex's promoted/global hub endpoint never includes music // libraries — append their per-library hubs so music rows - // reach home. No-op (zero extra calls) without a visible - // music library. - final musicFuture = _fetchLibraryHubsForClient( - client, - limit: hubItemLimit, - hiddenLibraryKeys: hiddenLibraryKeys, - includePlaybackHubs: includePlaybackHubs, - libraries: serverLibraries ?? const [], - kinds: const {MediaKind.artist}, - ); - hubs = [...await globalFuture, ...await musicFuture]; + // reach home. A failed prefetch cannot establish that there + // are no visible music libraries, so it is not a successful no-op. + final musicFuture = prefetchDegraded + ? null + : _fetchLibraryHubsForClient( + client, + limit: hubItemLimit, + hiddenLibraryKeys: hiddenLibraryKeys, + includePlaybackHubs: includePlaybackHubs, + libraries: serverLibraries ?? const [], + kinds: const {MediaKind.artist}, + ); + Object? globalError; + StackTrace? globalStackTrace; + List globalHubs = const []; + try { + globalHubs = await globalFuture; + } catch (error, stackTrace) { + globalDiagnostics.recordFailure(error); + globalError = error; + globalStackTrace = stackTrace; + } + final music = musicFuture == null ? null : await musicFuture; + if (globalHubs.isNotEmpty || (!globalDiagnostics.failed && !globalDiagnostics.cancelled)) { + legSucceededServerIds.add(serverId); + } + if (music != null) { + if (music.succeeded) legSucceededServerIds.add(serverId); + if (music.failed) legFailedServerIds.add(serverId); + if (music.cancelled) legCancelledServerIds.add(serverId); + } + if (globalError != null) Error.throwWithStackTrace(globalError, globalStackTrace!); + hubs = [...globalHubs, if (music != null) ...music.hubs]; } else { - hubs = await _fetchLibraryHubsForClient( + // A fallback backend cannot run a hub leg without the libraries its + // prefetch failed to discover. + if (useGlobalHubs && prefetchDegraded) return const []; + final libraryHubs = await _fetchLibraryHubsForClient( client, limit: hubItemLimit, hiddenLibraryKeys: hiddenLibraryKeys, includePlaybackHubs: includePlaybackHubs, libraries: useGlobalHubs ? serverLibraries : null, ); + if (libraryHubs.succeeded || (!libraryHubs.failed && !libraryHubs.cancelled)) { + legSucceededServerIds.add(serverId); + } + if (libraryHubs.failed) legFailedServerIds.add(serverId); + if (libraryHubs.cancelled) legCancelledServerIds.add(serverId); + hubs = libraryHubs.hubs; } - return _postProcessHubs(hubs, serverId: ServerId(serverId), hiddenLibraryKeys: hiddenLibraryKeys); + final processed = _postProcessHubs(hubs, serverId: ServerId(serverId), hiddenLibraryKeys: hiddenLibraryKeys); + final scope = client.cacheServerId; + observedRows.addAll([ + for (final hub in processed) + for (final item in hub.items) (item: item, clientScope: scope), + ]); + return processed; }, ); + for (final entry in globalDiagnosticsByServer.entries) { + if (entry.value.failed) legFailedServerIds.add(entry.key); + if (entry.value.cancelled) legCancelledServerIds.add(entry.key); + } + final all = fetched.items; final hubs = limit != null && limit < all.length ? all.sublist(0, limit) : all; - return (hubs: hubs, succeededServerIds: fetched.succeededServerIds, cancelledServerIds: fetched.cancelledServerIds); + return ( + hubs: hubs, + // Pre-limit, so a row trimmed off the tail still counts as observed. + observedItems: observedRows, + succeededServerIds: legSucceededServerIds, + cancelledServerIds: {...fetched.cancelledServerIds, ...legCancelledServerIds}, + failedServerIds: {...fetched.failedServerIds, ...legFailedServerIds}, + ); } /// Per-library hub fetch for a single client. Filters to visible libraries @@ -498,7 +605,12 @@ class DataAggregationService { /// musicvideos/homevideos, #1476; artist brings music rows to home) and /// concatenates the results. The rich-hub music append passes /// `{MediaKind.artist}` to fetch only what the global endpoint misses. - Future> _fetchLibraryHubsForClient( + /// + /// `succeeded` stays false when no library leg was attempted. This prevents + /// the optional rich-hub music append from masking a failed global leg; the + /// fallback caller separately recognizes a clean zero-library result as an + /// authoritative no-op. + Future<({List hubs, bool succeeded, bool failed, bool cancelled})> _fetchLibraryHubsForClient( MediaServerClient client, { required int limit, Set? hiddenLibraryKeys, @@ -524,29 +636,44 @@ class DataAggregationService { const concurrency = 3; final results = List>.filled(visible.length, const []); var next = 0; + var succeeded = false; + var failed = false; + var cancelled = false; Future worker() async { while (true) { final index = next++; if (index >= visible.length) return; final library = visible[index]; + final diagnostics = HubFetchDiagnostics(); try { - results[index] = await client.fetchLibraryHubs( + final hubs = await client.fetchLibraryHubs( library.id, libraryName: library.title, limit: limit, includePlaybackHubs: includePlaybackHubs, libraryKind: library.kind, + diagnostics: diagnostics, ); + results[index] = hubs; + if (hubs.isNotEmpty || (!diagnostics.failed && !diagnostics.cancelled)) succeeded = true; } catch (e, st) { - appLogger.e('Failed to fetch library hubs for ${library.globalKey}', error: e, stackTrace: st); + if (_isCancellation(e)) { + cancelled = true; + appLogger.d('Cancelled library hub fetch for ${library.globalKey}'); + } else { + failed = true; + appLogger.e('Failed to fetch library hubs for ${library.globalKey}', error: e, stackTrace: st); + } } + if (diagnostics.failed) failed = true; + if (diagnostics.cancelled) cancelled = true; } } await Future.wait([for (var i = 0; i < concurrency && i < visible.length; i++) worker()]); - return [for (final list in results) ...list]; + return (hubs: [for (final list in results) ...list], succeeded: succeeded, failed: failed, cancelled: cancelled); } /// Filter hidden-library items and drop empty hubs. diff --git a/lib/services/external_player_service.dart b/lib/services/external_player_service.dart index 34ed582a..11a3456a 100644 --- a/lib/services/external_player_service.dart +++ b/lib/services/external_player_service.dart @@ -177,6 +177,7 @@ class ExternalPlayerService { return; } + var startedSucceeded = false; try { await client.reportPlaybackStarted( itemId: metadata.id, @@ -185,6 +186,7 @@ class ExternalPlayerService { playMethod: 'DirectPlay', mediaSourceId: mediaSourceId, ); + startedSucceeded = true; } catch (e) { appLogger.d('External player progress: started call failed (continuing)', error: e); } @@ -210,6 +212,9 @@ class ExternalPlayerService { viewOffset: position.inMilliseconds, duration: duration.inMilliseconds, watchedThreshold: client.watchedThreshold, + // MediaBrowser persists stopped progress only for a session opened by + // Started; Plex persists every timeline report independently. + serverAcknowledged: !metadata.backend.usesMediaBrowserApi || startedSucceeded, ); if (isWatchedProgress( diff --git a/lib/services/jellyfin_client/parts/browse.dart b/lib/services/jellyfin_client/parts/browse.dart index 573a4a19..e74a27e9 100644 --- a/lib/services/jellyfin_client/parts/browse.dart +++ b/lib/services/jellyfin_client/parts/browse.dart @@ -1593,7 +1593,11 @@ mixin _JellyfinBrowseMethods on _JellyfinClientInternals { } @override - Future> fetchGlobalHubs({int limit = defaultHubPreviewLimit, bool includePlaybackHubs = true}) async { + Future> fetchGlobalHubs({ + int limit = defaultHubPreviewLimit, + bool includePlaybackHubs = true, + HubFetchDiagnostics? diagnostics, + }) async { // Jellyfin doesn't expose a single "hubs" endpoint, so we synthesise the // home rows from Latest plus optional playback rows. The richer Plex Discover surface // is intentionally left untranslated — see ServerCapabilities.richHubs. @@ -1607,6 +1611,7 @@ mixin _JellyfinBrowseMethods on _JellyfinClientInternals { continueTitle: t.discover.continueWatching, nextUpTitle: t.discover.nextUp, recentTitle: t.discover.recentlyAdded, + diagnostics: diagnostics, ); } @@ -1617,6 +1622,7 @@ mixin _JellyfinBrowseMethods on _JellyfinClientInternals { int limit = defaultHubPreviewLimit, bool includePlaybackHubs = true, MediaKind? libraryKind, + HubFetchDiagnostics? diagnostics, }) async { // Music libraries get their own hub set. Home passes // includePlaybackHubs=false because it already renders the app-level @@ -1630,6 +1636,7 @@ mixin _JellyfinBrowseMethods on _JellyfinClientInternals { libraryName: libraryName, limit: limit, includePlaybackHubs: includePlaybackHubs, + diagnostics: diagnostics, ); } @@ -1649,6 +1656,7 @@ mixin _JellyfinBrowseMethods on _JellyfinClientInternals { continueTitle: t.discover.continueWatchingIn(library: libraryName), nextUpTitle: t.discover.nextUpIn(library: libraryName), recentTitle: t.discover.recentlyAddedIn(library: libraryName), + diagnostics: diagnostics, ); } @@ -1670,14 +1678,20 @@ mixin _JellyfinBrowseMethods on _JellyfinClientInternals { required String recentTitle, String? parentId, String? latestItemTypes, + HubFetchDiagnostics? diagnostics, }) async { - final latestFuture = _safeFetchItemsArray('/Users/${_segment(connection.userId)}/Items/Latest', { - 'Limit': limit.toString(), - 'ParentId': ?parentId, - 'Fields': _hubRowFields, - 'IncludeItemTypes': ?latestItemTypes, - ...jellyfinImageQueryParameters, - }, retry: retry); + final latestFuture = _safeFetchItemsArray( + '/Users/${_segment(connection.userId)}/Items/Latest', + { + 'Limit': limit.toString(), + 'ParentId': ?parentId, + 'Fields': _hubRowFields, + 'IncludeItemTypes': ?latestItemTypes, + ...jellyfinImageQueryParameters, + }, + retry: retry, + diagnostics: diagnostics, + ); MediaHub hub(String suffix, String title, String type, List> items) => JellyfinMappers.syntheticHub( @@ -1698,28 +1712,37 @@ mixin _JellyfinBrowseMethods on _JellyfinClientInternals { final results = await Future.wait([ latestFuture, - _safeFetchItemsArray(_resumePath, { - 'userId': connection.userId, - ..._resumeFilterQuery, - 'ParentId': ?parentId, - 'Limit': limit.toString(), - 'Fields': _hubRowFields, - 'MediaTypes': 'Video', - 'Recursive': 'true', - 'EnableTotalRecordCount': 'false', - ...jellyfinImageQueryParameters, - }, retry: retry), + _safeFetchItemsArray( + _resumePath, + { + 'userId': connection.userId, + ..._resumeFilterQuery, + 'ParentId': ?parentId, + 'Limit': limit.toString(), + 'Fields': _hubRowFields, + 'MediaTypes': 'Video', + 'Recursive': 'true', + 'EnableTotalRecordCount': 'false', + ...jellyfinImageQueryParameters, + }, + retry: retry, + diagnostics: diagnostics, + ), includeNextUp - ? _fetchNextUpRows({ - 'userId': connection.userId, - 'ParentId': ?parentId, - 'Limit': limit.toString(), - 'Fields': _hubRowFields, - 'EnableResumable': 'false', - 'NextUpDateCutoff': _nextUpDateCutoff(), - 'EnableTotalRecordCount': 'false', - ...jellyfinImageQueryParameters, - }, retry: retry) + ? _fetchNextUpRows( + { + 'userId': connection.userId, + 'ParentId': ?parentId, + 'Limit': limit.toString(), + 'Fields': _hubRowFields, + 'EnableResumable': 'false', + 'NextUpDateCutoff': _nextUpDateCutoff(), + 'EnableTotalRecordCount': 'false', + ...jellyfinImageQueryParameters, + }, + retry: retry, + diagnostics: diagnostics, + ) : Future.value(const >[]), ]); @@ -1744,14 +1767,20 @@ mixin _JellyfinBrowseMethods on _JellyfinClientInternals { required String libraryName, required int limit, required bool includePlaybackHubs, + HubFetchDiagnostics? diagnostics, }) async { - final latestFuture = _safeFetchItemsArray('/Users/${_segment(connection.userId)}/Items/Latest', { - 'Limit': limit.toString(), - 'ParentId': libraryId, - 'Fields': _musicAlbumRowFields, - 'EnableUserData': 'false', - ...jellyfinImageQueryParameters, - }, retry: _libraryHubRetry); + final latestFuture = _safeFetchItemsArray( + '/Users/${_segment(connection.userId)}/Items/Latest', + { + 'Limit': limit.toString(), + 'ParentId': libraryId, + 'Fields': _musicAlbumRowFields, + 'EnableUserData': 'false', + ...jellyfinImageQueryParameters, + }, + retry: _libraryHubRetry, + diagnostics: diagnostics, + ); MediaHub latestAlbumsHub(List> items) => JellyfinMappers.syntheticHub( mapItem: _mapItem, @@ -1781,8 +1810,18 @@ mixin _JellyfinBrowseMethods on _JellyfinClientInternals { }; final results = await Future.wait([ latestFuture, - _safeFetchItemsArray('/Items', {...playedParams, 'SortBy': 'DatePlayed'}, retry: _libraryHubRetry), - _safeFetchItemsArray('/Items', {...playedParams, 'SortBy': 'PlayCount'}, retry: _libraryHubRetry), + _safeFetchItemsArray( + '/Items', + {...playedParams, 'SortBy': 'DatePlayed'}, + retry: _libraryHubRetry, + diagnostics: diagnostics, + ), + _safeFetchItemsArray( + '/Items', + {...playedParams, 'SortBy': 'PlayCount'}, + retry: _libraryHubRetry, + diagnostics: diagnostics, + ), ]); return [ @@ -2155,9 +2194,16 @@ mixin _JellyfinBrowseMethods on _JellyfinClientInternals { Map queryParameters, { _HubRetryPolicy? retry, AbortController? abort, + HubFetchDiagnostics? diagnostics, }) async { if (dialect.supportsGlobalNextUp) { - return _safeFetchItemsArray('/Shows/NextUp', queryParameters, retry: retry, abort: abort); + return _safeFetchItemsArray( + '/Shows/NextUp', + queryParameters, + retry: retry, + abort: abort, + diagnostics: diagnostics, + ); } final budgetAbort = AbortController(); @@ -2508,6 +2554,7 @@ mixin _JellyfinBrowseMethods on _JellyfinClientInternals { Map queryParameters, { _HubRetryPolicy? retry, AbortController? abort, + HubFetchDiagnostics? diagnostics, Duration? timeout, bool allowEndpointFailover = true, }) async { @@ -2532,6 +2579,7 @@ mixin _JellyfinBrowseMethods on _JellyfinClientInternals { // it propagate so the caller classifies the fetch as disrupted, not // empty. if (e is MediaServerHttpException && e.isCancellation) rethrow; + diagnostics?.recordFailure(e); appLogger.w('JellyfinClient: $path failed (treating as empty)', error: e, stackTrace: st); return const []; } diff --git a/lib/services/jellyfin_client/parts/live_tv.dart b/lib/services/jellyfin_client/parts/live_tv.dart index 76005f9b..2a53b4bd 100644 --- a/lib/services/jellyfin_client/parts/live_tv.dart +++ b/lib/services/jellyfin_client/parts/live_tv.dart @@ -12,6 +12,8 @@ mixin _JellyfinLiveTvMethods on _JellyfinClientInternals { Duration? timeout, // ignore: unused_element_parameter bool allowEndpointFailover, + // ignore: unused_element_parameter + HubFetchDiagnostics? diagnostics, }); /// Returns `true` when this server has Live TV configured (channels diff --git a/lib/services/offline_watch_sync_service.dart b/lib/services/offline_watch_sync_service.dart index 4fd02cfe..4c45b230 100644 --- a/lib/services/offline_watch_sync_service.dart +++ b/lib/services/offline_watch_sync_service.dart @@ -22,6 +22,17 @@ import 'settings_service.dart'; import 'trackers/tracker_coordinator.dart'; import 'watch_state_resolver.dart'; +typedef QueuedOfflineWatchAction = ({String? clientScopeId, String? profileId, int rowId, int revision}); + +typedef _OfflineWatchReplayResult = ({ + MediaItem item, + String? clientScopeId, + String? profileId, + int rowId, + int revision, + bool persisted, +}); + /// Service for managing offline watch progress and syncing it back to the /// owning server. Backend-neutral over [MediaServerClient] — Plex actions /// hit `/:/scrobble` and `/:/timeline`, while MediaBrowser actions use their @@ -110,16 +121,30 @@ class OfflineWatchSyncService extends ChangeNotifier { if (event.changeType != WatchStateChangeType.watched && event.changeType != WatchStateChangeType.unwatched) { return; } - unawaited(_discardQueuedProgress(ServerId(event.serverId), event.itemId)); + unawaited( + _discardQueuedProgress( + ServerId(event.serverId), + event.itemId, + beforeRevision: _offlineActionRevision(event.patchId), + ), + ); } - Future _discardQueuedProgress(ServerId serverId, String itemId) async { + int? _offlineActionRevision(WatchPatchId? patchId) { + final value = patchId?.value; + if (value == null || !value.startsWith('o:')) return null; + final separator = value.lastIndexOf(':'); + return separator < 2 ? null : int.tryParse(value.substring(separator + 1)); + } + + Future _discardQueuedProgress(ServerId serverId, String itemId, {int? beforeRevision}) async { try { final removed = await _database.deleteQueuedProgressForItem( profileId: _activeProfileId, serverId: serverId, clientScopeId: await _clientScopeIdForItem(serverId, itemId), ratingKey: itemId, + beforeRevision: beforeRevision, ); if (removed == 0) return; appLogger.d('Dropped $removed superseded queued progress action(s) for $serverId:$itemId'); @@ -258,7 +283,7 @@ class OfflineWatchSyncService extends ChangeNotifier { /// The `ratingKey:` parameter on the underlying `_database` calls is /// preserved as the on-disk column name; the in-memory parameter renamed /// here is just the API-level identifier. - Future queueProgressUpdate({ + Future queueProgressUpdate({ required ServerId serverId, required String itemId, required int viewOffset, @@ -267,9 +292,10 @@ class OfflineWatchSyncService extends ChangeNotifier { final shouldMarkWatched = duration != null && isWatchedByProgress(viewOffset, duration, serverId: ServerId(serverId)); final clientScopeId = await _clientScopeIdForItem(ServerId(serverId), itemId); + final profileId = _activeProfileId; - await _database.upsertProgressAction( - profileId: _activeProfileId, + final queued = await _database.upsertProgressAction( + profileId: profileId, serverId: serverId, clientScopeId: clientScopeId, ratingKey: itemId, @@ -287,23 +313,24 @@ class OfflineWatchSyncService extends ChangeNotifier { ); notifyListeners(); - return clientScopeId; + return (clientScopeId: clientScopeId, profileId: profileId, rowId: queued.rowId, revision: queued.revision); } - Future queueMarkWatched({required ServerId serverId, required String itemId}) => + Future queueMarkWatched({required ServerId serverId, required String itemId}) => _queueWatchStatusAction(serverId: serverId, itemId: itemId, actionType: OfflineActionType.watched.id); - Future queueMarkUnwatched({required ServerId serverId, required String itemId}) => + Future queueMarkUnwatched({required ServerId serverId, required String itemId}) => _queueWatchStatusAction(serverId: serverId, itemId: itemId, actionType: OfflineActionType.unwatched.id); - Future _queueWatchStatusAction({ + Future _queueWatchStatusAction({ required ServerId serverId, required String itemId, required String actionType, }) async { final clientScopeId = await _clientScopeIdForItem(ServerId(serverId), itemId); - await _database.insertWatchAction( - profileId: _activeProfileId, + final profileId = _activeProfileId; + final queued = await _database.insertWatchAction( + profileId: profileId, serverId: serverId, clientScopeId: clientScopeId, ratingKey: itemId, @@ -312,7 +339,7 @@ class OfflineWatchSyncService extends ChangeNotifier { appLogger.d('Queued offline mark $actionType: $serverId:$itemId'); notifyListeners(); - return clientScopeId; + return (clientScopeId: clientScopeId, profileId: profileId, rowId: queued.rowId, revision: queued.revision); } /// Check if an item should be considered watched based on progress percentage. @@ -446,14 +473,29 @@ class OfflineWatchSyncService extends ChangeNotifier { continue; } - final synced = await _withOnlineClientForAction(action, (client) async { + final synced = await _withOnlineClientForAction(action, (client, clientScopeId) async { try { - await _syncAction(client, action); - await _database.deleteWatchAction(action.id); + final result = await _syncAction(client, action, clientScopeId: clientScopeId); + if (!result.persisted) { + // The write did not land — MediaBrowser drops a stop for a + // session its Started never opened. Leave the row queued and + // count the attempt so a later pass retries it. + appLogger.d('Action ${action.id} did not persist; keeping it queued'); + await _database.updateSyncAttemptIfUnchanged(action.id, action.updatedAt, 'write did not persist'); + return; + } + final deleted = await _database.deleteWatchActionIfUnchanged(result.rowId, result.revision); + if (!deleted) { + appLogger.d('Synced action ${action.id} revision ${action.updatedAt}; a newer revision remains queued'); + return; + } + WatchPatchPromotionNotifier().promote( + WatchPatchId.offlineAction(profileId: result.profileId, rowId: result.rowId, revision: result.revision), + ); appLogger.d('Successfully synced action ${action.id}: ${action.actionType} for ${action.ratingKey}'); } catch (e) { appLogger.w('Failed to sync action ${action.id}: $e'); - await _database.updateSyncAttempt(action.id, e.toString()); + await _database.updateSyncAttemptIfUnchanged(action.id, action.updatedAt, e.toString()); } }); if (!synced) { @@ -507,7 +549,7 @@ class OfflineWatchSyncService extends ChangeNotifier { Future _withOnlineClientForAction( OfflineWatchProgressItem action, - Future Function(MediaServerClient client) callback, + Future Function(MediaServerClient client, String? clientScopeId) callback, ) async { final resolved = await _clientForAction(action); if (resolved == null) { @@ -520,7 +562,7 @@ class OfflineWatchSyncService extends ChangeNotifier { return false; } - await callback(resolved.client); + await callback(resolved.client, resolved.clientScopeId); return true; } @@ -575,14 +617,14 @@ class OfflineWatchSyncService extends ChangeNotifier { /// Uses the neutral [MediaServerClient] surface so Jellyfin's /// `/UserPlayedItems/{id}` and `/Sessions/Playing*` endpoints receive /// the same queued state Plex's `/:/scrobble` and `/:/timeline` do. - Future _syncAction(MediaServerClient client, OfflineWatchProgressItem action) async { - // Fetch metadata so trackers (and the stop-path watch event) get enough - // context — external ids, parent chain, library section. The plain - // watched/unwatched replays deliberately emit no WatchStateEvent: the - // offline provider already emitted it when the action was queued, and - // client markWatched/markUnwatched are transport-only. Best-effort: a - // missed metadata fetch falls back to a minimal MediaItem — the network - // call still goes through. + Future<_OfflineWatchReplayResult> _syncAction( + MediaServerClient client, + OfflineWatchProgressItem action, { + required String? clientScopeId, + }) async { + // Fetch metadata so tracker writes get enough context — external ids, + // parent chain and library section. Best-effort: a missed metadata fetch + // falls back to a minimal item while the server write still proceeds. final needsRichItem = action.actionType == OfflineActionType.watched.id || action.actionType == OfflineActionType.unwatched.id || @@ -602,14 +644,17 @@ class OfflineWatchSyncService extends ChangeNotifier { serverId: action.serverId, ); + var persisted = false; switch (action.actionType) { case 'watched': await client.markWatched(item); + persisted = true; await TrackerCoordinator.instance.markWatched(item, client); break; case 'unwatched': await client.markUnwatched(item); + persisted = true; await TrackerCoordinator.instance.markUnwatched(item, client); break; @@ -622,13 +667,19 @@ class OfflineWatchSyncService extends ChangeNotifier { final position = action.shouldMarkWatched && duration != null ? duration : Duration(milliseconds: action.viewOffset!); - if (!action.shouldMarkWatched || client.backend.usesMediaBrowserApi) { + // MediaBrowser ignores a stop for a session it never opened, so its + // Started call is a precondition for persistence, not best effort. + final requiresOpenSession = client.backend.usesMediaBrowserApi; + var startedSucceeded = !requiresOpenSession; + if (!action.shouldMarkWatched || requiresOpenSession) { try { await client.reportPlaybackStarted(itemId: action.ratingKey, position: position, duration: duration); + startedSucceeded = true; } catch (e) { - // Plex sometimes 5xxs the start when nothing follows; treat as - // best-effort and continue to the stop call which is the one - // that actually persists the resume position. + // Plex sometimes 5xxs the start when nothing follows; there the + // stop call is the one that persists the resume position, so + // continue. On MediaBrowser the stop will be dropped, so the + // action must stay queued for a later attempt. appLogger.d('Offline progress: started call failed (continuing)', error: e); } } @@ -640,18 +691,30 @@ class OfflineWatchSyncService extends ChangeNotifier { recordedAt: DateTime.fromMillisecondsSinceEpoch(action.updatedAt), ), ); + persisted = startedSucceeded; } - // If progress exceeded threshold, also mark as watched. On backends - // that mark played from the stopped report above (MediaBrowser), this - // only emits the local watch event — an explicit markWatched would - // double-scrobble via the Trakt plugin (#1287). if (action.shouldMarkWatched) { - await client.markWatchedFromPlaybackStop(item); + // MediaBrowser persists the played state from the stopped report. + // Plex still needs the explicit transport call, but replay must not + // emit another semantic event for an action already emitted offline. + if (!client.marksWatchedOnPlaybackStopped) { + await client.markWatched(item); + persisted = true; + } await TrackerCoordinator.instance.markWatched(item, client); } break; } + + return ( + item: item, + clientScopeId: clientScopeId, + profileId: action.profileId, + rowId: action.id, + revision: action.updatedAt, + persisted: persisted, + ); } /// Push a watch-state change Plezy observed on the server to the trackers. @@ -703,6 +766,7 @@ class OfflineWatchSyncService extends ChangeNotifier { item: episode, isNowWatched: isWatched, cacheServerId: client.cacheServerId, + serverAcknowledged: true, ); // The change came from the server (watched on another client), so no // playback or manual path has told the trackers about it. @@ -841,6 +905,7 @@ class OfflineWatchSyncService extends ChangeNotifier { item: metadata, isNowWatched: isWatched, cacheServerId: client.cacheServerId, + serverAcknowledged: true, ); await _mirrorWatchStateToTrackers(metadata, client, isWatched: isWatched); } diff --git a/lib/services/playback_progress_tracker.dart b/lib/services/playback_progress_tracker.dart index 0466c3a7..8e353bd2 100644 --- a/lib/services/playback_progress_tracker.dart +++ b/lib/services/playback_progress_tracker.dart @@ -146,6 +146,24 @@ class PlaybackProgressTracker { /// The post-watch hook has run; it fires at most once per tracker. bool _scrobbledHookRan = false; + /// Delivery provenance keyed by the exact snapshot handed to + /// [PlaybackReportSession]. Acceptance alone cannot identify delivery + /// because startup heartbeats may be dropped while still completing true. + final Map _deliveredProgressAcknowledgements = Map.identity(); + + /// Whether this backend reporting session has successfully opened. + bool _hasDeliveredStart = false; + + /// Whether the delivered stopped report persisted its position. + bool _stoppedProgressServerAcknowledged = false; + + /// Allows the first persisted Progress after a MediaBrowser Started to + /// upgrade local provenance even when the position delta is throttled. + bool _lastProgressNotificationServerAcknowledged = false; + + /// The exact report-derived watched patch that an explicit mark can settle. + WatchPatchId? _watchedPatchId; + /// Whether the final stopped progress event was already emitted locally. bool _stopProgressNotified = false; @@ -261,6 +279,8 @@ class PlaybackProgressTracker { _deliveredBelow = false; _serverObservedCrossing = false; _sessionEnded = false; + _hasDeliveredStart = false; + _stoppedProgressServerAcknowledged = false; } Future _sendProgress(String state, {Duration? positionOverride}) async { @@ -292,7 +312,7 @@ class PlaybackProgressTracker { // nothing. if (!canCommitStoppedProgress) return; await _sendOfflineProgress(position, duration); - _notifyProgressIfNeeded(position, duration, force: state == 'stopped'); + _notifyProgressIfNeeded(position, duration, force: state == 'stopped', serverAcknowledged: false); } else if (state == 'stopped') { // Stopped must complete before disposal. When reporting was disabled // by a fatal error, use the last position captured while output was @@ -304,17 +324,19 @@ class PlaybackProgressTracker { await _pendingSettle; _resetBackoff(); if (accepted && canCommitStoppedProgress) { - _notifyProgressIfNeeded(position, duration, force: true); + _notifyProgressIfNeeded( + position, + duration, + force: true, + serverAcknowledged: _stoppedProgressServerAcknowledged, + ); } } else { // Fire-and-forget for playing/paused — avoid blocking the Dart event loop unawaited( _sendOnlineProgress(state, position, duration) - .then((accepted) { + .then((_) { _resetBackoff(); - if (accepted) { - _notifyProgressIfNeeded(position, duration); - } }) .catchError((Object e) { _recordProgressFailure(e); @@ -370,7 +392,12 @@ class PlaybackProgressTracker { } } - void _notifyProgressIfNeeded(Duration position, Duration duration, {bool force = false}) { + void _notifyProgressIfNeeded( + Duration position, + Duration duration, { + bool force = false, + required bool serverAcknowledged, + }) { if (_scrobbled) return; if (position.inMilliseconds <= 0 || duration.inMilliseconds <= 0) return; if (force) { @@ -378,16 +405,20 @@ class PlaybackProgressTracker { _stopProgressNotified = true; } else { final last = _lastProgressNotifiedPosition; - if (last != null && (position - last).abs() < _progressNotifyDelta) return; + if (last != null && (position - last).abs() < _progressNotifyDelta) { + if (!serverAcknowledged || _lastProgressNotificationServerAcknowledged) return; + } } _lastProgressNotifiedPosition = position; + _lastProgressNotificationServerAcknowledged = serverAcknowledged; WatchStateNotifier().notifyProgress( item: metadata, cacheServerId: client?.cacheServerId, viewOffset: position.inMilliseconds, duration: duration.inMilliseconds, watchedThreshold: client?.watchedThreshold ?? 0.9, + serverAcknowledged: serverAcknowledged, ); } @@ -403,20 +434,26 @@ class PlaybackProgressTracker { final session = _reportSession; if (c == null || session == null) return false; - final accepted = await session.report( - PlaybackReportSnapshot( - state: state, - position: position, - duration: duration, - resolveStreamSelection: state == 'stopped' - ? _currentStreamSelectionForStopped - : _currentStreamSelectionForProgress, - ), + final snapshot = PlaybackReportSnapshot( + state: state, + position: position, + duration: duration, + resolveStreamSelection: state == 'stopped' + ? _currentStreamSelectionForStopped + : _currentStreamSelectionForProgress, ); + final accepted = await session.report(snapshot); if (accepted && allowScrobble) { await _maybeScrobble(c, position, duration); } + + if (!snapshot.isStopped) { + final serverAcknowledged = _deliveredProgressAcknowledgements.remove(snapshot); + if (serverAcknowledged != null) { + _notifyProgressIfNeeded(position, duration, serverAcknowledged: serverAcknowledged); + } + } return accepted; } @@ -434,7 +471,24 @@ class PlaybackProgressTracker { /// 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; + final c = client; + // The same backend as the client's, but read from the item so the + // classification matches the pattern already used for track selection + // below and does not depend on a client method. + final persistsPositionOnEveryReport = !metadata.backend.usesMediaBrowserApi; + if (snapshot.isStopped) { + // MediaBrowser needs a successfully opened session before Stopped can + // persist position; Plex timeline reports are independent. + _stoppedProgressServerAcknowledged = persistsPositionOnEveryReport || _hasDeliveredStart; + } else { + final isStarted = !_hasDeliveredStart; + _hasDeliveredStart = true; + // A MediaBrowser `Started` saves play count and last-played date but + // deliberately not the position, so it cannot acknowledge an offset. + _deliveredProgressAcknowledgements[snapshot] = !isStarted || persistsPositionOnEveryReport; + } + + final threshold = c?.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; @@ -503,6 +557,8 @@ class PlaybackProgressTracker { _serverMarkSettled = false; // Retry on the next delivered report. return; } + final patchId = _watchedPatchId; + if (patchId != null) WatchPatchPromotionNotifier().promote(patchId); await _runScrobbledHook(); } @@ -540,7 +596,7 @@ class PlaybackProgressTracker { // 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); + _watchedPatchId = c.notifyWatchedFromPlaybackSession(metadata); appLogger.d( 'Watched ${metadata.id} (${(percent * 100).toStringAsFixed(0)}% >= ${(threshold * 100).toStringAsFixed(0)}%)', ); diff --git a/lib/services/plex_client.dart b/lib/services/plex_client.dart index ef60a7f4..6bf6a8ea 100644 --- a/lib/services/plex_client.dart +++ b/lib/services/plex_client.dart @@ -2226,6 +2226,7 @@ class PlexClient int? librarySectionID, String? librarySectionTitle, bool Function(PlexMetadataDto)? filter, + HubFetchDiagnostics? diagnostics, }) async { try { final response = await retryTransientMediaServerCall( @@ -2253,6 +2254,7 @@ class PlexClient ), ); } catch (e) { + diagnostics?.recordFailure(e); appLogger.e('Failed to get $failureLabel: $e'); } return []; @@ -2264,6 +2266,7 @@ class PlexClient String sectionId, { int limit = defaultHubPreviewLimit, String? libraryName, + HubFetchDiagnostics? diagnostics, }) => _fetchHubs( path: '/hubs/sections/$sectionId', queryParameters: {'count': limit, 'includeGuids': 1}, @@ -2272,19 +2275,22 @@ class PlexClient failureLabel: 'library hubs', librarySectionID: _librarySectionIdFromString(sectionId), librarySectionTitle: libraryName, + diagnostics: diagnostics, filter: _videoOrMusicHubItem, ); /// Get global hubs (home page recommendations) /// Returns actual home page hubs like "Recently Added Movies", "Recently Added TV", etc. /// This matches the official Plex client's home page layout. - Future> _getGlobalHubs({int limit = defaultHubPreviewLimit}) => _fetchHubs( - path: _providerPromotedHubKey ?? _providerHomeHubKey ?? '/hubs', - queryParameters: {'count': limit, 'includeGuids': 1}, - operation: 'Plex global hubs', - deadline: MediaServerTimeouts.homeHubDeadline, - failureLabel: 'global hubs', - ); + Future> _getGlobalHubs({int limit = defaultHubPreviewLimit, HubFetchDiagnostics? diagnostics}) => + _fetchHubs( + path: _providerPromotedHubKey ?? _providerHomeHubKey ?? '/hubs', + queryParameters: {'count': limit, 'includeGuids': 1}, + operation: 'Plex global hubs', + deadline: MediaServerTimeouts.homeHubDeadline, + failureLabel: 'global hubs', + diagnostics: diagnostics, + ); /// Get related hubs for a specific metadata item (collections, similar, "more from" director/actor) Future> _getRelatedHubs(String ratingKey, {int count = 10}) => _fetchHubs( @@ -3614,8 +3620,12 @@ class PlexClient } @override - Future> fetchGlobalHubs({int limit = defaultHubPreviewLimit, bool includePlaybackHubs = true}) async { - final hubs = await _getGlobalHubs(limit: limit); + Future> fetchGlobalHubs({ + int limit = defaultHubPreviewLimit, + bool includePlaybackHubs = true, + HubFetchDiagnostics? diagnostics, + }) async { + final hubs = await _getGlobalHubs(limit: limit, diagnostics: diagnostics); return hubs.map((h) => PlexMappers.mediaHub(h)).toList(); } @@ -3626,10 +3636,11 @@ class PlexClient int limit = defaultHubPreviewLimit, bool includePlaybackHubs = true, MediaKind? libraryKind, + HubFetchDiagnostics? diagnostics, }) async { // libraryName is unused: Plex's /hubs/sections/{id} returns hubs already // titled per-library (e.g. "Recently Added in Movies"). - final hubs = await _getLibraryHubs(libraryId, limit: limit, libraryName: libraryName); + final hubs = await _getLibraryHubs(libraryId, limit: limit, libraryName: libraryName, diagnostics: diagnostics); return hubs.map((h) => PlexMappers.mediaHub(h)).toList(); } diff --git a/lib/services/watch_actions.dart b/lib/services/watch_actions.dart index dcc38e43..321df082 100644 --- a/lib/services/watch_actions.dart +++ b/lib/services/watch_actions.dart @@ -63,7 +63,12 @@ class WatchActions { } else { await client.markUnwatched(item); } - WatchStateNotifier().notifyWatched(item: item, isNowWatched: watched, cacheServerId: client.cacheServerId); + WatchStateNotifier().notifyWatched( + item: item, + isNowWatched: watched, + cacheServerId: client.cacheServerId, + serverAcknowledged: true, + ); unawaited( watched ? TrackerCoordinator.instance.markWatched(item, client) diff --git a/lib/utils/watch_state_notifier.dart b/lib/utils/watch_state_notifier.dart index 7e6813bc..85fff3e9 100644 --- a/lib/utils/watch_state_notifier.dart +++ b/lib/utils/watch_state_notifier.dart @@ -7,6 +7,39 @@ import 'global_key_utils.dart'; import 'hierarchical_event_mixin.dart'; import 'media_event_keys.dart'; +/// Identity of the store overlay entry a watch event creates. +/// +/// Two forms, because the two paths that need to settle later have different +/// lifetimes: +/// - [WatchPatchId.session] is minted per event and dies with the process, +/// which is right for a live playback crossing. +/// - [WatchPatchId.offlineAction] is *derived* from the persisted queue row +/// `(profileId, rowId, revision)` rather than minted, so emission, +/// hydration and promotion still join after a restart. A minted id could +/// not: the database persists the row and its revision, nothing more. +/// +/// Promotion matches on this, never on the global key — a same-item rewatch +/// must never be promoted or discarded in place of the action that settled. +class WatchPatchId { + final String value; + + const WatchPatchId._(this.value); + + factory WatchPatchId.session(int sequence) => WatchPatchId._('s:$sequence'); + + factory WatchPatchId.offlineAction({required String? profileId, required int rowId, required int revision}) => + WatchPatchId._('o:${profileId ?? ''}:$rowId:$revision'); + + @override + bool operator ==(Object other) => other is WatchPatchId && other.value == value; + + @override + int get hashCode => value.hashCode; + + @override + String toString() => 'WatchPatchId($value)'; +} + enum WatchStateChangeType { watched, unwatched, progressUpdate, removedFromContinueWatching } /// Event representing a watch state change with parent chain for hierarchical invalidation @@ -53,6 +86,20 @@ class WatchStateEvent with HierarchicalEventMixin { /// numeric id, Jellyfin sends a UUID; both round-trip as strings. final String? librarySectionID; + /// Whether the server had already accepted this exact state when the event + /// was emitted. + /// + /// Only an acknowledged patch may be superseded by a later authoritative + /// read (#1829): an unacknowledged one represents a write still owed to the + /// server, and a read must not retire it. Defaults to `false` so a new emit + /// site that forgets to classify itself degrades to today's behaviour rather + /// than silently becoming retireable. + final bool serverAcknowledged; + + /// Identity of the overlay entry this event creates, for later promotion. + /// Null for events nothing will ever settle. + final WatchPatchId? patchId; + WatchStateEvent({ required this.itemId, required this.serverId, @@ -63,6 +110,8 @@ class WatchStateEvent with HierarchicalEventMixin { this.viewOffset, this.isNowWatched, this.librarySectionID, + this.serverAcknowledged = false, + this.patchId, }) : globalKey = buildGlobalKey(ServerId(serverId), itemId); /// `serverId:librarySectionID`, matching [MediaLibrary.globalKey]. Null when @@ -98,9 +147,20 @@ class WatchStateNotifier extends BaseNotifier { } /// Helper to emit a watched/unwatched event from a [MediaItem]. - void notifyWatched({required MediaItem item, bool isNowWatched = true, String? cacheServerId}) { + /// + /// Returns the [WatchPatchId] of the overlay entry it created so a caller + /// that must settle later can promote that exact entry. Callers with + /// nothing to settle ignore it. + WatchPatchId? notifyWatched({ + required MediaItem item, + bool isNowWatched = true, + String? cacheServerId, + bool serverAcknowledged = false, + WatchPatchId? patchId, + }) { final serverId = serverIdForEvent(item, notifier: 'WatchStateNotifier', event: 'watched'); - if (serverId == null) return; + if (serverId == null) return null; + final id = patchId ?? WatchPatchId.session(++_sequence); notify( WatchStateEvent( itemId: item.id, @@ -111,23 +171,34 @@ class WatchStateNotifier extends BaseNotifier { mediaType: item.kind.id, isNowWatched: isNowWatched, librarySectionID: item.libraryId, + serverAcknowledged: serverAcknowledged, + patchId: id, ), ); + return id; } + /// Monotonic source for session-minted [WatchPatchId]s. + var _sequence = 0; + /// Helper to emit a progress update event. /// [watchedThreshold] defaults to 0.9 — pass the server's configured value /// (`client.watchedThreshold`) when available. - void notifyProgress({ + /// + /// Returns the created [WatchPatchId], as [notifyWatched] does. + WatchPatchId? notifyProgress({ required MediaItem item, required int viewOffset, required int duration, String? cacheServerId, double watchedThreshold = 0.9, + bool serverAcknowledged = false, + WatchPatchId? patchId, }) { final serverId = serverIdForEvent(item, notifier: 'WatchStateNotifier', event: 'progress'); - if (serverId == null) return; + if (serverId == null) return null; final isNowWatched = isWatchedProgress(positionMs: viewOffset, durationMs: duration, threshold: watchedThreshold); + final id = patchId ?? WatchPatchId.session(++_sequence); notify( WatchStateEvent( @@ -140,8 +211,11 @@ class WatchStateNotifier extends BaseNotifier { viewOffset: viewOffset, isNowWatched: isNowWatched, librarySectionID: item.libraryId, + serverAcknowledged: serverAcknowledged, + patchId: id, ), ); + return id; } /// Helper to emit a Continue Watching removal event. @@ -160,3 +234,37 @@ class WatchStateNotifier extends BaseNotifier { ); } } + +/// A patch the server has now definitively accepted. +/// +/// Deliberately *not* a [WatchStateEvent]: semantic subscribers must not see +/// promotions. `OfflineWatchSyncService` reacts to `watched`/`unwatched` by +/// purging queued progress, so replaying one here would delete a newer +/// rewatch — the very write promotion exists to protect. +/// +/// [BaseNotifier] is single-stream by construction, and `WatchStateNotifier` +/// fixes its type to [WatchStateEvent], so this needs its own channel rather +/// than a discriminated union on the existing one. +class WatchPatchPromotion { + final WatchPatchId patchId; + + const WatchPatchPromotion(this.patchId); + + @override + String toString() => 'WatchPatchPromotion($patchId)'; +} + +/// Sibling singleton to [WatchStateNotifier] carrying non-semantic +/// promotions, following the same pattern as `LibraryRefreshNotifier`. +class WatchPatchPromotionNotifier extends BaseNotifier { + static final WatchPatchPromotionNotifier _instance = WatchPatchPromotionNotifier._internal(); + + factory WatchPatchPromotionNotifier() => _instance; + + WatchPatchPromotionNotifier._internal(); + + void promote(WatchPatchId patchId) { + appLogger.d('WatchPatchPromotionNotifier: promoting $patchId'); + notify(WatchPatchPromotion(patchId)); + } +} diff --git a/test/providers/discover_provider_test.dart b/test/providers/discover_provider_test.dart index f96ffb61..3613e2b0 100644 --- a/test/providers/discover_provider_test.dart +++ b/test/providers/discover_provider_test.dart @@ -71,6 +71,8 @@ class _FakeAggregationService extends DataAggregationService { Set? hubSucceededServerIds; Set onDeckCancelledServerIds = const {}; Set hubCancelledServerIds = const {}; + Set onDeckFailedServerIds = const {}; + Set hubFailedServerIds = const {}; List Function() onDeckResult = () => const []; List Function() hubsResult = () => const []; Future? onDeckGate; @@ -93,8 +95,10 @@ class _FakeAggregationService extends DataAggregationService { final items = onDeckResult(); return ( items: limit != null && items.length > limit ? items.sublist(0, limit) : items, + observedItems: [for (final item in items) (item: item, clientScope: null)], succeededServerIds: onDeckSucceededServerIds ?? serverIds ?? const {'server_1'}, cancelledServerIds: onDeckCancelledServerIds, + failedServerIds: onDeckFailedServerIds, ); } @@ -112,10 +116,16 @@ class _FakeAggregationService extends DataAggregationService { if (started != null && !started.isCompleted) started.complete(); final gate = hubGate; if (gate != null) await gate; + final hubs = hubsResult(); return ( - hubs: hubsResult(), + hubs: hubs, + observedItems: [ + for (final hub in hubs) + for (final item in hub.items) (item: item, clientScope: null), + ], succeededServerIds: hubSucceededServerIds ?? serverIds ?? const {'server_1'}, cancelledServerIds: hubCancelledServerIds, + failedServerIds: hubFailedServerIds, ); } } @@ -697,6 +707,81 @@ void main() { expect(aggregation.onDeckCalls, greaterThan(callsBefore)); }); + group('manual refresh reports what actually happened (#1829)', () { + test('a zero-success refresh reports failure while keeping the rows visible', () async { + aggregation.onDeckResult = () => [_item('a')]; + aggregation.hubsResult = () => [_hub('hub-1')]; + await provider.load(); + + aggregation.onDeckSucceededServerIds = const {}; + aggregation.hubSucceededServerIds = const {}; + aggregation.onDeckFailedServerIds = const {'server_1'}; + aggregation.hubFailedServerIds = const {'server_1'}; + aggregation.onDeckResult = () => const []; + aggregation.hubsResult = () => const []; + + expect(await provider.refreshNow(), DiscoverRefreshOutcome.failed); + // Retained content still renders: the error is surfaced by the caller as + // a snackbar, never by blanking the screen. + expect(provider.onDeck.map((i) => i.id), ['a']); + expect(provider.hubs.map((h) => h.id), ['hub-1']); + expect(provider.errorMessage, isNull); + }); + + test('a partly failed refresh is degraded, not a success', () async { + aggregation.onDeckResult = () => [_item('a')]; + aggregation.hubsResult = () => [_hub('hub-1')]; + await provider.load(); + + aggregation.hubFailedServerIds = const {'server_2'}; + expect(await provider.refreshNow(), DiscoverRefreshOutcome.degraded); + }); + + test('a cancelled refresh is not reported as a failure', () async { + aggregation.onDeckSucceededServerIds = const {}; + aggregation.hubSucceededServerIds = const {}; + aggregation.onDeckCancelledServerIds = const {'server_1'}; + aggregation.hubCancelledServerIds = const {'server_1'}; + + expect(await provider.refreshNow(), DiscoverRefreshOutcome.cancelled); + }); + + test('a fully successful refresh reports success', () async { + aggregation.onDeckResult = () => [_item('a')]; + aggregation.hubsResult = () => [_hub('hub-1')]; + + expect(await provider.refreshNow(), DiscoverRefreshOutcome.refreshed); + }); + + test('a server that failed one leg stays eligible for retry', () async { + aggregation.onDeckResult = () => [_item('a')]; + aggregation.hubsResult = () => [_hub('hub-1')]; + // Succeeded and failed are unions, so one server can appear in both; + // loaded ids must exclude it or syncToOnlineServers never retries. + aggregation.hubFailedServerIds = const {'server_1'}; + await provider.load(); + + final callsBefore = aggregation.hubCalls; + await provider.syncToOnlineServers({'server_1'}); + expect(aggregation.hubCalls, greaterThan(callsBefore)); + }); + + test('a zero-success background refresh retains rows and stays silent', () async { + aggregation.onDeckResult = () => [_item('a')]; + await provider.load(); + + aggregation.onDeckSucceededServerIds = const {}; + aggregation.onDeckFailedServerIds = const {'server_1'}; + aggregation.onDeckResult = () => const []; + await provider.refreshContinueWatching(); + + // Previously this wiped the row outright. + expect(provider.onDeck.map((i) => i.id), ['a']); + expect(provider.errorMessage, isNull); + expect(provider.isLoading, isFalse); + }); + }); + test('a disrupted half is independent: on-deck commits while hubs stay loading', () async { aggregation.onDeckResult = () => [_item('a')]; aggregation.hubSucceededServerIds = const {}; diff --git a/test/providers/watch_state_store_test.dart b/test/providers/watch_state_store_test.dart index 612a3f5a..53c2b2e2 100644 --- a/test/providers/watch_state_store_test.dart +++ b/test/providers/watch_state_store_test.dart @@ -5,6 +5,7 @@ import 'package:plezy/media/media_backend.dart'; import 'package:plezy/media/media_kind.dart'; import 'package:plezy/providers/watch_state_store.dart'; import 'package:plezy/services/watch_state_resolver.dart'; +import 'package:plezy/utils/active_client_scope.dart'; import 'package:plezy/utils/watch_state_notifier.dart'; import '../test_helpers/media_items.dart'; @@ -22,6 +23,7 @@ WatchStateEvent _event({ int? viewOffset, List parentChain = const [], String mediaType = 'movie', + bool serverAcknowledged = false, }) { return WatchStateEvent( itemId: itemId, @@ -32,9 +34,13 @@ WatchStateEvent _event({ mediaType: mediaType, isNowWatched: isNowWatched, viewOffset: viewOffset, + serverAcknowledged: serverAcknowledged, + patchId: WatchPatchId.session(++_testSequence), ); } +var _testSequence = 0; + final _episode = testMediaItem( id: 'episode-1', backend: MediaBackend.jellyfin, @@ -231,4 +237,447 @@ void main() { store.setActiveClientScopesByServer({'jf-machine': 'jf-machine/user-b'}); expect(store.apply(_episode).isWatched, isFalse); }); + + group('authoritative observations supersede acknowledged patches (#1829)', () { + /// The reporter's sequence: the Mac pauses an episode, another device + /// advances the server, and a refresh must render the server's position + /// rather than the Mac's. + test('a fetch that started after the patch renders the server offset', () async { + final store = WatchStateStore(); + addTearDown(store.dispose); + + await _emit( + _event( + changeType: WatchStateChangeType.progressUpdate, + isNowWatched: false, + itemId: 'episode-1', + viewOffset: 1800000, + serverAcknowledged: true, + ), + ); + final fresh = _episode.copyWith(viewOffsetMs: 2700000, durationMs: 3600000); + expect(store.apply(fresh).viewOffsetMs, 1800000, reason: 'precondition: the stale patch still wins'); + + store.recordObservations( + [(item: fresh, clientScope: null)], + watermark: store.observationWatermark, + epoch: store.observationEpoch, + ); + + expect(store.apply(fresh).viewOffsetMs, 2700000); + expect(store.patchForItem(fresh), isNull); + }); + + test('a patch recorded during the fetch survives its observation', () async { + final store = WatchStateStore(); + addTearDown(store.dispose); + + // Watermark captured before the request, patch emitted while it is in + // flight: the response is older than the local action and must lose. + final watermark = store.observationWatermark; + await _emit( + _event( + changeType: WatchStateChangeType.progressUpdate, + isNowWatched: false, + itemId: 'episode-1', + viewOffset: 1800000, + serverAcknowledged: true, + ), + ); + final fresh = _episode.copyWith(viewOffsetMs: 2700000, durationMs: 3600000); + + store.recordObservations([(item: fresh, clientScope: null)], watermark: watermark, epoch: store.observationEpoch); + + expect(store.apply(fresh).viewOffsetMs, 1800000); + }); + + test('an unacknowledged patch is never superseded by a read', () async { + final store = WatchStateStore(); + addTearDown(store.dispose); + + // An offline write still owed to the server must outlive any fetch. + await _emit( + _event( + changeType: WatchStateChangeType.progressUpdate, + isNowWatched: false, + itemId: 'episode-1', + viewOffset: 1800000, + ), + ); + final fresh = _episode.copyWith(viewOffsetMs: 2700000, durationMs: 3600000); + + store.recordObservations( + [(item: fresh, clientScope: null)], + watermark: store.observationWatermark, + epoch: store.observationEpoch, + ); + + expect(store.apply(fresh).viewOffsetMs, 1800000); + }); + + test('an observation from another store epoch is rejected', () async { + final store = WatchStateStore(); + addTearDown(store.dispose); + + await _emit( + _event( + changeType: WatchStateChangeType.progressUpdate, + isNowWatched: false, + itemId: 'episode-1', + viewOffset: 1800000, + serverAcknowledged: true, + ), + ); + final fresh = _episode.copyWith(viewOffsetMs: 2700000, durationMs: 3600000); + + // A request issued before a profile switch, landing after it: its + // watermark belongs to a store that no longer exists. + store.recordObservations( + [(item: fresh, clientScope: null)], + watermark: store.observationWatermark, + epoch: Object(), + ); + + expect(store.apply(fresh).viewOffsetMs, 1800000); + }); + + test('an older response cannot lower an observation already recorded', () async { + final store = WatchStateStore(); + addTearDown(store.dispose); + + final stale = store.observationWatermark; + await _emit( + _event( + changeType: WatchStateChangeType.progressUpdate, + isNowWatched: false, + itemId: 'episode-1', + viewOffset: 1800000, + serverAcknowledged: true, + ), + ); + final fresh = _episode.copyWith(viewOffsetMs: 2700000, durationMs: 3600000); + + store.recordObservations( + [(item: fresh, clientScope: null)], + watermark: store.observationWatermark, + epoch: store.observationEpoch, + ); + // A concurrent request that started earlier completes last. + store.recordObservations([(item: fresh, clientScope: null)], watermark: stale, epoch: store.observationEpoch); + + expect(store.apply(fresh).viewOffsetMs, 2700000, reason: 'max() per key, so response order cannot matter'); + }); + + test('suppressing an item does not expose an older ancestor patch', () async { + final store = WatchStateStore(); + addTearDown(store.dispose); + + // Season marked unwatched first, then the episode played: without the + // ancestor barrier, suppressing the episode entry would let the older + // season patch win and render unwatched/0 — worse than either value. + await _emit( + _event( + changeType: WatchStateChangeType.unwatched, + isNowWatched: false, + itemId: 'season-1', + mediaType: 'season', + serverAcknowledged: true, + ), + ); + await _emit( + _event( + changeType: WatchStateChangeType.progressUpdate, + isNowWatched: false, + itemId: 'episode-1', + viewOffset: 1800000, + serverAcknowledged: true, + ), + ); + final fresh = _episode.copyWith(viewOffsetMs: 2700000, durationMs: 3600000, viewCount: 0); + + store.recordObservations( + [(item: fresh, clientScope: null)], + watermark: store.observationWatermark, + epoch: store.observationEpoch, + ); + + expect(store.patchForItem(fresh), isNull); + expect(store.apply(fresh).viewOffsetMs, 2700000); + }); + + test('an ancestor mark newer than the observation still reaches descendants', () async { + final store = WatchStateStore(); + addTearDown(store.dispose); + + final fresh = _episode.copyWith(viewOffsetMs: 2700000, durationMs: 3600000); + await _emit( + _event( + changeType: WatchStateChangeType.progressUpdate, + isNowWatched: false, + itemId: 'episode-1', + viewOffset: 1800000, + serverAcknowledged: true, + ), + ); + store.recordObservations( + [(item: fresh, clientScope: null)], + watermark: store.observationWatermark, + epoch: store.observationEpoch, + ); + // The container action happens after the fetch, so it must survive. + await _emit( + _event( + changeType: WatchStateChangeType.watched, + isNowWatched: true, + itemId: 'season-1', + mediaType: 'season', + serverAcknowledged: true, + ), + ); + + expect(store.apply(fresh).isWatched, isTrue); + }); + + test('an unobserved sibling still sees the container mark', () async { + final store = WatchStateStore(); + addTearDown(store.dispose); + + await _emit( + _event( + changeType: WatchStateChangeType.watched, + isNowWatched: true, + itemId: 'season-1', + mediaType: 'season', + serverAcknowledged: true, + ), + ); + final observed = _episode.copyWith(viewCount: 0); + final sibling = testMediaItem( + id: 'episode-2', + backend: MediaBackend.jellyfin, + kind: MediaKind.episode, + parentId: 'season-1', + grandparentId: 'show-1', + serverId: 'jf-machine', + ); + + store.recordObservations( + [(item: observed, clientScope: null)], + watermark: store.observationWatermark, + epoch: store.observationEpoch, + ); + + expect(store.apply(observed).isWatched, isFalse, reason: 'the observed child yields to the server'); + expect(store.apply(sibling).isWatched, isTrue, reason: 'the unobserved sibling keeps the container mark'); + }); + + test('an observation under one client scope cannot suppress another scope', () async { + final store = WatchStateStore(); + addTearDown(store.dispose); + store.setActiveClientScopesByServer({'jf-machine': 'jf-machine/user-b'}); + + await _emit( + _event( + changeType: WatchStateChangeType.progressUpdate, + isNowWatched: false, + itemId: 'episode-1', + viewOffset: 1800000, + cacheServerId: 'jf-machine/user-b', + serverAcknowledged: true, + ), + ); + final fresh = _episode.copyWith(viewOffsetMs: 2700000, durationMs: 3600000); + + // User A's request returning after the switch back to B: the public + // server id is shared, so keying by it would suppress B's watch state. + store.recordObservations( + [(item: fresh, clientScope: 'jf-machine/user-a')], + watermark: store.observationWatermark, + epoch: store.observationEpoch, + ); + + expect(store.apply(fresh).viewOffsetMs, 1800000); + }); + + test('a profile switch clears observations along with patches', () async { + final store = WatchStateStore(); + addTearDown(store.dispose); + + await _emit( + _event( + changeType: WatchStateChangeType.progressUpdate, + isNowWatched: false, + itemId: 'episode-1', + viewOffset: 1800000, + serverAcknowledged: true, + ), + ); + final fresh = _episode.copyWith(viewOffsetMs: 2700000, durationMs: 3600000); + store.recordObservations( + [(item: fresh, clientScope: null)], + watermark: store.observationWatermark, + epoch: store.observationEpoch, + ); + store.setActiveProfileId('profile-2'); + + await _emit( + _event( + changeType: WatchStateChangeType.progressUpdate, + isNowWatched: false, + itemId: 'episode-1', + viewOffset: 900000, + serverAcknowledged: true, + ), + ); + + expect(store.apply(fresh).viewOffsetMs, 900000, reason: 'a stale observation must not suppress the new patch'); + }); + }); + + group('promotion settles an owed write (#1829)', () { + test('a promoted patch becomes suppressible', () async { + final store = WatchStateStore(); + addTearDown(store.dispose); + + final patchId = WatchStateNotifier().notifyProgress(item: _episode, viewOffset: 1800000, duration: 3600000); + await Future.delayed(Duration.zero); + final fresh = _episode.copyWith(viewOffsetMs: 2700000, durationMs: 3600000); + + store.recordObservations( + [(item: fresh, clientScope: null)], + watermark: store.observationWatermark, + epoch: store.observationEpoch, + ); + expect(store.apply(fresh).viewOffsetMs, 1800000, reason: 'unacknowledged, so the read must not retire it'); + + WatchPatchPromotionNotifier().promote(patchId!); + await Future.delayed(Duration.zero); + + // Promotion assigns a fresh sequence, so the observation that preceded + // the write cannot suppress it; a later one can. + expect(store.apply(fresh).viewOffsetMs, 1800000); + store.recordObservations( + [(item: fresh, clientScope: null)], + watermark: store.observationWatermark, + epoch: store.observationEpoch, + ); + expect(store.apply(fresh).viewOffsetMs, 2700000); + }); + + test('promoting an unknown id is a no-op', () async { + final store = WatchStateStore(); + addTearDown(store.dispose); + + await _emit(_event(changeType: WatchStateChangeType.progressUpdate, isNowWatched: false, viewOffset: 1800000)); + WatchPatchPromotionNotifier().promote(WatchPatchId.session(999999)); + await Future.delayed(Duration.zero); + + expect(store.patchForGlobalKey('jf-machine:item-1')?.viewOffsetMs, 1800000); + }); + }); + + // The reporter's setup: Plex on macOS, where the client's cache scope is a + // profile scope that differs from the public server id. The observation and + // the patch must land on the same key or suppression silently never fires. + group("the reporter's Plex handoff (#1829)", () { + final plexScope = buildPlexProfileScopeId(serverId: ServerId('plex-machine'), profileId: 'profile-a'); + final plexEpisode = testMediaItem( + id: 'episode-1', + backend: MediaBackend.plex, + kind: MediaKind.episode, + serverId: 'plex-machine', + durationMs: 3600000, + ); + + test('a refreshed row overrides the offset the Mac paused at', () async { + final store = WatchStateStore(); + addTearDown(store.dispose); + // Exactly how ProfileSessionScreen wires the map: serverId -> the + // client's own cacheServerId. + store.setActiveClientScopesByServer({'plex-machine': plexScope}); + + // The Mac pauses with 30 minutes left. Plex persists position on every + // timeline report, so the tracker acknowledges it. + await _emit( + _event( + changeType: WatchStateChangeType.progressUpdate, + isNowWatched: false, + serverId: 'plex-machine', + itemId: 'episode-1', + cacheServerId: plexScope, + viewOffset: 1800000, + serverAcknowledged: true, + ), + ); + + // Precondition: without an observation the local patch pins the row. + // This is the bug as reported. + final serverRow = plexEpisode.copyWith(viewOffsetMs: 2700000); + expect(store.apply(serverRow).viewOffsetMs, 1800000); + + // The iPad advances the server to 15 minutes left; Refresh reads it back. + final watermark = store.observationWatermark; + store.recordObservations( + [(item: serverRow, clientScope: plexScope)], + watermark: watermark, + epoch: store.observationEpoch, + ); + + expect(store.apply(serverRow).viewOffsetMs, 2700000); + }); + + test('a pre-refresh pause still pins, so a later local action is not lost', () async { + final store = WatchStateStore(); + addTearDown(store.dispose); + store.setActiveClientScopesByServer({'plex-machine': plexScope}); + + final serverRow = plexEpisode.copyWith(viewOffsetMs: 2700000); + final watermark = store.observationWatermark; + store.recordObservations( + [(item: serverRow, clientScope: plexScope)], + watermark: watermark, + epoch: store.observationEpoch, + ); + + // Paused after that read: newer than the watermark, so it must survive. + await _emit( + _event( + changeType: WatchStateChangeType.progressUpdate, + isNowWatched: false, + serverId: 'plex-machine', + itemId: 'episode-1', + cacheServerId: plexScope, + viewOffset: 3000000, + serverAcknowledged: true, + ), + ); + + expect(store.apply(serverRow).viewOffsetMs, 3000000); + }); + + test('an unscoped store neither pins nor mis-suppresses', () async { + final store = WatchStateStore(); + addTearDown(store.dispose); + // DownloadMetadataStore replaces the scope map with only the servers it + // has downloads for, so a Plex server without downloads can end up + // unscoped. The patch keys off the event's own cacheServerId, so both + // sides miss together rather than leaving a stale pin behind. + store.setActiveClientScopesByServer(const {}); + + await _emit( + _event( + changeType: WatchStateChangeType.progressUpdate, + isNowWatched: false, + serverId: 'plex-machine', + itemId: 'episode-1', + cacheServerId: plexScope, + viewOffset: 1800000, + serverAcknowledged: true, + ), + ); + + final serverRow = plexEpisode.copyWith(viewOffsetMs: 2700000); + expect(store.apply(serverRow).viewOffsetMs, 2700000); + }); + }); } diff --git a/test/screens/discover_screen_test.dart b/test/screens/discover_screen_test.dart index 8f2d10af..8619f42c 100644 --- a/test/screens/discover_screen_test.dart +++ b/test/screens/discover_screen_test.dart @@ -651,8 +651,11 @@ class _FakeMediaServerClient implements MediaServerClient { Future> fetchContinueWatching({int? count = 20}) async => continueWatching; @override - Future> fetchGlobalHubs({int limit = defaultHubPreviewLimit, bool includePlaybackHubs = true}) async => - hubs; + Future> fetchGlobalHubs({ + int limit = defaultHubPreviewLimit, + bool includePlaybackHubs = true, + HubFetchDiagnostics? diagnostics, + }) async => hubs; @override dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation); @@ -692,7 +695,11 @@ class _GatedHubsFakeClient implements MediaServerClient { Future> fetchLibraries() async => const []; @override - Future> fetchGlobalHubs({int limit = defaultHubPreviewLimit, bool includePlaybackHubs = true}) { + Future> fetchGlobalHubs({ + int limit = defaultHubPreviewLimit, + bool includePlaybackHubs = true, + HubFetchDiagnostics? diagnostics, + }) { hubCalls++; final gate = Completer>(); _gates.add(gate); diff --git a/test/services/data_aggregation_bridge_test.dart b/test/services/data_aggregation_bridge_test.dart index 8f7ef495..fbd0c44d 100644 --- a/test/services/data_aggregation_bridge_test.dart +++ b/test/services/data_aggregation_bridge_test.dart @@ -126,6 +126,7 @@ class _GatedHubsClient implements MediaServerClient { int limit = defaultHubPreviewLimit, bool includePlaybackHubs = true, MediaKind? libraryKind, + HubFetchDiagnostics? diagnostics, }) { started.add(libraryId); return (_gates[libraryId] = Completer>()).future; diff --git a/test/services/offline_watch_sync_service_test.dart b/test/services/offline_watch_sync_service_test.dart index 1e557d1e..11e27c07 100644 --- a/test/services/offline_watch_sync_service_test.dart +++ b/test/services/offline_watch_sync_service_test.dart @@ -981,7 +981,7 @@ void main() { mgr.debugRegisterClientForTesting(clientA); final queuedScope = await svc.queueMarkWatched(serverId: ServerId('plex-machine'), itemId: 'item-1'); - expect(queuedScope, clientA.profileScopeId); + expect(queuedScope.clientScopeId, clientA.profileScopeId); expect((await db.getPendingWatchActions()).single.clientScopeId, clientA.profileScopeId); await svc.syncPendingItems(); @@ -1044,7 +1044,7 @@ void main() { final returnedScope = await svc.queueMarkWatched(serverId: ServerId('jf-machine'), itemId: 'item-1'); final queued = await db.getPendingWatchActions(); - expect(returnedScope, 'jf-machine/user-a'); + expect(returnedScope.clientScopeId, 'jf-machine/user-a'); expect(queued.single.clientScopeId, 'jf-machine/user-a'); }); @@ -1145,7 +1145,7 @@ void main() { final returnedScope = await svc.queueMarkWatched(serverId: ServerId('jf-machine'), itemId: 'item-1'); final queued = await db.getPendingWatchActions(); - expect(returnedScope, 'jf-machine/user-b'); + expect(returnedScope.clientScopeId, 'jf-machine/user-b'); expect(queued.single.clientScopeId, 'jf-machine/user-b'); });