824 lines
32 KiB
Dart
824 lines
32 KiB
Dart
import 'dart:async';
|
|
import '../media/ids.dart';
|
|
import 'dart:io' show Platform;
|
|
|
|
import 'package:flutter/foundation.dart';
|
|
|
|
import '../database/app_database.dart';
|
|
import '../database/download_operations.dart';
|
|
import '../media/media_backend.dart';
|
|
import '../media/media_item.dart';
|
|
import '../media/media_kind.dart';
|
|
import '../media/media_server_client.dart';
|
|
import '../media/playback_report_metadata.dart';
|
|
import '../media/watch_progress.dart';
|
|
import '../utils/app_logger.dart';
|
|
import '../utils/active_client_scope.dart';
|
|
import '../utils/global_key_utils.dart';
|
|
import 'offline_mode_source.dart';
|
|
import '../utils/watch_state_notifier.dart';
|
|
import 'multi_server_manager.dart';
|
|
import 'plex_client.dart';
|
|
import 'settings_service.dart';
|
|
import 'trackers/tracker_coordinator.dart';
|
|
import 'watch_state_resolver.dart';
|
|
|
|
/// Service for managing offline watch progress and syncing it back to the
|
|
/// owning server. Backend-neutral over [MediaServerClient] — Plex actions
|
|
/// hit `/:/scrobble` and `/:/timeline`, Jellyfin actions hit
|
|
/// `/UserPlayedItems/{id}` and `/Sessions/Playing*` through the same queue.
|
|
///
|
|
/// Handles:
|
|
/// - Queuing progress updates when offline
|
|
/// - Queuing manual watch/unwatch actions
|
|
/// - Auto-marking items as watched at the server's threshold
|
|
/// - Syncing queued actions when connectivity is restored
|
|
class OfflineWatchSyncService extends ChangeNotifier {
|
|
final AppDatabase _database;
|
|
final MultiServerManager _serverManager;
|
|
|
|
OfflineModeSource? _offlineModeSource;
|
|
VoidCallback? _offlineModeListener;
|
|
bool _isSyncing = false;
|
|
bool _isBidirectionalSyncing = false;
|
|
bool _isShutDown = false;
|
|
DateTime? _lastSyncTime;
|
|
bool _hasPerformedStartupSync = false;
|
|
String? _activeProfileId;
|
|
int? _availableProfileCount;
|
|
final Set<String> _legacyWatchActionsAdoptedForProfiles = <String>{};
|
|
|
|
/// Callback to refresh download provider metadata after sync
|
|
VoidCallback? onWatchStatesRefreshed;
|
|
|
|
/// Get watched threshold for a server. Cascades:
|
|
/// 1. Plex's fetched server prefs (`/:/prefs`)
|
|
/// 2. Jellyfin's fixed [MediaServerClient.watchedThreshold] (0.9)
|
|
/// 3. Cached value in SettingsService (mirrored by PlexClient.fetchServerPrefs)
|
|
/// 4. Default 90%
|
|
double getWatchedThreshold(ServerId serverId) {
|
|
final client = _serverManager.getClient(serverId);
|
|
if (client is PlexClient && client.serverPrefs.isNotEmpty) {
|
|
return client.watchedThreshold;
|
|
}
|
|
if (client != null && client.backend != MediaBackend.plex) {
|
|
// Jellyfin (and any future neutral backend) — the client exposes a
|
|
// fixed threshold that mirrors the wire-protocol behaviour.
|
|
return client.watchedThreshold;
|
|
}
|
|
// No client bound (offline) or Plex prefs not loaded yet — use the
|
|
// mirror written on the last successful prefs fetch.
|
|
final cached = SettingsService.instanceOrNull?.read(SettingsService.watchedThresholdPref(serverId)) ?? 90;
|
|
return cached / 100.0;
|
|
}
|
|
|
|
/// Minimum interval between syncs.
|
|
/// Mobile: no throttle (always sync on resume for cross-device updates)
|
|
/// Desktop: 2 minutes (reduced from 10 min to handle tab-switching better)
|
|
static Duration get minSyncInterval {
|
|
if (Platform.isIOS || Platform.isAndroid) {
|
|
return Duration.zero;
|
|
}
|
|
return const Duration(minutes: 2);
|
|
}
|
|
|
|
/// Maximum sync attempts before suppressing additional server-write retries.
|
|
/// Queued actions are retained so a transient outage or server bug never
|
|
/// silently drops local watch progress.
|
|
static const int maxSyncAttempts = 5;
|
|
|
|
OfflineWatchSyncService({required this._database, required this._serverManager});
|
|
|
|
/// Whether a sync is currently in progress
|
|
bool get isSyncing => _isSyncing;
|
|
|
|
void setActiveProfileId(String? profileId, {int? availableProfileCount}) {
|
|
_activeProfileId = profileId;
|
|
_availableProfileCount = availableProfileCount;
|
|
if (profileId != null && profileId.isNotEmpty) {
|
|
unawaited(_adoptLegacyWatchActionsForProfile(profileId, availableProfileCount: availableProfileCount));
|
|
}
|
|
}
|
|
|
|
Future<void> _adoptLegacyWatchActionsForProfile(String profileId, {int? availableProfileCount}) async {
|
|
if (profileId.isEmpty || !_legacyWatchActionsAdoptedForProfiles.add(profileId)) return;
|
|
if (availableProfileCount != 1) {
|
|
_legacyWatchActionsAdoptedForProfiles.remove(profileId);
|
|
return;
|
|
}
|
|
try {
|
|
await _database.adoptLegacyOfflineWatchActionsForProfile(profileId);
|
|
} catch (e, st) {
|
|
_legacyWatchActionsAdoptedForProfiles.remove(profileId);
|
|
appLogger.w('Failed to adopt legacy offline watch actions for $profileId', error: e, stackTrace: st);
|
|
}
|
|
}
|
|
|
|
Future<void> _adoptLegacyWatchActionsForActiveProfile() async {
|
|
final profileId = _activeProfileId;
|
|
if (profileId == null || profileId.isEmpty) return;
|
|
await _adoptLegacyWatchActionsForProfile(profileId, availableProfileCount: _availableProfileCount);
|
|
}
|
|
|
|
void startConnectivityMonitoring(OfflineModeSource source) {
|
|
if (_offlineModeSource != null && _offlineModeListener != null) {
|
|
_offlineModeSource!.removeListener(_offlineModeListener!);
|
|
}
|
|
|
|
_offlineModeSource = source;
|
|
_offlineModeListener = () {
|
|
if (!source.isOffline) {
|
|
// We just came online - trigger bidirectional sync
|
|
appLogger.i('Connectivity restored - starting bidirectional watch sync');
|
|
_performBidirectionalSync();
|
|
}
|
|
};
|
|
|
|
source.addListener(_offlineModeListener!);
|
|
|
|
// Don't sync on startup - servers aren't connected yet.
|
|
// Sync will happen when:
|
|
// - Connectivity is restored (listener triggers)
|
|
// - App resumes from background (onAppResumed)
|
|
}
|
|
|
|
/// Perform bidirectional sync: push local changes, then pull server states.
|
|
///
|
|
/// Push always happens immediately. Pull respects [minSyncInterval] unless [force] is true.
|
|
Future<void> _performBidirectionalSync({bool force = false}) async {
|
|
if (_isShutDown) return;
|
|
|
|
// Prevent overlapping bidirectional syncs
|
|
if (_isBidirectionalSyncing) {
|
|
appLogger.d('Bidirectional sync already in progress, skipping');
|
|
return;
|
|
}
|
|
|
|
if (_serverManager.onlineClients.isEmpty) {
|
|
appLogger.d('Skipping watch sync - no connected servers available yet');
|
|
return;
|
|
}
|
|
|
|
_isBidirectionalSyncing = true;
|
|
try {
|
|
// Always push local changes to server (never throttle outbound sync)
|
|
await syncPendingItems();
|
|
|
|
// Only throttle the pull from server
|
|
if (!force && _lastSyncTime != null) {
|
|
final elapsed = DateTime.now().difference(_lastSyncTime!);
|
|
if (elapsed < minSyncInterval) {
|
|
appLogger.d(
|
|
'Skipping server pull - last sync was ${elapsed.inMinutes}m ago (min: ${minSyncInterval.inMinutes}m)',
|
|
);
|
|
return;
|
|
}
|
|
}
|
|
|
|
// Pull latest states from server
|
|
await syncWatchStatesFromServer();
|
|
_lastSyncTime = DateTime.now();
|
|
} finally {
|
|
_isBidirectionalSyncing = false;
|
|
}
|
|
}
|
|
|
|
/// Called when app becomes active - syncs for cross-device updates.
|
|
/// On mobile, always syncs immediately (device-switching scenario).
|
|
/// On desktop, respects the throttle interval.
|
|
void onAppResumed() {
|
|
if (_isShutDown) return;
|
|
if (_offlineModeSource?.isOffline != true) {
|
|
final isMobile = Platform.isIOS || Platform.isAndroid;
|
|
appLogger.d('App resumed - ${isMobile ? "forcing" : "checking"} sync');
|
|
_performBidirectionalSync(force: isMobile);
|
|
}
|
|
}
|
|
|
|
/// Called when servers connect on app startup.
|
|
///
|
|
/// Triggers the initial sync now that PlexClients are available.
|
|
/// Only runs once per app session.
|
|
void onServersConnected() {
|
|
if (_isShutDown) return;
|
|
if (_hasPerformedStartupSync) return;
|
|
_hasPerformedStartupSync = true;
|
|
|
|
if (_offlineModeSource?.isOffline != true) {
|
|
appLogger.i('Servers connected - performing startup sync');
|
|
_performBidirectionalSync();
|
|
}
|
|
}
|
|
|
|
/// Queue a progress update for later sync.
|
|
///
|
|
/// This is called during offline playback to track the watch position.
|
|
/// If progress exceeds the server's threshold, shouldMarkWatched is set to true.
|
|
///
|
|
/// 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<String?> queueProgressUpdate({
|
|
required ServerId serverId,
|
|
required String itemId,
|
|
required int viewOffset,
|
|
required int? duration,
|
|
}) async {
|
|
final shouldMarkWatched =
|
|
duration != null && isWatchedByProgress(viewOffset, duration, serverId: ServerId(serverId));
|
|
final clientScopeId = await _clientScopeIdForItem(ServerId(serverId), itemId);
|
|
|
|
await _database.upsertProgressAction(
|
|
profileId: _activeProfileId,
|
|
serverId: serverId,
|
|
clientScopeId: clientScopeId,
|
|
ratingKey: itemId,
|
|
viewOffset: viewOffset,
|
|
duration: duration,
|
|
shouldMarkWatched: shouldMarkWatched,
|
|
);
|
|
|
|
final durationLabel = duration == null ? 'unknown' : '${(duration / 1000).toStringAsFixed(0)}s';
|
|
final percentLabel = duration == null || duration <= 0
|
|
? 'unknown'
|
|
: '${((viewOffset / duration) * 100).toStringAsFixed(1)}%';
|
|
appLogger.d(
|
|
'Queued offline progress: $serverId:$itemId at ${(viewOffset / 1000).toStringAsFixed(0)}s / $durationLabel ($percentLabel)',
|
|
);
|
|
|
|
notifyListeners();
|
|
return clientScopeId;
|
|
}
|
|
|
|
Future<String?> queueMarkWatched({required ServerId serverId, required String itemId}) =>
|
|
_queueWatchStatusAction(serverId: serverId, itemId: itemId, actionType: OfflineActionType.watched.id);
|
|
|
|
Future<String?> queueMarkUnwatched({required ServerId serverId, required String itemId}) =>
|
|
_queueWatchStatusAction(serverId: serverId, itemId: itemId, actionType: OfflineActionType.unwatched.id);
|
|
|
|
Future<String?> _queueWatchStatusAction({
|
|
required ServerId serverId,
|
|
required String itemId,
|
|
required String actionType,
|
|
}) async {
|
|
final clientScopeId = await _clientScopeIdForItem(ServerId(serverId), itemId);
|
|
await _database.insertWatchAction(
|
|
profileId: _activeProfileId,
|
|
serverId: serverId,
|
|
clientScopeId: clientScopeId,
|
|
ratingKey: itemId,
|
|
actionType: actionType,
|
|
);
|
|
|
|
appLogger.d('Queued offline mark $actionType: $serverId:$itemId');
|
|
notifyListeners();
|
|
return clientScopeId;
|
|
}
|
|
|
|
/// Check if an item should be considered watched based on progress percentage.
|
|
bool isWatchedByProgress(int viewOffset, int duration, {ServerId? serverId}) {
|
|
final threshold = serverId != null ? getWatchedThreshold(serverId) : 0.9;
|
|
return isWatchedProgress(positionMs: viewOffset, durationMs: duration, threshold: threshold);
|
|
}
|
|
|
|
/// Get the local watch status for a media item.
|
|
///
|
|
/// Returns:
|
|
/// - `true` if item was marked as watched locally or progress >= server threshold
|
|
/// - `false` if item was marked as unwatched locally
|
|
/// - `null` if no local watched/unwatched action exists (use cached server data)
|
|
Future<bool?> getLocalWatchStatus(String globalKey, {String? clientScopeId}) async {
|
|
await _adoptLegacyWatchActionsForActiveProfile();
|
|
final expectedScope = clientScopeId ?? _activeClientScopeIdForGlobalKey(globalKey);
|
|
final profileId = _activeProfileId;
|
|
final actions = await _database.getWatchActionsForKey(
|
|
globalKey,
|
|
profileId: profileId,
|
|
filterProfile: profileId != null,
|
|
clientScopeId: expectedScope,
|
|
filterClientScope: expectedScope != null,
|
|
);
|
|
return WatchStateResolver.fromActions(actions).isWatched;
|
|
}
|
|
|
|
/// Get local watch statuses for multiple items in a single database query.
|
|
///
|
|
/// Returns a map of globalKey -> watch status (true/false/null).
|
|
/// More efficient than calling getLocalWatchStatus multiple times.
|
|
Future<Map<String, bool?>> getLocalWatchStatusesBatched(
|
|
Set<String> globalKeys, {
|
|
Map<String, String?>? clientScopeIdsByGlobalKey,
|
|
}) async {
|
|
if (globalKeys.isEmpty) return {};
|
|
await _adoptLegacyWatchActionsForActiveProfile();
|
|
|
|
final scopes = clientScopeIdsByGlobalKey ?? _activeClientScopeIdsForGlobalKeys(globalKeys);
|
|
final profileId = _activeProfileId;
|
|
final actions = await _database.getWatchActionsForKeys(
|
|
globalKeys,
|
|
profileId: profileId,
|
|
filterProfile: profileId != null,
|
|
clientScopeIdsByGlobalKey: scopes,
|
|
);
|
|
final result = <String, bool?>{};
|
|
|
|
for (final key in globalKeys) {
|
|
result[key] = WatchStateResolver.fromActions(actions[key] ?? const []).isWatched;
|
|
}
|
|
|
|
return result;
|
|
}
|
|
|
|
/// Get the local view offset (resume position) for a media item.
|
|
///
|
|
/// Returns the locally tracked position, or null if none exists. Explicit
|
|
/// watched/unwatched actions clear resume by resolving to a zero offset.
|
|
Future<int?> getLocalViewOffset(String globalKey, {String? clientScopeId}) async {
|
|
await _adoptLegacyWatchActionsForActiveProfile();
|
|
final expectedScope = clientScopeId ?? _activeClientScopeIdForGlobalKey(globalKey);
|
|
final profileId = _activeProfileId;
|
|
final actions = await _database.getWatchActionsForKey(
|
|
globalKey,
|
|
profileId: profileId,
|
|
filterProfile: profileId != null,
|
|
clientScopeId: expectedScope,
|
|
filterClientScope: expectedScope != null,
|
|
);
|
|
final snapshot = WatchStateResolver.fromActions(actions);
|
|
final offset = snapshot.hasViewOffsetMs ? snapshot.viewOffsetMs : null;
|
|
return offset != null && offset > 0 ? offset : null;
|
|
}
|
|
|
|
Future<int> getPendingSyncCount() async {
|
|
await _adoptLegacyWatchActionsForActiveProfile();
|
|
final profileId = _activeProfileId;
|
|
return profileId == null || profileId.isEmpty
|
|
? _database.getPendingSyncCount(maxSyncAttempts: maxSyncAttempts)
|
|
: _database.getPendingSyncCount(profileId: profileId, maxSyncAttempts: maxSyncAttempts);
|
|
}
|
|
|
|
/// Sync all pending items to their respective servers.
|
|
///
|
|
/// Called automatically when connectivity is restored, or manually.
|
|
/// Actions are batched by server to reduce connectivity lookups.
|
|
Future<void> syncPendingItems() async {
|
|
if (_isSyncing) {
|
|
appLogger.d('Sync already in progress, skipping');
|
|
return;
|
|
}
|
|
|
|
_isSyncing = true;
|
|
notifyListeners();
|
|
|
|
try {
|
|
await _adoptLegacyWatchActionsForActiveProfile();
|
|
final profileId = _activeProfileId;
|
|
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');
|
|
return;
|
|
}
|
|
|
|
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 '
|
|
'(${action.syncAttempts} attempts). Last error: ${action.lastError}',
|
|
);
|
|
continue;
|
|
}
|
|
|
|
final synced = await _withOnlineClientForAction(action, (client) async {
|
|
try {
|
|
await _syncAction(client, action);
|
|
await _database.deleteWatchAction(action.id);
|
|
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());
|
|
}
|
|
});
|
|
if (!synced) {
|
|
appLogger.d('Keeping action ${action.id} queued until server is available');
|
|
continue;
|
|
}
|
|
}
|
|
} finally {
|
|
_isSyncing = false;
|
|
notifyListeners();
|
|
}
|
|
}
|
|
|
|
Future<String?> _clientScopeIdForItem(ServerId serverId, String itemId) async {
|
|
// A downloaded row's clientScopeId is a cache/source hint, not an owner.
|
|
// Offline watch actions are user-owned, so a new local action follows the
|
|
// currently active scoped Jellyfin client. Once queued, _clientForAction
|
|
// replays that exact scope even if the active user changes later.
|
|
final client = _serverManager.getClient(serverId);
|
|
final activeScopeId = resolveActiveClientScopeId(serverId: serverId, cacheServerId: client?.cacheServerId);
|
|
if (activeScopeId != null) return activeScopeId;
|
|
final download = await _database.getDownloadedMedia(buildGlobalKey(ServerId(serverId), itemId));
|
|
return resolveActiveClientScopeId(serverId: serverId, cacheServerId: download?.clientScopeId);
|
|
}
|
|
|
|
Future<({MediaServerClient client, String? clientScopeId})?> _clientForAction(OfflineWatchProgressItem action) async {
|
|
final scopeId = resolveActiveClientScopeId(
|
|
serverId: ServerId(action.serverId),
|
|
cacheServerId: action.clientScopeId,
|
|
);
|
|
if (scopeId != null) {
|
|
final scoped = _serverManager.getJellyfinClientByCompoundId(scopeId);
|
|
if (scoped != null) return (client: scoped, clientScopeId: scopeId);
|
|
}
|
|
final client = _serverManager.getClient(ServerId(action.serverId));
|
|
if (client == null) return null;
|
|
if (client.backend == MediaBackend.jellyfin && client.cacheServerId != action.serverId) {
|
|
appLogger.w(
|
|
'Refusing to sync unscoped Jellyfin action ${action.id} for ${action.serverId}:${action.ratingKey}; '
|
|
'no queued client scope is available',
|
|
);
|
|
return null;
|
|
}
|
|
return (client: client, clientScopeId: action.clientScopeId);
|
|
}
|
|
|
|
Future<bool> _withOnlineClientForAction(
|
|
OfflineWatchProgressItem action,
|
|
Future<void> Function(MediaServerClient client) callback,
|
|
) async {
|
|
final resolved = await _clientForAction(action);
|
|
if (resolved == null) {
|
|
appLogger.d('No client for server ${action.serverId} scope ${action.clientScopeId}, skipping');
|
|
return false;
|
|
}
|
|
|
|
if (!_serverManager.isClientOnline(ServerId(action.serverId), clientScopeId: resolved.clientScopeId)) {
|
|
appLogger.d('Server ${action.serverId} scope ${resolved.clientScopeId} is offline, skipping');
|
|
return false;
|
|
}
|
|
|
|
await callback(resolved.client);
|
|
return true;
|
|
}
|
|
|
|
Map<String, String?> _activeClientScopeIdsForGlobalKeys(Set<String> globalKeys) {
|
|
final scopes = <String, String?>{};
|
|
for (final globalKey in globalKeys) {
|
|
final scopeId = _activeClientScopeIdForGlobalKey(globalKey);
|
|
if (scopeId != null) scopes[globalKey] = scopeId;
|
|
}
|
|
return scopes;
|
|
}
|
|
|
|
String? _activeClientScopeIdForGlobalKey(String globalKey) {
|
|
final parsed = parseGlobalKey(globalKey);
|
|
if (parsed == null) return null;
|
|
return _activeClientScopeIdForServer(parsed.serverId);
|
|
}
|
|
|
|
String? _activeClientScopeIdForServer(ServerId serverId) {
|
|
final client = _serverManager.getClient(serverId);
|
|
return resolveActiveClientScopeId(serverId: serverId, cacheServerId: client?.cacheServerId);
|
|
}
|
|
|
|
Future<MediaServerClient?> _clientForDownloadScope(ServerId serverId, String? clientScopeId) async {
|
|
if (clientScopeId != null && clientScopeId.isNotEmpty) {
|
|
final scoped = _serverManager.getJellyfinClientByCompoundId(clientScopeId);
|
|
if (scoped != null) return scoped;
|
|
}
|
|
return _serverManager.getClient(serverId);
|
|
}
|
|
|
|
Future<T?> _withOnlineClientForDownloadScope<T>(
|
|
ServerId serverId,
|
|
String? clientScopeId,
|
|
Future<T> Function(MediaServerClient client) callback,
|
|
) async {
|
|
final client = await _clientForDownloadScope(ServerId(serverId), clientScopeId);
|
|
if (client == null) {
|
|
appLogger.d('No client for server $serverId scope $clientScopeId, skipping');
|
|
return null;
|
|
}
|
|
|
|
if (!_serverManager.isClientOnline(ServerId(serverId), clientScopeId: clientScopeId)) {
|
|
appLogger.d('Server $serverId scope $clientScopeId is offline, skipping');
|
|
return null;
|
|
}
|
|
|
|
return callback(client);
|
|
}
|
|
|
|
/// Sync a single action to the server.
|
|
///
|
|
/// 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<void> _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.
|
|
final needsRichItem =
|
|
action.actionType == OfflineActionType.watched.id ||
|
|
action.actionType == OfflineActionType.unwatched.id ||
|
|
(action.actionType == OfflineActionType.progress.id && action.shouldMarkWatched);
|
|
MediaItem? item;
|
|
if (needsRichItem) {
|
|
try {
|
|
item = await client.fetchItem(action.ratingKey);
|
|
} catch (_) {
|
|
// Fall through to the synthetic item below.
|
|
}
|
|
}
|
|
item ??= MediaItem(
|
|
id: action.ratingKey,
|
|
backend: client.backend,
|
|
kind: MediaKind.unknown,
|
|
serverId: action.serverId,
|
|
);
|
|
|
|
switch (action.actionType) {
|
|
case 'watched':
|
|
await client.markWatched(item);
|
|
await TrackerCoordinator.instance.markWatched(item, client);
|
|
break;
|
|
|
|
case 'unwatched':
|
|
await client.markUnwatched(item);
|
|
await TrackerCoordinator.instance.markUnwatched(item, client);
|
|
break;
|
|
|
|
case 'progress':
|
|
// Push resumable progress, or a completed offline playback. Jellyfin's
|
|
// `/Sessions/Playing/Stopped` ignores events without an open session
|
|
// row, so non-Plex backends still get a lightweight Started call.
|
|
if (action.viewOffset != null) {
|
|
final duration = action.duration == null ? null : Duration(milliseconds: action.duration!);
|
|
final position = action.shouldMarkWatched && duration != null
|
|
? duration
|
|
: Duration(milliseconds: action.viewOffset!);
|
|
if (!action.shouldMarkWatched || client.backend != MediaBackend.plex) {
|
|
try {
|
|
await client.reportPlaybackStarted(itemId: action.ratingKey, position: position, duration: duration);
|
|
} 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.
|
|
appLogger.d('Offline progress: started call failed (continuing)', error: e);
|
|
}
|
|
}
|
|
await client.reportPlaybackStopped(
|
|
itemId: action.ratingKey,
|
|
position: position,
|
|
duration: duration,
|
|
report: PlaybackReportMetadata.offlineReplay(
|
|
recordedAt: DateTime.fromMillisecondsSinceEpoch(action.updatedAt),
|
|
),
|
|
);
|
|
}
|
|
|
|
// If progress exceeded threshold, also mark as watched. On backends
|
|
// that mark played from the stopped report above (Jellyfin) 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);
|
|
await TrackerCoordinator.instance.markWatched(item, client);
|
|
}
|
|
break;
|
|
}
|
|
}
|
|
|
|
/// Sync watch states for all episodes in a single season.
|
|
///
|
|
/// Returns the number of episodes synced, or -1 on failure.
|
|
Future<int> _syncSeasonEpisodes(
|
|
MediaServerClient client,
|
|
ServerId serverId,
|
|
String seasonRatingKey,
|
|
Set<String> downloadedEpisodeKeys,
|
|
) async {
|
|
try {
|
|
final seasonEpisodes = await client.fetchChildren(seasonRatingKey);
|
|
int synced = 0;
|
|
|
|
for (final episode in seasonEpisodes) {
|
|
if (!downloadedEpisodeKeys.contains(episode.id)) continue;
|
|
|
|
final cacheServerId = client.cacheServerId;
|
|
final prior = await client.cache.getMetadata(ServerId(cacheServerId), episode.id);
|
|
final isWatched = (episode.viewCount ?? 0) > 0;
|
|
|
|
if (prior != null) {
|
|
// Existing cached row — patch in the watch-state and rollup
|
|
// fields without disturbing Media/Chapter blobs.
|
|
final wasWatched = (prior.viewCount ?? 0) > 0;
|
|
await client.cache.applyWatchState(
|
|
serverId: ServerId(cacheServerId),
|
|
itemId: episode.id,
|
|
isWatched: isWatched,
|
|
viewOffsetMs: episode.viewOffsetMs,
|
|
lastViewedAt: episode.lastViewedAt,
|
|
viewedLeafCount: episode.viewedLeafCount,
|
|
);
|
|
if (wasWatched != isWatched) {
|
|
WatchStateNotifier().notifyWatched(
|
|
item: episode,
|
|
isNowWatched: isWatched,
|
|
cacheServerId: client.cacheServerId,
|
|
);
|
|
}
|
|
} else {
|
|
// No cached row yet — populate the canonical row via fetchItem
|
|
// (cache-fallback mixin writes the response to cache for free)
|
|
// and then stamp the watch snapshot. Transition is unknowable
|
|
// without a prior row, so we skip the notify to match the
|
|
// previous synthesize-fresh-row behaviour.
|
|
try {
|
|
await client.fetchItem(episode.id);
|
|
await client.cache.applyWatchState(
|
|
serverId: ServerId(cacheServerId),
|
|
itemId: episode.id,
|
|
isWatched: isWatched,
|
|
viewOffsetMs: episode.viewOffsetMs,
|
|
lastViewedAt: episode.lastViewedAt,
|
|
viewedLeafCount: episode.viewedLeafCount,
|
|
);
|
|
} catch (e) {
|
|
appLogger.d('Cache populate failed for ${episode.id}', error: e);
|
|
}
|
|
}
|
|
synced++;
|
|
}
|
|
|
|
return synced;
|
|
} catch (e) {
|
|
appLogger.d('Failed to sync watch states for season $seasonRatingKey: $e');
|
|
return -1;
|
|
}
|
|
}
|
|
|
|
/// Fetch latest watch states from server and update local cache.
|
|
///
|
|
/// Called when coming online or on app startup to pull any watch state
|
|
/// changes made on other devices.
|
|
///
|
|
/// Optimized to fetch episodes by season (one API call per season)
|
|
/// instead of one API call per episode.
|
|
Future<void> syncWatchStatesFromServer() async {
|
|
try {
|
|
final profileId = _activeProfileId;
|
|
if (profileId == null || profileId.isEmpty) {
|
|
appLogger.d('Skipping watch-state pull — no active profile');
|
|
return;
|
|
}
|
|
|
|
await _database.adoptLegacyDownloadsForProfile(profileId);
|
|
final ownedKeys = await _database.getDownloadOwnerKeysForProfile(profileId);
|
|
if (ownedKeys.isEmpty) {
|
|
appLogger.d('No active-profile downloads to sync watch states for');
|
|
return;
|
|
}
|
|
|
|
// Pull watch state only for downloads visible to the active profile. The
|
|
// physical rows are shared across profiles, but server watch state is not.
|
|
final downloadedItems = (await _database.getAllDownloadedMetadata())
|
|
.where((item) => ownedKeys.contains(item.globalKey))
|
|
.toList();
|
|
|
|
if (downloadedItems.isEmpty) {
|
|
appLogger.d('No downloaded items to sync watch states for');
|
|
return;
|
|
}
|
|
|
|
appLogger.i('Syncing watch states from server for ${downloadedItems.length} items');
|
|
|
|
// Separate episodes (with season parent) from other items (movies,
|
|
// etc.), using the active scoped client for user-owned Jellyfin watch
|
|
// state. Downloads are shared, but server watch state is per user.
|
|
// Structure: (serverId, clientScopeId) -> seasonRatingKey -> Set<episodeRatingKey>
|
|
final episodesByScopeAndSeason = <({ServerId serverId, String? clientScopeId}), Map<String, Set<String>>>{};
|
|
// Structure: (serverId, clientScopeId) -> List<ratingKey>
|
|
final nonEpisodeItems = <({ServerId serverId, String? clientScopeId}), List<String>>{};
|
|
|
|
for (final item in downloadedItems) {
|
|
final scope = (
|
|
serverId: ServerId(item.serverId),
|
|
clientScopeId: _activeClientScopeIdForServer(ServerId(item.serverId)),
|
|
);
|
|
if (item.type == 'episode' && item.parentRatingKey != null) {
|
|
// Group episodes by server and season for batch fetching
|
|
episodesByScopeAndSeason
|
|
.putIfAbsent(scope, () => {})
|
|
.putIfAbsent(item.parentRatingKey!, () => {})
|
|
.add(item.ratingKey);
|
|
} else {
|
|
// Movies, or episodes without parent (fallback to individual fetch)
|
|
nonEpisodeItems.putIfAbsent(scope, () => []).add(item.ratingKey);
|
|
}
|
|
}
|
|
|
|
int syncedCount = 0;
|
|
int seasonCount = 0;
|
|
|
|
// Fetch episodes by season (batch) - one API call per season
|
|
for (final scopeEntry in episodesByScopeAndSeason.entries) {
|
|
final scope = scopeEntry.key;
|
|
final seasonMap = scopeEntry.value;
|
|
|
|
await _withOnlineClientForDownloadScope(scope.serverId, scope.clientScopeId, (client) async {
|
|
for (final seasonEntry in seasonMap.entries) {
|
|
final result = await _syncSeasonEpisodes(client, scope.serverId, seasonEntry.key, seasonEntry.value);
|
|
if (result >= 0) {
|
|
syncedCount += result;
|
|
seasonCount++;
|
|
}
|
|
}
|
|
});
|
|
}
|
|
|
|
// Fetch non-episode items individually (movies, etc.)
|
|
for (final entry in nonEpisodeItems.entries) {
|
|
final scope = entry.key;
|
|
final ratingKeys = entry.value;
|
|
|
|
await _withOnlineClientForDownloadScope(scope.serverId, scope.clientScopeId, (client) async {
|
|
for (final ratingKey in ratingKeys) {
|
|
try {
|
|
// Snapshot prior viewCount through the neutral cache so we
|
|
// can detect a watched-status change from another device.
|
|
final prior = await client.cache.getMetadata(ServerId(client.cacheServerId), ratingKey);
|
|
final wasWatched = (prior?.viewCount ?? 0) > 0;
|
|
|
|
// fetchItem already caches the full API response (with
|
|
// chapters/markers) via the client's internal cache layer.
|
|
final metadata = await client.fetchItem(ratingKey);
|
|
if (metadata != null) {
|
|
syncedCount++;
|
|
final isWatched = (metadata.viewCount ?? 0) > 0;
|
|
if (wasWatched != isWatched) {
|
|
WatchStateNotifier().notifyWatched(
|
|
item: metadata,
|
|
isNowWatched: isWatched,
|
|
cacheServerId: client.cacheServerId,
|
|
);
|
|
}
|
|
}
|
|
} catch (e) {
|
|
appLogger.d('Failed to sync watch state for $ratingKey: $e');
|
|
}
|
|
}
|
|
});
|
|
}
|
|
|
|
final movieCount = nonEpisodeItems.values.fold(0, (a, b) => a + b.length);
|
|
appLogger.i('Synced watch states: $seasonCount seasons, $movieCount other items ($syncedCount total)');
|
|
|
|
// Notify download provider to refresh metadata from updated cache
|
|
if (syncedCount > 0) {
|
|
onWatchStatesRefreshed?.call();
|
|
}
|
|
|
|
notifyListeners();
|
|
} catch (e) {
|
|
appLogger.w('Error syncing watch states from server: $e');
|
|
}
|
|
}
|
|
|
|
/// Clear all pending watch actions (e.g., when logging out).
|
|
Future<void> clearAll() async {
|
|
await _database.clearAllWatchActions();
|
|
notifyListeners();
|
|
}
|
|
|
|
@override
|
|
void dispose() {
|
|
_isShutDown = true;
|
|
if (_offlineModeSource != null && _offlineModeListener != null) {
|
|
_offlineModeSource!.removeListener(_offlineModeListener!);
|
|
}
|
|
super.dispose();
|
|
}
|
|
}
|