edde746
a56b9a3dfb
Consolidates duplicated logic behind shared implementations — paginated grid tabs, focus chrome, cached remote stores, sheet selection columns, the server artifact store and a test fixture layer — and removes code that had become unreachable. Net reduction of about 5,500 lines with no behaviour change. Where a fix had landed separately in code that moved into a shared helper, the fix was re-applied inside the helper rather than left behind in the copy that went away.
1450 lines
57 KiB
Dart
1450 lines
57 KiB
Dart
import 'dart:async';
|
|
import 'dart:convert';
|
|
import 'dart:io';
|
|
import '../media/ids.dart';
|
|
import 'package:drift/drift.dart';
|
|
import 'package:drift/native.dart';
|
|
import 'package:flutter/foundation.dart';
|
|
import 'package:shared_preferences/shared_preferences.dart';
|
|
import 'package:path_provider/path_provider.dart';
|
|
import 'package:path/path.dart' as p;
|
|
|
|
import 'tables.dart';
|
|
import 'plex_metadata_recovery.dart';
|
|
import 'tvos_database_recovery_store.dart';
|
|
import '../models/download_models.dart';
|
|
import '../services/base_shared_preferences_service.dart';
|
|
import '../services/credential_vault.dart';
|
|
import '../utils/app_logger.dart';
|
|
import '../utils/serial_future_queue.dart';
|
|
import '../utils/global_key_utils.dart';
|
|
|
|
part 'app_database.g.dart';
|
|
|
|
/// Action queued in the offline watch-progress sync table. The serialized
|
|
/// form ([id]) is what gets persisted in [OfflineWatchProgress.actionType];
|
|
/// keep these strings stable across renames so existing rows resolve.
|
|
enum OfflineActionType {
|
|
progress,
|
|
watched,
|
|
unwatched;
|
|
|
|
/// Stable string id used for persistence. Survives an enum-name rename
|
|
/// (e.g. `progress` → `inProgress`) — `.name` would corrupt every row.
|
|
String get id => switch (this) {
|
|
OfflineActionType.progress => 'progress',
|
|
OfflineActionType.watched => 'watched',
|
|
OfflineActionType.unwatched => 'unwatched',
|
|
};
|
|
|
|
/// Inverse of [id]. Throws on unknown so a typo in production doesn't
|
|
/// silently fall back to the wrong action.
|
|
static OfflineActionType fromId(String id) => switch (id) {
|
|
'progress' => OfflineActionType.progress,
|
|
'watched' => OfflineActionType.watched,
|
|
'unwatched' => OfflineActionType.unwatched,
|
|
_ => throw ArgumentError('Unknown OfflineActionType id: $id'),
|
|
};
|
|
}
|
|
|
|
final class AppDatabaseBootstrap {
|
|
const AppDatabaseBootstrap({required this.database, required this.recoveryOutcome});
|
|
|
|
final AppDatabase database;
|
|
final TvosDatabaseRecoveryOutcome recoveryOutcome;
|
|
}
|
|
|
|
@DriftDatabase(
|
|
tables: [
|
|
DownloadedMedia,
|
|
DownloadOwners,
|
|
DownloadQueue,
|
|
ApiCache,
|
|
OfflineWatchProgress,
|
|
SyncRules,
|
|
SyncRuleDownloads,
|
|
Connections,
|
|
Profiles,
|
|
ProfileConnections,
|
|
],
|
|
)
|
|
class AppDatabase extends _$AppDatabase {
|
|
AppDatabase._(QueryExecutor executor, {TvosDatabaseRecoveryStore? recoveryStore})
|
|
: this._withRecovery(executor, recoveryStore);
|
|
|
|
/// Test-only constructor — inject an in-memory [QueryExecutor]
|
|
/// (e.g. `NativeDatabase.memory()`) so tests don't touch real disk.
|
|
@visibleForTesting
|
|
AppDatabase.forTesting(QueryExecutor executor, {TvosDatabaseRecoveryStore? recoveryStore})
|
|
: this._withRecovery(executor, recoveryStore);
|
|
AppDatabase._withRecovery(super.e, this._recoveryStore);
|
|
|
|
final TvosDatabaseRecoveryStore? _recoveryStore;
|
|
final SerialFutureQueue _durabilityQueue = SerialFutureQueue();
|
|
static final Object _durabilityZoneKey = Object();
|
|
static final SerialFutureQueue _tvosRecoveryQueue = SerialFutureQueue();
|
|
|
|
/// Resolves and opens the production database, eagerly completing Drift
|
|
/// setup and migrations on non-tvOS before returning. tvOS recovery retains
|
|
/// ownership of database access ordering.
|
|
static Future<AppDatabaseBootstrap> open({
|
|
bool isTvos = const bool.fromEnvironment('TVOS_BUILD'),
|
|
File? databaseFile,
|
|
SharedPreferencesWithCache? preferences,
|
|
QueryExecutor Function(File file)? executorFactory,
|
|
TvosDatabaseRecoveryStore? recoveryStore,
|
|
TvosDatabaseRecoveryPriorInstallEvidence? priorInstallEvidence,
|
|
}) async {
|
|
final file = databaseFile ?? await _resolveProductionDatabaseFile();
|
|
if (!await file.parent.exists()) {
|
|
await file.parent.create(recursive: true);
|
|
}
|
|
if (databaseFile == null && !Platform.isAndroid && !Platform.isIOS && !await file.exists()) {
|
|
await migrateLegacyDesktopDatabase(target: file);
|
|
}
|
|
|
|
final databaseExisted = await file.exists();
|
|
if (isTvos && !databaseExisted) {
|
|
await _removeOrphanedDatabaseSidecars(file);
|
|
}
|
|
|
|
final prefs = preferences ?? await BaseSharedPreferencesService.sharedCache();
|
|
final store = recoveryStore ?? TvosDatabaseRecoveryStore(prefs, isTvos: isTvos);
|
|
final database = AppDatabase._((executorFactory ?? _createNativeDatabase)(file), recoveryStore: store);
|
|
try {
|
|
if (!isTvos) {
|
|
// Drift executors open lazily. Force the connection through setup and
|
|
// migrations while failures are still covered by this close/rethrow
|
|
// boundary and the caller's startup download-recovery decision.
|
|
await database.customSelect('SELECT 1').get();
|
|
// It deliberately does not claim capacity for a later write.
|
|
}
|
|
final outcome = await _tvosRecoveryQueue.run(
|
|
() => store.reconcile(
|
|
databaseExisted: databaseExisted,
|
|
readIdentity: database._readProtectedIdentityRecoveryRows,
|
|
readPending: database._readPendingRecoveryRows,
|
|
restore: database._restoreRecoverySnapshot,
|
|
hasPriorInstallEvidence:
|
|
priorInstallEvidence ??
|
|
() async {
|
|
return (prefs.getString('active_app_profile_id')?.isNotEmpty ?? false) ||
|
|
(prefs.getBool('profile_migration_v1_done') ?? false) ||
|
|
(prefs.getString('credential_vault_key_v1')?.isNotEmpty ?? false);
|
|
},
|
|
),
|
|
);
|
|
return AppDatabaseBootstrap(database: database, recoveryOutcome: outcome);
|
|
} catch (_) {
|
|
await database.close();
|
|
rethrow;
|
|
}
|
|
}
|
|
|
|
/// Wraps one complete registry identity mutation. Nested registry helpers
|
|
/// share the outer commit and all identity/pending commits are serialized.
|
|
Future<T> runIdentityMutation<T>(Future<T> Function() mutation) {
|
|
return _runDurableMutation(TvosDatabaseRecoveryGroup.identity, mutation);
|
|
}
|
|
|
|
/// Establishes a fresh committed recovery generation only after a user has
|
|
/// acknowledged [TvosDatabaseRecoveryOutcome.recoveryRequired] by starting
|
|
/// a new sign-in. This keeps invalid evidence blocking automatic bootstrap
|
|
/// while allowing the explicit recovery path to persist new identity rows.
|
|
Future<void> acknowledgeTvosDatabaseRecoveryRequired() {
|
|
final store = _recoveryStore;
|
|
if (store == null || !store.isTvos) return Future<void>.value();
|
|
|
|
return _durabilityQueue.run(
|
|
() => _tvosRecoveryQueue.run(
|
|
() => store.acknowledgeRecoveryRequired(
|
|
readIdentity: _readProtectedIdentityRecoveryRows,
|
|
readPending: _readPendingRecoveryRows,
|
|
),
|
|
),
|
|
);
|
|
}
|
|
|
|
Future<T> _runPendingMutation<T>(Future<T> Function() mutation) {
|
|
return _runDurableMutation(TvosDatabaseRecoveryGroup.pending, mutation);
|
|
}
|
|
|
|
Future<T> _runDurableMutation<T>(TvosDatabaseRecoveryGroup group, Future<T> Function() mutation) {
|
|
final store = _recoveryStore;
|
|
if (store == null || !store.isTvos) return mutation();
|
|
if (Zone.current[_durabilityZoneKey] == this) return mutation();
|
|
|
|
return _durabilityQueue.run(
|
|
() => _tvosRecoveryQueue.run(
|
|
() => runZoned(
|
|
() => store.runDurableMutation(
|
|
group: group,
|
|
mutation: mutation,
|
|
readIdentity: _readProtectedIdentityRecoveryRows,
|
|
readPending: _readPendingRecoveryRows,
|
|
),
|
|
zoneValues: {_durabilityZoneKey: this},
|
|
),
|
|
),
|
|
);
|
|
}
|
|
|
|
/// Recovery preferences are a second persisted copy of identity rows. Run
|
|
/// the same credential-vault cutover before reading those rows so a legacy
|
|
/// plaintext database can never become an authoritative plaintext image.
|
|
Future<Map<String, Object?>> _readProtectedIdentityRecoveryRows() async {
|
|
await _migrateLegacyCredentialsBeforeRecoverySnapshot();
|
|
return _readIdentityRecoveryRows();
|
|
}
|
|
|
|
Future<void> _migrateLegacyCredentialsBeforeRecoverySnapshot() async {
|
|
final connectionUpdates = <(String, String)>[];
|
|
for (final row in await select(connections).get()) {
|
|
final decoded = jsonDecode(row.configJson);
|
|
if (decoded is! Map<String, dynamic>) {
|
|
throw const FormatException('Invalid connection configuration');
|
|
}
|
|
if (!_containsPlaintextConnectionCredential(row.kind, decoded)) continue;
|
|
final protected = await CredentialVault.protectConnectionConfig(row.kind, decoded);
|
|
connectionUpdates.add((row.id, jsonEncode(protected)));
|
|
}
|
|
|
|
final tokenUpdates = <(String, String, String)>[];
|
|
for (final row in await select(profileConnections).get()) {
|
|
if (row.userToken.isEmpty || CredentialVault.isProtected(row.userToken)) continue;
|
|
tokenUpdates.add((row.profileId, row.connectionId, await CredentialVault.protect(row.userToken)));
|
|
}
|
|
if (connectionUpdates.isEmpty && tokenUpdates.isEmpty) return;
|
|
|
|
await transaction(() async {
|
|
for (final (id, configJson) in connectionUpdates) {
|
|
await (update(
|
|
connections,
|
|
)..where((table) => table.id.equals(id))).write(ConnectionsCompanion(configJson: Value(configJson)));
|
|
}
|
|
for (final (profileId, connectionId, token) in tokenUpdates) {
|
|
await (update(profileConnections)
|
|
..where((table) => table.profileId.equals(profileId) & table.connectionId.equals(connectionId)))
|
|
.write(ProfileConnectionsCompanion(userToken: Value(token)));
|
|
}
|
|
});
|
|
}
|
|
|
|
static bool _containsPlaintextConnectionCredential(String kind, Map<String, dynamic> config) {
|
|
bool isPlaintext(Object? value) => value is String && value.isNotEmpty && !CredentialVault.isProtected(value);
|
|
|
|
if (kind == 'jellyfin') return isPlaintext(config['accessToken']);
|
|
if (kind != 'plex') return false;
|
|
if (isPlaintext(config['accountToken'])) return true;
|
|
final servers = config['servers'];
|
|
return servers is List && servers.any((server) => server is Map && isPlaintext(server['accessToken']));
|
|
}
|
|
|
|
Future<Map<String, Object?>> _readIdentityRecoveryRows() async {
|
|
final connectionRows = await (select(connections)..orderBy([(t) => OrderingTerm.asc(t.id)])).get();
|
|
final profileRows = await (select(profiles)..orderBy([(t) => OrderingTerm.asc(t.id)])).get();
|
|
final joinRows = await (select(
|
|
profileConnections,
|
|
)..orderBy([(t) => OrderingTerm.asc(t.profileId), (t) => OrderingTerm.asc(t.connectionId)])).get();
|
|
// Drift's generated serializer is the recovery image's column schema:
|
|
// `toJson`/`fromJson` use these camelCase keys, so the read and restore
|
|
// sides can never drift apart when a column is added or renamed.
|
|
return {
|
|
'connections': [for (final row in connectionRows) row.toJson()],
|
|
'profiles': [for (final row in profileRows) row.toJson()],
|
|
'profileConnections': [for (final row in joinRows) row.toJson()],
|
|
};
|
|
}
|
|
|
|
Future<Map<String, Object?>> _readPendingRecoveryRows() async {
|
|
final rows = await (select(offlineWatchProgress)..orderBy([(t) => OrderingTerm.asc(t.id)])).get();
|
|
return {
|
|
'offlineWatchProgress': [for (final row in rows) row.toJson()],
|
|
};
|
|
}
|
|
|
|
Future<void> _restoreRecoverySnapshot(TvosDatabaseRecoverySnapshot snapshot) async {
|
|
final connectionRows = _decodeRecoveryRows(snapshot.identity, 'connections', ConnectionRow.fromJson);
|
|
final profileRows = _decodeRecoveryRows(snapshot.identity, 'profiles', ProfileRow.fromJson);
|
|
final joinRows = _decodeRecoveryRows(snapshot.identity, 'profileConnections', ProfileConnectionRow.fromJson);
|
|
final pendingRows = _decodeRecoveryRows(
|
|
snapshot.pending,
|
|
'offlineWatchProgress',
|
|
OfflineWatchProgressItem.fromJson,
|
|
);
|
|
|
|
// Recovery images from releases before the credential vault may contain
|
|
// plaintext secrets. Protect them before they cross into Drift; already
|
|
// protected values remain byte-identical because vault protection is
|
|
// idempotent.
|
|
for (var index = 0; index < connectionRows.length; index++) {
|
|
final row = connectionRows[index];
|
|
final decoded = jsonDecode(row.configJson);
|
|
if (decoded is! Map<String, dynamic>) {
|
|
throw const FormatException('Invalid connection configuration');
|
|
}
|
|
if (_containsPlaintextConnectionCredential(row.kind, decoded)) {
|
|
connectionRows[index] = row.copyWith(
|
|
configJson: jsonEncode(await CredentialVault.protectConnectionConfig(row.kind, decoded)),
|
|
);
|
|
}
|
|
}
|
|
for (var index = 0; index < joinRows.length; index++) {
|
|
final row = joinRows[index];
|
|
if (row.userToken.isNotEmpty && !CredentialVault.isProtected(row.userToken)) {
|
|
joinRows[index] = row.copyWith(userToken: await CredentialVault.protect(row.userToken));
|
|
}
|
|
}
|
|
|
|
await transaction(() async {
|
|
// Recovery completion (the durable marker removal) is deliberately
|
|
// separate from this transaction. Replace the snapshot-owned rows so a
|
|
// restart can replay the same committed image after a crash or marker
|
|
// removal failure without hitting primary-key conflicts.
|
|
await delete(profileConnections).go();
|
|
await delete(profiles).go();
|
|
await delete(connections).go();
|
|
await delete(offlineWatchProgress).go();
|
|
// `toCompanion(false)` writes every column explicitly, including the
|
|
// nulls, so a restored row is byte-identical to the captured one rather
|
|
// than picking up column defaults.
|
|
for (final row in connectionRows) {
|
|
await into(connections).insert(row.toCompanion(false));
|
|
}
|
|
for (final row in profileRows) {
|
|
await into(profiles).insert(row.toCompanion(false));
|
|
}
|
|
for (final row in joinRows) {
|
|
await into(profileConnections).insert(row.toCompanion(false));
|
|
}
|
|
for (final row in pendingRows) {
|
|
await into(offlineWatchProgress).insert(row.toCompanion(false));
|
|
}
|
|
});
|
|
}
|
|
|
|
static List<T> _decodeRecoveryRows<T extends DataClass>(
|
|
Map<String, Object?> group,
|
|
String key,
|
|
T Function(Map<String, dynamic> json) fromJson,
|
|
) {
|
|
final value = group[key];
|
|
if (value is! List) throw _invalidRecoveryImage;
|
|
return [
|
|
for (final row in value)
|
|
if (row is Map<String, dynamic>) _decodeRecoveryRow(row, fromJson) else throw _invalidRecoveryImage,
|
|
];
|
|
}
|
|
|
|
/// Reads one row through drift's generated deserializer and rejects anything
|
|
/// that does not round-trip back to the exact same map. Drift already throws
|
|
/// on a missing or mistyped required column; the round-trip additionally
|
|
/// rejects unknown and missing-but-nullable columns, which the serializer
|
|
/// would otherwise accept silently.
|
|
static T _decodeRecoveryRow<T extends DataClass>(
|
|
Map<String, dynamic> row,
|
|
T Function(Map<String, dynamic> json) fromJson,
|
|
) {
|
|
final T decoded;
|
|
try {
|
|
decoded = fromJson(row);
|
|
} catch (_) {
|
|
throw _invalidRecoveryImage;
|
|
}
|
|
if (!mapEquals(decoded.toJson(), row)) throw _invalidRecoveryImage;
|
|
return decoded;
|
|
}
|
|
|
|
static const FormatException _invalidRecoveryImage = FormatException('Invalid tvOS database recovery image');
|
|
|
|
@override
|
|
int get schemaVersion => 20;
|
|
|
|
@override
|
|
MigrationStrategy get migration {
|
|
return MigrationStrategy(
|
|
// Enforce ProfileConnections → Profiles/Connections cascades.
|
|
// Drift turns FKs *off* during migrations, so the per-connection
|
|
// pragma we set in `_openConnection` is wiped on first open. This
|
|
// hook runs after migrations and re-enables it for subsequent
|
|
// queries — also applies to in-memory test databases that don't go
|
|
// through `_openConnection`.
|
|
beforeOpen: (details) async {
|
|
await customStatement('PRAGMA foreign_keys = ON');
|
|
},
|
|
onCreate: (Migrator m) async {
|
|
await m.createAll();
|
|
},
|
|
onUpgrade: (Migrator m, int from, int to) async {
|
|
if (from < 7) {
|
|
appLogger.i('Adding OfflineWatchProgress table (v7 migration)');
|
|
await m.createTable(offlineWatchProgress);
|
|
}
|
|
if (from < 8) {
|
|
appLogger.i('Adding bgTaskId column to DownloadedMedia (v8 migration)');
|
|
await _ignoreAlreadyExists(
|
|
'DownloadedMedia.bgTaskId column',
|
|
() => m.addColumn(downloadedMedia, downloadedMedia.bgTaskId),
|
|
);
|
|
}
|
|
if (from < 9) {
|
|
appLogger.i('Adding mediaIndex column to DownloadedMedia (v9 migration)');
|
|
await _ignoreAlreadyExists(
|
|
'DownloadedMedia.mediaIndex column',
|
|
() => m.addColumn(downloadedMedia, downloadedMedia.mediaIndex),
|
|
);
|
|
}
|
|
if (from < 10) {
|
|
appLogger.i('Adding SyncRules table (v10 migration)');
|
|
await m.createTable(syncRules);
|
|
}
|
|
if (from < 11) {
|
|
appLogger.i('Adding enabled column to SyncRules (v11 migration)');
|
|
await _ignoreAlreadyExists('SyncRules.enabled column', () => m.addColumn(syncRules, syncRules.enabled));
|
|
}
|
|
if (from < 12) {
|
|
appLogger.i('Adding downloadFilter column to SyncRules (v12 migration)');
|
|
await _ignoreAlreadyExists(
|
|
'SyncRules.downloadFilter column',
|
|
() => m.addColumn(syncRules, syncRules.downloadFilter),
|
|
);
|
|
}
|
|
if (from < 13) {
|
|
appLogger.i('Adding indexes on DownloadedMedia hot-queried columns (v13 migration)');
|
|
final indexes = {
|
|
'idx_downloaded_media_status': idxDownloadedMediaStatus,
|
|
'idx_downloaded_media_server': idxDownloadedMediaServer,
|
|
'idx_downloaded_media_parent': idxDownloadedMediaParent,
|
|
'idx_downloaded_media_grandparent': idxDownloadedMediaGrandparent,
|
|
};
|
|
for (final entry in indexes.entries) {
|
|
await _ignoreAlreadyExists('Index ${entry.key}', () => m.create(entry.value));
|
|
}
|
|
}
|
|
if (from < 14) {
|
|
appLogger.i(
|
|
'Adding Connections, Profiles, ProfileConnections, DownloadOwners + scope/profile columns (v14 migration)',
|
|
);
|
|
|
|
await m.createTable(connections);
|
|
await _ignoreAlreadyExists('Index idx_connections_kind', () => m.create(idxConnectionsKind));
|
|
|
|
await m.createTable(profiles);
|
|
await _ignoreAlreadyExists('Index idx_profiles_kind', () => m.create(idxProfilesKind));
|
|
|
|
await m.createTable(profileConnections);
|
|
await _ignoreAlreadyExists(
|
|
'Index idx_profile_connections_connection_id',
|
|
() => m.create(idxProfileConnectionsConnectionId),
|
|
);
|
|
await _ignoreAlreadyExists(
|
|
'Index idx_profile_connections_profile_id',
|
|
() => m.create(idxProfileConnectionsProfileId),
|
|
);
|
|
|
|
await _ignoreAlreadyExists('DownloadOwners table', () => m.createTable(downloadOwners));
|
|
await _ignoreAlreadyExists('Index idx_download_owners_profile', () => m.create(idxDownloadOwnersProfile));
|
|
await _ignoreAlreadyExists(
|
|
'Index idx_download_owners_global_key',
|
|
() => m.create(idxDownloadOwnersGlobalKey),
|
|
);
|
|
|
|
await _ignoreAlreadyExists(
|
|
'DownloadedMedia.clientScopeId column',
|
|
() => m.addColumn(downloadedMedia, downloadedMedia.clientScopeId),
|
|
);
|
|
await _ignoreAlreadyExists(
|
|
'OfflineWatchProgress.clientScopeId column',
|
|
() => m.addColumn(offlineWatchProgress, offlineWatchProgress.clientScopeId),
|
|
);
|
|
await _ignoreAlreadyExists('SyncRules.profileId column', () => m.addColumn(syncRules, syncRules.profileId));
|
|
await _ignoreAlreadyExists(
|
|
'OfflineWatchProgress.profileId column',
|
|
() => m.addColumn(offlineWatchProgress, offlineWatchProgress.profileId),
|
|
);
|
|
|
|
await customStatement('''
|
|
UPDATE downloaded_media
|
|
SET client_scope_id = (
|
|
SELECT id FROM connections
|
|
WHERE kind = 'jellyfin'
|
|
AND substr(id, 1, length(downloaded_media.server_id) + 1) = downloaded_media.server_id || '/'
|
|
LIMIT 1
|
|
)
|
|
WHERE client_scope_id IS NULL
|
|
AND EXISTS (
|
|
SELECT 1 FROM connections
|
|
WHERE kind = 'jellyfin'
|
|
AND substr(id, 1, length(downloaded_media.server_id) + 1) = downloaded_media.server_id || '/'
|
|
)
|
|
''');
|
|
await customStatement('''
|
|
UPDATE offline_watch_progress
|
|
SET client_scope_id = (
|
|
SELECT id FROM connections
|
|
WHERE kind = 'jellyfin'
|
|
AND substr(id, 1, length(offline_watch_progress.server_id) + 1) = offline_watch_progress.server_id || '/'
|
|
LIMIT 1
|
|
)
|
|
WHERE client_scope_id IS NULL
|
|
AND EXISTS (
|
|
SELECT 1 FROM connections
|
|
WHERE kind = 'jellyfin'
|
|
AND substr(id, 1, length(offline_watch_progress.server_id) + 1) = offline_watch_progress.server_id || '/'
|
|
)
|
|
''');
|
|
|
|
await _ignoreAlreadyExists(
|
|
'Index idx_offline_watch_progress_server',
|
|
() => m.create(idxOfflineWatchProgressServer),
|
|
);
|
|
await _ignoreAlreadyExists('Index idx_sync_rules_profile', () => m.create(idxSyncRulesProfile));
|
|
await _ignoreAlreadyExists(
|
|
'Index idx_offline_watch_progress_profile',
|
|
() => 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),
|
|
);
|
|
}
|
|
if (from < 16) {
|
|
appLogger.i('Adding includeSpecials column to SyncRules (v16 migration)');
|
|
await _ignoreAlreadyExists(
|
|
'SyncRules.includeSpecials column',
|
|
() => m.addColumn(syncRules, syncRules.includeSpecials),
|
|
);
|
|
}
|
|
if (from < 17) {
|
|
appLogger.i('Scoping pinned legacy Plex metadata before removing bare cache rows (v17 migration)');
|
|
await customStatement(
|
|
_rescopePinnedPlexMetadataStatement(
|
|
namespaceExpression: "'/~plex-profile/' || owner.profile_id || ':'",
|
|
ownerJoin: '''JOIN download_owners AS owner
|
|
ON owner.global_key = metadata.global_key''',
|
|
),
|
|
);
|
|
// A direct pre-v14 upgrade has no owners yet: profiles and owner
|
|
// adoption are bootstrapped only after the database opens. Preserve
|
|
// those downloads in the neutral Plex transfer namespace so the
|
|
// first profile can adopt them without inheriting legacy watch data.
|
|
await customStatement(
|
|
_rescopePinnedPlexMetadataStatement(
|
|
namespaceExpression: "'/~plex-transfer:'",
|
|
ownerFilter: '''AND NOT EXISTS (
|
|
SELECT 1
|
|
FROM download_owners AS owner
|
|
WHERE owner.global_key = metadata.global_key
|
|
)''',
|
|
),
|
|
);
|
|
|
|
final transferRows = await customSelect('''
|
|
SELECT cache_key, data
|
|
FROM api_cache
|
|
WHERE instr(
|
|
substr(cache_key, 1, instr(cache_key, ':') - 1),
|
|
'/~plex-transfer'
|
|
) > 0
|
|
AND instr(cache_key, ':/library/metadata/') > 0
|
|
''').get();
|
|
for (final row in transferRows) {
|
|
final cacheKey = row.read<String>('cache_key');
|
|
try {
|
|
final sanitized = sanitizePlexMetadataForOwnerlessTransfer(row.read<String>('data'));
|
|
await customStatement('UPDATE api_cache SET data = ? WHERE cache_key = ?', [sanitized, cacheKey]);
|
|
} on FormatException catch (error, stackTrace) {
|
|
appLogger.w(
|
|
'Discarding invalid legacy Plex transfer metadata for $cacheKey',
|
|
error: error,
|
|
stackTrace: stackTrace,
|
|
);
|
|
await customStatement('DELETE FROM api_cache WHERE cache_key = ?', [cacheKey]);
|
|
}
|
|
}
|
|
|
|
// Only mark a physical row as transferable when its sanitized leaf
|
|
// exists. Parent-only cache remnants cannot hydrate an offline item.
|
|
await customStatement('''
|
|
UPDATE downloaded_media
|
|
SET client_scope_id = server_id || '/~plex-transfer'
|
|
WHERE NOT EXISTS (
|
|
SELECT 1
|
|
FROM download_owners AS owner
|
|
WHERE owner.global_key = downloaded_media.global_key
|
|
)
|
|
AND EXISTS (
|
|
SELECT 1
|
|
FROM api_cache AS transfer
|
|
WHERE transfer.cache_key =
|
|
downloaded_media.server_id
|
|
|| '/~plex-transfer:/library/metadata/'
|
|
|| downloaded_media.rating_key
|
|
AND transfer.pinned = 1
|
|
)
|
|
''');
|
|
await customStatement('''
|
|
WITH legacy_plex_metadata AS (
|
|
SELECT
|
|
cache_key,
|
|
substr(cache_key, instr(cache_key, ':') + 1) AS endpoint
|
|
FROM api_cache
|
|
WHERE instr(cache_key, ':/library/metadata/') > 0
|
|
AND instr(
|
|
substr(cache_key, 1, instr(cache_key, ':') - 1),
|
|
'/~plex-profile/'
|
|
) = 0
|
|
AND instr(
|
|
substr(cache_key, 1, instr(cache_key, ':') - 1),
|
|
'/~plex-transfer'
|
|
) = 0
|
|
)
|
|
DELETE FROM api_cache
|
|
WHERE cache_key IN (
|
|
SELECT cache_key
|
|
FROM legacy_plex_metadata
|
|
WHERE (
|
|
endpoint GLOB '/library/metadata/?*'
|
|
AND endpoint NOT GLOB '/library/metadata/*/*'
|
|
) OR (
|
|
endpoint GLOB '/library/metadata/?*/children'
|
|
AND endpoint NOT GLOB '/library/metadata/*/*/*'
|
|
)
|
|
)
|
|
''');
|
|
}
|
|
if (from < 18) {
|
|
appLogger.i('Adding safRootUri column to DownloadedMedia (v18 migration)');
|
|
await _ignoreAlreadyExists(
|
|
'DownloadedMedia.safRootUri column',
|
|
() => m.addColumn(downloadedMedia, downloadedMedia.safRootUri),
|
|
);
|
|
}
|
|
if (from < 19) {
|
|
appLogger.i('Adding backend metadata scope columns to DownloadOwners (v19 migration)');
|
|
await _ignoreAlreadyExists(
|
|
'DownloadOwners.backend column',
|
|
() => m.addColumn(downloadOwners, downloadOwners.backend),
|
|
);
|
|
await _ignoreAlreadyExists(
|
|
'DownloadOwners.clientScopeId column',
|
|
() => m.addColumn(downloadOwners, downloadOwners.clientScopeId),
|
|
);
|
|
await customStatement('''
|
|
UPDATE download_owners
|
|
SET client_scope_id = CASE
|
|
WHEN EXISTS (
|
|
SELECT 1
|
|
FROM downloaded_media
|
|
WHERE downloaded_media.global_key = download_owners.global_key
|
|
AND downloaded_media.client_scope_id LIKE '%/~plex-profile/%'
|
|
) THEN (
|
|
SELECT downloaded_media.server_id || '/~plex-profile/' || download_owners.profile_id
|
|
FROM downloaded_media
|
|
WHERE downloaded_media.global_key = download_owners.global_key
|
|
)
|
|
WHEN EXISTS (
|
|
SELECT 1
|
|
FROM downloaded_media
|
|
JOIN profile_connections
|
|
ON profile_connections.profile_id = download_owners.profile_id
|
|
JOIN connections
|
|
ON connections.id = profile_connections.connection_id
|
|
WHERE downloaded_media.global_key = download_owners.global_key
|
|
AND connections.kind = 'jellyfin'
|
|
AND profile_connections.user_identifier != ''
|
|
AND (
|
|
connections.id = downloaded_media.server_id
|
|
OR substr(connections.id, 1, length(downloaded_media.server_id) + 1)
|
|
= downloaded_media.server_id || '/'
|
|
)
|
|
) THEN (
|
|
SELECT CASE
|
|
WHEN connections.id = downloaded_media.server_id
|
|
THEN downloaded_media.server_id || '/' || profile_connections.user_identifier
|
|
ELSE connections.id
|
|
END
|
|
FROM downloaded_media
|
|
JOIN profile_connections
|
|
ON profile_connections.profile_id = download_owners.profile_id
|
|
JOIN connections
|
|
ON connections.id = profile_connections.connection_id
|
|
WHERE downloaded_media.global_key = download_owners.global_key
|
|
AND connections.kind = 'jellyfin'
|
|
AND profile_connections.user_identifier != ''
|
|
AND (
|
|
connections.id = downloaded_media.server_id
|
|
OR substr(connections.id, 1, length(downloaded_media.server_id) + 1)
|
|
= downloaded_media.server_id || '/'
|
|
)
|
|
ORDER BY profile_connections.is_default DESC,
|
|
profile_connections.last_used_at DESC,
|
|
connections.id
|
|
LIMIT 1
|
|
)
|
|
ELSE NULL
|
|
END
|
|
WHERE client_scope_id IS NULL
|
|
''');
|
|
await customStatement('''
|
|
UPDATE download_owners
|
|
SET backend = CASE
|
|
WHEN client_scope_id LIKE '%/~plex-profile/%' THEN 'plex'
|
|
WHEN client_scope_id IS NOT NULL AND EXISTS (
|
|
SELECT 1
|
|
FROM downloaded_media
|
|
JOIN profile_connections
|
|
ON profile_connections.profile_id = download_owners.profile_id
|
|
JOIN connections
|
|
ON connections.id = profile_connections.connection_id
|
|
WHERE downloaded_media.global_key = download_owners.global_key
|
|
AND connections.kind = 'jellyfin'
|
|
AND (
|
|
connections.id = downloaded_media.server_id
|
|
OR substr(connections.id, 1, length(downloaded_media.server_id) + 1)
|
|
= downloaded_media.server_id || '/'
|
|
)
|
|
) THEN 'jellyfin'
|
|
END
|
|
WHERE backend IS NULL
|
|
''');
|
|
}
|
|
if (from < 20) {
|
|
appLogger.i('Adding sync rule download associations (v20 migration)');
|
|
await _ignoreAlreadyExists(
|
|
'SyncRules.downloadLinksInitialized column',
|
|
() => m.addColumn(syncRules, syncRules.downloadLinksInitialized),
|
|
);
|
|
await _ignoreAlreadyExists('SyncRuleDownloads table', () => m.createTable(syncRuleDownloads));
|
|
await _ignoreAlreadyExists(
|
|
'Index idx_sync_rule_downloads_profile_key',
|
|
() => m.create(idxSyncRuleDownloadsProfileKey),
|
|
);
|
|
}
|
|
},
|
|
);
|
|
}
|
|
|
|
Future<void> _ignoreAlreadyExists(String label, Future<void> Function() operation) async {
|
|
try {
|
|
await operation();
|
|
} catch (e) {
|
|
final message = e.toString().toLowerCase();
|
|
if (message.contains('already exists') || message.contains('duplicate column name')) {
|
|
appLogger.w('$label already exists during migration: $e');
|
|
return;
|
|
}
|
|
rethrow;
|
|
}
|
|
}
|
|
|
|
Expression<bool> _nullableTextPredicate(GeneratedColumn<String> column, String? value) {
|
|
return value == null ? column.isNull() : column.equals(value);
|
|
}
|
|
|
|
/// Get all pending offline watch actions for sync
|
|
Future<List<OfflineWatchProgressItem>> getPendingWatchActions({String? profileId}) {
|
|
final query = select(offlineWatchProgress)..orderBy([(t) => OrderingTerm.asc(t.createdAt)]);
|
|
if (profileId != null) {
|
|
query.where((t) => t.profileId.equals(profileId));
|
|
}
|
|
return query.get();
|
|
}
|
|
|
|
/// Claim pre-v18 offline watch actions for [profileId]. Those rows predate
|
|
/// profile ownership and have `NULL profile_id`; the first active profile
|
|
/// inherits them so already-watched offline progress is not stranded.
|
|
Future<void> adoptLegacyOfflineWatchActionsForProfile(String profileId) async {
|
|
if (profileId.isEmpty) return;
|
|
await _runPendingMutation(
|
|
() => (update(
|
|
offlineWatchProgress,
|
|
)..where((t) => t.profileId.isNull())).write(OfflineWatchProgressCompanion(profileId: Value(profileId))),
|
|
);
|
|
}
|
|
|
|
/// Get pending watch actions for a specific server
|
|
@visibleForTesting
|
|
Future<List<OfflineWatchProgressItem>> getPendingWatchActionsForServer(ServerId serverId, {String? profileId}) {
|
|
return (select(offlineWatchProgress)
|
|
..where(
|
|
(t) =>
|
|
t.serverId.equals(serverId) &
|
|
(profileId == null ? const Constant(true) : t.profileId.equals(profileId)),
|
|
)
|
|
..orderBy([(t) => OrderingTerm.asc(t.createdAt)]))
|
|
.get();
|
|
}
|
|
|
|
SimpleSelectStatement<$OfflineWatchProgressTable, OfflineWatchProgressItem> _watchActionsQuery(
|
|
Expression<bool> Function($OfflineWatchProgressTable table) matchesKey, {
|
|
String? profileId,
|
|
bool filterProfile = false,
|
|
String? clientScopeId,
|
|
bool filterClientScope = false,
|
|
}) {
|
|
return select(offlineWatchProgress)
|
|
..where(
|
|
(t) =>
|
|
matchesKey(t) &
|
|
(filterProfile ? _nullableTextPredicate(t.profileId, profileId) : const Constant(true)) &
|
|
(filterClientScope ? _nullableTextPredicate(t.clientScopeId, clientScopeId) : const Constant(true)),
|
|
)
|
|
..orderBy([(t) => OrderingTerm.desc(t.updatedAt), (t) => OrderingTerm.desc(t.id)]);
|
|
}
|
|
|
|
/// Get the latest action for a specific item
|
|
@visibleForTesting
|
|
Future<OfflineWatchProgressItem?> getLatestWatchAction(
|
|
String globalKey, {
|
|
String? profileId,
|
|
bool filterProfile = false,
|
|
String? clientScopeId,
|
|
bool filterClientScope = false,
|
|
}) {
|
|
return (_watchActionsQuery(
|
|
(t) => t.globalKey.equals(globalKey),
|
|
profileId: profileId,
|
|
filterProfile: filterProfile,
|
|
clientScopeId: clientScopeId,
|
|
filterClientScope: filterClientScope,
|
|
)..limit(1)).getSingleOrNull();
|
|
}
|
|
|
|
Future<List<OfflineWatchProgressItem>> getWatchActionsForKey(
|
|
String globalKey, {
|
|
String? profileId,
|
|
bool filterProfile = false,
|
|
String? clientScopeId,
|
|
bool filterClientScope = false,
|
|
}) {
|
|
return _watchActionsQuery(
|
|
(t) => t.globalKey.equals(globalKey),
|
|
profileId: profileId,
|
|
filterProfile: filterProfile,
|
|
clientScopeId: clientScopeId,
|
|
filterClientScope: filterClientScope,
|
|
).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 _watchActionsQuery(
|
|
(t) => t.globalKey.isIn(globalKeys),
|
|
profileId: profileId,
|
|
filterProfile: filterProfile,
|
|
).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.
|
|
/// Keys with no actions will not be present in the returned map.
|
|
Future<Map<String, OfflineWatchProgressItem>> getLatestWatchActionsForKeys(
|
|
Set<String> globalKeys, {
|
|
String? profileId,
|
|
bool filterProfile = false,
|
|
Map<String, String?>? clientScopeIdsByGlobalKey,
|
|
}) async {
|
|
final actionsByKey = await getWatchActionsForKeys(
|
|
globalKeys,
|
|
profileId: profileId,
|
|
filterProfile: filterProfile,
|
|
clientScopeIdsByGlobalKey: clientScopeIdsByGlobalKey,
|
|
);
|
|
return {for (final entry in actionsByKey.entries) entry.key: entry.value.first};
|
|
}
|
|
|
|
bool _clientScopeValuesMatch(String? actual, String? expected) {
|
|
final normalizedActual = actual == null || actual.isEmpty ? null : actual;
|
|
final normalizedExpected = expected == null || expected.isEmpty ? null : expected;
|
|
return normalizedActual == normalizedExpected;
|
|
}
|
|
|
|
/// Insert or update a progress action (merges with existing).
|
|
Future<void> upsertProgressAction({
|
|
String? profileId,
|
|
required ServerId serverId,
|
|
String? clientScopeId,
|
|
required String ratingKey,
|
|
required int viewOffset,
|
|
required int? duration,
|
|
required bool shouldMarkWatched,
|
|
}) async {
|
|
return _runPendingMutation(() async {
|
|
final globalKey = buildGlobalKey(ServerId(serverId), ratingKey);
|
|
final now = DateTime.now().millisecondsSinceEpoch;
|
|
|
|
await transaction(() async {
|
|
final existing =
|
|
await (select(offlineWatchProgress)
|
|
..where(
|
|
(t) =>
|
|
t.globalKey.equals(globalKey) &
|
|
_nullableTextPredicate(t.profileId, profileId) &
|
|
_nullableTextPredicate(t.clientScopeId, clientScopeId) &
|
|
t.actionType.equals(OfflineActionType.progress.id),
|
|
)
|
|
..orderBy([(t) => OrderingTerm.asc(t.id)]))
|
|
.get();
|
|
|
|
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).
|
|
/// Removes conflicting actions for the same item.
|
|
Future<void> insertWatchAction({
|
|
String? profileId,
|
|
required ServerId serverId,
|
|
String? clientScopeId,
|
|
required String ratingKey,
|
|
required String actionType, // 'watched' or 'unwatched'
|
|
}) async {
|
|
return _runPendingMutation(() async {
|
|
final globalKey = buildGlobalKey(ServerId(serverId), ratingKey);
|
|
final now = DateTime.now().millisecondsSinceEpoch;
|
|
|
|
await transaction(() async {
|
|
// Remove conflicting actions (opposite action type and progress).
|
|
await (delete(offlineWatchProgress)..where(
|
|
(t) =>
|
|
t.globalKey.equals(globalKey) &
|
|
_nullableTextPredicate(t.profileId, profileId) &
|
|
_nullableTextPredicate(t.clientScopeId, clientScopeId),
|
|
))
|
|
.go();
|
|
|
|
await into(offlineWatchProgress).insert(
|
|
OfflineWatchProgressCompanion.insert(
|
|
serverId: serverId,
|
|
profileId: Value(profileId),
|
|
clientScopeId: Value(clientScopeId),
|
|
ratingKey: ratingKey,
|
|
globalKey: globalKey,
|
|
actionType: actionType,
|
|
createdAt: now,
|
|
updatedAt: now,
|
|
),
|
|
);
|
|
});
|
|
});
|
|
}
|
|
|
|
/// Delete a specific watch action after successful sync
|
|
Future<void> deleteWatchAction(int id) {
|
|
return _runPendingMutation(() async {
|
|
await (delete(offlineWatchProgress)..where((t) => t.id.equals(id))).go();
|
|
});
|
|
}
|
|
|
|
/// Update sync attempt count and error message
|
|
Future<void> updateSyncAttempt(int id, String? errorMessage) async {
|
|
return _runPendingMutation(() async {
|
|
final existing = await (select(offlineWatchProgress)..where((t) => t.id.equals(id))).getSingleOrNull();
|
|
|
|
if (existing != null) {
|
|
await (update(offlineWatchProgress)..where((t) => t.id.equals(id))).write(
|
|
OfflineWatchProgressCompanion(syncAttempts: Value(existing.syncAttempts + 1), lastError: Value(errorMessage)),
|
|
);
|
|
}
|
|
});
|
|
}
|
|
|
|
/// Get count of pending sync items
|
|
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;
|
|
}
|
|
|
|
/// Clear all pending watch actions (e.g., after logout)
|
|
Future<void> clearAllWatchActions() {
|
|
return _runPendingMutation(() async {
|
|
await delete(offlineWatchProgress).go();
|
|
});
|
|
}
|
|
|
|
/// Drop a removed profile's queued watch actions (profile teardown).
|
|
Future<void> deleteWatchActionsForProfile(String profileId) async {
|
|
await _runPendingMutation(() async {
|
|
await (delete(offlineWatchProgress)..where((t) => t.profileId.equals(profileId))).go();
|
|
});
|
|
}
|
|
|
|
Future<List<SyncRuleItem>> getSyncRules({String? profileId}) {
|
|
final query = select(syncRules);
|
|
if (profileId != null) {
|
|
query.where((t) => t.profileId.equals(profileId));
|
|
}
|
|
return query.get();
|
|
}
|
|
|
|
Future<SyncRuleItem?> getSyncRule(String globalKey) {
|
|
return (select(syncRules)..where((t) => t.globalKey.equals(globalKey))).getSingleOrNull();
|
|
}
|
|
|
|
Future<void> associateSyncRuleDownload(SyncRuleItem rule, String downloadGlobalKey) {
|
|
return into(syncRuleDownloads).insertOnConflictUpdate(
|
|
SyncRuleDownloadsCompanion.insert(
|
|
syncRuleId: rule.id,
|
|
profileId: rule.profileId,
|
|
downloadGlobalKey: downloadGlobalKey,
|
|
),
|
|
);
|
|
}
|
|
|
|
Future<void> markSyncRuleDownloadLinksInitialized(String globalKey) {
|
|
return (update(syncRules)..where((t) => t.globalKey.equals(globalKey))).write(
|
|
const SyncRulesCompanion(downloadLinksInitialized: Value(true)),
|
|
);
|
|
}
|
|
|
|
Future<List<SyncRuleItem>> getUninitializedSyncRulesForServer({
|
|
required String profileId,
|
|
required ServerId serverId,
|
|
}) {
|
|
return (select(syncRules)..where(
|
|
(t) => t.profileId.equals(profileId) & t.serverId.equals(serverId) & t.downloadLinksInitialized.equals(false),
|
|
))
|
|
.get();
|
|
}
|
|
|
|
Future<List<SyncRuleDownloadItem>> getSyncRuleDownloadLinks(int syncRuleId) {
|
|
return (select(syncRuleDownloads)..where((t) => t.syncRuleId.equals(syncRuleId))).get();
|
|
}
|
|
|
|
Future<List<String>> getOwnedDownloadKeysForAncestorRule({
|
|
required String profileId,
|
|
required ServerId serverId,
|
|
required String ratingKey,
|
|
required bool matchGrandparent,
|
|
}) async {
|
|
final query = select(
|
|
downloadedMedia,
|
|
).join([innerJoin(downloadOwners, downloadOwners.globalKey.equalsExp(downloadedMedia.globalKey))]);
|
|
final ancestorMatches = matchGrandparent
|
|
? downloadedMedia.grandparentRatingKey.equals(ratingKey) | downloadedMedia.parentRatingKey.equals(ratingKey)
|
|
: downloadedMedia.parentRatingKey.equals(ratingKey);
|
|
query.where(
|
|
downloadOwners.profileId.equals(profileId) &
|
|
downloadedMedia.serverId.equals(serverId) &
|
|
downloadedMedia.status.isIn([
|
|
DownloadStatus.queued.index,
|
|
DownloadStatus.downloading.index,
|
|
DownloadStatus.completed.index,
|
|
DownloadStatus.paused.index,
|
|
]) &
|
|
ancestorMatches,
|
|
);
|
|
final rows = await query.get();
|
|
return rows.map((row) => row.readTable(downloadedMedia).globalKey).toList(growable: false);
|
|
}
|
|
|
|
Future<List<String>> getExclusiveSyncRuleDownloadKeys(SyncRuleItem rule) async {
|
|
final links = await getSyncRuleDownloadLinks(rule.id);
|
|
if (links.isEmpty) return const [];
|
|
|
|
final keys = links.map((link) => link.downloadGlobalKey).toSet();
|
|
final allLinks = await (select(
|
|
syncRuleDownloads,
|
|
)..where((t) => t.profileId.equals(rule.profileId) & t.downloadGlobalKey.isIn(keys))).get();
|
|
final linkedRuleCounts = <String, int>{};
|
|
for (final link in allLinks) {
|
|
linkedRuleCounts.update(link.downloadGlobalKey, (count) => count + 1, ifAbsent: () => 1);
|
|
}
|
|
return [
|
|
for (final key in keys)
|
|
if (linkedRuleCounts[key] == 1) key,
|
|
];
|
|
}
|
|
|
|
Future<void> insertSyncRule({
|
|
String profileId = '',
|
|
required ServerId serverId,
|
|
required String ratingKey,
|
|
required String globalKey,
|
|
required String targetType,
|
|
required int episodeCount,
|
|
int mediaIndex = 0,
|
|
String downloadFilter = 'unwatched',
|
|
bool includeSpecials = true,
|
|
}) async {
|
|
// [insertOnConflictUpdate] defaults the conflict target to the primary
|
|
// key (`id`), which is auto-incremented — the conflict never triggers
|
|
// and the row's UNIQUE [globalKey] constraint blows up instead. Drive
|
|
// the upsert off the public [globalKey] so re-creating a rule for the same
|
|
// shared target updates the existing row.
|
|
await into(syncRules).insert(
|
|
SyncRulesCompanion.insert(
|
|
serverId: serverId,
|
|
profileId: Value(profileId),
|
|
ratingKey: ratingKey,
|
|
globalKey: globalKey,
|
|
targetType: targetType,
|
|
episodeCount: episodeCount,
|
|
createdAt: DateTime.now().millisecondsSinceEpoch,
|
|
mediaIndex: Value(mediaIndex),
|
|
downloadFilter: Value(downloadFilter),
|
|
includeSpecials: Value(includeSpecials),
|
|
),
|
|
onConflict: DoUpdate(
|
|
(_) => SyncRulesCompanion(
|
|
serverId: Value(serverId),
|
|
profileId: Value(profileId),
|
|
ratingKey: Value(ratingKey),
|
|
targetType: Value(targetType),
|
|
episodeCount: Value(episodeCount),
|
|
mediaIndex: Value(mediaIndex),
|
|
downloadFilter: Value(downloadFilter),
|
|
includeSpecials: Value(includeSpecials),
|
|
),
|
|
target: [syncRules.globalKey],
|
|
),
|
|
);
|
|
}
|
|
|
|
/// Claim pre-v16 public sync rules for [profileId]. Rules created before
|
|
/// profile ownership have an empty profile id and a public global key.
|
|
Future<void> adoptLegacySyncRulesForProfile(String profileId) async {
|
|
if (profileId.isEmpty) return;
|
|
final legacyRules = await (select(syncRules)..where((t) => t.profileId.equals(''))).get();
|
|
for (final rule in legacyRules) {
|
|
final scopedKey = buildProfileScopedGlobalKey(profileId, ServerId(rule.serverId), rule.ratingKey);
|
|
final duplicate = await getSyncRule(scopedKey);
|
|
if (duplicate != null) {
|
|
await (delete(syncRules)..where((t) => t.id.equals(rule.id))).go();
|
|
continue;
|
|
}
|
|
await (update(syncRules)..where((t) => t.id.equals(rule.id))).write(
|
|
SyncRulesCompanion(profileId: Value(profileId), globalKey: Value(scopedKey)),
|
|
);
|
|
await (update(
|
|
syncRuleDownloads,
|
|
)..where((t) => t.syncRuleId.equals(rule.id))).write(SyncRuleDownloadsCompanion(profileId: Value(profileId)));
|
|
}
|
|
}
|
|
|
|
Future<void> _writeSyncRule(String globalKey, SyncRulesCompanion values) async {
|
|
await (update(syncRules)..where((t) => t.globalKey.equals(globalKey))).write(values);
|
|
}
|
|
|
|
Future<void> updateSyncRuleCount(String globalKey, int episodeCount) =>
|
|
_writeSyncRule(globalKey, SyncRulesCompanion(episodeCount: Value(episodeCount)));
|
|
|
|
Future<void> updateSyncRuleFilter(String globalKey, String downloadFilter) =>
|
|
_writeSyncRule(globalKey, SyncRulesCompanion(downloadFilter: Value(downloadFilter)));
|
|
|
|
Future<void> updateSyncRuleEnabled(String globalKey, bool enabled) =>
|
|
_writeSyncRule(globalKey, SyncRulesCompanion(enabled: Value(enabled)));
|
|
|
|
Future<void> updateSyncRuleLastExecuted(String globalKey) =>
|
|
_writeSyncRule(globalKey, SyncRulesCompanion(lastExecutedAt: Value(DateTime.now().millisecondsSinceEpoch)));
|
|
|
|
Future<void> completeSyncRuleExecution(String globalKey) {
|
|
return (update(syncRules)..where((t) => t.globalKey.equals(globalKey))).write(
|
|
SyncRulesCompanion(
|
|
lastExecutedAt: Value(DateTime.now().millisecondsSinceEpoch),
|
|
downloadLinksInitialized: const Value(true),
|
|
),
|
|
);
|
|
}
|
|
|
|
Future<void> deleteSyncRule(String globalKey) async {
|
|
await (delete(syncRules)..where((t) => t.globalKey.equals(globalKey))).go();
|
|
}
|
|
|
|
/// Drop a removed profile's sync rules (profile teardown).
|
|
Future<void> deleteSyncRulesForProfile(String profileId) async {
|
|
await (delete(syncRules)..where((t) => t.profileId.equals(profileId))).go();
|
|
}
|
|
|
|
/// Drop every sync rule (full logout).
|
|
Future<void> clearAllSyncRules() async {
|
|
await delete(syncRules).go();
|
|
}
|
|
|
|
/// Get all downloaded media items (for syncing watch states)
|
|
Future<List<DownloadedMediaItem>> getAllDownloadedMetadata() {
|
|
return (select(downloadedMedia)..where((t) => t.status.equals(DownloadStatus.completed.index))).get();
|
|
}
|
|
}
|
|
|
|
/// Builds the v17 statement that re-keys pinned legacy Plex metadata rows into
|
|
/// a scoped cache namespace.
|
|
///
|
|
/// The owned and the ownerless branch run the same operation over the same
|
|
/// `download_metadata_ids` set and differ only in three spots: the expression
|
|
/// spliced into the new `cache_key` ([namespaceExpression]), an optional join
|
|
/// that exposes the owning profile ([ownerJoin]), and an optional extra
|
|
/// predicate that keeps each branch to its own rows ([ownerFilter]).
|
|
String _rescopePinnedPlexMetadataStatement({
|
|
required String namespaceExpression,
|
|
String ownerJoin = '',
|
|
String ownerFilter = '',
|
|
}) =>
|
|
'''
|
|
WITH download_metadata_ids AS (
|
|
SELECT global_key, server_id, rating_key AS metadata_id
|
|
FROM downloaded_media
|
|
UNION
|
|
SELECT global_key, server_id, parent_rating_key AS metadata_id
|
|
FROM downloaded_media
|
|
WHERE parent_rating_key IS NOT NULL
|
|
AND parent_rating_key != ''
|
|
UNION
|
|
SELECT global_key, server_id, grandparent_rating_key AS metadata_id
|
|
FROM downloaded_media
|
|
WHERE grandparent_rating_key IS NOT NULL
|
|
AND grandparent_rating_key != ''
|
|
)
|
|
INSERT INTO api_cache (cache_key, data, pinned, cached_at)
|
|
SELECT DISTINCT
|
|
metadata.server_id
|
|
|| $namespaceExpression
|
|
|| substr(source.cache_key, length(metadata.server_id) + 2),
|
|
source.data,
|
|
source.pinned,
|
|
source.cached_at
|
|
FROM download_metadata_ids AS metadata
|
|
$ownerJoin
|
|
JOIN api_cache AS source
|
|
ON source.cache_key =
|
|
metadata.server_id || ':/library/metadata/' || metadata.metadata_id
|
|
OR source.cache_key =
|
|
metadata.server_id || ':/library/metadata/' || metadata.metadata_id || '/children'
|
|
WHERE source.pinned = 1
|
|
$ownerFilter
|
|
ON CONFLICT(cache_key) DO UPDATE SET
|
|
data = excluded.data,
|
|
pinned = excluded.pinned,
|
|
cached_at = excluded.cached_at
|
|
''';
|
|
|
|
Future<File> _resolveProductionDatabaseFile() async {
|
|
final dbFolder = (Platform.isAndroid || Platform.isIOS)
|
|
? await getApplicationDocumentsDirectory()
|
|
: await getApplicationSupportDirectory();
|
|
return File(p.join(dbFolder.path, 'plezy_downloads.db'));
|
|
}
|
|
|
|
Future<void> _removeOrphanedDatabaseSidecars(File databaseFile) async {
|
|
for (final suffix in const ['-wal', '-shm']) {
|
|
final sidecar = File('${databaseFile.path}$suffix');
|
|
if (await sidecar.exists()) await sidecar.delete();
|
|
}
|
|
}
|
|
|
|
QueryExecutor _createNativeDatabase(File file) {
|
|
return NativeDatabase.createInBackground(
|
|
file,
|
|
setup: (db) {
|
|
db.execute('PRAGMA journal_mode=WAL');
|
|
db.execute('PRAGMA synchronous=NORMAL');
|
|
// Enforce ProfileConnections → Connections cascades.
|
|
// SQLite requires this on every connection — it is not persisted.
|
|
db.execute('PRAGMA foreign_keys = ON');
|
|
},
|
|
);
|
|
}
|
|
|
|
/// Move the legacy desktop DB from `Documents/` to `ApplicationSupport/`.
|
|
/// `File.rename` only works within a single volume — Windows users with
|
|
/// OneDrive-redirected Documents (or any cross-drive setup) hit
|
|
/// `ERROR_NOT_SAME_DEVICE` (errno 17), and the uncaught throw used to
|
|
/// strand the splash on "Loading servers..." forever (#1022). Falls back
|
|
/// to a synced sibling temporary copy followed by an atomic rename on any
|
|
/// [FileSystemException], and swallows all errors so a failed migration
|
|
/// never propagates fatally. The canonical-path lock file is intentionally
|
|
/// retained: deleting it could let a new process lock a different inode while
|
|
/// an existing waiter still holds the old one.
|
|
///
|
|
/// [sourceOverride], [renameOverride], [copyOverride], and [publishOverride]
|
|
/// are test seams — production callers leave them null.
|
|
Future<void> migrateLegacyDesktopDatabase({
|
|
required File target,
|
|
File? sourceOverride,
|
|
Future<void> Function(File source, String targetPath)? renameOverride,
|
|
Future<void> Function(File source, File temporary)? copyOverride,
|
|
Future<void> Function(File temporary, File target)? publishOverride,
|
|
}) async {
|
|
final File oldFile;
|
|
try {
|
|
if (sourceOverride != null) {
|
|
oldFile = sourceOverride;
|
|
} else {
|
|
final oldFolder = await getApplicationDocumentsDirectory();
|
|
oldFile = File(p.join(oldFolder.path, 'plezy_downloads.db'));
|
|
}
|
|
if (!await oldFile.exists()) return;
|
|
} catch (e, st) {
|
|
appLogger.w('Legacy DB migration skipped before source lookup completed', error: e, stackTrace: st);
|
|
return;
|
|
}
|
|
|
|
try {
|
|
final moved = await _withLegacyDatabasePublishLock(target, () async {
|
|
if (await target.exists()) {
|
|
appLogger.w('Legacy DB migration skipped because ${target.path} now exists');
|
|
return false;
|
|
}
|
|
if (renameOverride != null) {
|
|
await renameOverride(oldFile, target.path);
|
|
} else {
|
|
await oldFile.rename(target.path);
|
|
}
|
|
return true;
|
|
});
|
|
if (!moved) return;
|
|
appLogger.i('Moved legacy DB from ${oldFile.path} → ${target.path}');
|
|
return;
|
|
} on FileSystemException catch (e) {
|
|
appLogger.w('Legacy DB rename failed (osError=${e.osError?.errorCode}); falling back to copy', error: e);
|
|
}
|
|
|
|
final temporary = File(
|
|
p.join(
|
|
target.parent.path,
|
|
'.${p.basename(target.path)}.legacy-migration-$pid-${DateTime.now().microsecondsSinceEpoch}.tmp',
|
|
),
|
|
);
|
|
try {
|
|
if (copyOverride != null) {
|
|
await copyOverride(oldFile, temporary);
|
|
} else {
|
|
await _copyFileAndSync(oldFile, temporary);
|
|
}
|
|
|
|
final published = await _withLegacyDatabasePublishLock(target, () async {
|
|
// Recheck only while holding the inter-process lock. On POSIX, rename
|
|
// replaces an existing destination, so an unlocked check can race a
|
|
// concurrent publisher and overwrite its now-canonical database.
|
|
if (await target.exists()) {
|
|
appLogger.w('Legacy DB migration skipped because ${target.path} now exists');
|
|
return false;
|
|
}
|
|
|
|
// The temporary file is a sibling, so this rename stays on one volume
|
|
// and publishes the complete, synced copy atomically.
|
|
if (publishOverride != null) {
|
|
await publishOverride(temporary, target);
|
|
} else {
|
|
await temporary.rename(target.path);
|
|
}
|
|
return true;
|
|
});
|
|
if (!published) return;
|
|
try {
|
|
await oldFile.delete();
|
|
} catch (e) {
|
|
// Leaving the source behind is non-fatal — the new file is canonical.
|
|
appLogger.w('Legacy DB copied but old file delete failed: $e');
|
|
}
|
|
appLogger.i('Copied legacy DB from ${oldFile.path} → ${target.path}');
|
|
} catch (e, st) {
|
|
// A failed copy or final rename never touches the canonical path. Keep
|
|
// the legacy source so a future launch can retry.
|
|
appLogger.e('Legacy DB migration failed entirely', error: e, stackTrace: st);
|
|
} finally {
|
|
try {
|
|
if (await temporary.exists()) await temporary.delete();
|
|
} catch (e, st) {
|
|
appLogger.w('Failed to clean legacy DB migration temporary file', error: e, stackTrace: st);
|
|
}
|
|
}
|
|
}
|
|
|
|
Future<T> _withLegacyDatabasePublishLock<T>(File target, Future<T> Function() action) async {
|
|
final lockFile = File(p.join(target.parent.path, '.${p.basename(target.path)}.legacy-migration.lock'));
|
|
final handle = await lockFile.open(mode: FileMode.append);
|
|
var locked = false;
|
|
try {
|
|
await handle.lock(FileLock.blockingExclusive);
|
|
locked = true;
|
|
return await action();
|
|
} finally {
|
|
try {
|
|
if (locked) await handle.unlock();
|
|
} finally {
|
|
await handle.close();
|
|
}
|
|
}
|
|
}
|
|
|
|
Future<void> _copyFileAndSync(File source, File destination) async {
|
|
final input = await source.open();
|
|
try {
|
|
final output = await destination.open(mode: FileMode.writeOnly);
|
|
try {
|
|
final buffer = Uint8List(64 * 1024);
|
|
while (true) {
|
|
final bytesRead = await input.readInto(buffer);
|
|
if (bytesRead == 0) break;
|
|
await output.writeFrom(buffer, 0, bytesRead);
|
|
}
|
|
await output.flush();
|
|
} finally {
|
|
await output.close();
|
|
}
|
|
} finally {
|
|
await input.close();
|
|
}
|
|
}
|