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

fix(playback): harden offline source reporting

Browse Source
This commit is contained in:
edde746
2026-05-29 19:55:01 +02:00
parent 7501f461b9
commit d93ea9813f
34 changed files with 739 additions and 273 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/database/app_database.dart
+104 -44
View File
@@ -60,7 +60,7 @@ class AppDatabase extends _$AppDatabase {
AppDatabase.forTesting(super.e);
@override
int get schemaVersion => 14;
int get schemaVersion => 15;
@override
MigrationStrategy get migration {
@@ -206,6 +206,13 @@ class AppDatabase extends _$AppDatabase {
() => m.create(idxOfflineWatchProgressProfile),
);
}
if (from < 15) {
appLogger.i('Adding mediaSourceId column to DownloadedMedia (v15 migration)');
await _ignoreAlreadyExists(
'DownloadedMedia.mediaSourceId column',
() => m.addColumn(downloadedMedia, downloadedMedia.mediaSourceId),
);
}
},
);
}
@@ -282,6 +289,52 @@ class AppDatabase extends _$AppDatabase {
.getSingleOrNull();
}
Future<List<OfflineWatchProgressItem>> getWatchActionsForKey(
String globalKey, {
String? profileId,
bool filterProfile = false,
String? clientScopeId,
bool filterClientScope = false,
}) {
return (select(offlineWatchProgress)
..where(
(t) =>
t.globalKey.equals(globalKey) &
(filterProfile ? _nullableTextPredicate(t.profileId, profileId) : const Constant(true)) &
(filterClientScope ? _clientScopePredicate(t.clientScopeId, clientScopeId) : const Constant(true)),
)
..orderBy([(t) => OrderingTerm.desc(t.updatedAt)]))
.get();
}
Future<Map<String, List<OfflineWatchProgressItem>>> getWatchActionsForKeys(
Set<String> globalKeys, {
String? profileId,
bool filterProfile = false,
Map<String, String?>? clientScopeIdsByGlobalKey,
}) async {
if (globalKeys.isEmpty) return const {};
final rows =
await (select(offlineWatchProgress)
..where(
(t) =>
t.globalKey.isIn(globalKeys) &
(filterProfile ? _nullableTextPredicate(t.profileId, profileId) : const Constant(true)),
)
..orderBy([(t) => OrderingTerm.desc(t.updatedAt)]))
.get();
final result = <String, List<OfflineWatchProgressItem>>{};
for (final action in rows) {
if (clientScopeIdsByGlobalKey != null && clientScopeIdsByGlobalKey.containsKey(action.globalKey)) {
final expectedScope = clientScopeIdsByGlobalKey[action.globalKey];
if (!_clientScopeValuesMatch(action.clientScopeId, expectedScope)) continue;
}
result.putIfAbsent(action.globalKey, () => <OfflineWatchProgressItem>[]).add(action);
}
return result;
}
/// Get the latest actions for multiple items in a single query
///
/// Returns a map of globalKey -> latest action for each key.
@@ -338,49 +391,53 @@ class AppDatabase extends _$AppDatabase {
final globalKey = buildGlobalKey(serverId, ratingKey);
final now = DateTime.now().millisecondsSinceEpoch;
// Check for existing progress entry
final existing =
await (select(offlineWatchProgress)
..where(
(t) =>
t.globalKey.equals(globalKey) &
_nullableTextPredicate(t.profileId, profileId) &
_clientScopePredicate(t.clientScopeId, clientScopeId) &
t.actionType.equals(OfflineActionType.progress.id),
)
..limit(1))
.getSingleOrNull();
await transaction(() async {
final existing =
await (select(offlineWatchProgress)
..where(
(t) =>
t.globalKey.equals(globalKey) &
_nullableTextPredicate(t.profileId, profileId) &
_clientScopePredicate(t.clientScopeId, clientScopeId) &
t.actionType.equals(OfflineActionType.progress.id),
)
..orderBy([(t) => OrderingTerm.asc(t.id)]))
.get();
if (existing != null) {
// Update existing progress entry
await (update(offlineWatchProgress)..where((t) => t.id.equals(existing.id))).write(
OfflineWatchProgressCompanion(
viewOffset: Value(viewOffset),
duration: Value(duration),
shouldMarkWatched: Value(shouldMarkWatched),
profileId: Value(profileId),
clientScopeId: Value(clientScopeId),
updatedAt: Value(now),
),
);
} else {
// Insert new progress entry
await into(offlineWatchProgress).insert(
OfflineWatchProgressCompanion.insert(
serverId: serverId,
profileId: Value(profileId),
clientScopeId: Value(clientScopeId),
ratingKey: ratingKey,
globalKey: globalKey,
actionType: OfflineActionType.progress.id,
viewOffset: Value(viewOffset),
duration: Value(duration),
shouldMarkWatched: Value(shouldMarkWatched),
createdAt: now,
updatedAt: now,
),
);
}
final keep = existing.isEmpty ? null : existing.first;
if (keep != null) {
await (update(offlineWatchProgress)..where((t) => t.id.equals(keep.id))).write(
OfflineWatchProgressCompanion(
viewOffset: Value(viewOffset),
duration: Value(duration),
shouldMarkWatched: Value(shouldMarkWatched),
profileId: Value(profileId),
clientScopeId: Value(clientScopeId),
updatedAt: Value(now),
),
);
final duplicateIds = existing.skip(1).map((row) => row.id).toList(growable: false);
if (duplicateIds.isNotEmpty) {
await (delete(offlineWatchProgress)..where((t) => t.id.isIn(duplicateIds))).go();
}
} else {
await into(offlineWatchProgress).insert(
OfflineWatchProgressCompanion.insert(
serverId: serverId,
profileId: Value(profileId),
clientScopeId: Value(clientScopeId),
ratingKey: ratingKey,
globalKey: globalKey,
actionType: OfflineActionType.progress.id,
viewOffset: Value(viewOffset),
duration: Value(duration),
shouldMarkWatched: Value(shouldMarkWatched),
createdAt: now,
updatedAt: now,
),
);
}
});
}
/// Insert a manual watch action (watched or unwatched).
@@ -436,11 +493,14 @@ class AppDatabase extends _$AppDatabase {
}
/// Get count of pending sync items
Future<int> getPendingSyncCount({String? profileId}) async {
Future<int> getPendingSyncCount({String? profileId, int? maxSyncAttempts}) async {
final query = selectOnly(offlineWatchProgress)..addColumns([offlineWatchProgress.id.count()]);
if (profileId != null) {
query.where(offlineWatchProgress.profileId.equals(profileId));
}
if (maxSyncAttempts != null) {
query.where(offlineWatchProgress.syncAttempts.isSmallerThanValue(maxSyncAttempts));
}
final count = await query.map((row) => row.read(offlineWatchProgress.id.count())).getSingle();
return count ?? 0;
}