yorickr/plezy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: trakt/download hardening and cleanup sweep

Browse Source
This commit is contained in:
edde746
2026-04-21 13:57:03 +02:00
parent 93d196572f
commit ee020d81bd
66 changed files with 1526 additions and 2167 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/services/download_manager_service.dart
+43 -9
View File
@@ -12,6 +12,7 @@ import 'saf_storage_service.dart';
import '../models/download_models.dart';
import '../models/plex_metadata.dart';
import '../models/plex_media_info.dart';
import '../services/offline_mode_source.dart';
import '../services/plex_client.dart';
import '../services/download_storage_service.dart';
import '../services/plex_api_cache.dart';
@@ -71,6 +72,8 @@ class DownloadManagerService {
PlexClient? Function(String serverId)? _clientResolver;
PlexClient? _fallbackClient;
OfflineModeSource? _offlineSource;
// background_downloader state
bool _fileDownloaderInitialized = false;
static const _downloadGroup = 'video_downloads';
@@ -145,6 +148,14 @@ class DownloadManagerService {
_clientResolver = resolver;
}
/// Inject the offline-mode source. When `isOffline`, queue/resume paths skip
/// network work and defer until connectivity returns.
void setOfflineSource(OfflineModeSource? source) {
_offlineSource = source;
}
bool get _isOffline => _offlineSource?.isOffline ?? false;
/// Look up the correct client for [serverId].
/// Returns null if the server is offline — callers should skip/defer the work.
PlexClient? _getClient(String? serverId) {
@@ -285,6 +296,11 @@ class DownloadManagerService {
void resumeQueuedDownloads(PlexClient client) {
_fallbackClient = client;
if (_isOffline) {
appLogger.d('Skipping resumeQueuedDownloads — offline');
return;
}
// Attempt deferred supplementary downloads for recovered items
_processPendingSupplementaryDownloads(client);
@@ -349,13 +365,24 @@ class DownloadManagerService {
}
}
/// Cancel any per-download timers (progress debounce + auto-retry) for [key].
/// Idempotent; safe to call from any terminal/pause path.
void _cancelDownloadTimers(String key) {
_progressDebounceTimers.remove(key)?.cancel();
_autoRetryTimers.remove(key)?.cancel();
}
/// Delete a file if it exists and log the deletion
/// Returns true if file was deleted, false otherwise
Future<bool> _deleteFileIfExists(File file, String description) async {
if (await file.exists()) {
await file.delete();
appLogger.i('Deleted $description: ${file.path}');
return true;
try {
if (await file.exists()) {
await file.delete();
appLogger.i('Deleted $description: ${file.path}');
return true;
}
} catch (e) {
appLogger.w('Failed to delete $description: ${file.path}', error: e);
}
return false;
}
@@ -715,6 +742,7 @@ class DownloadManagerService {
/// Handle a system-initiated cancel — re-queue unless already completed.
Future<void> _onDownloadCanceled(String globalKey) async {
_cancelDownloadTimers(globalKey);
final ctx = _pendingDownloadContext.remove(globalKey);
if (ctx == null) return;
if (_completingKeys.contains(globalKey)) return;
@@ -737,6 +765,7 @@ class DownloadManagerService {
appLogger.d('Ignoring failure event for $globalKey: completion in progress');
return;
}
_cancelDownloadTimers(globalKey);
_pendingDownloadContext.remove(globalKey);
final existing = await _database.getDownloadedMedia(globalKey);
@@ -767,7 +796,6 @@ class DownloadManagerService {
);
await _transitionStatus(globalKey, DownloadStatus.failed, errorMessage: errorMessage);
await _database.removeFromQueue(globalKey);
_autoRetryTimers[globalKey]?.cancel();
_autoRetryTimers[globalKey] = Timer(_autoRetryDelay, () {
_autoRetryTimers.remove(globalKey);
_performAutoRetry(globalKey);
@@ -791,6 +819,7 @@ class DownloadManagerService {
appLogger.d('Ignoring permanent failure event for $globalKey: completion in progress');
return;
}
_cancelDownloadTimers(globalKey);
_pendingDownloadContext.remove(globalKey);
final existing = await _database.getDownloadedMedia(globalKey);
@@ -840,8 +869,8 @@ class DownloadManagerService {
}
_completingKeys.add(globalKey);
try {
// Flush any pending debounced progress write before completing
_progressDebounceTimers.remove(globalKey)?.cancel();
// Flush any pending debounced progress write + cancel any scheduled retry
_cancelDownloadTimers(globalKey);
// Fresh DB check — bail if already completed (guards against race with orphan scan)
final existingCheck = await _database.getDownloadedMedia(globalKey);
@@ -1242,6 +1271,7 @@ class DownloadManagerService {
_pausingKeys.add(globalKey);
try {
_cancelDownloadTimers(globalKey);
final bgTaskId = await _database.getBgTaskId(globalKey);
if (bgTaskId != null) {
final task = await FileDownloader().taskForId(bgTaskId);
@@ -1302,7 +1332,7 @@ class DownloadManagerService {
/// Cancel a download
Future<void> cancelDownload(String globalKey) async {
_autoRetryTimers.remove(globalKey)?.cancel();
_cancelDownloadTimers(globalKey);
final bgTaskId = await _database.getBgTaskId(globalKey);
if (bgTaskId != null) {
await FileDownloader().cancelTaskWithId(bgTaskId);
@@ -1314,7 +1344,7 @@ class DownloadManagerService {
/// Delete a downloaded item and its files
Future<void> deleteDownload(String globalKey) async {
_autoRetryTimers.remove(globalKey)?.cancel();
_cancelDownloadTimers(globalKey);
// Cancel if actively downloading via background_downloader
final bgTaskId = await _database.getBgTaskId(globalKey);
if (bgTaskId != null) {
@@ -1892,6 +1922,10 @@ class DownloadManagerService {
timer.cancel();
}
_autoRetryTimers.clear();
_pendingDownloadContext.clear();
_pendingSupplementaryDownloads.clear();
_completingKeys.clear();
_pausingKeys.clear();
_progressController.close();
_deletionProgressController.close();
}
lib/services/download_storage_service.dart
+21 -3
View File
@@ -9,6 +9,19 @@ import '../utils/app_logger.dart';
import '../utils/formatters.dart';
import 'settings_service.dart';
/// Thrown when the downloads storage layer cannot create or access a directory
/// (permission denied, quota exceeded, SAF permission revoked, etc.).
class DownloadStorageException implements Exception {
final String message;
final String path;
final Object cause;
DownloadStorageException(this.message, this.path, this.cause);
@override
String toString() => 'DownloadStorageException: $message (path: $path, cause: $cause)';
}
class DownloadStorageService {
static DownloadStorageService? _instance;
static DownloadStorageService get instance => _instance ??= DownloadStorageService._();
@@ -219,12 +232,17 @@ class DownloadStorageService {
.trim();
}
/// Ensure a directory exists, creating it if necessary
/// Ensure a directory exists, creating it if necessary.
/// `Directory.create(recursive: true)` is idempotent — it no-ops if the
/// directory already exists.
Future<Directory> _ensureDirectoryExists(Directory dir) async {
if (!await dir.exists()) {
try {
await dir.create(recursive: true);
return dir;
} catch (e, st) {
appLogger.e('Failed to ensure directory exists: ${dir.path}', error: e, stackTrace: st);
throw DownloadStorageException('Cannot create directory', dir.path, e);
}
return dir;
}
/// Format a media title with optional year: "Title (YYYY)" or "Title"
lib/services/sync_rule_executor.dart
+24 -32
View File
@@ -5,9 +5,11 @@ import '../models/download_models.dart';
import '../models/plex_metadata.dart';
import '../utils/app_logger.dart';
import '../utils/content_utils.dart';
import '../utils/episode_collection.dart';
import '../utils/global_key_utils.dart';
import 'download_manager_service.dart';
import 'multi_server_manager.dart';
import 'offline_mode_source.dart';
import 'plex_client.dart';
/// Sync-rule filter values stored in `SyncRules.downloadFilter`.
@@ -40,10 +42,18 @@ class SyncRuleExecutor {
static const Duration _cooldownWifi = Duration(minutes: 30);
static const Duration _cooldownCellular = Duration(hours: 3);
OfflineModeSource? _offlineSource;
SyncRuleExecutor({required AppDatabase database}) : _database = database;
bool get isExecuting => _isExecuting;
/// Inject the offline-mode source so we can skip running rules when the
/// device has no Plex connectivity (every `getChildren` call would fail).
void setOfflineSource(OfflineModeSource? source) {
_offlineSource = source;
}
/// Execute every enabled sync rule.
///
/// The adaptive cooldown (30 min on WiFi/Ethernet, 3 h on cellular) only
@@ -67,6 +77,11 @@ class SyncRuleExecutor {
return [];
}
if (_offlineSource?.isOffline ?? false) {
appLogger.d('Skipping sync rules — offline');
return [];
}
// Read connectivity once for both the WiFi-only gate and the cooldown pick.
final List<ConnectivityResult> connectivity = await _readConnectivity();
if (await DownloadManagerService.shouldBlockDownloadOnCellularWith(connectivity)) {
@@ -134,6 +149,11 @@ class SyncRuleExecutor {
return null;
}
if (_offlineSource?.isOffline ?? false) {
appLogger.d('Skipping single sync rule $globalKey — offline');
return null;
}
if (await DownloadManagerService.shouldBlockDownloadOnCellular()) {
appLogger.d('Skipping single sync rule $globalKey — cellular download blocked');
return null;
@@ -210,9 +230,9 @@ class SyncRuleExecutor {
}) async {
final unwatchedEpisodes = <PlexMetadata>[];
if (rule.targetType == ContentTypes.show) {
await _collectEpisodesForShow(client, rule.ratingKey, unwatchedOnly: true, out: unwatchedEpisodes);
await collectEpisodesForShow(client, rule.ratingKey, unwatchedOnly: true, out: unwatchedEpisodes);
} else {
await _collectEpisodesForSeason(client, rule.ratingKey, unwatchedOnly: true, out: unwatchedEpisodes);
await collectEpisodesForSeason(client, rule.ratingKey, unwatchedOnly: true, out: unwatchedEpisodes);
}
if (unwatchedEpisodes.isEmpty) {
@@ -333,9 +353,9 @@ class SyncRuleExecutor {
if (unwatchedOnly && item.isWatched && !item.hasActiveProgress) break;
out.add(item);
case ContentTypes.show:
await _collectEpisodesForShow(client, item.ratingKey, unwatchedOnly: unwatchedOnly, out: out);
await collectEpisodesForShow(client, item.ratingKey, unwatchedOnly: unwatchedOnly, out: out);
case ContentTypes.season:
await _collectEpisodesForSeason(client, item.ratingKey, unwatchedOnly: unwatchedOnly, out: out);
await collectEpisodesForSeason(client, item.ratingKey, unwatchedOnly: unwatchedOnly, out: out);
default:
// Skip music, clips, nested collections/playlists, unknown types.
break;
@@ -343,34 +363,6 @@ class SyncRuleExecutor {
}
}
Future<void> _collectEpisodesForShow(
PlexClient client,
String showRatingKey, {
required bool unwatchedOnly,
required List<PlexMetadata> out,
}) async {
final seasons = await client.getChildren(showRatingKey);
for (final season in seasons) {
if (season.type == ContentTypes.season) {
await _collectEpisodesForSeason(client, season.ratingKey, unwatchedOnly: unwatchedOnly, out: out);
}
}
}
Future<void> _collectEpisodesForSeason(
PlexClient client,
String seasonRatingKey, {
required bool unwatchedOnly,
required List<PlexMetadata> out,
}) async {
final episodes = await client.getChildren(seasonRatingKey);
for (final ep in episodes) {
if (ep.type != ContentTypes.episode) continue;
if (unwatchedOnly && ep.isWatched && !ep.hasActiveProgress) continue;
out.add(ep);
}
}
static bool _isActiveDownload(DownloadProgress? p) =>
p != null &&
(p.status == DownloadStatus.completed ||
lib/services/trakt/trakt_sync_queue.dart
+50 -7
View File
@@ -1,8 +1,10 @@
import 'dart:async';
import 'dart:convert';
import 'package:shared_preferences/shared_preferences.dart';
import '../../models/trakt/trakt_ids.dart';
import '../../utils/app_logger.dart';
import 'trakt_constants.dart';
/// One pending watched/unwatched push waiting to be drained to Trakt.
@@ -73,23 +75,43 @@ class TraktSyncQueueItem {
///
/// Cap at [maxAttempts] before dropping permanently — matches
/// `OfflineWatchSyncService.maxSyncAttempts`.
///
/// Serialises all writes (`add`, `save`, `drainWith`) through a Completer chain
/// so concurrent `add()` calls don't interleave read-modify-write and lose items.
class TraktSyncQueue {
static const String _baseKey = 'trakt_sync_queue';
static const int maxAttempts = 5;
Future<void> _writeLock = Future<void>.value();
Future<T> _locked<T>(Future<T> Function() action) {
final previous = _writeLock;
final completer = Completer<void>();
_writeLock = completer.future;
return previous.then((_) => action()).whenComplete(completer.complete);
}
Future<List<TraktSyncQueueItem>> load(String userUuid) async {
final prefs = await SharedPreferences.getInstance();
final raw = prefs.getString(traktUserKey(userUuid, _baseKey));
final key = traktUserKey(userUuid, _baseKey);
final raw = prefs.getString(key);
if (raw == null) return [];
try {
final list = json.decode(raw) as List<dynamic>;
return list.map((e) => TraktSyncQueueItem.fromJson(e as Map<String, dynamic>)).toList();
} catch (_) {
} catch (e, st) {
appLogger.e('Trakt sync queue parse failed, discarding', error: e, stackTrace: st);
await prefs.setString(traktUserKey(userUuid, '${_baseKey}_corrupt'), raw);
await prefs.remove(key);
return [];
}
}
Future<void> save(String userUuid, List<TraktSyncQueueItem> items) async {
Future<void> save(String userUuid, List<TraktSyncQueueItem> items) {
return _locked(() => _saveRaw(userUuid, items));
}
Future<void> _saveRaw(String userUuid, List<TraktSyncQueueItem> items) async {
final prefs = await SharedPreferences.getInstance();
final key = traktUserKey(userUuid, _baseKey);
if (items.isEmpty) {
@@ -99,9 +121,30 @@ class TraktSyncQueue {
}
}
Future<void> add(String userUuid, TraktSyncQueueItem item) async {
final items = await load(userUuid);
items.add(item);
await save(userUuid, items);
Future<void> add(String userUuid, TraktSyncQueueItem item) {
return _locked(() async {
final items = await load(userUuid);
items.add(item);
await _saveRaw(userUuid, items);
});
}
/// Atomic drain: load the queue, run [processor] for each item, and save the
/// items the processor decided to retain. Holds the write lock for the whole
/// cycle so concurrent `add()`s wait until the drain completes (no lost items).
///
/// [processor] returns `null` to drop the item, or a (possibly mutated) item
/// to retain for the next drain (e.g. `item.incrementAttempts()`).
Future<void> drainWith(String userUuid, Future<TraktSyncQueueItem?> Function(TraktSyncQueueItem) processor) {
return _locked(() async {
final items = await load(userUuid);
if (items.isEmpty) return;
final remaining = <TraktSyncQueueItem>[];
for (final item in items) {
final keep = await processor(item);
if (keep != null) remaining.add(keep);
}
await _saveRaw(userUuid, remaining);
});
}
}
lib/services/trakt/trakt_sync_service.dart
+53 -12
View File
@@ -1,4 +1,5 @@
import 'dart:async';
import 'dart:collection';
import '../../models/trakt/trakt_ids.dart';
import '../../models/trakt/trakt_scrobble_request.dart';
@@ -40,6 +41,12 @@ class TraktSyncService {
/// key GUID cache survives a binge-watch session.
final Map<String, TraktGuidResolver> _resolvers = {};
/// Fallback buffer for items that failed to persist to the on-disk queue
/// (e.g. SharedPreferences write threw). Retried on next `flushQueue`.
/// Bounded to keep memory pressure finite; oldest items drop first.
static const int _maxInMemoryFallback = 100;
final Queue<TraktSyncQueueItem> _inMemoryFallback = Queue<TraktSyncQueueItem>();
bool _isFlushing = false;
Future<void> initialize({required MultiServerManager serverManager}) async {
@@ -50,7 +57,11 @@ class TraktSyncService {
final settings = await SettingsService.getInstance();
_isEnabled = settings.getEnableTraktWatchedSync();
_subscription = WatchStateNotifier().stream.listen(_onWatchStateEvent);
_subscription = WatchStateNotifier().stream.listen(
_onWatchStateEvent,
onError: (Object e, StackTrace st) =>
appLogger.w('Trakt sync: watch event handler error', error: e, stackTrace: st),
);
}
Future<void> setEnabled(bool enabled) async {
@@ -166,7 +177,7 @@ class TraktSyncService {
Future<void> _trySendOrQueue(TraktSyncQueueItem item, TraktScrobbleRequest body) async {
final client = _client;
if (client == null) {
await _queue.add(_activeUserUuid, item);
await _persistOrBuffer(item);
return;
}
try {
@@ -174,7 +185,27 @@ class TraktSyncService {
appLogger.d('Trakt sync: ${item.op.name} ${item.ratingKey} → ok');
} catch (e) {
appLogger.d('Trakt sync: ${item.op.name} ${item.ratingKey} failed, queuing', error: e);
await _persistOrBuffer(item);
}
}
/// Persist an item to the on-disk queue; fall back to a bounded in-memory
/// buffer if the disk write throws (e.g. disk full, SAF permission revoked).
/// Retried at the start of the next `flushQueue` run.
Future<void> _persistOrBuffer(TraktSyncQueueItem item) async {
try {
await _queue.add(_activeUserUuid, item);
} catch (e, st) {
appLogger.e(
'Trakt sync: queue persist failed for ${item.op.name} ${item.ratingKey}, buffering in memory',
error: e,
stackTrace: st,
);
if (_inMemoryFallback.length >= _maxInMemoryFallback) {
final dropped = _inMemoryFallback.removeFirst();
appLogger.w('Trakt sync: in-memory fallback full, dropping ${dropped.op.name} ${dropped.ratingKey}');
}
_inMemoryFallback.addLast(item);
}
}
@@ -193,31 +224,41 @@ class TraktSyncService {
if (client == null) return;
_isFlushing = true;
try {
final items = await _queue.load(_activeUserUuid);
if (items.isEmpty) return;
await _recoverInMemoryFallback();
final remaining = <TraktSyncQueueItem>[];
for (final item in items) {
await _queue.drainWith(_activeUserUuid, (item) async {
if (item.attempts >= TraktSyncQueue.maxAttempts) {
appLogger.w('Trakt sync: dropping ${item.op.name} ${item.ratingKey} after ${item.attempts} attempts');
continue;
return null;
}
try {
await _dispatch(client, item, _bodyFor(item));
appLogger.d('Trakt sync: drained ${item.op.name} ${item.ratingKey}');
await Future<void>.delayed(_queueRequestSpacing);
return null;
} catch (e) {
appLogger.d('Trakt sync: drain failed for ${item.ratingKey}, will retry', error: e);
remaining.add(item.incrementAttempts());
await Future<void>.delayed(_queueRequestSpacing);
return item.incrementAttempts();
}
await Future<void>.delayed(_queueRequestSpacing);
}
await _queue.save(_activeUserUuid, remaining);
});
} finally {
_isFlushing = false;
}
}
/// Try to move items buffered in memory (because prior disk writes failed)
/// back onto the persistent queue. Best-effort; items that still can't be
/// persisted stay in the buffer for the next flush.
Future<void> _recoverInMemoryFallback() async {
if (_inMemoryFallback.isEmpty) return;
final snapshot = List<TraktSyncQueueItem>.from(_inMemoryFallback);
_inMemoryFallback.clear();
for (final item in snapshot) {
await _persistOrBuffer(item);
}
}
TraktScrobbleRequest _bodyFor(TraktSyncQueueItem item) {
return switch (item.kind) {
TraktMediaKind.movie => TraktScrobbleRequest.movie(ids: item.ids),