From 460f8ebc2d82c679a82aebc89e7ba625ed881506 Mon Sep 17 00:00:00 2001 From: zarzet <42882290+zarzet@users.noreply.github.com> Date: Thu, 8 Oct 2026 03:19:45 +0700 Subject: [PATCH] perf(collections): stage large playlist additions off the UI thread --- .../playlist_batch_write_test.dart | 259 ++++++++++++++++++ .../library_collections_provider.dart | 89 +++--- .../library_collections_database.dart | 40 ++- lib/services/playlist_batch_writer.dart | 201 ++++++++++++++ rust_backend/crates/mobile/src/data_jobs.rs | 3 + rust_backend/crates/mobile/src/lib.rs | 1 + .../mobile/src/native_playlist_write.rs | 210 ++++++++++++++ test/library_collections_mutation_test.dart | 72 +++++ test/playlist_batch_benchmark_test.dart | 59 ++++ test/playlist_batch_publication_test.dart | 210 ++++++++++++++ test/playlist_batch_writer_test.dart | 132 +++++++++ test/support/playlist_batch_benchmark.dart | 61 +++++ 12 files changed, 1284 insertions(+), 53 deletions(-) create mode 100644 integration_test/playlist_batch_write_test.dart create mode 100644 lib/services/playlist_batch_writer.dart create mode 100644 rust_backend/crates/mobile/src/native_playlist_write.rs create mode 100644 test/playlist_batch_benchmark_test.dart create mode 100644 test/playlist_batch_publication_test.dart create mode 100644 test/playlist_batch_writer_test.dart create mode 100644 test/support/playlist_batch_benchmark.dart diff --git a/integration_test/playlist_batch_write_test.dart b/integration_test/playlist_batch_write_test.dart new file mode 100644 index 00000000..25f45043 --- /dev/null +++ b/integration_test/playlist_batch_write_test.dart @@ -0,0 +1,259 @@ +import 'dart:convert'; +import 'dart:io'; +import 'dart:isolate'; + +import 'package:crypto/crypto.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:integration_test/integration_test.dart'; +import 'package:sqflite/sqflite.dart'; +import 'package:spotiflac_android/models/library_collections.dart'; +import 'package:spotiflac_android/services/library_collections_database.dart'; +import 'package:spotiflac_android/services/platform_bridge.dart'; +import 'package:spotiflac_android/services/playlist_batch_writer.dart'; + +import '../test/support/library_collections_benchmark.dart'; +import '../test/support/playlist_batch_benchmark.dart'; + +Future _digest(Object value) => Isolate.run( + () => sha256.convert(utf8.encode(jsonEncode(value))).toString(), +); + +void main() { + final binding = IntegrationTestWidgetsFlutterBinding.ensureInitialized(); + final report = { + 'platform': Platform.operatingSystem, + 'note': 'Five alternating paired samples. Heartbeat gaps are not FPS.', + }; + late Directory root; + late Database db; + final now = DateTime.utc(2026, 10, 8); + final stamp = now.toIso8601String(); + var completed = 0; + + setUp(() async { + root = await Directory.systemTemp.createTemp('playlist-device-'); + final extensions = await Directory('${root.path}/extensions').create(); + final data = await Directory('${root.path}/data').create(); + final output = await Directory('${root.path}/output').create(); + await PlatformBridge.initExtensionSystem( + extensions.path, + data.path, + masterKey: base64Encode(List.filled(32, 1)), + lyricsProviders: const [], + lyricsFetchOptions: const {}, + allowedDirectories: [output.path], + ); + final implementations = await PlatformBridge.getBackendImplementations(); + expect(implementations['extensions'], 'rust'); + db = await openDatabase( + '${root.path}/library_collections.db', + version: 2, + onConfigure: (db) async { + await db.execute('PRAGMA foreign_keys=ON'); + await db.execute('PRAGMA recursive_triggers=ON'); + await db.execute('PRAGMA auto_vacuum=INCREMENTAL'); + await db.rawQuery('PRAGMA journal_mode=WAL'); + await db.execute('PRAGMA synchronous=NORMAL'); + await db.rawQuery('PRAGMA busy_timeout=5000'); + }, + onCreate: (db, _) async { + await db.execute( + 'CREATE TABLE playlists(id TEXT PRIMARY KEY,name TEXT NOT NULL,cover_image_path TEXT,created_at TEXT NOT NULL,updated_at TEXT NOT NULL)', + ); + await db.execute( + 'CREATE TABLE playlist_tracks(playlist_id TEXT NOT NULL,track_key TEXT NOT NULL,track_json TEXT NOT NULL,added_at TEXT NOT NULL,PRIMARY KEY(playlist_id,track_key),FOREIGN KEY(playlist_id) REFERENCES playlists(id) ON DELETE CASCADE)', + ); + await db.execute( + 'CREATE INDEX idx_playlist_tracks_playlist_id ON playlist_tracks(playlist_id)', + ); + for (final id in ['p', 'other']) { + await db.insert('playlists', { + 'id': id, + 'name': id, + 'created_at': 'old', + 'updated_at': 'old', + }); + } + }, + ); + }); + + tearDown(() async { + await db.close(); + await PlatformBridge.cleanupExtensions(); + await root.delete(recursive: true); + }); + + tearDownAll(() { + report['completed_cases'] = completed; + binding.reportData = {'playlist_batch_write': report}; + }); + + Future publish(PreparedPlaylistBatch batch, {String id = 'p'}) async { + if (batch.rowsPath == null) { + await LibraryCollectionsDatabase.writeDatabasePlaylistTracks( + db, + playlistId: id, + playlistUpdatedAt: stamp, + tracks: batch.rows, + ); + } else { + await publishPlaylistBatch( + database: db, + playlistId: id, + updatedAt: stamp, + rowsPath: batch.rowsPath!, + expectedCount: batch.entries.length, + temporaryDirectory: root, + ); + } + } + + // Preserve the pre-optimization single large sqflite batch as the baseline. + Future originalWrite(List> rows) => + db.transaction((txn) async { + final batch = txn.batch(); + for (final row in rows) { + batch.insert('playlist_tracks', { + 'playlist_id': 'p', + ...row, + }, conflictAlgorithm: ConflictAlgorithm.replace); + } + batch.update( + 'playlists', + {'updated_at': stamp}, + where: 'id=?', + whereArgs: ['p'], + ); + await batch.commit(noResult: true); + }); + + for (final count in [20, 256, 1024, 2048, 12000]) { + testWidgets('playlist publication parity and paired timing ($count)', ( + tester, + ) async { + final unique = await playlistBatchFixture(count); + final input = [...unique, unique.first, unique.last]; + final expected = inlinePlaylistBatch(input, const {}, now); + final expectedDigest = await _digest(expected.rows); + final baseline = >[]; + final optimized = >[]; + final playlist = UserPlaylistCollection( + id: 'p', + name: 'p', + createdAt: now, + updatedAt: now, + tracks: const [], + ); + for (var sample = 0; sample < 5; sample++) { + for (final candidate in sample.isEven ? [false, true] : [true, false]) { + await db.delete('playlist_tracks'); + await db.update('playlists', {'updated_at': 'old'}); + final result = await measureCollectionOperation(() async { + late List additions; + if (candidate) { + final batch = await preparePlaylistBatch( + tracks: input, + existingKeys: playlist.trackKeys, + addedAt: now, + temporaryDirectory: root, + ); + try { + expect(batch.duplicates, 2); + expect(batch.rowsPath != null, count >= 2048); + await publish(batch); + additions = batch.entries; + } finally { + await batch.dispose(); + } + } else { + final batch = inlinePlaylistBatch(input, playlist.trackKeys, now); + await originalWrite(batch.rows); + additions = batch.entries; + } + // Include the same model publication in both timing paths. + return playlist.copyWith( + tracks: [...playlist.tracks, ...additions], + updatedAt: now, + ); + }); + expect(result.value.trackKeys.length, count); + expect( + result.value.tracks.map((entry) => entry.key), + expected.entries.map((entry) => entry.key), + ); + final persisted = await db.rawQuery( + 'SELECT track_key,track_json,added_at FROM playlist_tracks ORDER BY added_at ASC,rowid ASC', + ); + expect(await _digest(persisted), expectedDigest); + expect(await db.rawQuery('PRAGMA foreign_key_check'), isEmpty); + expect((await db.rawQuery('PRAGMA database_list')).length, 1); + expect( + await db.rawQuery( + 'SELECT id,updated_at FROM playlists ORDER BY id', + ), + [ + {'id': 'other', 'updated_at': 'old'}, + {'id': 'p', 'updated_at': stamp}, + ], + ); + expect( + await root + .list() + .where( + (entry) => + entry.path.contains('playlist-input-') || + entry.path.contains('playlist-stage-'), + ) + .isEmpty, + isTrue, + ); + (candidate ? optimized : baseline).add(result.timing); + } + } + report['$count'] = {'baseline': baseline, 'optimized': optimized}; + completed++; + binding.reportData = {'playlist_batch_write': report}; + }); + } + + for (final count in [300, 2100]) { + testWidgets('rollback and recovery through actual SQLite ($count)', ( + tester, + ) async { + final batch = await preparePlaylistBatch( + tracks: await playlistBatchFixture(count), + existingKeys: const {}, + addedAt: now, + temporaryDirectory: root, + ); + try { + await db.execute( + "CREATE TRIGGER reject_update BEFORE UPDATE ON playlists BEGIN SELECT RAISE(ABORT,'fixture rollback'); END", + ); + await expectLater(publish(batch), throwsA(isA())); + expect(await db.rawQuery('SELECT * FROM playlist_tracks'), isEmpty); + expect((await db.rawQuery('PRAGMA database_list')).length, 1); + await db.execute('DROP TRIGGER reject_update'); + await expectLater( + publish(batch, id: 'missing'), + throwsA(isA()), + ); + expect(await db.rawQuery('SELECT * FROM playlist_tracks'), isEmpty); + await publish(batch); + expect( + Sqflite.firstIntValue( + await db.rawQuery('SELECT COUNT(*) FROM playlist_tracks'), + ), + count, + ); + expect(await db.rawQuery('PRAGMA integrity_check'), [ + {'integrity_check': 'ok'}, + ]); + completed++; + } finally { + await batch.dispose(); + } + }); + } +} diff --git a/lib/providers/library_collections_provider.dart b/lib/providers/library_collections_provider.dart index 7841a8c0..7a88bcd0 100644 --- a/lib/providers/library_collections_provider.dart +++ b/lib/providers/library_collections_provider.dart @@ -9,6 +9,7 @@ import 'package:spotiflac_android/models/track.dart'; import 'package:spotiflac_android/services/ffmpeg_service.dart'; import 'package:spotiflac_android/services/library_collections_database.dart'; import 'package:spotiflac_android/services/library_collections_hydration.dart'; +import 'package:spotiflac_android/services/playlist_batch_writer.dart'; import 'package:spotiflac_android/utils/logger.dart'; export 'package:spotiflac_android/models/library_collections.dart'; @@ -421,60 +422,48 @@ class LibraryCollectionsNotifier extends Notifier { } final now = DateTime.now(); - final knownKeys = {...playlist.trackKeys}; - final entriesToAdd = []; - var alreadyInPlaylistCount = 0; - - for (final track in tracks) { - final key = trackCollectionKey(track); - if (!knownKeys.add(key)) { - alreadyInPlaylistCount++; - continue; + final prepared = await preparePlaylistBatch( + tracks: tracks, + existingKeys: playlist.trackKeys, + addedAt: now, + ); + try { + if (!ref.mounted || prepared.entries.isEmpty) { + return PlaylistAddBatchResult( + addedCount: 0, + alreadyInPlaylistCount: prepared.duplicates, + ); } - - entriesToAdd.add( - CollectionTrackEntry(key: key, track: track, addedAt: now), - ); - } - - if (entriesToAdd.isEmpty) { + final rowsPath = prepared.rowsPath; + if (rowsPath != null) { + await _db.upsertPlaylistTracksFile( + playlistId: playlistId, + playlistUpdatedAt: now.toIso8601String(), + rowsPath: rowsPath, + expectedCount: prepared.entries.length, + ); + } else { + await _db.upsertPlaylistTracksBatch( + playlistId: playlistId, + playlistUpdatedAt: now.toIso8601String(), + tracks: prepared.rows, + ); + } + final changed = _replacePlaylistById(playlistId, (current) { + return current.copyWith( + // Append in playlist order, matching the ASC snapshot ordering. + tracks: [...current.tracks, ...prepared.entries], + updatedAt: now, + ); + }); + if (changed) _invalidatePlaylistPickerSummaries(); return PlaylistAddBatchResult( - addedCount: 0, - alreadyInPlaylistCount: alreadyInPlaylistCount, + addedCount: changed ? prepared.entries.length : 0, + alreadyInPlaylistCount: prepared.duplicates, ); + } finally { + await prepared.dispose(); } - - await _db.upsertPlaylistTracksBatch( - playlistId: playlistId, - playlistUpdatedAt: now.toIso8601String(), - tracks: entriesToAdd - .map( - (entry) => { - 'track_key': entry.key, - 'track_json': jsonEncode(entry.track.toJson()), - 'added_at': entry.addedAt.toIso8601String(), - }, - ) - .toList(growable: false), - ); - final changed = _replacePlaylistById(playlistId, (current) { - return current.copyWith( - // Append in playlist order, matching the ASC snapshot ordering. - tracks: [...current.tracks, ...entriesToAdd], - updatedAt: now, - ); - }); - if (!changed) { - return PlaylistAddBatchResult( - addedCount: 0, - alreadyInPlaylistCount: alreadyInPlaylistCount, - ); - } - _invalidatePlaylistPickerSummaries(); - return PlaylistAddBatchResult( - addedCount: entriesToAdd.length, - alreadyInPlaylistCount: alreadyInPlaylistCount, - ); }); Future removeTrackFromPlaylist(String playlistId, String trackKey) => diff --git a/lib/services/library_collections_database.dart b/lib/services/library_collections_database.dart index 057ce0dd..cacb80ed 100644 --- a/lib/services/library_collections_database.dart +++ b/lib/services/library_collections_database.dart @@ -8,6 +8,7 @@ import 'package:sqflite/sqflite.dart'; import 'package:spotiflac_android/models/library_collections.dart'; import 'package:spotiflac_android/services/collection_restore_codec.dart'; import 'package:spotiflac_android/services/platform_bridge.dart'; +import 'package:spotiflac_android/services/playlist_batch_writer.dart'; import 'package:spotiflac_android/services/sqlite_helpers.dart' as sqlite; import 'package:spotiflac_android/services/sqlite_native_snapshot.dart'; import 'package:spotiflac_android/utils/logger.dart'; @@ -725,9 +726,24 @@ class LibraryCollectionsDatabase { required List> tracks, }) async { if (tracks.isEmpty) return; - final db = await database; - await db.transaction((txn) async { - final batch = txn.batch(); + await writeDatabasePlaylistTracks( + await database, + playlistId: playlistId, + playlistUpdatedAt: playlistUpdatedAt, + tracks: tracks, + ); + } + + static Future writeDatabasePlaylistTracks( + Database db, { + required String playlistId, + required String playlistUpdatedAt, + required List> tracks, + }) async { + if (tracks.isEmpty) return; + await sqlite.transactionWithBusyRetry(db, (txn) async { + var batch = txn.batch(); + var pending = 0; for (final track in tracks) { batch.insert(_tablePlaylistTracks, { 'playlist_id': playlistId, @@ -735,6 +751,11 @@ class LibraryCollectionsDatabase { 'track_json': track['track_json'], 'added_at': track['added_at'], }, conflictAlgorithm: ConflictAlgorithm.replace); + if (++pending == 128) { + await batch.commit(noResult: true); + batch = txn.batch(); + pending = 0; + } } batch.update( _tablePlaylists, @@ -746,6 +767,19 @@ class LibraryCollectionsDatabase { }); } + Future upsertPlaylistTracksFile({ + required String playlistId, + required String playlistUpdatedAt, + required String rowsPath, + required int expectedCount, + }) async => publishPlaylistBatch( + database: await database, + playlistId: playlistId, + updatedAt: playlistUpdatedAt, + rowsPath: rowsPath, + expectedCount: expectedCount, + ); + Future deletePlaylistTrack({ required String playlistId, required String trackKey, diff --git a/lib/services/playlist_batch_writer.dart b/lib/services/playlist_batch_writer.dart new file mode 100644 index 00000000..ef426d5b --- /dev/null +++ b/lib/services/playlist_batch_writer.dart @@ -0,0 +1,201 @@ +import 'dart:convert'; +import 'dart:io'; +import 'dart:isolate'; + +import 'package:path_provider/path_provider.dart'; +import 'package:sqflite/sqflite.dart'; +import 'package:spotiflac_android/models/library_collections.dart'; +import 'package:spotiflac_android/models/track.dart'; +import 'package:spotiflac_android/services/platform_bridge.dart'; +import 'package:spotiflac_android/services/sqlite_helpers.dart' as sqlite; +import 'package:spotiflac_android/services/sqlite_native_snapshot.dart'; +import 'package:spotiflac_android/utils/logger.dart'; + +// Small selections cost less inline; reserve worker setup for larger lists. +const _workerThreshold = 1024; +const _nativeThreshold = 2048; +var _attachmentSequence = 0; +final _log = AppLogger('PlaylistBatch'); + +class PreparedPlaylistBatch { + const PreparedPlaylistBatch({ + required this.entries, + required this.duplicates, + required this.rows, + this.rowsPath, + this.inputDirectory, + }); + + final List entries; + final int duplicates; + final List> rows; + final String? rowsPath; + final String? inputDirectory; + + Future dispose() async { + final directory = inputDirectory; + if (directory != null) await _removeDirectory(directory); + } +} + +/// Deduplication and encoding share one worker. Large native batches return a +/// file path instead of transferring thousands of JSON maps back through UI. +Future preparePlaylistBatch({ + required Iterable tracks, + required Set existingKeys, + required DateTime addedAt, + bool? useNative, + Directory? temporaryDirectory, +}) async { + final snapshot = tracks.toList(growable: false); + String? inputDirectory; + if (snapshot.length >= _nativeThreshold && + (useNative ?? PlatformBridge.supportsCoreBackend)) { + inputDirectory = + (await (temporaryDirectory ?? await getTemporaryDirectory()).createTemp( + 'playlist-input-', + )).path; + } + try { + if (snapshot.length < _workerThreshold) { + return _prepare(snapshot, existingKeys, addedAt, inputDirectory); + } + return await Isolate.run( + () => _prepare(snapshot, existingKeys, addedAt, inputDirectory), + ); + } catch (_) { + if (inputDirectory != null) await _removeDirectory(inputDirectory); + rethrow; + } +} + +PreparedPlaylistBatch _prepare( + List tracks, + Set existingKeys, + DateTime addedAt, + String? inputDirectory, +) { + final keys = {}; + final entries = []; + var duplicates = 0; + for (final track in tracks) { + final key = trackCollectionKey(track); + if (existingKeys.contains(key) || !keys.add(key)) { + duplicates++; + } else { + entries.add( + CollectionTrackEntry(key: key, track: track, addedAt: addedAt), + ); + } + } + final rows = >[]; + String? rowsPath; + if (inputDirectory != null && entries.length >= _nativeThreshold) { + rowsPath = '$inputDirectory/rows.ndjson'; + final file = File(rowsPath).openSync(mode: FileMode.write); + try { + var buffer = StringBuffer(); + for (final entry in entries) { + buffer.writeln( + jsonEncode({'key': entry.key, 'track': entry.track.toJson()}), + ); + if (buffer.length >= 64 * 1024) { + file.writeStringSync(buffer.toString()); + buffer = StringBuffer(); + } + } + if (buffer.isNotEmpty) file.writeStringSync(buffer.toString()); + } finally { + file.closeSync(); + } + } else { + final timestamp = addedAt.toIso8601String(); + for (final entry in entries) { + rows.add({ + 'track_key': entry.key, + 'track_json': jsonEncode(entry.track.toJson()), + 'added_at': timestamp, + }); + } + } + return PreparedPlaylistBatch( + entries: entries, + duplicates: duplicates, + rows: rows, + rowsPath: rowsPath, + inputDirectory: inputDirectory, + ); +} + +/// Rust exclusively builds a new private database. Publication and the +/// playlist timestamp stay in one transaction on the existing sqflite owner. +Future publishPlaylistBatch({ + required Database database, + required String playlistId, + required String updatedAt, + required String rowsPath, + required int expectedCount, + Future> Function(Map)? nativeJobRunner, + Directory? temporaryDirectory, +}) async { + final directory = await (temporaryDirectory ?? await getTemporaryDirectory()) + .createTemp('playlist-stage-'); + final stagedPath = '${directory.path}/playlist_additions.db'; + var detached = true; + try { + final result = await (nativeJobRunner ?? PlatformBridge.runNativeDataJob)({ + 'operation': 'collections_stage_playlist_additions', + 'rows_path': rowsPath, + 'output_path': stagedPath, + 'expected_count': expectedCount, + }); + if (result['committed'] != true || + result['published'] != true || + result['count'] != expectedCount || + result['path'] != stagedPath) { + throw const FormatException('Incomplete playlist staging'); + } + final alias = 'playlist_additions_${_attachmentSequence++}'; + await database.execute('ATTACH DATABASE ? AS $alias', [stagedPath]); + detached = false; + var committed = false; + try { + await sqlite.transactionWithBusyRetry(database, (txn) async { + await txn.execute( + 'INSERT OR REPLACE INTO main.playlist_tracks ' + '(playlist_id,track_key,track_json,added_at) ' + 'SELECT ?,track_key,track_json,? FROM $alias.playlist_additions ' + 'ORDER BY position', + [playlistId, updatedAt], + ); + await txn.update( + 'playlists', + {'updated_at': updatedAt}, + where: 'id = ?', + whereArgs: [playlistId], + ); + }); + committed = true; + } finally { + try { + await detachNativeSqliteSnapshot(database, alias, stagedPath); + detached = true; + } on NativeSqliteSnapshotDetachException { + if (!committed) rethrow; + _log.w('Playlist saved; retaining staging after failed DETACH'); + } + } + } finally { + // This output directory is separate from the input batch so its owner can + // always remove the input even if a failed DETACH retains the SQLite file. + if (detached) await _removeDirectory(directory.path); + } +} + +Future _removeDirectory(String path) async { + try { + await Directory(path).delete(recursive: true); + } on FileSystemException catch (error) { + _log.w('Could not remove playlist staging: $error'); + } +} diff --git a/rust_backend/crates/mobile/src/data_jobs.rs b/rust_backend/crates/mobile/src/data_jobs.rs index f57f5840..a515dff0 100644 --- a/rust_backend/crates/mobile/src/data_jobs.rs +++ b/rust_backend/crates/mobile/src/data_jobs.rs @@ -126,6 +126,9 @@ fn execute( check, ), "collections_snapshot" => crate::native_collections::execute(request, check), + "collections_stage_playlist_additions" => { + crate::native_playlist_write::execute(request, check) + } operation if operation.starts_with("library_") => { crate::native_library::execute(request, check) } diff --git a/rust_backend/crates/mobile/src/lib.rs b/rust_backend/crates/mobile/src/lib.rs index c8537d28..3e070f4b 100644 --- a/rust_backend/crates/mobile/src/lib.rs +++ b/rust_backend/crates/mobile/src/lib.rs @@ -18,6 +18,7 @@ mod native_id3; mod native_library; mod native_listing; mod native_lyrics; +mod native_playlist_write; mod native_playlists; mod native_xml; mod network_tags; diff --git a/rust_backend/crates/mobile/src/native_playlist_write.rs b/rust_backend/crates/mobile/src/native_playlist_write.rs new file mode 100644 index 00000000..6e0f6619 --- /dev/null +++ b/rust_backend/crates/mobile/src/native_playlist_write.rs @@ -0,0 +1,210 @@ +//! Build an isolated playlist write batch. Only the live database's owner may +//! attach this closed file and publish its rows into the application's tables. +use rusqlite::Connection; +use serde::Deserialize; +use serde_json::{Value, json, value::RawValue}; +use std::{ + borrow::Cow, + fs::File, + io::{BufRead, BufReader}, + path::Path, +}; + +#[derive(Deserialize)] +struct TrackRow<'a> { + #[serde(borrow)] + key: Cow<'a, str>, + #[serde(borrow)] + track: &'a RawValue, +} + +pub(crate) fn execute( + request: &Value, + check: &dyn Fn() -> Result<(), String>, +) -> Result { + check()?; + let input = request["rows_path"] + .as_str() + .ok_or("Missing playlist rows_path")?; + let output = request["output_path"] + .as_str() + .ok_or("Missing playlist output_path")?; + let expected = request["expected_count"] + .as_u64() + .ok_or("Invalid playlist expected_count")?; + let output = Path::new(output); + if !Path::new(input).is_absolute() + || !output.is_absolute() + || output.file_name().and_then(|name| name.to_str()) != Some("playlist_additions.db") + { + return Err( + "Playlist staging requires absolute rows_path and private playlist_additions.db".into(), + ); + } + let mut input = + BufReader::with_capacity(64 * 1024, File::open(input).map_err(|e| e.to_string())?); + let staged = tempfile::NamedTempFile::new_in( + output + .parent() + .ok_or("Missing playlist staging directory")?, + ) + .map_err(|e| e.to_string())?; + // Never open output_path itself: it may already exist. The temporary file + // is exclusively owned by this job and is published without replacement. + let mut db = Connection::open(staged.path()).map_err(|e| e.to_string())?; + db.execute_batch( + "PRAGMA user_version=1; + CREATE TABLE playlist_additions( + position INTEGER PRIMARY KEY, + track_key TEXT NOT NULL UNIQUE, + track_json TEXT NOT NULL + );", + ) + .map_err(|e| e.to_string())?; + let transaction = db.transaction().map_err(|e| e.to_string())?; + let mut count = 0_u64; + { + let mut insert = transaction + .prepare("INSERT INTO playlist_additions(track_key,track_json) VALUES (?,?)") + .map_err(|e| e.to_string())?; + let mut line = Vec::new(); + loop { + check()?; + line.clear(); + if input + .read_until(b'\n', &mut line) + .map_err(|e| e.to_string())? + == 0 + { + break; + } + let row: TrackRow<'_> = serde_json::from_slice(&line).map_err(|e| e.to_string())?; + if row.key.is_empty() || !row.track.get().starts_with('{') { + return Err("Invalid playlist track row".into()); + } + if count == expected { + return Err("Playlist staging received too many rows".into()); + } + // Preserve the original track JSON byte-for-byte instead of + // allocating a metadata map and serializing every field again. + insert + .execute([row.key.as_ref(), row.track.get()]) + .map_err(|e| e.to_string())?; + count += 1; + } + } + if count != expected { + return Err(format!( + "Incomplete playlist staging: expected {expected}, got {count}" + )); + } + check()?; + transaction.commit().map_err(|e| e.to_string())?; + db.close().map_err(|(_, error)| error.to_string())?; + staged.as_file().sync_all().map_err(|e| e.to_string())?; + check()?; + staged + .persist_noclobber(output) + .map_err(|e| e.to_string())?; + Ok(json!({"path":output,"count":count,"committed":true,"published":true})) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::cell::Cell; + + #[test] + fn stages_ordered_unicode_rows_and_preserves_raw_track_json() { + let root = tempfile::tempdir().unwrap(); + let input = root.path().join("rows.ndjson"); + let output = root.path().join("playlist_additions.db"); + let raw = r#"{ "name": "音楽 🎵", "unmodeled": {"value":1e-12}, "id":"one" }"#; + std::fs::write(&input, format!("{{\"key\":\"source:one\\n\\\"\",\"track\":{raw}}}\n{{\"key\":\"source:two\",\"track\":{{\"id\":\"two\"}}}}\n")).unwrap(); + let result = execute( + &json!({"rows_path":input,"output_path":output,"expected_count":2}), + &|| Ok(()), + ) + .unwrap(); + assert_eq!(result["count"], 2); + let db = Connection::open(&output).unwrap(); + let rows = db + .prepare("SELECT track_key,track_json FROM playlist_additions ORDER BY position") + .unwrap() + .query_map([], |row| { + Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)) + }) + .unwrap() + .collect::, _>>() + .unwrap(); + assert_eq!( + rows, + [ + ("source:one\n\"".into(), raw.into()), + ("source:two".into(), r#"{"id":"two"}"#.into()) + ] + ); + assert_eq!( + db.pragma_query_value(None, "user_version", |row| row.get::<_, i64>(0)) + .unwrap(), + 1 + ); + } + + #[test] + fn rejects_invalid_duplicate_truncated_and_cancelled_batches_without_publication() { + let valid = "{\"key\":\"one\",\"track\":{\"id\":\"one\"}}\n"; + for (payload, expected, cancel) in [ + (valid.to_owned(), 2, false), + (valid.repeat(2), 1, false), + (valid.repeat(2), 2, false), + (format!("{valid}{{bad"), 2, false), + ("{\"key\":\"one\",\"track\":null}\n".into(), 1, false), + ("{\"key\":\"\",\"track\":{}}\n".into(), 1, false), + (valid.to_owned(), 1, true), + ] { + let root = tempfile::tempdir().unwrap(); + let input = root.path().join("rows.ndjson"); + let output = root.path().join("playlist_additions.db"); + std::fs::write(&input, payload).unwrap(); + let checks = Cell::new(0); + let result = execute( + &json!({"rows_path":input,"output_path":output,"expected_count":expected}), + &|| { + checks.set(checks.get() + 1); + if cancel && checks.get() == 3 { + Err("cancelled".into()) + } else { + Ok(()) + } + }, + ); + assert!(result.is_err()); + assert!(!output.exists()); + assert_eq!(std::fs::read_dir(root.path()).unwrap().count(), 1); + } + } + + #[test] + fn never_replaces_an_existing_destination_or_opens_the_live_database() { + for filename in ["playlist_additions.db", "library_collections.db"] { + let root = tempfile::tempdir().unwrap(); + let input = root.path().join("rows.ndjson"); + let output = root.path().join(filename); + std::fs::write(&input, "").unwrap(); + std::fs::write(&output, "existing database").unwrap(); + assert!( + execute( + &json!({"rows_path":input,"output_path":output,"expected_count":0}), + &|| Ok(()) + ) + .is_err() + ); + assert_eq!( + std::fs::read_to_string(output).unwrap(), + "existing database" + ); + assert_eq!(std::fs::read_dir(root.path()).unwrap().count(), 2); + } + } +} diff --git a/test/library_collections_mutation_test.dart b/test/library_collections_mutation_test.dart index eb634b74..73d0219d 100644 --- a/test/library_collections_mutation_test.dart +++ b/test/library_collections_mutation_test.dart @@ -163,6 +163,20 @@ class _CollectionsDatabase implements LibraryCollectionsDatabase { Future deletePlaylist(String playlistId) => _write('delete:playlist:$playlistId', () => playlists.remove(playlistId)); + @override + Future upsertPlaylistTracksBatch({ + required String playlistId, + required String playlistUpdatedAt, + required List> tracks, + }) => _write('batch:$playlistId', () { + final rows = playlistTracks.putIfAbsent(playlistId, () => []); + for (final track in tracks) { + rows.removeWhere((row) => row['track_key'] == track['track_key']); + rows.add({...track, 'playlist_id': playlistId}); + } + playlists[playlistId]!['updated_at'] = playlistUpdatedAt; + }); + @override dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation); } @@ -189,6 +203,64 @@ ProviderContainer _container(_CollectionsDatabase db) { } void main() { + for (final failFirst in [false, true]) { + test( + 'bulk additions publish after persistence and serialize duplicates (failure: $failFirst)', + () async { + final db = _CollectionsDatabase(); + db.playlists['playlist'] = { + 'id': 'playlist', + 'name': 'Playlist', + 'created_at': '2026-01-01T00:00:00Z', + 'updated_at': '2026-01-01T00:00:00Z', + }; + db.releasePlaylistRead.complete(); + db.failNextWrite = failFirst; + final container = _container(db); + final notifier = container.read(libraryCollectionsProvider.notifier); + final first = notifier.addTracksToPlaylist('playlist', [ + _track('1'), + _track('1'), + _track('2'), + ]); + final failure = failFirst ? expectLater(first, throwsStateError) : null; + await db.writeStarted.future; + expect( + container + .read(libraryCollectionsProvider) + .playlistById('playlist')! + .trackCount, + 0, + ); + final second = notifier.addTracksToPlaylist('playlist', [ + _track('2'), + _track('3'), + ]); + expect(db.events, ['batch:playlist']); + db.releaseWrite.complete(); + if (failure != null) { + await failure; + } else { + final result = await first; + expect(result.addedCount, 2); + expect(result.alreadyInPlaylistCount, 1); + } + final result = await second; + expect(result.addedCount, failFirst ? 2 : 1); + expect(result.alreadyInPlaylistCount, failFirst ? 0 : 1); + final playlist = container + .read(libraryCollectionsProvider) + .playlistById('playlist')!; + expect( + playlist.tracks.map((entry) => entry.track.id), + failFirst ? ['2', '3'] : ['1', '2', '3'], + ); + expect(playlist.trackCount, failFirst ? 2 : 3); + expect(db.events, ['batch:playlist', 'batch:playlist']); + }, + ); + } + test('disposed snapshot read removes native rows before returning', () async { final directory = await Directory.systemTemp.createTemp( 'collection-read-disposed-', diff --git a/test/playlist_batch_benchmark_test.dart b/test/playlist_batch_benchmark_test.dart new file mode 100644 index 00000000..467eb194 --- /dev/null +++ b/test/playlist_batch_benchmark_test.dart @@ -0,0 +1,59 @@ +import 'dart:convert'; +import 'dart:io'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:spotiflac_android/services/playlist_batch_writer.dart'; + +import 'support/library_collections_benchmark.dart'; +import 'support/playlist_batch_benchmark.dart'; + +void main() { + test( + 'compare playlist batch preparation with the original inline path', + () async { + final tracks = await playlistBatchFixture(12000); + final root = await Directory.systemTemp.createTemp('playlist-benchmark-'); + addTearDown(() => root.delete(recursive: true)); + final samples = >[]; + final workers = >[]; + for (var index = 0; index < 5; index++) { + for (final worker in index.isEven ? [false, true] : [true, false]) { + if (worker) { + final result = await measureCollectionOperation( + () => preparePlaylistBatch( + tracks: tracks, + existingKeys: const {}, + addedAt: DateTime.utc(2026, 10, 8), + useNative: true, + temporaryDirectory: root, + ), + ); + expect(result.value.entries.length, tracks.length); + workers.add(result.timing); + await result.value.dispose(); + continue; + } + final result = await measureCollectionOperation( + () => inlinePlaylistBatch( + tracks, + const {}, + DateTime.utc(2026, 10, 8), + ), + ); + expect(result.value.rows.length, tracks.length); + samples.add(result.timing); + } + } + // Host diagnostics only; the device profile benchmark owns speed claims. + // ignore: avoid_print + print( + jsonEncode({ + 'rows': tracks.length, + 'host_inline': samples, + 'host_worker': workers, + }), + ); + }, + skip: !const bool.fromEnvironment('RUN_PLAYLIST_BATCH_BENCHMARK'), + ); +} diff --git a/test/playlist_batch_publication_test.dart b/test/playlist_batch_publication_test.dart new file mode 100644 index 00000000..bc4b81bd --- /dev/null +++ b/test/playlist_batch_publication_test.dart @@ -0,0 +1,210 @@ +import 'dart:convert'; +import 'dart:io'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:sqflite/sqflite.dart'; +import 'package:spotiflac_android/services/playlist_batch_writer.dart'; + +import 'support/sqlite_process_database.dart'; + +class _DetachFailure implements Database { + _DetachFailure(this.database); + @override + final SqliteProcessDatabase database; + + @override + Future execute(String sql, [List? args]) async { + if (sql.startsWith('DETACH DATABASE')) throw StateError('detach failed'); + await database.execute(sql, args); + } + + @override + Future transaction( + Future Function(Transaction) action, { + bool? exclusive, + }) => database.transaction(action, exclusive: exclusive); + + @override + dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation); +} + +void main() { + late Directory root; + late SqliteProcessDatabase live; + late String input; + String? retainedStage; + + setUp(() async { + root = await Directory.systemTemp.createTemp('playlist-publication-'); + live = await SqliteProcessDatabase.open(path: '${root.path}/live.db'); + await live.execute('PRAGMA foreign_keys=ON'); + await live.execute( + 'CREATE TABLE playlists(id TEXT PRIMARY KEY, name TEXT, updated_at TEXT)', + ); + await live.execute( + 'CREATE TABLE playlist_tracks(playlist_id TEXT NOT NULL, track_key TEXT NOT NULL, track_json TEXT NOT NULL, added_at TEXT NOT NULL, PRIMARY KEY(playlist_id,track_key), FOREIGN KEY(playlist_id) REFERENCES playlists(id))', + ); + await live.execute('CREATE TABLE published(track_key TEXT)'); + await live.execute( + 'CREATE TRIGGER audit AFTER INSERT ON playlist_tracks BEGIN INSERT INTO published VALUES(new.track_key); END', + ); + await live.execute("INSERT INTO playlists VALUES('p','Original','old')"); + await live.execute("INSERT INTO playlists VALUES('other','Other','old')"); + await live.execute( + "INSERT INTO playlist_tracks VALUES('p','existing','{}','old')", + ); + await live.execute('DELETE FROM published'); + input = '${root.path}/rows.ndjson'; + await File(input).writeAsString(''); + retainedStage = null; + }); + tearDown(() async { + await live.close(); + await root.delete(recursive: true); + }); + + // This fixture implements only the native boundary. SQL publication below + // runs on real SQLite; the native parser itself has Rust and device tests. + Future> stage(Map request) async { + expect(request['operation'], 'collections_stage_playlist_additions'); + expect(request.values, isNot(contains(live.path))); + expect((await live.rawQuery('PRAGMA database_list')).length, 1); + final path = request['output_path'] as String; + retainedStage = path; + final private = await SqliteProcessDatabase.open(path: path); + try { + await private.execute( + 'CREATE TABLE playlist_additions(position INTEGER PRIMARY KEY,track_key TEXT UNIQUE,track_json TEXT)', + ); + for (final key in ['new-a', 'new-b']) { + await private.execute( + 'INSERT INTO playlist_additions(track_key,track_json) VALUES(?,?)', + [ + key, + jsonEncode({'id': key, 'name': '音楽 🎵'}), + ], + ); + } + } finally { + await private.close(); + } + return {'committed': true, 'published': true, 'path': path, 'count': 2}; + } + + Future publish({ + Database? database, + String playlist = 'p', + Future> Function(Map)? runner, + }) => publishPlaylistBatch( + database: database ?? live, + playlistId: playlist, + updatedAt: 'new', + rowsPath: input, + expectedCount: 2, + temporaryDirectory: root, + nativeJobRunner: runner ?? stage, + ); + + test( + 'publishes in order through live triggers and changes only target timestamp', + () async { + await publish(); + expect( + (await live.rawQuery( + 'SELECT track_key FROM playlist_tracks ORDER BY rowid', + )).map((row) => row['track_key']), + ['existing', 'new-a', 'new-b'], + ); + expect(await live.rawQuery('SELECT * FROM published'), [ + {'track_key': 'new-a'}, + {'track_key': 'new-b'}, + ]); + expect( + await live.rawQuery( + 'SELECT id,name,updated_at FROM playlists ORDER BY rowid', + ), + [ + {'id': 'p', 'name': 'Original', 'updated_at': 'new'}, + {'id': 'other', 'name': 'Other', 'updated_at': 'old'}, + ], + ); + expect(await live.rawQuery('PRAGMA foreign_key_check'), isEmpty); + expect((await live.rawQuery('PRAGMA database_list')).length, 1); + expect(await File(retainedStage!).exists(), false); + }, + ); + + test( + 'timestamp failure rolls back all inserted tracks and trigger effects', + () async { + await live.execute( + "CREATE TRIGGER fail_update BEFORE UPDATE ON playlists BEGIN SELECT RAISE(ABORT,'denied'); END", + ); + await expectLater(publish(), throwsA(isA())); + expect( + (await live.rawQuery( + 'SELECT track_key FROM playlist_tracks', + )).single['track_key'], + 'existing', + ); + expect(await live.rawQuery('SELECT * FROM published'), isEmpty); + expect( + (await live.rawQuery( + "SELECT updated_at FROM playlists WHERE id='p'", + )).single['updated_at'], + 'old', + ); + expect((await live.rawQuery('PRAGMA database_list')).length, 1); + expect(await File(retainedStage!).exists(), false); + }, + ); + + test( + 'missing playlist preserves foreign-key enforcement and existing data', + () async { + await expectLater( + publish(playlist: 'missing'), + throwsA(isA()), + ); + expect( + (await live.rawQuery('SELECT track_key FROM playlist_tracks')).length, + 1, + ); + expect((await live.rawQuery('PRAGMA database_list')).length, 1); + }, + ); + + test('incomplete native result cannot enter the live transaction', () async { + await expectLater( + publish( + runner: (request) async { + final result = await stage(request); + result['count'] = 1; + return result; + }, + ), + throwsFormatException, + ); + expect( + (await live.rawQuery('SELECT track_key FROM playlist_tracks')).length, + 1, + ); + expect(await File(retainedStage!).exists(), false); + expect((await live.rawQuery('PRAGMA database_list')).length, 1); + }); + + test( + 'failed detach retains its file without reporting a durable commit as failure', + () async { + await publish(database: _DetachFailure(live)); + expect( + (await live.rawQuery('SELECT track_key FROM playlist_tracks')).length, + 3, + ); + expect(await File(retainedStage!).exists(), true); + expect((await live.rawQuery('PRAGMA database_list')).length, 2); + await File(input).delete(); + expect(await File(retainedStage!).exists(), true); + }, + ); +} diff --git a/test/playlist_batch_writer_test.dart b/test/playlist_batch_writer_test.dart new file mode 100644 index 00000000..285c3d86 --- /dev/null +++ b/test/playlist_batch_writer_test.dart @@ -0,0 +1,132 @@ +import 'dart:convert'; +import 'dart:io'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:spotiflac_android/models/track.dart'; +import 'package:spotiflac_android/services/playlist_batch_writer.dart'; + +import 'support/playlist_batch_benchmark.dart'; + +class _InvalidTrack extends Track { + _InvalidTrack() + : super( + id: 'invalid', + name: 'Invalid', + artistName: '', + albumName: '', + duration: 0, + ); + + @override + Map toJson() => {'value': double.nan}; +} + +void main() { + late Directory root; + setUp(() async { + root = await Directory.systemTemp.createTemp('playlist-batch-test-'); + }); + tearDown(() => root.delete(recursive: true)); + final now = DateTime.utc(2026, 10, 8); + + for (final count in [12, 1024, 2500]) { + test( + 'preparation preserves order, metadata and duplicate counts ($count)', + () async { + final fixture = await playlistBatchFixture(count); + final tracks = [...fixture, fixture[1], fixture[3]]; + final keys = {'example:0', 'example:1'}; + final baseline = inlinePlaylistBatch(tracks, keys, now); + final prepared = await preparePlaylistBatch( + tracks: tracks, + existingKeys: keys, + addedAt: now, + useNative: true, + temporaryDirectory: root, + ); + try { + expect( + prepared.entries.map((entry) => entry.key), + baseline.entries.map((entry) => entry.key), + ); + expect(prepared.duplicates, baseline.duplicates); + expect(prepared.entries.every((entry) => entry.addedAt == now), true); + expect(keys, {'example:0', 'example:1'}); + if (prepared.rowsPath == null) { + expect(prepared.rows, baseline.rows); + } else { + expect(prepared.rows, isEmpty); + final rows = await File(prepared.rowsPath!).readAsLines(); + expect(rows.length, baseline.rows.length); + for (var index = 0; index < rows.length; index++) { + final row = jsonDecode(rows[index]) as Map; + expect(row['key'], baseline.rows[index]['track_key']); + expect( + jsonEncode(row['track']), + baseline.rows[index]['track_json'], + ); + } + } + } finally { + await prepared.dispose(); + } + expect(await root.list().toList(), isEmpty); + }, + ); + } + + test('ISRC wins across providers and first batch metadata wins', () async { + Track track(String id, String source, String isrc) => Track( + id: id, + source: source, + isrc: isrc, + name: id, + artistName: 'Artist', + albumName: 'Album', + duration: 1, + ); + final prepared = await preparePlaylistBatch( + tracks: [ + track('first', 'one', ' xx001 '), + track('second', 'two', 'XX001'), + track('third', 'two', 'xx002'), + ], + existingKeys: {'isrc:XX002'}, + addedAt: now, + ); + expect(prepared.duplicates, 2); + expect(prepared.entries.single.track.id, 'first'); + expect(prepared.entries.single.key, 'isrc:XX001'); + }); + + test('mostly duplicate batches retain the small persistence route', () async { + final fixture = await playlistBatchFixture(12); + final prepared = await preparePlaylistBatch( + tracks: [for (var index = 0; index < 256; index++) ...fixture], + existingKeys: const {}, + addedAt: now, + useNative: true, + temporaryDirectory: root, + ); + expect(prepared.rowsPath, isNull); + expect(prepared.entries.length, 12); + expect(prepared.duplicates, 3060); + await prepared.dispose(); + expect(await root.list().toList(), isEmpty); + }); + + test('failed encoding removes its partial row file', () async { + final fixture = await playlistBatchFixture(2500); + await expectLater( + preparePlaylistBatch( + tracks: [...fixture, _InvalidTrack()], + existingKeys: const {}, + addedAt: now, + useNative: true, + temporaryDirectory: root, + ), + throwsA(anything), + ); + expect(await root.list().toList(), isEmpty); + }); +} diff --git a/test/support/playlist_batch_benchmark.dart b/test/support/playlist_batch_benchmark.dart new file mode 100644 index 00000000..dbe67b22 --- /dev/null +++ b/test/support/playlist_batch_benchmark.dart @@ -0,0 +1,61 @@ +import 'dart:convert'; +import 'dart:isolate'; + +import 'package:spotiflac_android/models/library_collections.dart'; +import 'package:spotiflac_android/models/track.dart'; + +Future> playlistBatchFixture(int count) => Isolate.run( + () => List.generate( + count, + (index) => Track( + id: '$index', + source: 'example', + name: 'Track $index — 音楽 🎵', + artistName: 'Artist ${index % 100}', + albumName: 'Album ${index ~/ 12}', + albumArtist: 'Album artist ${index % 100}', + duration: 180, + coverUrl: 'https://example.test/cover/${index ~/ 12}.jpg', + trackNumber: index % 12 + 1, + discNumber: 1, + genre: 'Rock', + comment: List.filled(12, 'Metadata $index').join(' '), + ), + growable: false, + ), +); + +// Production preparation before the batch-write optimization. Keep the same +// identity precedence, duplicate counting, ordering and serialization here. +({ + List entries, + List> rows, + int duplicates, +}) +inlinePlaylistBatch( + List tracks, + Set existingKeys, + DateTime now, +) { + final knownKeys = {...existingKeys}; + final entries = []; + var duplicates = 0; + for (final track in tracks) { + final key = trackCollectionKey(track); + if (!knownKeys.add(key)) { + duplicates++; + continue; + } + entries.add(CollectionTrackEntry(key: key, track: track, addedAt: now)); + } + final rows = entries + .map( + (entry) => { + 'track_key': entry.key, + 'track_json': jsonEncode(entry.track.toJson()), + 'added_at': entry.addedAt.toIso8601String(), + }, + ) + .toList(growable: false); + return (entries: entries, rows: rows, duplicates: duplicates); +}