edde746
4307c49cd2
Drops dead code across services, models, utils and widgets, including the connection auth service, which had no implementer, and the Live TV DVR provisioning models, which had no caller. Tests that only covered deleted behaviour are removed or trimmed. No behaviour change.
1503 lines
59 KiB
Dart
1503 lines
59 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,
|
|
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, then reconciles tvOS
|
|
/// recovery before returning it to startup consumers.
|
|
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 {
|
|
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();
|
|
return {
|
|
'connections': [
|
|
for (final row in connectionRows)
|
|
{
|
|
'id': row.id,
|
|
'kind': row.kind,
|
|
'displayName': row.displayName,
|
|
'configJson': row.configJson,
|
|
'isDefault': row.isDefault,
|
|
'createdAt': row.createdAt,
|
|
'lastAuthenticatedAt': row.lastAuthenticatedAt,
|
|
},
|
|
],
|
|
'profiles': [
|
|
for (final row in profileRows)
|
|
{
|
|
'id': row.id,
|
|
'kind': row.kind,
|
|
'displayName': row.displayName,
|
|
'avatarThumbUrl': row.avatarThumbUrl,
|
|
'configJson': row.configJson,
|
|
'sortOrder': row.sortOrder,
|
|
'createdAt': row.createdAt,
|
|
'lastUsedAt': row.lastUsedAt,
|
|
},
|
|
],
|
|
'profileConnections': [
|
|
for (final row in joinRows)
|
|
{
|
|
'profileId': row.profileId,
|
|
'connectionId': row.connectionId,
|
|
'userToken': row.userToken,
|
|
'userIdentifier': row.userIdentifier,
|
|
'isDefault': row.isDefault,
|
|
'tokenAcquiredAt': row.tokenAcquiredAt,
|
|
'lastUsedAt': row.lastUsedAt,
|
|
},
|
|
],
|
|
};
|
|
}
|
|
|
|
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)
|
|
{
|
|
'id': row.id,
|
|
'profileId': row.profileId,
|
|
'serverId': row.serverId,
|
|
'clientScopeId': row.clientScopeId,
|
|
'ratingKey': row.ratingKey,
|
|
'globalKey': row.globalKey,
|
|
'actionType': row.actionType,
|
|
'viewOffset': row.viewOffset,
|
|
'duration': row.duration,
|
|
'shouldMarkWatched': row.shouldMarkWatched,
|
|
'createdAt': row.createdAt,
|
|
'updatedAt': row.updatedAt,
|
|
'syncAttempts': row.syncAttempts,
|
|
'lastError': row.lastError,
|
|
},
|
|
],
|
|
};
|
|
}
|
|
|
|
Future<void> _restoreRecoverySnapshot(TvosDatabaseRecoverySnapshot snapshot) async {
|
|
final connectionRows = _decodeRecoveryRows(snapshot.identity, 'connections', const {
|
|
'id',
|
|
'kind',
|
|
'displayName',
|
|
'configJson',
|
|
'isDefault',
|
|
'createdAt',
|
|
'lastAuthenticatedAt',
|
|
});
|
|
final profileRows = _decodeRecoveryRows(snapshot.identity, 'profiles', const {
|
|
'id',
|
|
'kind',
|
|
'displayName',
|
|
'avatarThumbUrl',
|
|
'configJson',
|
|
'sortOrder',
|
|
'createdAt',
|
|
'lastUsedAt',
|
|
});
|
|
final joinRows = _decodeRecoveryRows(snapshot.identity, 'profileConnections', const {
|
|
'profileId',
|
|
'connectionId',
|
|
'userToken',
|
|
'userIdentifier',
|
|
'isDefault',
|
|
'tokenAcquiredAt',
|
|
'lastUsedAt',
|
|
});
|
|
final pendingRows = _decodeRecoveryRows(snapshot.pending, 'offlineWatchProgress', const {
|
|
'id',
|
|
'profileId',
|
|
'serverId',
|
|
'clientScopeId',
|
|
'ratingKey',
|
|
'globalKey',
|
|
'actionType',
|
|
'viewOffset',
|
|
'duration',
|
|
'shouldMarkWatched',
|
|
'createdAt',
|
|
'updatedAt',
|
|
'syncAttempts',
|
|
'lastError',
|
|
});
|
|
|
|
// 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 (final row in connectionRows) {
|
|
final kind = _requiredRecoveryValue<String>(row, 'kind');
|
|
final configJson = _requiredRecoveryValue<String>(row, 'configJson');
|
|
final decoded = jsonDecode(configJson);
|
|
if (decoded is! Map<String, dynamic>) {
|
|
throw const FormatException('Invalid connection configuration');
|
|
}
|
|
if (_containsPlaintextConnectionCredential(kind, decoded)) {
|
|
row['configJson'] = jsonEncode(await CredentialVault.protectConnectionConfig(kind, decoded));
|
|
}
|
|
}
|
|
for (final row in joinRows) {
|
|
final token = _requiredRecoveryValue<String>(row, 'userToken');
|
|
if (token.isNotEmpty && !CredentialVault.isProtected(token)) {
|
|
row['userToken'] = await CredentialVault.protect(token);
|
|
}
|
|
}
|
|
|
|
final connectionCompanions = [
|
|
for (final row in connectionRows)
|
|
ConnectionsCompanion(
|
|
id: Value(_requiredRecoveryValue<String>(row, 'id')),
|
|
kind: Value(_requiredRecoveryValue<String>(row, 'kind')),
|
|
displayName: Value(_requiredRecoveryValue<String>(row, 'displayName')),
|
|
configJson: Value(_requiredRecoveryValue<String>(row, 'configJson')),
|
|
isDefault: Value(_requiredRecoveryValue<bool>(row, 'isDefault')),
|
|
createdAt: Value(_requiredRecoveryValue<int>(row, 'createdAt')),
|
|
lastAuthenticatedAt: Value(_nullableRecoveryValue<int>(row, 'lastAuthenticatedAt')),
|
|
),
|
|
];
|
|
final profileCompanions = [
|
|
for (final row in profileRows)
|
|
ProfilesCompanion(
|
|
id: Value(_requiredRecoveryValue<String>(row, 'id')),
|
|
kind: Value(_requiredRecoveryValue<String>(row, 'kind')),
|
|
displayName: Value(_requiredRecoveryValue<String>(row, 'displayName')),
|
|
avatarThumbUrl: Value(_nullableRecoveryValue<String>(row, 'avatarThumbUrl')),
|
|
configJson: Value(_requiredRecoveryValue<String>(row, 'configJson')),
|
|
sortOrder: Value(_requiredRecoveryValue<int>(row, 'sortOrder')),
|
|
createdAt: Value(_requiredRecoveryValue<int>(row, 'createdAt')),
|
|
lastUsedAt: Value(_nullableRecoveryValue<int>(row, 'lastUsedAt')),
|
|
),
|
|
];
|
|
final joinCompanions = [
|
|
for (final row in joinRows)
|
|
ProfileConnectionsCompanion(
|
|
profileId: Value(_requiredRecoveryValue<String>(row, 'profileId')),
|
|
connectionId: Value(_requiredRecoveryValue<String>(row, 'connectionId')),
|
|
userToken: Value(_requiredRecoveryValue<String>(row, 'userToken')),
|
|
userIdentifier: Value(_requiredRecoveryValue<String>(row, 'userIdentifier')),
|
|
isDefault: Value(_requiredRecoveryValue<bool>(row, 'isDefault')),
|
|
tokenAcquiredAt: Value(_nullableRecoveryValue<int>(row, 'tokenAcquiredAt')),
|
|
lastUsedAt: Value(_nullableRecoveryValue<int>(row, 'lastUsedAt')),
|
|
),
|
|
];
|
|
final pendingCompanions = [
|
|
for (final row in pendingRows)
|
|
OfflineWatchProgressCompanion(
|
|
id: Value(_requiredRecoveryValue<int>(row, 'id')),
|
|
profileId: Value(_nullableRecoveryValue<String>(row, 'profileId')),
|
|
serverId: Value(_requiredRecoveryValue<String>(row, 'serverId')),
|
|
clientScopeId: Value(_nullableRecoveryValue<String>(row, 'clientScopeId')),
|
|
ratingKey: Value(_requiredRecoveryValue<String>(row, 'ratingKey')),
|
|
globalKey: Value(_requiredRecoveryValue<String>(row, 'globalKey')),
|
|
actionType: Value(_requiredRecoveryValue<String>(row, 'actionType')),
|
|
viewOffset: Value(_nullableRecoveryValue<int>(row, 'viewOffset')),
|
|
duration: Value(_nullableRecoveryValue<int>(row, 'duration')),
|
|
shouldMarkWatched: Value(_requiredRecoveryValue<bool>(row, 'shouldMarkWatched')),
|
|
createdAt: Value(_requiredRecoveryValue<int>(row, 'createdAt')),
|
|
updatedAt: Value(_requiredRecoveryValue<int>(row, 'updatedAt')),
|
|
syncAttempts: Value(_requiredRecoveryValue<int>(row, 'syncAttempts')),
|
|
lastError: Value(_nullableRecoveryValue<String>(row, 'lastError')),
|
|
),
|
|
];
|
|
|
|
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();
|
|
for (final row in connectionCompanions) {
|
|
await into(connections).insert(row);
|
|
}
|
|
for (final row in profileCompanions) {
|
|
await into(profiles).insert(row);
|
|
}
|
|
for (final row in joinCompanions) {
|
|
await into(profileConnections).insert(row);
|
|
}
|
|
for (final row in pendingCompanions) {
|
|
await into(offlineWatchProgress).insert(row);
|
|
}
|
|
});
|
|
}
|
|
|
|
static List<Map<String, Object?>> _decodeRecoveryRows(
|
|
Map<String, Object?> group,
|
|
String key,
|
|
Set<String> expectedKeys,
|
|
) {
|
|
final value = group[key];
|
|
if (value is! List) throw const FormatException('Invalid tvOS database recovery image');
|
|
return [
|
|
for (final value in value)
|
|
if (value is Map<String, Object?> &&
|
|
value.keys.toSet().containsAll(expectedKeys) &&
|
|
value.length == expectedKeys.length)
|
|
value
|
|
else
|
|
throw const FormatException('Invalid tvOS database recovery image'),
|
|
];
|
|
}
|
|
|
|
static T _requiredRecoveryValue<T>(Map<String, Object?> row, String key) {
|
|
final value = row[key];
|
|
if (!row.containsKey(key) || value is! T) {
|
|
throw const FormatException('Invalid tvOS database recovery image');
|
|
}
|
|
return value;
|
|
}
|
|
|
|
static T? _nullableRecoveryValue<T>(Map<String, Object?> row, String key) {
|
|
if (!row.containsKey(key)) throw const FormatException('Invalid tvOS database recovery image');
|
|
final value = row[key];
|
|
if (value == null) return null;
|
|
if (value is! T) throw const FormatException('Invalid tvOS database recovery image');
|
|
return value as T;
|
|
}
|
|
|
|
@override
|
|
int get schemaVersion => 19;
|
|
|
|
@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('''
|
|
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
|
|
|| '/~plex-profile/'
|
|
|| owner.profile_id
|
|
|| ':'
|
|
|| substr(source.cache_key, length(metadata.server_id) + 2),
|
|
source.data,
|
|
source.pinned,
|
|
source.cached_at
|
|
FROM download_metadata_ids AS metadata
|
|
JOIN download_owners AS owner
|
|
ON owner.global_key = metadata.global_key
|
|
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
|
|
ON CONFLICT(cache_key) DO UPDATE SET
|
|
data = excluded.data,
|
|
pinned = excluded.pinned,
|
|
cached_at = excluded.cached_at
|
|
''');
|
|
// 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('''
|
|
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
|
|
|| '/~plex-transfer:'
|
|
|| substr(source.cache_key, length(metadata.server_id) + 2),
|
|
source.data,
|
|
source.pinned,
|
|
source.cached_at
|
|
FROM download_metadata_ids AS metadata
|
|
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
|
|
AND NOT EXISTS (
|
|
SELECT 1
|
|
FROM download_owners AS owner
|
|
WHERE owner.global_key = metadata.global_key
|
|
)
|
|
ON CONFLICT(cache_key) DO UPDATE SET
|
|
data = excluded.data,
|
|
pinned = excluded.pinned,
|
|
cached_at = excluded.cached_at
|
|
''');
|
|
|
|
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
|
|
''');
|
|
}
|
|
},
|
|
);
|
|
}
|
|
|
|
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> _clientScopePredicate(GeneratedColumn<String> column, String? clientScopeId) {
|
|
return clientScopeId == null ? column.isNull() : column.equals(clientScopeId);
|
|
}
|
|
|
|
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 ? _clientScopePredicate(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) &
|
|
_clientScopePredicate(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) &
|
|
_clientScopePredicate(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> 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)),
|
|
);
|
|
}
|
|
}
|
|
|
|
Future<void> updateSyncRuleCount(String globalKey, int episodeCount) async {
|
|
await (update(
|
|
syncRules,
|
|
)..where((t) => t.globalKey.equals(globalKey))).write(SyncRulesCompanion(episodeCount: Value(episodeCount)));
|
|
}
|
|
|
|
Future<void> updateSyncRuleFilter(String globalKey, String downloadFilter) async {
|
|
await (update(
|
|
syncRules,
|
|
)..where((t) => t.globalKey.equals(globalKey))).write(SyncRulesCompanion(downloadFilter: Value(downloadFilter)));
|
|
}
|
|
|
|
Future<void> updateSyncRuleEnabled(String globalKey, bool enabled) async {
|
|
await (update(
|
|
syncRules,
|
|
)..where((t) => t.globalKey.equals(globalKey))).write(SyncRulesCompanion(enabled: Value(enabled)));
|
|
}
|
|
|
|
Future<void> updateSyncRuleLastExecuted(String globalKey) async {
|
|
await (update(syncRules)..where((t) => t.globalKey.equals(globalKey))).write(
|
|
SyncRulesCompanion(lastExecutedAt: Value(DateTime.now().millisecondsSinceEpoch)),
|
|
);
|
|
}
|
|
|
|
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();
|
|
}
|
|
}
|
|
|
|
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();
|
|
}
|
|
}
|