From eeb010cd5277f84d663a82c497fdf9144a408894 Mon Sep 17 00:00:00 2001 From: zarzet <42882290+zarzet@users.noreply.github.com> Date: Sun, 18 Oct 2026 22:54:28 +0700 Subject: [PATCH] refactor(library): extract the ingestion store from library_database.dart LibraryIngestionStore holds scan and incremental ingestion writes with bodies unchanged. LibraryDatabase keeps thin delegating methods with the same public signatures and passes in its connection and the lazy History attachment. SQL text is unchanged. --- lib/services/library_database.dart | 432 ++------------------ lib/services/library_ingestion_store.dart | 472 ++++++++++++++++++++++ 2 files changed, 505 insertions(+), 399 deletions(-) create mode 100644 lib/services/library_ingestion_store.dart diff --git a/lib/services/library_database.dart b/lib/services/library_database.dart index 6b006170..d0d0dcf5 100644 --- a/lib/services/library_database.dart +++ b/lib/services/library_database.dart @@ -13,6 +13,7 @@ import 'package:spotiflac_android/services/library_cleanup.dart'; import 'package:spotiflac_android/services/library_search.dart'; import 'package:spotiflac_android/services/library_database_models.dart'; import 'package:spotiflac_android/services/library_duplicates.dart'; +import 'package:spotiflac_android/services/library_ingestion_store.dart'; import 'package:spotiflac_android/services/library_native_adapter.dart'; import 'package:spotiflac_android/services/library_row_mapper.dart'; import 'package:spotiflac_android/services/library_schema.dart'; @@ -35,13 +36,6 @@ class LibraryDatabase { static const String visibleLibraryView = LibrarySchema.visibleView; static const String searchFtsTable = LibrarySchema.searchFtsTable; static const String lookupSummaryTable = LibrarySchema.lookupSummaryTable; - static const String _scanStageTable = 'library_scan_stage'; - static const String _scanStagePathKeysTable = 'library_scan_path_keys_stage'; - static const String _incrementalStageTable = 'library_incremental_stage'; - static const String _incrementalStagePathKeysTable = - 'library_incremental_path_keys_stage'; - static const String _downloadedLibraryIdsStageTable = - 'library_downloaded_ids_stage'; // v4 records ReplayGain availability; older rows rescan once. static const int audioMetadataScanVersion = libraryAudioMetadataScanVersion; @@ -50,6 +44,10 @@ class LibraryDatabase { static final sqlite.SingleFlightInitializer _database = sqlite.SingleFlightInitializer(); bool _historyAttached = false; + late final LibraryIngestionStore _ingestion = LibraryIngestionStore( + openDatabase: () => database, + ensureHistoryAttached: _ensureHistoryAttached, + ); // null means this database connection has not attempted FTS setup yet. // false is a completed capability/setup result and must not be retried. bool? _searchFtsAvailable; @@ -87,45 +85,12 @@ class LibraryDatabase { LibraryScanNDJSONFile scan, { required String requestId, required bool Function() isCancelled, - }) async { - if (isCancelled()) throw StateError('Library scan cancelled'); - if (PlatformBridge.supportsCoreBackend) { - final db = await database; - await _ensureHistoryAttached(db); - final result = await LibraryNativeAdapter(db).importScanFile( - sourceId, - scan, - requestId: requestId, - isCancelled: isCancelled, - ); - return ( - inserted: (result['inserted'] as num).toInt(), - skipped: (result['skipped'] as num).toInt(), - ); - } - Stream> validatedRows() async* { - var count = 0; - await for (final row in scan.rows()) { - if (isCancelled()) { - throw StateError('Library scan cancelled during ingestion'); - } - count++; - yield row; - } - if (count != scan.expectedCount) { - throw FormatException( - 'Library scan row count mismatch: decoded $count, ' - 'expected ${scan.expectedCount}', - ); - } - } - - return replaceSourceStream( - sourceId, - validatedRows(), - preserveMissing: scan.errorCount > 0, - ); - } + }) => _ingestion.replaceSourceScanFile( + sourceId, + scan, + requestId: requestId, + isCancelled: isCancelled, + ); /// Exports timestamps directly from the DB on the worker. Android's adapter /// backfills zero SAF timestamps in bounded ContentResolver batches first. @@ -150,18 +115,14 @@ class LibraryDatabase { required bool isSaf, required String requestId, required bool Function() isCancelled, - }) async { - final db = await database; - await _ensureHistoryAttached(db); - return LibraryNativeAdapter(db).scanIncremental( - sourceId, - folderPath, - snapshotPath, - isSaf: isSaf, - requestId: requestId, - isCancelled: isCancelled, - ); - } + }) => _ingestion.scanIncrementalNative( + sourceId, + folderPath, + snapshotPath, + isSaf: isSaf, + requestId: requestId, + isCancelled: isCancelled, + ); Future> searchLibrary({ required String query, @@ -220,95 +181,18 @@ class LibraryDatabase { return '${normalizeLookupText(albumName)}|${normalizeLookupText(albumArtist ?? artistName)}'; } - Future upsert(Map json, {String? sourceId}) async { - final db = await database; - await sqlite.transactionWithBusyRetry(db, (txn) async { - await txn.insert( - 'library', - LibraryRowMapper.encode(json, sourceId: sourceId), - conflictAlgorithm: ConflictAlgorithm.replace, - ); - final batch = txn.batch(); - _putPathKeysInBatch( - batch, - json['id'] as String, - json['filePath'] as String?, - ); - await batch.commit(noResult: true); - }); - } + Future upsert(Map json, {String? sourceId}) => + _ingestion.upsert(json, sourceId: sourceId); Future upsertBatch( List> items, { String? sourceId, - }) async { - if (items.isEmpty) return; - final db = await database; - await sqlite.transactionWithBusyRetry(db, (txn) async { - final batch = txn.batch(); - for (final json in items) { - batch.insert( - 'library', - LibraryRowMapper.encode(json, sourceId: sourceId), - conflictAlgorithm: ConflictAlgorithm.replace, - ); - _putPathKeysInBatch( - batch, - json['id'] as String, - json['filePath'] as String?, - ); - } - await batch.commit(noResult: true); - }); - _log.i('Batch inserted ${items.length} items'); - } + }) => _ingestion.upsertBatch(items, sourceId: sourceId); /// Removes rows from one source whose normalized path keys already exist in /// download History, without materializing either database's paths in Dart. - Future deleteDownloadedRowsForSource(String sourceId) async { - final db = await database; - await _ensureHistoryAttached(db); - const downloadedMatch = ''' - EXISTS ( - SELECT 1 - FROM library_path_keys lk - JOIN history_db.history_path_keys hk ON hk.path_key = lk.path_key - WHERE lk.item_id = library.id - ) - '''; - await db.execute('DROP TABLE IF EXISTS $_downloadedLibraryIdsStageTable'); - await db.execute( - 'CREATE TEMP TABLE $_downloadedLibraryIdsStageTable ' - '(id TEXT PRIMARY KEY)', - ); - try { - await db.rawInsert( - 'INSERT INTO $_downloadedLibraryIdsStageTable(id) ' - 'SELECT id FROM library ' - 'WHERE source_id = ? AND $downloadedMatch', - [sourceId], - ); - final countRows = await db.rawQuery( - 'SELECT COUNT(*) AS count FROM $_downloadedLibraryIdsStageTable', - ); - final count = Sqflite.firstIntValue(countRows) ?? 0; - if (count == 0) return 0; - - await sqlite.transactionWithBusyRetry(db, (txn) async { - await txn.rawDelete( - 'DELETE FROM library_path_keys WHERE item_id IN ' - '(SELECT id FROM $_downloadedLibraryIdsStageTable)', - ); - await txn.rawDelete( - 'DELETE FROM library WHERE id IN ' - '(SELECT id FROM $_downloadedLibraryIdsStageTable)', - ); - }); - return count; - } finally { - await db.execute('DROP TABLE IF EXISTS $_downloadedLibraryIdsStageTable'); - } - } + Future deleteDownloadedRowsForSource(String sourceId) => + _ingestion.deleteDownloadedRowsForSource(sourceId); /// Upserts incremental scan rows after filtering them through the attached /// History path-key index. This keeps the hot incremental path independent @@ -316,137 +200,10 @@ class LibraryDatabase { Future<({int upserted, int skipped})> upsertBatchExcludingHistory( List> items, { required String sourceId, - }) async { - if (items.isEmpty) return (upserted: 0, skipped: 0); - final db = await database; - await _ensureHistoryAttached(db); - await db.execute('DROP TABLE IF EXISTS $_incrementalStageTable'); - await db.execute('DROP TABLE IF EXISTS $_incrementalStagePathKeysTable'); - await db.execute( - 'CREATE TEMP TABLE $_incrementalStageTable ' - 'AS SELECT * FROM library WHERE 0', - ); - await db.execute( - 'CREATE UNIQUE INDEX idx_${_incrementalStageTable}_id ' - 'ON $_incrementalStageTable(id)', - ); - await db.execute( - 'CREATE UNIQUE INDEX idx_${_incrementalStageTable}_path ' - 'ON $_incrementalStageTable(file_path)', - ); - await db.execute(''' - CREATE TEMP TABLE $_incrementalStagePathKeysTable ( - item_id TEXT NOT NULL, - path_key TEXT NOT NULL, - PRIMARY KEY (item_id, path_key) - ) - '''); - await db.execute( - 'CREATE INDEX idx_${_incrementalStagePathKeysTable}_key ' - 'ON $_incrementalStagePathKeysTable(path_key)', - ); + }) => _ingestion.upsertBatchExcludingHistory(items, sourceId: sourceId); - try { - final batch = db.batch(); - final stagedIds = {}; - for (final json in items) { - final id = json['id'] as String?; - if (id == null || id.trim().isEmpty) { - throw const FormatException('Library scan row has no valid id'); - } - batch.insert( - _incrementalStageTable, - LibraryRowMapper.encode(json, sourceId: sourceId), - conflictAlgorithm: ConflictAlgorithm.replace, - ); - sqlite.putPathKeysInBatch( - batch, - _incrementalStagePathKeysTable, - id, - json['filePath'] as String?, - // The stage starts empty: only a repeated id has keys to replace. - replaceExisting: !stagedIds.add(id), - ); - } - await batch.commit(noResult: true); - - const historyMatch = - ''' - EXISTS ( - SELECT 1 - FROM $_incrementalStagePathKeysTable sk - JOIN history_db.history_path_keys hk ON hk.path_key = sk.path_key - WHERE sk.item_id = s.id - ) - '''; - final skippedRows = await db.rawQuery( - 'SELECT COUNT(*) AS count FROM $_incrementalStageTable s ' - 'WHERE $historyMatch', - ); - final skipped = Sqflite.firstIntValue(skippedRows) ?? 0; - final stagedRows = await db.rawQuery( - 'SELECT COUNT(*) AS count FROM $_incrementalStageTable', - ); - final staged = Sqflite.firstIntValue(stagedRows) ?? 0; - final columns = (await db.rawQuery( - 'PRAGMA table_info(library)', - )).map((row) => row['name'] as String).toList(growable: false); - final columnList = columns.join(', '); - final selectedColumns = columns.map((column) => 's.$column').join(', '); - - await sqlite.transactionWithBusyRetry(db, (txn) async { - await txn.rawDelete(''' - DELETE FROM library_path_keys - WHERE item_id IN ( - SELECT s.id FROM $_incrementalStageTable s WHERE NOT $historyMatch - ) - '''); - await txn.rawInsert(''' - INSERT OR REPLACE INTO library ($columnList) - SELECT $selectedColumns FROM $_incrementalStageTable s - WHERE NOT $historyMatch - '''); - await txn.rawInsert(''' - INSERT OR IGNORE INTO library_path_keys(item_id, path_key) - SELECT sk.item_id, sk.path_key - FROM $_incrementalStagePathKeysTable sk - JOIN $_incrementalStageTable s ON s.id = sk.item_id - WHERE NOT $historyMatch - '''); - }); - return (upserted: staged - skipped, skipped: skipped); - } finally { - await db.execute('DROP TABLE IF EXISTS $_incrementalStagePathKeysTable'); - await db.execute('DROP TABLE IF EXISTS $_incrementalStageTable'); - } - } - - Future replaceAll(List> items) async { - final db = await database; - await sqlite.transactionWithBusyRetry(db, (txn) async { - await txn.delete('library_path_keys'); - await txn.delete('library'); - if (items.isEmpty) { - return; - } - - final batch = txn.batch(); - for (final json in items) { - batch.insert( - 'library', - LibraryRowMapper.encode(json), - conflictAlgorithm: ConflictAlgorithm.replace, - ); - _putPathKeysInBatch( - batch, - json['id'] as String, - json['filePath'] as String?, - ); - } - await batch.commit(noResult: true); - }); - _log.i('Replaced library with ${items.length} items'); - } + Future replaceAll(List> items) => + _ingestion.replaceAll(items); /// Stages scan rows in bounded, independently committed batches, then swaps /// only this source in one short transaction. Download-history exclusion is @@ -456,135 +213,12 @@ class LibraryDatabase { Stream> items, { int batchSize = 300, bool preserveMissing = false, - }) async { - if (batchSize <= 0) { - throw ArgumentError.value(batchSize, 'batchSize', 'Must be positive'); - } - final db = await database; - await _ensureHistoryAttached(db); - await db.execute('DROP TABLE IF EXISTS $_scanStageTable'); - await db.execute('DROP TABLE IF EXISTS $_scanStagePathKeysTable'); - await db.execute( - 'CREATE TEMP TABLE $_scanStageTable AS SELECT * FROM library WHERE 0', - ); - await db.execute( - 'CREATE UNIQUE INDEX idx_${_scanStageTable}_id ' - 'ON $_scanStageTable(id)', - ); - await db.execute( - 'CREATE UNIQUE INDEX idx_${_scanStageTable}_path ' - 'ON $_scanStageTable(file_path)', - ); - await db.execute(''' - CREATE TEMP TABLE $_scanStagePathKeysTable ( - item_id TEXT NOT NULL, - path_key TEXT NOT NULL, - PRIMARY KEY (item_id, path_key) - ) - '''); - await db.execute( - 'CREATE INDEX idx_${_scanStagePathKeysTable}_key ' - 'ON $_scanStagePathKeysTable(path_key)', - ); - - var streamed = 0; - try { - var batch = db.batch(); - var pending = 0; - Future flush() async { - if (pending == 0) return; - await batch.commit(noResult: true); - batch = db.batch(); - pending = 0; - } - - final stagedIds = {}; - await for (final json in items) { - final id = json['id'] as String?; - if (id == null || id.trim().isEmpty) { - throw const FormatException('Library scan row has no valid id'); - } - batch.insert( - _scanStageTable, - LibraryRowMapper.encode(json, sourceId: sourceId), - conflictAlgorithm: ConflictAlgorithm.replace, - ); - sqlite.putPathKeysInBatch( - batch, - _scanStagePathKeysTable, - id, - json['filePath'] as String?, - // The stage starts empty: only a repeated id has keys to replace. - replaceExisting: !stagedIds.add(id), - ); - streamed++; - pending++; - if (pending >= batchSize) await flush(); - } - await flush(); - - const historyMatch = - ''' - EXISTS ( - SELECT 1 - FROM $_scanStagePathKeysTable sk - JOIN history_db.history_path_keys hk ON hk.path_key = sk.path_key - WHERE sk.item_id = s.id - ) - '''; - final skippedRows = await db.rawQuery( - 'SELECT COUNT(*) AS count FROM $_scanStageTable s ' - 'WHERE $historyMatch', - ); - final skipped = Sqflite.firstIntValue(skippedRows) ?? 0; - final stagedRows = await db.rawQuery( - 'SELECT COUNT(*) AS count FROM $_scanStageTable', - ); - final staged = Sqflite.firstIntValue(stagedRows) ?? 0; - final columns = (await db.rawQuery( - 'PRAGMA table_info(library)', - )).map((row) => row['name'] as String).toList(growable: false); - final columnList = columns.join(', '); - final selectedColumns = columns.map((column) => 's.$column').join(', '); - - await sqlite.transactionWithBusyRetry(db, (txn) async { - await deleteReplacedLibraryScanRows( - txn, - sourceId, - stageTable: _scanStageTable, - preserveMissing: preserveMissing, - ); - await txn.rawDelete(''' - DELETE FROM library_path_keys - WHERE item_id IN ( - SELECT s.id FROM $_scanStageTable s WHERE NOT $historyMatch - ) - '''); - await txn.rawInsert(''' - INSERT OR REPLACE INTO library ($columnList) - SELECT $selectedColumns FROM $_scanStageTable s - WHERE NOT $historyMatch - '''); - await txn.rawInsert(''' - INSERT OR IGNORE INTO library_path_keys(item_id, path_key) - SELECT sk.item_id, sk.path_key - FROM $_scanStagePathKeysTable sk - JOIN $_scanStageTable s ON s.id = sk.item_id - WHERE NOT $historyMatch - '''); - }); - final inserted = staged - skipped; - _log.i( - 'Streamed $streamed rows, staged $staged unique rows, and swapped ' - '$inserted into Library source ' - '$sourceId ($skipped downloads excluded)', - ); - return (inserted: inserted, skipped: skipped); - } finally { - await db.execute('DROP TABLE IF EXISTS $_scanStagePathKeysTable'); - await db.execute('DROP TABLE IF EXISTS $_scanStageTable'); - } - } + }) => _ingestion.replaceSourceStream( + sourceId, + items, + batchSize: batchSize, + preserveMissing: preserveMissing, + ); Future> getSources() async { final db = await database; diff --git a/lib/services/library_ingestion_store.dart b/lib/services/library_ingestion_store.dart new file mode 100644 index 00000000..46764a53 --- /dev/null +++ b/lib/services/library_ingestion_store.dart @@ -0,0 +1,472 @@ +import 'package:sqflite/sqflite.dart'; +import 'package:spotiflac_android/services/library_cleanup.dart'; +import 'package:spotiflac_android/services/library_native_adapter.dart'; +import 'package:spotiflac_android/services/library_row_mapper.dart'; +import 'package:spotiflac_android/services/platform_bridge.dart'; +import 'package:spotiflac_android/services/sqlite_helpers.dart' as sqlite; +import 'package:spotiflac_android/utils/logger.dart'; + +final _log = AppLogger('LibraryDatabase'); + +/// Scan and incremental ingestion writes. The connection and the lazy +/// History attachment stay owned by LibraryDatabase and are passed in. +class LibraryIngestionStore { + static const String _scanStageTable = 'library_scan_stage'; + static const String _scanStagePathKeysTable = 'library_scan_path_keys_stage'; + static const String _incrementalStageTable = 'library_incremental_stage'; + static const String _incrementalStagePathKeysTable = + 'library_incremental_path_keys_stage'; + static const String _downloadedLibraryIdsStageTable = + 'library_downloaded_ids_stage'; + + final Future Function() _openDatabase; + final Future Function(Database db) _attachHistory; + + LibraryIngestionStore({ + required Future Function() openDatabase, + required Future Function(Database db) ensureHistoryAttached, + }) : _openDatabase = openDatabase, + _attachHistory = ensureHistoryAttached; + + Future get database => _openDatabase(); + + Future _ensureHistoryAttached(Database db) => _attachHistory(db); + + /// Keeps native scan output native through parsing, indexing and the atomic + /// source swap. The stream implementation remains the desktop/test adapter. + Future<({int inserted, int skipped})> replaceSourceScanFile( + String sourceId, + LibraryScanNDJSONFile scan, { + required String requestId, + required bool Function() isCancelled, + }) async { + if (isCancelled()) throw StateError('Library scan cancelled'); + if (PlatformBridge.supportsCoreBackend) { + final db = await database; + await _ensureHistoryAttached(db); + final result = await LibraryNativeAdapter(db).importScanFile( + sourceId, + scan, + requestId: requestId, + isCancelled: isCancelled, + ); + return ( + inserted: (result['inserted'] as num).toInt(), + skipped: (result['skipped'] as num).toInt(), + ); + } + Stream> validatedRows() async* { + var count = 0; + await for (final row in scan.rows()) { + if (isCancelled()) { + throw StateError('Library scan cancelled during ingestion'); + } + count++; + yield row; + } + if (count != scan.expectedCount) { + throw FormatException( + 'Library scan row count mismatch: decoded $count, ' + 'expected ${scan.expectedCount}', + ); + } + } + + return replaceSourceStream( + sourceId, + validatedRows(), + preserveMissing: scan.errorCount > 0, + ); + } + + /// Native folder scanning and delta ingestion share one operation, so delta + /// track maps never return to Flutter. SAF traversal stays in its OS adapter. + Future> scanIncrementalNative( + String sourceId, + String folderPath, + String snapshotPath, { + required bool isSaf, + required String requestId, + required bool Function() isCancelled, + }) async { + final db = await database; + await _ensureHistoryAttached(db); + return LibraryNativeAdapter(db).scanIncremental( + sourceId, + folderPath, + snapshotPath, + isSaf: isSaf, + requestId: requestId, + isCancelled: isCancelled, + ); + } + + void _putPathKeysInBatch(Batch batch, String id, String? filePath) => + sqlite.putPathKeysInBatch(batch, 'library_path_keys', id, filePath); + + Future upsert(Map json, {String? sourceId}) async { + final db = await database; + await sqlite.transactionWithBusyRetry(db, (txn) async { + await txn.insert( + 'library', + LibraryRowMapper.encode(json, sourceId: sourceId), + conflictAlgorithm: ConflictAlgorithm.replace, + ); + final batch = txn.batch(); + _putPathKeysInBatch( + batch, + json['id'] as String, + json['filePath'] as String?, + ); + await batch.commit(noResult: true); + }); + } + + Future upsertBatch( + List> items, { + String? sourceId, + }) async { + if (items.isEmpty) return; + final db = await database; + await sqlite.transactionWithBusyRetry(db, (txn) async { + final batch = txn.batch(); + for (final json in items) { + batch.insert( + 'library', + LibraryRowMapper.encode(json, sourceId: sourceId), + conflictAlgorithm: ConflictAlgorithm.replace, + ); + _putPathKeysInBatch( + batch, + json['id'] as String, + json['filePath'] as String?, + ); + } + await batch.commit(noResult: true); + }); + _log.i('Batch inserted ${items.length} items'); + } + + /// Removes rows from one source whose normalized path keys already exist in + /// download History, without materializing either database's paths in Dart. + Future deleteDownloadedRowsForSource(String sourceId) async { + final db = await database; + await _ensureHistoryAttached(db); + const downloadedMatch = ''' + EXISTS ( + SELECT 1 + FROM library_path_keys lk + JOIN history_db.history_path_keys hk ON hk.path_key = lk.path_key + WHERE lk.item_id = library.id + ) + '''; + await db.execute('DROP TABLE IF EXISTS $_downloadedLibraryIdsStageTable'); + await db.execute( + 'CREATE TEMP TABLE $_downloadedLibraryIdsStageTable ' + '(id TEXT PRIMARY KEY)', + ); + try { + await db.rawInsert( + 'INSERT INTO $_downloadedLibraryIdsStageTable(id) ' + 'SELECT id FROM library ' + 'WHERE source_id = ? AND $downloadedMatch', + [sourceId], + ); + final countRows = await db.rawQuery( + 'SELECT COUNT(*) AS count FROM $_downloadedLibraryIdsStageTable', + ); + final count = Sqflite.firstIntValue(countRows) ?? 0; + if (count == 0) return 0; + + await sqlite.transactionWithBusyRetry(db, (txn) async { + await txn.rawDelete( + 'DELETE FROM library_path_keys WHERE item_id IN ' + '(SELECT id FROM $_downloadedLibraryIdsStageTable)', + ); + await txn.rawDelete( + 'DELETE FROM library WHERE id IN ' + '(SELECT id FROM $_downloadedLibraryIdsStageTable)', + ); + }); + return count; + } finally { + await db.execute('DROP TABLE IF EXISTS $_downloadedLibraryIdsStageTable'); + } + } + + /// Upserts incremental scan rows after filtering them through the attached + /// History path-key index. This keeps the hot incremental path independent + /// of total History size. + Future<({int upserted, int skipped})> upsertBatchExcludingHistory( + List> items, { + required String sourceId, + }) async { + if (items.isEmpty) return (upserted: 0, skipped: 0); + final db = await database; + await _ensureHistoryAttached(db); + await db.execute('DROP TABLE IF EXISTS $_incrementalStageTable'); + await db.execute('DROP TABLE IF EXISTS $_incrementalStagePathKeysTable'); + await db.execute( + 'CREATE TEMP TABLE $_incrementalStageTable ' + 'AS SELECT * FROM library WHERE 0', + ); + await db.execute( + 'CREATE UNIQUE INDEX idx_${_incrementalStageTable}_id ' + 'ON $_incrementalStageTable(id)', + ); + await db.execute( + 'CREATE UNIQUE INDEX idx_${_incrementalStageTable}_path ' + 'ON $_incrementalStageTable(file_path)', + ); + await db.execute(''' + CREATE TEMP TABLE $_incrementalStagePathKeysTable ( + item_id TEXT NOT NULL, + path_key TEXT NOT NULL, + PRIMARY KEY (item_id, path_key) + ) + '''); + await db.execute( + 'CREATE INDEX idx_${_incrementalStagePathKeysTable}_key ' + 'ON $_incrementalStagePathKeysTable(path_key)', + ); + + try { + final batch = db.batch(); + final stagedIds = {}; + for (final json in items) { + final id = json['id'] as String?; + if (id == null || id.trim().isEmpty) { + throw const FormatException('Library scan row has no valid id'); + } + batch.insert( + _incrementalStageTable, + LibraryRowMapper.encode(json, sourceId: sourceId), + conflictAlgorithm: ConflictAlgorithm.replace, + ); + sqlite.putPathKeysInBatch( + batch, + _incrementalStagePathKeysTable, + id, + json['filePath'] as String?, + // The stage starts empty: only a repeated id has keys to replace. + replaceExisting: !stagedIds.add(id), + ); + } + await batch.commit(noResult: true); + + const historyMatch = + ''' + EXISTS ( + SELECT 1 + FROM $_incrementalStagePathKeysTable sk + JOIN history_db.history_path_keys hk ON hk.path_key = sk.path_key + WHERE sk.item_id = s.id + ) + '''; + final skippedRows = await db.rawQuery( + 'SELECT COUNT(*) AS count FROM $_incrementalStageTable s ' + 'WHERE $historyMatch', + ); + final skipped = Sqflite.firstIntValue(skippedRows) ?? 0; + final stagedRows = await db.rawQuery( + 'SELECT COUNT(*) AS count FROM $_incrementalStageTable', + ); + final staged = Sqflite.firstIntValue(stagedRows) ?? 0; + final columns = (await db.rawQuery( + 'PRAGMA table_info(library)', + )).map((row) => row['name'] as String).toList(growable: false); + final columnList = columns.join(', '); + final selectedColumns = columns.map((column) => 's.$column').join(', '); + + await sqlite.transactionWithBusyRetry(db, (txn) async { + await txn.rawDelete(''' + DELETE FROM library_path_keys + WHERE item_id IN ( + SELECT s.id FROM $_incrementalStageTable s WHERE NOT $historyMatch + ) + '''); + await txn.rawInsert(''' + INSERT OR REPLACE INTO library ($columnList) + SELECT $selectedColumns FROM $_incrementalStageTable s + WHERE NOT $historyMatch + '''); + await txn.rawInsert(''' + INSERT OR IGNORE INTO library_path_keys(item_id, path_key) + SELECT sk.item_id, sk.path_key + FROM $_incrementalStagePathKeysTable sk + JOIN $_incrementalStageTable s ON s.id = sk.item_id + WHERE NOT $historyMatch + '''); + }); + return (upserted: staged - skipped, skipped: skipped); + } finally { + await db.execute('DROP TABLE IF EXISTS $_incrementalStagePathKeysTable'); + await db.execute('DROP TABLE IF EXISTS $_incrementalStageTable'); + } + } + + Future replaceAll(List> items) async { + final db = await database; + await sqlite.transactionWithBusyRetry(db, (txn) async { + await txn.delete('library_path_keys'); + await txn.delete('library'); + if (items.isEmpty) { + return; + } + + final batch = txn.batch(); + for (final json in items) { + batch.insert( + 'library', + LibraryRowMapper.encode(json), + conflictAlgorithm: ConflictAlgorithm.replace, + ); + _putPathKeysInBatch( + batch, + json['id'] as String, + json['filePath'] as String?, + ); + } + await batch.commit(noResult: true); + }); + _log.i('Replaced library with ${items.length} items'); + } + + /// Stages scan rows in bounded, independently committed batches, then swaps + /// only this source in one short transaction. Download-history exclusion is + /// an indexed SQLite anti-join, avoiding a full History path set in Dart. + Future<({int inserted, int skipped})> replaceSourceStream( + String sourceId, + Stream> items, { + int batchSize = 300, + bool preserveMissing = false, + }) async { + if (batchSize <= 0) { + throw ArgumentError.value(batchSize, 'batchSize', 'Must be positive'); + } + final db = await database; + await _ensureHistoryAttached(db); + await db.execute('DROP TABLE IF EXISTS $_scanStageTable'); + await db.execute('DROP TABLE IF EXISTS $_scanStagePathKeysTable'); + await db.execute( + 'CREATE TEMP TABLE $_scanStageTable AS SELECT * FROM library WHERE 0', + ); + await db.execute( + 'CREATE UNIQUE INDEX idx_${_scanStageTable}_id ' + 'ON $_scanStageTable(id)', + ); + await db.execute( + 'CREATE UNIQUE INDEX idx_${_scanStageTable}_path ' + 'ON $_scanStageTable(file_path)', + ); + await db.execute(''' + CREATE TEMP TABLE $_scanStagePathKeysTable ( + item_id TEXT NOT NULL, + path_key TEXT NOT NULL, + PRIMARY KEY (item_id, path_key) + ) + '''); + await db.execute( + 'CREATE INDEX idx_${_scanStagePathKeysTable}_key ' + 'ON $_scanStagePathKeysTable(path_key)', + ); + + var streamed = 0; + try { + var batch = db.batch(); + var pending = 0; + Future flush() async { + if (pending == 0) return; + await batch.commit(noResult: true); + batch = db.batch(); + pending = 0; + } + + final stagedIds = {}; + await for (final json in items) { + final id = json['id'] as String?; + if (id == null || id.trim().isEmpty) { + throw const FormatException('Library scan row has no valid id'); + } + batch.insert( + _scanStageTable, + LibraryRowMapper.encode(json, sourceId: sourceId), + conflictAlgorithm: ConflictAlgorithm.replace, + ); + sqlite.putPathKeysInBatch( + batch, + _scanStagePathKeysTable, + id, + json['filePath'] as String?, + // The stage starts empty: only a repeated id has keys to replace. + replaceExisting: !stagedIds.add(id), + ); + streamed++; + pending++; + if (pending >= batchSize) await flush(); + } + await flush(); + + const historyMatch = + ''' + EXISTS ( + SELECT 1 + FROM $_scanStagePathKeysTable sk + JOIN history_db.history_path_keys hk ON hk.path_key = sk.path_key + WHERE sk.item_id = s.id + ) + '''; + final skippedRows = await db.rawQuery( + 'SELECT COUNT(*) AS count FROM $_scanStageTable s ' + 'WHERE $historyMatch', + ); + final skipped = Sqflite.firstIntValue(skippedRows) ?? 0; + final stagedRows = await db.rawQuery( + 'SELECT COUNT(*) AS count FROM $_scanStageTable', + ); + final staged = Sqflite.firstIntValue(stagedRows) ?? 0; + final columns = (await db.rawQuery( + 'PRAGMA table_info(library)', + )).map((row) => row['name'] as String).toList(growable: false); + final columnList = columns.join(', '); + final selectedColumns = columns.map((column) => 's.$column').join(', '); + + await sqlite.transactionWithBusyRetry(db, (txn) async { + await deleteReplacedLibraryScanRows( + txn, + sourceId, + stageTable: _scanStageTable, + preserveMissing: preserveMissing, + ); + await txn.rawDelete(''' + DELETE FROM library_path_keys + WHERE item_id IN ( + SELECT s.id FROM $_scanStageTable s WHERE NOT $historyMatch + ) + '''); + await txn.rawInsert(''' + INSERT OR REPLACE INTO library ($columnList) + SELECT $selectedColumns FROM $_scanStageTable s + WHERE NOT $historyMatch + '''); + await txn.rawInsert(''' + INSERT OR IGNORE INTO library_path_keys(item_id, path_key) + SELECT sk.item_id, sk.path_key + FROM $_scanStagePathKeysTable sk + JOIN $_scanStageTable s ON s.id = sk.item_id + WHERE NOT $historyMatch + '''); + }); + final inserted = staged - skipped; + _log.i( + 'Streamed $streamed rows, staged $staged unique rows, and swapped ' + '$inserted into Library source ' + '$sourceId ($skipped downloads excluded)', + ); + return (inserted: inserted, skipped: skipped); + } finally { + await db.execute('DROP TABLE IF EXISTS $_scanStagePathKeysTable'); + await db.execute('DROP TABLE IF EXISTS $_scanStageTable'); + } + } +}