From 2b7142bdcd16cdae2327e7129684dd284ee8a10e Mon Sep 17 00:00:00 2001 From: edde746 <86283021+edde746@users.noreply.github.com> Date: Thu, 2 Jul 2026 07:30:20 +0200 Subject: [PATCH] fix(downloads): profile-scoped ownership and watch-sync integrity --- lib/database/app_database.dart | 5 ++ lib/database/download_operations.dart | 25 ++++++- lib/main.dart | 11 ++- lib/providers/download_provider.dart | 72 +++++++++++++------ lib/screens/profile/profile_teardown.dart | 5 ++ lib/services/offline_watch_sync_service.dart | 19 ++++- lib/services/sync_rule_executor.dart | 12 ++-- .../offline_watch_sync_service_test.dart | 68 ++++++++++++++++++ 8 files changed, 186 insertions(+), 31 deletions(-) diff --git a/lib/database/app_database.dart b/lib/database/app_database.dart index b0e751f5..699196a5 100644 --- a/lib/database/app_database.dart +++ b/lib/database/app_database.dart @@ -631,6 +631,11 @@ class AppDatabase extends _$AppDatabase { await (delete(syncRules)..where((t) => t.profileId.equals(profileId))).go(); } + /// Drop every sync rule (full logout). + Future clearAllSyncRules() async { + await delete(syncRules).go(); + } + /// Get all downloaded media items (for syncing watch states) Future> getAllDownloadedMetadata() { return (select(downloadedMedia)..where((t) => t.status.equals(DownloadStatus.completed.index))).get(); diff --git a/lib/database/download_operations.dart b/lib/database/download_operations.dart index b6de07eb..a78fd857 100644 --- a/lib/database/download_operations.dart +++ b/lib/database/download_operations.dart @@ -65,12 +65,33 @@ extension DownloadDatabaseOperations on AppDatabase { } /// Claim pre-v17 shared download rows for [profileId]. Rows that already - /// have any owner are left untouched so later profiles do not inherit them. + /// have any valid owner are left untouched so later profiles do not + /// inherit them. + /// + /// Runs on every profile switch — validity context is computed once and + /// applied in memory instead of the per-download full-table rescan + /// `getDownloadOwnerCount` would do. Future adoptLegacyDownloadsForProfile(String profileId) async { if (profileId.isEmpty) return; final rows = await select(downloadedMedia).get(); + if (rows.isEmpty) return; + + final owners = await select(downloadOwners).get(); + final localProfileIds = (await select(profiles).get()).map((row) => row.id).toSet(); + final connectionIds = (await select(connections).get()).map((row) => row.id).toSet(); + bool valid(DownloadOwnerItem owner) { + if (localProfileIds.contains(owner.profileId)) return true; + final plexHome = parsePlexHomeProfileId(owner.profileId); + if (plexHome != null) return connectionIds.contains(plexHome.accountConnectionId); + return localProfileIds.isEmpty; + } + + final ownedKeys = { + for (final owner in owners) + if (valid(owner)) owner.globalKey, + }; for (final row in rows) { - if (await getDownloadOwnerCount(row.globalKey) == 0) { + if (!ownedKeys.contains(row.globalKey)) { await addDownloadOwner(profileId: profileId, globalKey: row.globalKey); } } diff --git a/lib/main.dart b/lib/main.dart index 520763f7..4f1c8753 100644 --- a/lib/main.dart +++ b/lib/main.dart @@ -780,7 +780,11 @@ class _MainAppState extends State with WidgetsBindingObserver { final activeProfile = context.read(); _offlineWatchSyncService.setActiveProfileId( activeProfile.activeId, - availableProfileCount: activeProfile.profiles.length, + // Legacy-adoption gate: only trust the count once the provider + // has hydrated (locals + cached home users) — a transient + // count of 1 mid-load would permanently mis-adopt pre-profile + // watch actions. + availableProfileCount: activeProfile.isInitialized ? activeProfile.profiles.length : null, ); // Offline-sync drain replays a batch of queued watch actions without @@ -818,7 +822,10 @@ class _MainAppState extends State with WidgetsBindingObserver { }, update: (_, activeProfile, previous) { final provider = previous ?? _offlineWatchSyncService; - provider.setActiveProfileId(activeProfile.activeId, availableProfileCount: activeProfile.profiles.length); + provider.setActiveProfileId( + activeProfile.activeId, + availableProfileCount: activeProfile.isInitialized ? activeProfile.profiles.length : null, + ); return provider; }, ), diff --git a/lib/providers/download_provider.dart b/lib/providers/download_provider.dart index 70b91e95..c6019532 100644 --- a/lib/providers/download_provider.dart +++ b/lib/providers/download_provider.dart @@ -165,10 +165,13 @@ class DownloadProvider extends ChangeNotifier with DisposableChangeNotifierMixin bool _ownsProgressEntry(MapEntry entry) => _ownsDownloadKey(entry.key); - Future _claimDownloadForActiveProfile(String globalKey) async { - final profileId = _requireActiveProfileId(); - if (_ownedDownloadKeys.contains(globalKey)) return false; + /// Claim [globalKey] for an explicit [profileId] — sync rules claim for + /// the RULE'S owner, not whoever is active when the pass lands, so a + /// mid-run profile switch can't leak ownership across profiles. + Future _claimDownloadForProfile(String globalKey, String profileId) async { + if (_activeProfileId == profileId && _ownedDownloadKeys.contains(globalKey)) return false; await _database.addDownloadOwner(profileId: profileId, globalKey: globalKey); + // _ownedDownloadKeys mirrors only the active profile's rows. if (_activeProfileId != profileId) return false; _ownedDownloadKeys.add(globalKey); return true; @@ -469,15 +472,24 @@ class DownloadProvider extends ChangeNotifier with DisposableChangeNotifierMixin // (`_loadPersistedDownloads` rehydrates `_metadata` from the cache). if (shouldPersistToCache) { unawaited( - ApiCache.forBackend(base.backend) - .applyWatchState( - serverId: ServerId(event.cacheServerId ?? event.serverId), - itemId: event.itemId, - isWatched: isWatched, - ) - .catchError((Object e) { - appLogger.w('Failed to apply watch state to cache for $globalKey', error: e); - }), + () async { + // Jellyfin cache rows are per-user (cacheServerId embeds the user); + // Plex rows are keyed by server only, so persisting one user's flip + // into a download SHARED with another profile would surface as that + // profile's watch state too. Skip the shared case — each profile's + // own queued watch actions still re-apply its state on reload. + if (base.backend == MediaBackend.plex && + await _database.hasDownloadOwner(globalKey, excludingProfileId: _activeProfileId)) { + return; + } + await ApiCache.forBackend(base.backend).applyWatchState( + serverId: ServerId(event.cacheServerId ?? event.serverId), + itemId: event.itemId, + isWatched: isWatched, + ); + }().catchError((Object e) { + appLogger.w('Failed to apply watch state to cache for $globalKey', error: e); + }), ); } safeNotifyListeners(); @@ -968,21 +980,22 @@ class DownloadProvider extends ChangeNotifier with DisposableChangeNotifierMixin int mediaIndex = 0, DownloadVersionConfig? versionConfig, _RelatedMetadataDownloadContext? relatedContext, + String? claimForProfileId, }) async { if (!_downloadManager.downloadsSupported) return false; - _requireActiveProfileId(); + final ownerProfileId = claimForProfileId ?? _requireActiveProfileId(); final globalKey = metadata.globalKey; // Don't duplicate the physical download. If another profile already owns - // the shared row, claiming it makes it visible for the active profile. + // the shared row, claiming it makes it visible for the owning profile. if (_downloads.containsKey(globalKey)) { final existing = _downloads[globalKey]!; if (existing.status == DownloadStatus.downloading || existing.status == DownloadStatus.completed || existing.status == DownloadStatus.queued || existing.status == DownloadStatus.paused) { - final claimed = await _claimDownloadForActiveProfile(globalKey); + final claimed = await _claimDownloadForProfile(globalKey, ownerProfileId); if (claimed) safeNotifyListeners(); return claimed; } @@ -1044,7 +1057,7 @@ class DownloadProvider extends ChangeNotifier with DisposableChangeNotifierMixin // Store full metadata for display _metadata[globalKey] = metadataToStore; - await _claimDownloadForActiveProfile(globalKey); + await _claimDownloadForProfile(globalKey, ownerProfileId); // Update local state immediately for UI feedback _downloads[globalKey] = DownloadProgress(globalKey: globalKey, status: DownloadStatus.queued); @@ -1681,8 +1694,19 @@ class DownloadProvider extends ChangeNotifier with DisposableChangeNotifierMixin serverManager: serverManager, downloads: downloads, metadata: Map.unmodifiable(_metadata), - queueSingleDownload: (episode, client, {int mediaIndex = 0}) => - _queueSingleDownload(episode, client, mediaIndex: mediaIndex, relatedContext: relatedContext), + queueSingleDownload: (episode, client, {int mediaIndex = 0}) async { + // A profile switch mid-pass must not keep queueing the old + // profile's rules; whatever does get queued is claimed for the + // rule's owner, never the new active profile. + if (_activeProfileId != profileId) return false; + return _queueSingleDownload( + episode, + client, + mediaIndex: mediaIndex, + relatedContext: relatedContext, + claimForProfileId: profileId, + ); + }, isOffline: _offlineSource?.isOffline ?? false, force: force, ); @@ -1709,8 +1733,16 @@ class DownloadProvider extends ChangeNotifier with DisposableChangeNotifierMixin serverManager: serverManager, downloads: downloads, metadata: Map.unmodifiable(_metadata), - queueSingleDownload: (episode, client, {int mediaIndex = 0}) => - _queueSingleDownload(episode, client, mediaIndex: mediaIndex, relatedContext: relatedContext), + queueSingleDownload: (episode, client, {int mediaIndex = 0}) async { + if (_activeProfileId != profileId) return false; + return _queueSingleDownload( + episode, + client, + mediaIndex: mediaIndex, + relatedContext: relatedContext, + claimForProfileId: profileId, + ); + }, isOffline: _offlineSource?.isOffline ?? false, ); } diff --git a/lib/screens/profile/profile_teardown.dart b/lib/screens/profile/profile_teardown.dart index efc4b012..6ac6cd2f 100644 --- a/lib/screens/profile/profile_teardown.dart +++ b/lib/screens/profile/profile_teardown.dart @@ -240,6 +240,11 @@ Future logoutAllProfiles(BuildContext context) async { await scope.storage.clearActiveProfileId(); await scope.storage.clearAllProfileLastUsed(); await scope.storage.clearAllUserScopedPreferences(); + // Queued watch actions and sync rules are keyed by the profiles that just + // ceased to exist; left behind they'd strand forever (or worse, replay + // through the next sign-in's clients). + await scope.database.clearAllWatchActions(); + await scope.database.clearAllSyncRules(); // The API cache is app-global and Plex rows are keyed by server only, so // a later sign-in as a different user must not inherit them. await ApiCache.instance.clearVolatile(); diff --git a/lib/services/offline_watch_sync_service.dart b/lib/services/offline_watch_sync_service.dart index 7673c283..d326bdcc 100644 --- a/lib/services/offline_watch_sync_service.dart +++ b/lib/services/offline_watch_sync_service.dart @@ -372,9 +372,14 @@ class OfflineWatchSyncService extends ChangeNotifier { try { await _adoptLegacyWatchActionsForActiveProfile(); final profileId = _activeProfileId; - final pendingActions = profileId == null || profileId.isEmpty - ? await _database.getPendingWatchActions() - : await _database.getPendingWatchActions(profileId: profileId); + if (profileId == null || profileId.isEmpty) { + // No active profile: dropping the filter would replay EVERY + // profile's queued actions through whatever clients happen to be + // bound — the wrong user's account. Actions stay queued. + appLogger.d('No active profile — deferring pending watch sync'); + return; + } + final pendingActions = await _database.getPendingWatchActions(profileId: profileId); if (pendingActions.isEmpty) { appLogger.d('No pending watch actions to sync'); @@ -384,6 +389,14 @@ class OfflineWatchSyncService extends ChangeNotifier { appLogger.i('Syncing ${pendingActions.length} pending watch actions'); for (final action in pendingActions) { + if (_activeProfileId != profileId) { + // A profile switch mid-loop rebinds server clients to the NEW + // user's tokens under the same server ids — replaying the rest + // would write this profile's watch history to another account. + // Remaining actions stay queued for the next sync. + appLogger.i('Active profile changed mid-sync — requeueing remaining watch actions'); + return; + } if (action.syncAttempts >= maxSyncAttempts) { appLogger.w( 'Skipping action ${action.id} - exceeded retry limit ' diff --git a/lib/services/sync_rule_executor.dart b/lib/services/sync_rule_executor.dart index 47f9373b..23bdbb2a 100644 --- a/lib/services/sync_rule_executor.dart +++ b/lib/services/sync_rule_executor.dart @@ -39,7 +39,10 @@ class SyncRuleResult { class SyncRuleExecutor { final AppDatabase _database; bool _isExecuting = false; - DateTime? _lastFullRunAt; + + /// Per profile: one profile's background pass must not consume another + /// profile's cooldown window after a switch. + final Map _lastFullRunAtByProfile = {}; static const Duration _cooldownWifi = Duration(minutes: 30); static const Duration _cooldownCellular = Duration(hours: 3); @@ -85,11 +88,12 @@ class SyncRuleExecutor { return []; } - if (!force && _lastFullRunAt != null) { + final lastFullRunAt = _lastFullRunAtByProfile[profileId]; + if (!force && lastFullRunAt != null) { final hasWifi = connectivity.contains(ConnectivityResult.wifi) || connectivity.contains(ConnectivityResult.ethernet); final cooldown = hasWifi ? _cooldownWifi : _cooldownCellular; - final elapsed = DateTime.now().difference(_lastFullRunAt!); + final elapsed = DateTime.now().difference(lastFullRunAt); if (elapsed < cooldown) { appLogger.d( 'Sync rules cooldown active (${elapsed.inMinutes}m < ${cooldown.inMinutes}m, hasWifi=$hasWifi) — skipping', @@ -124,7 +128,7 @@ class SyncRuleExecutor { } } - _lastFullRunAt = DateTime.now(); + _lastFullRunAtByProfile[profileId] = DateTime.now(); return results; } finally { _isExecuting = false; diff --git a/test/services/offline_watch_sync_service_test.dart b/test/services/offline_watch_sync_service_test.dart index cdd2e337..3e4e7837 100644 --- a/test/services/offline_watch_sync_service_test.dart +++ b/test/services/offline_watch_sync_service_test.dart @@ -346,6 +346,7 @@ void main() { mgr.dispose(); await db.close(); }); + svc.setActiveProfileId('p1'); final client = _RecordingMediaClient(serverId: ServerId('srv'), backend: MediaBackend.plex); mgr.debugRegisterClientForTesting(client); @@ -371,6 +372,7 @@ void main() { mgr.dispose(); await db.close(); }); + svc.setActiveProfileId('p1'); final client = _RecordingMediaClient(serverId: ServerId('srv'), backend: MediaBackend.plex); mgr.debugRegisterClientForTesting(client); @@ -395,6 +397,7 @@ void main() { mgr.dispose(); await db.close(); }); + svc.setActiveProfileId('p1'); final client = _RecordingMediaClient(serverId: ServerId('srv'), backend: MediaBackend.plex); mgr.debugRegisterClientForTesting(client); @@ -421,6 +424,7 @@ void main() { mgr.dispose(); await db.close(); }); + svc.setActiveProfileId('p1'); final client = _RecordingMediaClient(serverId: ServerId('srv'), backend: MediaBackend.jellyfin); mgr.debugRegisterClientForTesting(client); @@ -443,6 +447,69 @@ void main() { // of getLocalWatchStatus / getLocalViewOffset). // ============================================================ + group('syncPendingItems profile scoping', () { + test('defers entirely when no profile is active', () async { + final (svc: svc, db: db, mgr: mgr) = _makeService(); + addTearDown(() async { + svc.dispose(); + mgr.dispose(); + await db.close(); + }); + + svc.setActiveProfileId('p1'); + final client = _RecordingMediaClient(serverId: ServerId('srv'), backend: MediaBackend.plex); + mgr.debugRegisterClientForTesting(client); + await svc.queueMarkWatched(serverId: ServerId('srv'), itemId: '42'); + + // Active profile cleared (sign-out/teardown window): replaying the + // queue through whatever clients are bound would hit the wrong user. + svc.setActiveProfileId(null); + await svc.syncPendingItems(); + + expect(client.watched, isEmpty); + expect(await db.getPendingSyncCount(), 1); + }); + + test('requeues remaining actions when the active profile changes mid-sync', () async { + final (svc: svc, db: db, mgr: mgr) = _makeService(); + addTearDown(() async { + svc.dispose(); + mgr.dispose(); + await db.close(); + }); + + svc.setActiveProfileId('p1'); + final posts = []; + final client = JellyfinClient.forTesting( + connection: _jellyfinConnection('user-a'), + httpClient: MockClient((request) async { + if (request.method == 'GET' && request.url.path.startsWith('/Users/user-a/Items/')) { + final id = request.url.pathSegments.last; + return http.Response('{"Id":"$id","Type":"Movie","Name":"Movie $id"}', 200); + } + if (request.method == 'POST' && request.url.path.startsWith('/UserPlayedItems/')) { + posts.add(request.url.path); + // The switch lands while action 1 is mid-flight — the binder + // would now be rebinding this server id to another user. + svc.setActiveProfileId('p2'); + return http.Response('', 204); + } + return http.Response('not found', 404); + }), + ); + addTearDown(client.close); + mgr.debugRegisterJellyfinClientForTesting(client); + + await svc.queueMarkWatched(serverId: ServerId('jf-machine'), itemId: 'item-1'); + await svc.queueMarkWatched(serverId: ServerId('jf-machine'), itemId: 'item-2'); + + await svc.syncPendingItems(); + + expect(posts, hasLength(1)); + expect(await db.getPendingSyncCount(), 1); + }); + }); + group('queueProgressUpdate', () { test('persists a progress row with shouldMarkWatched=false below threshold', () async { final (svc: svc, db: db, mgr: mgr) = _makeService(); @@ -881,6 +948,7 @@ void main() { mgr.dispose(); await db.close(); }); + svc.setActiveProfileId('p1'); final pathsByUser = >{'user-a': [], 'user-b': []};