fix(downloads): propagate hierarchical watch state

This commit is contained in:
edde746
2026-07-12 08:42:25 +02:00
parent 352aa04d3e
commit 699e73bcd5
4 changed files with 435 additions and 43 deletions
+99 -34
View File
@@ -20,6 +20,7 @@ import '../services/download_storage_service.dart';
import '../services/multi_server_manager.dart'; import '../services/multi_server_manager.dart';
import '../services/offline_mode_source.dart'; import '../services/offline_mode_source.dart';
import '../services/watch_state_resolver.dart'; import '../services/watch_state_resolver.dart';
import 'watch_state_store.dart';
import '../media/media_server_client.dart'; import '../media/media_server_client.dart';
import '../services/sync_rule_executor.dart'; import '../services/sync_rule_executor.dart';
import '../utils/app_logger.dart'; import '../utils/app_logger.dart';
@@ -62,6 +63,7 @@ class DownloadProvider extends ChangeNotifier with DisposableChangeNotifierMixin
StreamSubscription<DownloadProgress>? _progressSubscription; StreamSubscription<DownloadProgress>? _progressSubscription;
StreamSubscription<DeletionProgress>? _deletionProgressSubscription; StreamSubscription<DeletionProgress>? _deletionProgressSubscription;
StreamSubscription<WatchStateEvent>? _watchStateSubscription; StreamSubscription<WatchStateEvent>? _watchStateSubscription;
final WatchStateStore _watchStateStore = WatchStateStore();
late final Future<void> _initFuture; late final Future<void> _initFuture;
// Track download progress by public globalKey (serverId:ratingKey). // Track download progress by public globalKey (serverId:ratingKey).
@@ -74,6 +76,7 @@ class DownloadProvider extends ChangeNotifier with DisposableChangeNotifierMixin
// Store Plex thumb paths for offline display (actual file path computed from hash) // Store Plex thumb paths for offline display (actual file path computed from hash)
final Map<String, DownloadedArtwork> _artworkPaths = {}; final Map<String, DownloadedArtwork> _artworkPaths = {};
final Map<String, String?> _watchScopesByServer = {};
// Track items currently being queued (building download queue) // Track items currently being queued (building download queue)
final Set<String> _queueing = {}; final Set<String> _queueing = {};
@@ -106,6 +109,7 @@ class DownloadProvider extends ChangeNotifier with DisposableChangeNotifierMixin
// Keep cached metadata fresh when items get marked watched/unwatched anywhere // Keep cached metadata fresh when items get marked watched/unwatched anywhere
// in the app, so re-entering a screen reflects the latest state. // in the app, so re-entering a screen reflects the latest state.
_watchStateSubscription = WatchStateNotifier().stream.listen(_onWatchStateChanged); _watchStateSubscription = WatchStateNotifier().stream.listen(_onWatchStateChanged);
_watchStateStore.addListener(_onWatchStateOverlayChanged);
// Load persisted downloads from database // Load persisted downloads from database
_initFuture = _loadPersistedDownloads(); _initFuture = _loadPersistedDownloads();
@@ -125,6 +129,8 @@ class DownloadProvider extends ChangeNotifier with DisposableChangeNotifierMixin
_progressSubscription = _downloadManager.progressStream.listen(_onProgressUpdate); _progressSubscription = _downloadManager.progressStream.listen(_onProgressUpdate);
_deletionProgressSubscription = _downloadManager.deletionProgressStream.listen(_onDeletionProgressUpdate); _deletionProgressSubscription = _downloadManager.deletionProgressStream.listen(_onDeletionProgressUpdate);
_watchStateSubscription = WatchStateNotifier().stream.listen(_onWatchStateChanged); _watchStateSubscription = WatchStateNotifier().stream.listen(_onWatchStateChanged);
_watchStateStore.addListener(_onWatchStateOverlayChanged);
_watchStateStore.setActiveProfileId(_activeProfileId);
_initFuture = _loadProfileScopedState(); _initFuture = _loadProfileScopedState();
} }
@@ -144,6 +150,9 @@ class DownloadProvider extends ChangeNotifier with DisposableChangeNotifierMixin
void setActiveProfileId(String? profileId) { void setActiveProfileId(String? profileId) {
if (_activeProfileId == profileId) return; if (_activeProfileId == profileId) return;
_activeProfileId = profileId; _activeProfileId = profileId;
_watchStateStore.setActiveProfileId(profileId);
_watchScopesByServer.clear();
_watchStateStore.setActiveClientScopesByServer(const {});
_profileGeneration++; _profileGeneration++;
final reload = _reloadProfileScopedStateForActiveProfile(); final reload = _reloadProfileScopedStateForActiveProfile();
_profileScopedReloadFuture = reload; _profileScopedReloadFuture = reload;
@@ -152,10 +161,12 @@ class DownloadProvider extends ChangeNotifier with DisposableChangeNotifierMixin
Future<void> _reloadProfileScopedStateForActiveProfile() async { Future<void> _reloadProfileScopedStateForActiveProfile() async {
final targetProfileId = _activeProfileId; final targetProfileId = _activeProfileId;
final targetGeneration = _profileGeneration;
await _initFuture; await _initFuture;
if (_activeProfileId != targetProfileId) return; if (_activeProfileId != targetProfileId || _profileGeneration != targetGeneration) return;
await _loadProfileScopedState(); await _loadProfileScopedState();
if (_activeProfileId == targetProfileId) { await _applyOfflineWatchOverlay(expectedProfileGeneration: targetGeneration);
if (_activeProfileId == targetProfileId && _profileGeneration == targetGeneration) {
safeNotifyListeners(); safeNotifyListeners();
} }
} }
@@ -365,48 +376,91 @@ class DownloadProvider extends ChangeNotifier with DisposableChangeNotifierMixin
} }
} }
/// Patch `_metadata` viewCount/viewOffsetMs from queued OfflineWatchProgress /// Hydrate queued OfflineWatchProgress actions into the canonical
/// actions. Idempotent and cheap (one batched DB read). /// hierarchy-aware watch-state layer.
Future<void> _applyOfflineWatchOverlay({int? expectedProfileGeneration}) async { Future<void> _applyOfflineWatchOverlay({int? expectedProfileGeneration}) async {
if (_metadata.isEmpty) return;
bool isStale() => bool isStale() =>
expectedProfileGeneration != null && (isDisposed || expectedProfileGeneration != _profileGeneration); expectedProfileGeneration != null && (isDisposed || expectedProfileGeneration != _profileGeneration);
try { try {
final keys = _metadata.keys.toSet(); final profileId = _activeProfileId;
if (profileId == null || profileId.isEmpty) {
_watchStateStore.setHydratedPatches(const []);
return;
}
final keys = <String>{};
for (final item in _metadata.values) {
keys.add(item.globalKey);
final serverId = serverIdOrNull(item.serverId);
if (serverId == null) continue;
for (final parentId in item.parentChain) {
keys.add(buildGlobalKey(serverId, parentId));
}
}
if (keys.isEmpty) {
_watchStateStore.setHydratedPatches(const []);
return;
}
final scopes = <String, String?>{}; final scopes = <String, String?>{};
final scopesByServer = <String, String?>{};
for (final key in keys) { for (final key in keys) {
scopes[key] = await _offlineWatchScopeForGlobalKey(key); final parsed = parseGlobalKey(key);
if (parsed == null) continue;
var scope = scopesByServer[parsed.serverId];
if (!scopesByServer.containsKey(parsed.serverId)) {
scope = await _offlineWatchScopeForServer(parsed.serverId);
scopesByServer[parsed.serverId] = scope;
}
scopes[key] = scope;
if (isStale()) return; if (isStale()) return;
} }
final profileId = _activeProfileId; _watchScopesByServer
..clear()
..addAll(scopesByServer);
_watchStateStore.setActiveClientScopesByServer(_watchScopesByServer);
final actions = await _database.getWatchActionsForKeys( final actions = await _database.getWatchActionsForKeys(
keys, keys,
profileId: profileId, profileId: profileId,
filterProfile: profileId != null, filterProfile: true,
clientScopeIdsByGlobalKey: scopes, clientScopeIdsByGlobalKey: scopes,
); );
if (isStale()) return; if (isStale()) return;
if (actions.isEmpty) return; final hydrated = <HydratedWatchStatePatch>[];
for (final entry in actions.entries) { for (final entry in actions.entries) {
final base = _metadata[entry.key];
if (base == null) continue;
final snapshot = WatchStateResolver.fromActions(entry.value); final snapshot = WatchStateResolver.fromActions(entry.value);
if (snapshot.isEmpty) continue; if (snapshot.isEmpty) continue;
_metadata[entry.key] = snapshot.apply(base); final latest = entry.value.firstWhere(
(action) =>
action.actionType == 'watched' || action.actionType == 'unwatched' || action.actionType == 'progress',
);
final scopedKey = latest.clientScopeId != null && latest.clientScopeId!.isNotEmpty
? buildGlobalKey(ServerId(latest.clientScopeId!), latest.ratingKey)
: latest.globalKey;
hydrated.add(
HydratedWatchStatePatch(
globalKey: scopedKey,
patch: WatchStatePatch.fromSnapshot(snapshot),
updatedAt: latest.updatedAt,
order: latest.id,
),
);
} }
_watchStateStore.setHydratedPatches(hydrated);
} catch (e) { } catch (e) {
appLogger.w('Failed to apply offline watch overlay', error: e); appLogger.w('Failed to apply offline watch overlay', error: e);
} }
} }
Future<String?> _offlineWatchScopeForGlobalKey(String globalKey) async { Future<String?> _offlineWatchScopeForServer(String serverId) async {
final parsed = parseGlobalKey(globalKey); final activeScope = _downloadManager.activeClientScopeIdForServer(ServerId(serverId));
if (parsed == null) return null;
final activeScope = _downloadManager.activeClientScopeIdForServer(parsed.serverId);
if (activeScope != null && activeScope.isNotEmpty) return activeScope; if (activeScope != null && activeScope.isNotEmpty) return activeScope;
final downloaded = await _database.getDownloadedMedia(globalKey); for (final globalKey in _downloads.keys) {
final downloadedScope = downloaded?.clientScopeId; if (!_ownsDownloadKey(globalKey)) continue;
return downloadedScope == null || downloadedScope.isEmpty ? null : downloadedScope; final parsed = parseGlobalKey(globalKey);
if (parsed?.serverId != serverId) continue;
final downloadedScope = (await _database.getDownloadedMedia(globalKey))?.clientScopeId;
if (downloadedScope != null && downloadedScope.isNotEmpty) return downloadedScope;
}
return null;
} }
/// Load parent metadata (show + season for episodes, artist + album for /// Load parent metadata (show + season for episodes, artist + album for
@@ -459,24 +513,30 @@ class DownloadProvider extends ChangeNotifier with DisposableChangeNotifierMixin
_progressSubscription?.cancel(); _progressSubscription?.cancel();
_deletionProgressSubscription?.cancel(); _deletionProgressSubscription?.cancel();
_watchStateSubscription?.cancel(); _watchStateSubscription?.cancel();
_watchStateStore.removeListener(_onWatchStateOverlayChanged);
_watchStateStore.dispose();
super.dispose(); super.dispose();
} }
void _onWatchStateOverlayChanged() => safeNotifyListeners();
void _onWatchStateChanged(WatchStateEvent event) { void _onWatchStateChanged(WatchStateEvent event) {
final snapshot = WatchStateResolver.fromEvent(event); final snapshot = WatchStateResolver.fromEvent(event);
if (snapshot.isEmpty) return; if (snapshot.isEmpty) return;
final globalKey = buildGlobalKey(ServerId(event.serverId), event.itemId); final globalKey = buildGlobalKey(ServerId(event.serverId), event.itemId);
final base = _metadata[globalKey]; final base = _metadata[globalKey];
if (base == null) return;
final eventScope = event.cacheServerId; final eventScope = event.cacheServerId;
final activeScope = _downloadManager.activeClientScopeIdForServer(ServerId(event.serverId)); final activeScope = _downloadManager.activeClientScopeIdForServer(ServerId(event.serverId));
if (activeScope != null && activeScope.isNotEmpty) {
_watchScopesByServer[event.serverId] = activeScope;
_watchStateStore.setActiveClientScopesByServer(_watchScopesByServer);
}
if (base == null) return;
if (eventScope != null && eventScope.isNotEmpty && eventScope != event.serverId && eventScope != activeScope) { if (eventScope != null && eventScope.isNotEmpty && eventScope != event.serverId && eventScope != activeScope) {
return; return;
} }
_metadata[globalKey] = snapshot.apply(base);
final isWatched = snapshot.isWatched; final isWatched = snapshot.isWatched;
// Sub-threshold progress ticks are frequent; offline reloads re-apply them // Sub-threshold progress ticks are frequent; offline reloads re-apply them
// from queued watch actions, so only durable watch flips hit the cache here. // from queued watch actions, so only durable watch flips hit the cache here.
@@ -507,7 +567,6 @@ class DownloadProvider extends ChangeNotifier with DisposableChangeNotifierMixin
}), }),
); );
} }
safeNotifyListeners();
} }
/// Ensure metadata has a serverId, falling back to a parent's serverId. /// Ensure metadata has a serverId, falling back to a parent's serverId.
@@ -519,7 +578,8 @@ class DownloadProvider extends ChangeNotifier with DisposableChangeNotifierMixin
Map.unmodifiable(Map.fromEntries(_downloads.entries.where(_ownsProgressEntry))); Map.unmodifiable(Map.fromEntries(_downloads.entries.where(_ownsProgressEntry)));
/// All metadata for downloads /// All metadata for downloads
Map<String, MediaItem> get metadata => Map.unmodifiable(_metadata); Map<String, MediaItem> get metadata =>
Map.unmodifiable({for (final entry in _metadata.entries) entry.key: _watchStateStore.apply(entry.value)});
/// Get unique TV shows that have downloaded episodes /// Get unique TV shows that have downloaded episodes
/// Returns stored show metadata, or synthesizes from episode metadata as fallback /// Returns stored show metadata, or synthesizes from episode metadata as fallback
@@ -529,7 +589,7 @@ class DownloadProvider extends ChangeNotifier with DisposableChangeNotifierMixin
for (final entry in _metadata.entries) { for (final entry in _metadata.entries) {
final globalKey = entry.key; final globalKey = entry.key;
if (!_ownsDownloadKey(globalKey)) continue; if (!_ownsDownloadKey(globalKey)) continue;
final meta = entry.value; final meta = _watchStateStore.apply(entry.value);
final progress = _downloads[globalKey]; final progress = _downloads[globalKey];
if (progress?.status == DownloadStatus.completed && meta.isEpisode) { if (progress?.status == DownloadStatus.completed && meta.isEpisode) {
@@ -537,7 +597,7 @@ class DownloadProvider extends ChangeNotifier with DisposableChangeNotifierMixin
if (showRatingKey != null && !shows.containsKey(showRatingKey)) { if (showRatingKey != null && !shows.containsKey(showRatingKey)) {
// Try to get stored show metadata first // Try to get stored show metadata first
final showGlobalKey = buildGlobalKey(ServerId(meta.serverId!), showRatingKey); final showGlobalKey = buildGlobalKey(ServerId(meta.serverId!), showRatingKey);
final storedShow = _metadata[showGlobalKey]; final storedShow = _resolvedMetadata(showGlobalKey);
if (storedShow != null && storedShow.isShow) { if (storedShow != null && storedShow.isShow) {
// Use stored show metadata (has year, summary, clearLogo) // Use stored show metadata (has year, summary, clearLogo)
@@ -577,7 +637,7 @@ class DownloadProvider extends ChangeNotifier with DisposableChangeNotifierMixin
final progress = _downloads[entry.key]; final progress = _downloads[entry.key];
return progress?.status == DownloadStatus.completed && entry.value.isMovie; return progress?.status == DownloadStatus.completed && entry.value.isMovie;
}) })
.map((entry) => entry.value) .map((entry) => _watchStateStore.apply(entry.value))
.toList(); .toList();
} }
@@ -590,7 +650,7 @@ class DownloadProvider extends ChangeNotifier with DisposableChangeNotifierMixin
for (final entry in _metadata.entries) { for (final entry in _metadata.entries) {
final globalKey = entry.key; final globalKey = entry.key;
if (!_ownsDownloadKey(globalKey)) continue; if (!_ownsDownloadKey(globalKey)) continue;
final meta = entry.value; final meta = _watchStateStore.apply(entry.value);
if (meta.kind != MediaKind.track) continue; if (meta.kind != MediaKind.track) continue;
if (_downloads[globalKey]?.status != DownloadStatus.completed) continue; if (_downloads[globalKey]?.status != DownloadStatus.completed) continue;
@@ -598,7 +658,7 @@ class DownloadProvider extends ChangeNotifier with DisposableChangeNotifierMixin
if (albumRatingKey == null || albums.containsKey(albumRatingKey)) continue; if (albumRatingKey == null || albums.containsKey(albumRatingKey)) continue;
final albumGlobalKey = buildGlobalKey(ServerId(meta.serverId!), albumRatingKey); final albumGlobalKey = buildGlobalKey(ServerId(meta.serverId!), albumRatingKey);
final storedAlbum = _metadata[albumGlobalKey]; final storedAlbum = _resolvedMetadata(albumGlobalKey);
if (storedAlbum != null && storedAlbum.kind == MediaKind.album) { if (storedAlbum != null && storedAlbum.kind == MediaKind.album) {
albums[albumRatingKey] = storedAlbum; albums[albumRatingKey] = storedAlbum;
} else { } else {
@@ -635,7 +695,7 @@ class DownloadProvider extends ChangeNotifier with DisposableChangeNotifierMixin
meta.parentId == albumRatingKey && meta.parentId == albumRatingKey &&
_downloads[entry.key]?.status == DownloadStatus.completed; _downloads[entry.key]?.status == DownloadStatus.completed;
}) })
.map((entry) => entry.value) .map((entry) => _watchStateStore.apply(entry.value))
.toList(); .toList();
tracks.sort((a, b) { tracks.sort((a, b) {
final byDisc = (a.discNumber ?? 1).compareTo(b.discNumber ?? 1); final byDisc = (a.discNumber ?? 1).compareTo(b.discNumber ?? 1);
@@ -646,7 +706,12 @@ class DownloadProvider extends ChangeNotifier with DisposableChangeNotifierMixin
} }
/// Get metadata for a specific download /// Get metadata for a specific download
MediaItem? getMetadata(String globalKey) => _metadata[globalKey]; MediaItem? _resolvedMetadata(String globalKey) {
final item = _metadata[globalKey];
return item == null ? null : _watchStateStore.apply(item);
}
MediaItem? getMetadata(String globalKey) => _resolvedMetadata(globalKey);
/// Get artwork paths for a specific download (for offline display) /// Get artwork paths for a specific download (for offline display)
DownloadedArtwork? getArtworkPaths(String globalKey) => _artworkPaths[globalKey]; DownloadedArtwork? getArtworkPaths(String globalKey) => _artworkPaths[globalKey];
@@ -667,7 +732,7 @@ class DownloadProvider extends ChangeNotifier with DisposableChangeNotifierMixin
final meta = entry.value; final meta = entry.value;
return progress?.status == DownloadStatus.completed && meta.isEpisode && meta.grandparentId == showRatingKey; return progress?.status == DownloadStatus.completed && meta.isEpisode && meta.grandparentId == showRatingKey;
}) })
.map((entry) => entry.value) .map((entry) => _watchStateStore.apply(entry.value))
.toList(); .toList();
} }
@@ -1656,7 +1721,7 @@ class DownloadProvider extends ChangeNotifier with DisposableChangeNotifierMixin
.toList(); .toList();
for (final globalKey in completedKeys) { for (final globalKey in completedKeys) {
final meta = _metadata[globalKey]; final meta = _resolvedMetadata(globalKey);
if (meta == null) continue; if (meta == null) continue;
if (!meta.isEpisode && !meta.isMovie) continue; if (!meta.isEpisode && !meta.isMovie) continue;
if (!meta.isWatched) continue; if (!meta.isWatched) continue;
+86 -9
View File
@@ -37,11 +37,50 @@ class WatchStatePatch {
int get hashCode => Object.hash(isWatched, hasViewOffsetMs, viewOffsetMs); int get hashCode => Object.hash(isWatched, hasViewOffsetMs, viewOffsetMs);
} }
@immutable
class HydratedWatchStatePatch {
final String globalKey;
final WatchStatePatch patch;
final int updatedAt;
final int order;
const HydratedWatchStatePatch({
required this.globalKey,
required this.patch,
required this.updatedAt,
required this.order,
});
}
class _WatchStatePatchEntry { class _WatchStatePatchEntry {
final WatchStatePatch patch; final WatchStatePatch patch;
final int updatedAt;
final int sequence; final int sequence;
final bool isSessionEvent;
const _WatchStatePatchEntry(this.patch, this.sequence); const _WatchStatePatchEntry(
this.patch, {
required this.updatedAt,
required this.sequence,
required this.isSessionEvent,
});
bool isNewerThan(_WatchStatePatchEntry other) {
if (updatedAt != other.updatedAt) return updatedAt > other.updatedAt;
if (isSessionEvent != other.isSessionEvent) return isSessionEvent;
return sequence > other.sequence;
}
@override
bool operator ==(Object other) =>
other is _WatchStatePatchEntry &&
other.patch == patch &&
other.updatedAt == updatedAt &&
other.sequence == sequence &&
other.isSessionEvent == isSessionEvent;
@override
int get hashCode => Object.hash(patch, updatedAt, sequence, isSessionEvent);
} }
/// The single session-local layer for watch-state freshness. /// The single session-local layer for watch-state freshness.
@@ -60,23 +99,32 @@ class WatchStateStore extends ChangeNotifier with DisposableChangeNotifierMixin
StreamSubscription<WatchStateEvent>? _subscription; StreamSubscription<WatchStateEvent>? _subscription;
final Map<String, _WatchStatePatchEntry> _patches = {}; final Map<String, _WatchStatePatchEntry> _patches = {};
final Map<String, _WatchStatePatchEntry> _hydratedPatches = {};
String? _activeProfileId; String? _activeProfileId;
Map<String, String?> _activeClientScopesByServer = const {}; Map<String, String?> _activeClientScopesByServer = const {};
int _sequence = 0; int _sequence = 0;
_WatchStatePatchEntry? _exactEntryFor(String globalKey) {
final session = _patches[globalKey];
final hydrated = _hydratedPatches[globalKey];
if (session == null) return hydrated;
if (hydrated == null) return session;
return session.isNewerThan(hydrated) ? session : hydrated;
}
_WatchStatePatchEntry? _entryFor(String globalKey) { _WatchStatePatchEntry? _entryFor(String globalKey) {
_WatchStatePatchEntry? scopedEntry; _WatchStatePatchEntry? scopedEntry;
final parsed = parseGlobalKey(globalKey); final parsed = parseGlobalKey(globalKey);
if (parsed != null) { if (parsed != null) {
final scoped = _activeClientScopesByServer[parsed.serverId]; final scoped = _activeClientScopesByServer[parsed.serverId];
if (scoped != null && scoped.isNotEmpty) { if (scoped != null && scoped.isNotEmpty) {
scopedEntry = _patches[buildGlobalKey(ServerId(scoped), parsed.ratingKey)]; scopedEntry = _exactEntryFor(buildGlobalKey(ServerId(scoped), parsed.ratingKey));
} }
} }
final unscopedEntry = _patches[globalKey]; final unscopedEntry = _exactEntryFor(globalKey);
if (scopedEntry == null) return unscopedEntry; if (scopedEntry == null) return unscopedEntry;
if (unscopedEntry == null) return scopedEntry; if (unscopedEntry == null) return scopedEntry;
return scopedEntry.sequence >= unscopedEntry.sequence ? scopedEntry : unscopedEntry; return scopedEntry.isNewerThan(unscopedEntry) ? scopedEntry : unscopedEntry;
} }
WatchStatePatch? patchForGlobalKey(String globalKey) => _entryFor(globalKey)?.patch; WatchStatePatch? patchForGlobalKey(String globalKey) => _entryFor(globalKey)?.patch;
@@ -88,7 +136,7 @@ class WatchStateStore extends ChangeNotifier with DisposableChangeNotifierMixin
for (final parentId in item.parentChain) { for (final parentId in item.parentChain) {
// Mirror MediaItem.globalKey's bare-id fallback when serverId is missing. // 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);
if (entry != null && (best == null || entry.sequence > best.sequence)) best = entry; if (entry != null && (best == null || entry.isNewerThan(best))) best = entry;
} }
} }
return best?.patch; return best?.patch;
@@ -99,7 +147,7 @@ class WatchStateStore extends ChangeNotifier with DisposableChangeNotifierMixin
} }
List<MediaItem> applyAll(List<MediaItem> items) { List<MediaItem> applyAll(List<MediaItem> items) {
if (_patches.isEmpty) return items; if (_patches.isEmpty && _hydratedPatches.isEmpty) return items;
return [for (final item in items) apply(item)]; return [for (final item in items) apply(item)];
} }
@@ -115,8 +163,9 @@ class WatchStateStore extends ChangeNotifier with DisposableChangeNotifierMixin
void setActiveProfileId(String? profileId) { void setActiveProfileId(String? profileId) {
if (_activeProfileId == profileId) return; if (_activeProfileId == profileId) return;
_activeProfileId = profileId; _activeProfileId = profileId;
if (_patches.isEmpty) return; if (_patches.isEmpty && _hydratedPatches.isEmpty) return;
_patches.clear(); _patches.clear();
_hydratedPatches.clear();
safeNotifyListeners(); safeNotifyListeners();
} }
@@ -127,7 +176,30 @@ class WatchStateStore extends ChangeNotifier with DisposableChangeNotifierMixin
}; };
if (mapEquals(_activeClientScopesByServer, normalized)) return; if (mapEquals(_activeClientScopesByServer, normalized)) return;
_activeClientScopesByServer = Map.unmodifiable(normalized); _activeClientScopesByServer = Map.unmodifiable(normalized);
if (_patches.isNotEmpty) safeNotifyListeners(); if (_patches.isNotEmpty || _hydratedPatches.isNotEmpty) safeNotifyListeners();
}
/// Replace the persisted local-action layer without disturbing newer
/// session events. Timestamps preserve freshness across item/ancestor keys.
void setHydratedPatches(Iterable<HydratedWatchStatePatch> patches) {
final next = <String, _WatchStatePatchEntry>{};
for (final hydrated in patches) {
final candidate = _WatchStatePatchEntry(
hydrated.patch,
updatedAt: hydrated.updatedAt,
sequence: hydrated.order,
isSessionEvent: false,
);
final existing = next[hydrated.globalKey];
if (existing == null || candidate.isNewerThan(existing)) {
next[hydrated.globalKey] = candidate;
}
}
if (mapEquals(_hydratedPatches, next)) return;
_hydratedPatches
..clear()
..addAll(next);
safeNotifyListeners();
} }
void _onWatchStateEvent(WatchStateEvent event) { void _onWatchStateEvent(WatchStateEvent event) {
@@ -139,7 +211,12 @@ class WatchStateStore extends ChangeNotifier with DisposableChangeNotifierMixin
final key = cacheServerId != null && cacheServerId.isNotEmpty && cacheServerId != event.serverId final key = cacheServerId != null && cacheServerId.isNotEmpty && cacheServerId != event.serverId
? buildGlobalKey(ServerId(cacheServerId), event.itemId) ? buildGlobalKey(ServerId(cacheServerId), event.itemId)
: event.globalKey; : event.globalKey;
_patches[key] = _WatchStatePatchEntry(patch, ++_sequence); _patches[key] = _WatchStatePatchEntry(
patch,
updatedAt: DateTime.now().millisecondsSinceEpoch,
sequence: ++_sequence,
isSessionEvent: true,
);
safeNotifyListeners(); safeNotifyListeners();
} }
+203
View File
@@ -1114,6 +1114,209 @@ void main() {
p.dispose(); p.dispose();
}); });
test('show, season, and episode events resolve by hierarchical freshness', () async {
final p = DownloadProvider.forTesting(downloadManager: downloadManager, database: db);
await p.ensureInitialized();
final show = MediaItem(
id: 'show-1',
backend: MediaBackend.plex,
kind: MediaKind.show,
serverId: ServerId('srv'),
leafCount: 3,
viewedLeafCount: 0,
);
final season1 = MediaItem(
id: 'season-1',
backend: MediaBackend.plex,
kind: MediaKind.season,
parentId: 'show-1',
serverId: ServerId('srv'),
leafCount: 2,
viewedLeafCount: 0,
);
final episode1 = MediaItem(
id: 'episode-1',
backend: MediaBackend.plex,
kind: MediaKind.episode,
parentId: 'season-1',
grandparentId: 'show-1',
serverId: ServerId('srv'),
viewCount: 0,
viewOffsetMs: 40000,
);
final episode2 = episode1.copyWith(id: 'episode-2', viewOffsetMs: 50000);
final episode3 = episode1.copyWith(id: 'episode-3', parentId: 'season-2', viewOffsetMs: 60000);
p.debugSeedState(
metadata: {
show.globalKey: show,
season1.globalKey: season1,
episode1.globalKey: episode1,
episode2.globalKey: episode2,
episode3.globalKey: episode3,
},
);
WatchStateNotifier().notifyWatched(item: show);
await Future<void>.delayed(Duration.zero);
expect(p.getMetadata(episode1.globalKey)?.isWatched, isTrue);
expect(p.getMetadata(episode2.globalKey)?.viewOffsetMs, 0);
expect(p.getMetadata(episode3.globalKey)?.isWatched, isTrue);
WatchStateNotifier().notifyWatched(item: season1, isNowWatched: false);
await Future<void>.delayed(Duration.zero);
expect(p.getMetadata(episode1.globalKey)?.isWatched, isFalse);
expect(p.getMetadata(episode2.globalKey)?.isWatched, isFalse);
expect(p.getMetadata(episode3.globalKey)?.isWatched, isTrue);
WatchStateNotifier().notifyWatched(item: episode1);
await Future<void>.delayed(Duration.zero);
expect(p.getMetadata(episode1.globalKey)?.isWatched, isTrue);
expect(p.getMetadata(episode2.globalKey)?.isWatched, isFalse);
WatchStateNotifier().notifyWatched(item: show, isNowWatched: false);
await Future<void>.delayed(Duration.zero);
expect(p.getMetadata(episode1.globalKey)?.isWatched, isFalse);
expect(p.getMetadata(episode2.globalKey)?.isWatched, isFalse);
expect(p.getMetadata(episode3.globalKey)?.isWatched, isFalse);
expect(p.getMetadata(episode3.globalKey)?.viewOffsetMs, 0);
p.dispose();
});
test('queued parent and episode overrides survive provider reload', () async {
await db.insertWatchAction(
profileId: 'profile-a',
serverId: ServerId('srv'),
ratingKey: 'show-1',
actionType: 'watched',
);
await db.insertWatchAction(
profileId: 'profile-a',
serverId: ServerId('srv'),
ratingKey: 'episode-1',
actionType: 'unwatched',
);
final episode1 = MediaItem(
id: 'episode-1',
backend: MediaBackend.plex,
kind: MediaKind.episode,
parentId: 'season-1',
grandparentId: 'show-1',
serverId: ServerId('srv'),
viewCount: 0,
viewOffsetMs: 45000,
);
final episode2 = episode1.copyWith(id: 'episode-2', viewOffsetMs: 55000);
Future<DownloadProvider> hydrate() async {
final provider = DownloadProvider.forTesting(
downloadManager: downloadManager,
database: db,
activeProfileId: 'profile-a',
);
await provider.ensureInitialized();
provider.debugSeedState(metadata: {episode1.globalKey: episode1, episode2.globalKey: episode2});
await provider.refreshMetadataFromCache();
return provider;
}
var p = await hydrate();
expect(p.getMetadata(episode1.globalKey)?.isWatched, isFalse);
expect(p.getMetadata(episode1.globalKey)?.viewOffsetMs, 0);
expect(p.getMetadata(episode2.globalKey)?.isWatched, isTrue);
expect(p.getMetadata(episode2.globalKey)?.viewOffsetMs, 0);
p.dispose();
p = await hydrate();
expect(p.getMetadata(episode1.globalKey)?.isWatched, isFalse);
expect(p.getMetadata(episode2.globalKey)?.isWatched, isTrue);
p.dispose();
});
test('queued overlays are isolated to the active profile', () async {
await db.insertWatchAction(
profileId: 'profile-a',
serverId: ServerId('srv'),
ratingKey: 'show-1',
actionType: 'watched',
);
await db.insertWatchAction(
profileId: 'profile-b',
serverId: ServerId('srv'),
ratingKey: 'show-1',
actionType: 'unwatched',
);
final episode = MediaItem(
id: 'episode-1',
backend: MediaBackend.plex,
kind: MediaKind.episode,
parentId: 'season-1',
grandparentId: 'show-1',
serverId: ServerId('srv'),
viewCount: 0,
);
final p = DownloadProvider.forTesting(
downloadManager: downloadManager,
database: db,
activeProfileId: 'profile-a',
);
await p.ensureInitialized();
p.debugSeedState(metadata: {episode.globalKey: episode});
await p.refreshMetadataFromCache();
expect(p.getMetadata(episode.globalKey)?.isWatched, isTrue);
p.setActiveProfileId('profile-b');
await p.refreshMetadataFromCache();
expect(p.getMetadata(episode.globalKey)?.isWatched, isFalse);
p.dispose();
});
test('queued overlays are isolated to the active Jellyfin client scope', () async {
await db.insertWatchAction(
profileId: 'test-profile',
serverId: ServerId('jf-machine'),
clientScopeId: 'jf-machine/user-a',
ratingKey: 'show-1',
actionType: 'watched',
);
await db.insertWatchAction(
profileId: 'test-profile',
serverId: ServerId('jf-machine'),
clientScopeId: 'jf-machine/user-b',
ratingKey: 'show-1',
actionType: 'unwatched',
);
final episode = MediaItem(
id: 'episode-1',
backend: MediaBackend.jellyfin,
kind: MediaKind.episode,
parentId: 'season-1',
grandparentId: 'show-1',
serverId: ServerId('jf-machine'),
viewCount: 0,
);
var activeScope = 'jf-machine/user-a';
testClientResolver = (serverId, {clientScopeId}) => serverId == 'jf-machine'
? _ScopedTestClient(serverId: ServerId('jf-machine'), scopedServerId: activeScope)
: null;
final p = DownloadProvider.forTesting(downloadManager: downloadManager, database: db);
await p.ensureInitialized();
p.debugSeedState(metadata: {episode.globalKey: episode});
await p.refreshMetadataFromCache();
expect(p.getMetadata(episode.globalKey)?.isWatched, isTrue);
activeScope = 'jf-machine/user-b';
await p.refreshMetadataFromCache();
expect(p.getMetadata(episode.globalKey)?.isWatched, isFalse);
p.dispose();
});
}); });
group('DownloadProvider — progress stream', () { group('DownloadProvider — progress stream', () {
@@ -157,4 +157,51 @@ void main() {
expect(resolved.viewedLeafCount, 10); expect(resolved.viewedLeafCount, 10);
expect(resolved.isWatched, isTrue); expect(resolved.isWatched, isTrue);
}); });
test('hydrated parent and item patches retain persisted freshness', () {
final store = WatchStateStore();
addTearDown(store.dispose);
store.setHydratedPatches(const [
HydratedWatchStatePatch(
globalKey: 'jf-machine:show-1',
patch: WatchStatePatch(isWatched: true, hasViewOffsetMs: true, viewOffsetMs: 0),
updatedAt: 100,
order: 1,
),
HydratedWatchStatePatch(
globalKey: 'jf-machine:episode-1',
patch: WatchStatePatch(isWatched: false, hasViewOffsetMs: true, viewOffsetMs: 0),
updatedAt: 200,
order: 2,
),
]);
final resolved = store.apply(_episode.copyWith(viewOffsetMs: 30000));
expect(resolved.isWatched, isFalse);
expect(resolved.viewOffsetMs, 0);
});
test('hydrated patches are isolated to the active client scope', () {
final store = WatchStateStore();
addTearDown(store.dispose);
store.setHydratedPatches(const [
HydratedWatchStatePatch(
globalKey: 'jf-machine/user-a:show-1',
patch: WatchStatePatch(isWatched: true),
updatedAt: 100,
order: 1,
),
HydratedWatchStatePatch(
globalKey: 'jf-machine/user-b:show-1',
patch: WatchStatePatch(isWatched: false),
updatedAt: 100,
order: 2,
),
]);
store.setActiveClientScopesByServer({'jf-machine': 'jf-machine/user-a'});
expect(store.apply(_episode).isWatched, isTrue);
store.setActiveClientScopesByServer({'jf-machine': 'jf-machine/user-b'});
expect(store.apply(_episode).isWatched, isFalse);
});
} }