perf(collections): restore backups on Rust SQLite workers

integration_test/native_collection_restore_test.dart

@@ -0,0 +1,261 @@
+import 'dart:async';
+import 'dart:convert';
+import 'dart:io';
+import 'dart:isolate';
+
+import 'package:crypto/crypto.dart';
+import 'package:flutter/material.dart';
+import 'package:flutter_test/flutter_test.dart';
+import 'package:integration_test/integration_test.dart';
+import 'package:path_provider/path_provider.dart';
+import 'package:sqflite/sqflite.dart';
+import 'package:spotiflac_android/services/collection_restore_codec.dart';
+import 'package:spotiflac_android/services/library_collections_database.dart';
+import 'package:spotiflac_android/services/platform_bridge.dart';
+
+const _date = '2026-10-07T05:00:00.000';
+const _tables = {
+  'wishlist': 'wishlist_tracks',
+  'loved': 'loved_tracks',
+  'artist': 'favorite_artists',
+  'playlist': 'playlists',
+  'playlist_track': 'playlist_tracks',
+};
+
+Future<Database> _database(String path) => openDatabase(
+  path,
+  version: 2,
+  onConfigure: (db) async {
+    await db.execute('PRAGMA foreign_keys=ON');
+    await db.execute('PRAGMA recursive_triggers=ON');
+    await db.rawQuery('PRAGMA journal_mode=WAL');
+    await db.execute('PRAGMA synchronous=NORMAL');
+    await db.rawQuery('PRAGMA busy_timeout=5000');
+  },
+  onCreate: (db, _) async {
+    for (final sql in [
+      'CREATE TABLE wishlist_tracks(track_key TEXT PRIMARY KEY,track_json TEXT NOT NULL,added_at TEXT NOT NULL)',
+      'CREATE TABLE loved_tracks(track_key TEXT PRIMARY KEY,track_json TEXT NOT NULL,added_at TEXT NOT NULL)',
+      'CREATE TABLE favorite_artists(artist_key TEXT PRIMARY KEY,artist_json TEXT NOT NULL,added_at TEXT NOT NULL)',
+      '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)',
+      '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)',
+      for (final table in [
+        'wishlist_tracks',
+        'loved_tracks',
+        'favorite_artists',
+        'playlist_tracks',
+      ])
+        'CREATE INDEX idx_${table}_added_at ON $table(added_at DESC)',
+      'CREATE INDEX idx_playlists_created_at ON playlists(created_at DESC)',
+      'CREATE INDEX idx_playlist_tracks_playlist_id ON playlist_tracks(playlist_id)',
+    ]) {
+      await db.execute(sql);
+    }
+  },
+);
+
+Future<Map<String, dynamic>> _fixture(int count) => Isolate.run(() {
+  final entries = [
+    for (var i = 0; i < count; i++)
+      {
+        'key': 'example:$i',
+        'track': {
+          'id': '$i',
+          'name': '音楽 🎵 Track $i',
+          'artistName': 'Artist ${i % 127}',
+          'albumName': 'Album ${i ~/ 12}',
+          'coverUrl': 'https://example.test/${i ~/ 12}.jpg',
+          'source': 'example',
+          'duration': 180,
+        },
+        'addedAt': _date,
+      },
+  ];
+  return {
+    'wishlist': entries.take(count ~/ 10).toList(),
+    'loved': entries.take(count ~/ 10).toList(),
+    'favoriteArtists': [
+      {'key': 'example:artist', 'name': 'İstanbul', 'addedAt': _date},
+    ],
+    'playlists': [
+      {
+        'id': 'p',
+        'name': 'old',
+        'tracks': [entries.first],
+        'createdAt': _date,
+      },
+      {
+        'id': 'p',
+        'name': '最新',
+        'coverImagePath': '/covers/a.jpg',
+        'createdAt': _date,
+        'tracks': entries,
+      },
+      {'id': 'empty', 'name': '', 'createdAt': _date},
+    ],
+  };
+});
+
+/// Matches the old per-row transaction, including inline JSON encoding, and
+/// uses the same database/schema/indexes as the production native route.
+Future<void> _legacyRestore(Database db, Map<String, dynamic> input) =>
+    db.transaction((txn) async {
+      for (final table in [
+        'playlist_tracks',
+        'playlists',
+        'wishlist_tracks',
+        'loved_tracks',
+        'favorite_artists',
+      ]) {
+        await txn.delete(table);
+      }
+      for (final command in collectionRestoreCommands(input, _date)) {
+        await txn.insert(
+          _tables[command['kind']]!,
+          command['values'] as Map<String, dynamic>,
+          conflictAlgorithm: ConflictAlgorithm.replace,
+        );
+      }
+    });
+
+Future<String> _digest(Database db) async {
+  final rows = <String, Object?>{};
+  for (final table in _tables.values) {
+    rows[table] = await db.rawQuery('SELECT * FROM $table ORDER BY rowid');
+  }
+  return Isolate.run(
+    () => sha256.convert(utf8.encode(jsonEncode(rows))).toString(),
+  );
+}
+
+void main() {
+  final binding = IntegrationTestWidgetsFlutterBinding.ensureInitialized();
+  late Directory root;
+  late Database db;
+  setUp(() async {
+    root = await Directory.systemTemp.createTemp('native-collections-');
+    db = await _database('${root.path}/library_collections.db');
+  });
+  tearDown(() async {
+    await db.close();
+    await root.delete(recursive: true);
+  });
+
+  testWidgets(
+    'production restore preserves exact rows and rejects incomplete input atomically',
+    (tester) async {
+      final input = await _fixture(20);
+      await _legacyRestore(db, input);
+      final oldDigest = await _digest(db);
+      await LibraryCollectionsDatabase.replaceDatabaseFromBackup(
+        db,
+        input,
+        nowIso: _date,
+      );
+      expect(await _digest(db), oldDigest);
+      final spool = '${root.path}/commands.ndjson';
+      final count = await writeCollectionRestoreCommands(spool, input, _date);
+      await expectLater(
+        PlatformBridge.runNativeDataJob({
+          'operation': 'backup_collections_import',
+          'collections_path': db.path,
+          'ndjson_path': spool,
+          'expected_count': count + 1,
+        }),
+        throwsA(isA<Exception>()),
+      );
+      expect(await _digest(db), oldDigest);
+      await LibraryCollectionsDatabase.replaceDatabaseFromBackup(
+        db,
+        {},
+        nowIso: _date,
+      );
+      for (final table in _tables.values) {
+        expect(await db.query(table), isEmpty);
+      }
+    },
+  );
+
+  testWidgets(
+    'paired large collection restore records total latency and UI heartbeat',
+    (tester) async {
+      await tester.pumpWidget(
+        const MaterialApp(
+          home: Scaffold(body: Center(child: CircularProgressIndicator())),
+        ),
+      );
+      final input = await _fixture(10000);
+      await LibraryCollectionsDatabase.replaceDatabaseFromBackup(
+        db,
+        await _fixture(20),
+        nowIso: _date,
+      );
+      final timings = <String, List<int>>{'dart': [], 'native': []};
+      final heartbeats = <String, List<Map<String, int>>>{
+        'dart': [],
+        'native': [],
+      };
+      String? expected;
+      for (var sample = 0; sample < 5; sample++) {
+        for (final native in sample.isEven ? [false, true] : [true, false]) {
+          final label = native ? 'native' : 'dart';
+          final elapsed = Stopwatch()..start();
+          var lastBeat = 0;
+          var maximumGap = 0;
+          var beats = 0;
+          final timer = Timer.periodic(const Duration(milliseconds: 10), (_) {
+            final now = elapsed.elapsedMicroseconds;
+            final gap = now - lastBeat;
+            if (gap > maximumGap) maximumGap = gap;
+            lastBeat = now;
+            beats++;
+          });
+          try {
+            if (native) {
+              await LibraryCollectionsDatabase.replaceDatabaseFromBackup(
+                db,
+                input,
+                nowIso: _date,
+              );
+            } else {
+              await _legacyRestore(db, input);
+            }
+          } finally {
+            elapsed.stop();
+            timer.cancel();
+          }
+          timings[label]!.add(elapsed.elapsedMicroseconds);
+          heartbeats[label]!.add({
+            'count': beats,
+            'maximum_gap_us': maximumGap,
+          });
+          final digest = await _digest(db);
+          expected ??= digest;
+          expect(digest, expected);
+          expect(beats, greaterThan(0));
+        }
+      }
+      final report = {
+        'platform': Platform.operatingSystem,
+        'persisted_rows': 12003,
+        'input_commands': 12005,
+        'input_tracks': 10000,
+        'mode':
+            'five alternating paired samples; original per-row sqflite versus complete production worker+spool+native transaction',
+        'times_us': timings,
+        'heartbeat': heartbeats,
+        'sha256_rows': expected,
+        'scope':
+            'Collection restore latency; unchanged UI; no claim of whole-app FPS or device RAM.',
+      };
+      binding.reportData ??= {};
+      binding.reportData!['collection_restore'] = report;
+      final output = File(
+        '${(await getApplicationSupportDirectory()).path}/collection-restore-benchmark.json',
+      );
+      await output.writeAsString(jsonEncode(report));
+      debugPrint('COLLECTION_RESTORE_REPORT: ${output.path}');
+      debugPrint('COLLECTION_RESTORE_BENCHMARK: ${jsonEncode(report)}');
+    },
+  );
+}

lib/services/collection_restore_codec.dart

@@ -0,0 +1,105 @@
+import 'dart:convert';
+import 'dart:io';
+import 'dart:isolate';
+
+/// Keeps the existing collection JSON/date codec in Dart, away from frames.
+/// Native receives only a path and streams these ordered database commands.
+Iterable<Map<String, dynamic>> collectionRestoreCommands(
+  Map<String, dynamic> collections,
+  String nowIso,
+) sync* {
+  Iterable<Map<String, dynamic>> entries(Object? value) sync* {
+    if (value is! List) return;
+    for (final entry in value.whereType<Map<Object?, Object?>>()) {
+      yield Map<String, dynamic>.from(entry);
+    }
+  }
+
+  Map<String, dynamic>? trackCommand(
+    String kind,
+    Map<String, dynamic> entry, {
+    String? playlistId,
+  }) {
+    final key = entry['key'] as String?;
+    final track = entry['track'];
+    if (key == null || key.isEmpty || track is! Map) return null;
+    return {
+      'kind': kind,
+      'values': {
+        'playlist_id': ?playlistId,
+        'track_key': key,
+        'track_json': jsonEncode(track),
+        'added_at': (entry['addedAt'] as String?) ?? nowIso,
+      },
+    };
+  }
+
+  for (final kind in ['wishlist', 'loved']) {
+    for (final entry in entries(collections[kind])) {
+      final command = trackCommand(kind, entry);
+      if (command != null) yield command;
+    }
+  }
+  for (final entry in entries(collections['favoriteArtists'])) {
+    final key = entry['key'] as String?;
+    if (key == null || key.isEmpty) continue;
+    yield {
+      'kind': 'artist',
+      'values': {
+        'artist_key': key,
+        'artist_json': jsonEncode(entry),
+        'added_at': (entry['addedAt'] as String?) ?? nowIso,
+      },
+    };
+  }
+  for (final playlist in entries(collections['playlists'])) {
+    final id = playlist['id'] as String?;
+    if (id == null || id.isEmpty) continue;
+    final createdAt = (playlist['createdAt'] as String?) ?? nowIso;
+    yield {
+      'kind': 'playlist',
+      'values': {
+        'id': id,
+        'name': (playlist['name'] as String?) ?? '',
+        'cover_image_path': playlist['coverImagePath'] as String?,
+        'created_at': createdAt,
+        'updated_at': (playlist['updatedAt'] as String?) ?? createdAt,
+      },
+    };
+    for (final entry in entries(playlist['tracks'])) {
+      final command = trackCommand('playlist_track', entry, playlistId: id);
+      if (command != null) yield command;
+    }
+  }
+}
+
+Future<List<Map<String, dynamic>>> prepareCollectionRestoreCommands(
+  Map<String, dynamic> collections,
+  String nowIso,
+) => Isolate.run(
+  () => collectionRestoreCommands(collections, nowIso).toList(growable: false),
+);
+
+Future<int> writeCollectionRestoreCommands(
+  String path,
+  Map<String, dynamic> collections,
+  String nowIso,
+) => Isolate.run(() {
+  final output = File(path).openSync(mode: FileMode.write);
+  try {
+    var count = 0;
+    var buffer = StringBuffer();
+    for (final command in collectionRestoreCommands(collections, nowIso)) {
+      buffer.writeln(jsonEncode(command));
+      count++;
+      if (buffer.length >= 64 * 1024) {
+        output.writeStringSync(buffer.toString());
+        buffer = StringBuffer();
+      }
+    }
+    if (buffer.isNotEmpty) output.writeStringSync(buffer.toString());
+    return count;
+  } finally {
+    output.closeSync();
+  }
+});

lib/services/library_collections_database.dart

@@ -1,8 +1,12 @@
 import 'dart:convert';
+import 'dart:io';
 
+import 'package:path_provider/path_provider.dart';
 import 'package:shared_preferences/shared_preferences.dart';
 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/sqlite_helpers.dart' as sqlite;
 import 'package:spotiflac_android/utils/logger.dart';
 
@@ -488,22 +492,56 @@ class LibraryCollectionsDatabase {
     Map<String, dynamic> collectionsJson,
   ) async {
     final nowIso = DateTime.now().toIso8601String();
+    await replaceDatabaseFromBackup(
+      await database,
+      collectionsJson,
+      nowIso: nowIso,
+    );
+  }
 
-    List<Map<String, dynamic>> listOf(String key) {
-      final raw = collectionsJson[key];
-      if (raw is! List) return const [];
-      return raw
-          .whereType<Map<Object?, Object?>>()
-          .map((e) => Map<String, dynamic>.from(e))
-          .toList(growable: false);
+  /// Shares the production restore route with isolated database integration
+  /// fixtures. The caller initializes the app-owned schema before native IO.
+  static Future<void> replaceDatabaseFromBackup(
+    Database db,
+    Map<String, dynamic> collectionsJson, {
+    String? nowIso,
+  }) async {
+    final restoreTime = nowIso ?? DateTime.now().toIso8601String();
+    if (PlatformBridge.supportsCoreBackend) {
+      final temporary = await Directory(
+        (await getTemporaryDirectory()).path,
+      ).createTemp('collection-restore-');
+      try {
+        final path = '${temporary.path}/commands.ndjson';
+        final count = await writeCollectionRestoreCommands(
+          path,
+          collectionsJson,
+          restoreTime,
+        );
+        final result = await PlatformBridge.runNativeDataJob({
+          'operation': 'backup_collections_import',
+          'collections_path': db.path,
+          'ndjson_path': path,
+          'expected_count': count,
+        });
+        if (result['committed'] != true) {
+          throw const FormatException('Incomplete collection restore');
+        }
+      } finally {
+        try {
+          await temporary.delete(recursive: true);
+        } on FileSystemException catch (error) {
+          _log.w('Could not remove collection restore staging: $error');
+        }
+      }
+      _log.i('Restored collections from backup');
+      return;
     }
 
-    final wishlist = listOf('wishlist');
-    final loved = listOf('loved');
-    final playlists = listOf('playlists');
-    final favoriteArtists = listOf('favoriteArtists');
-
-    final db = await database;
+    final commands = await prepareCollectionRestoreCommands(
+      collectionsJson,
+      restoreTime,
+    );
     await db.transaction((txn) async {
       await txn.delete(_tablePlaylistTracks);
       await txn.delete(_tablePlaylists);
@@ -511,63 +549,29 @@ class LibraryCollectionsDatabase {
       await txn.delete(_tableLoved);
       await txn.delete(_tableFavoriteArtists);
 
-      Future<void> insertTrackEntries(
-        String table,
-        List<Map<String, dynamic>> entries,
-      ) async {
-        for (final entry in entries) {
-          final key = entry['key'] as String?;
-          final track = entry['track'];
-          if (key == null || key.isEmpty || track is! Map) continue;
-          await txn.insert(table, {
-            'track_key': key,
-            'track_json': jsonEncode(track),
-            'added_at': (entry['addedAt'] as String?) ?? nowIso,
-          }, conflictAlgorithm: ConflictAlgorithm.replace);
-        }
-      }
-
-      await insertTrackEntries(_tableWishlist, wishlist);
-      await insertTrackEntries(_tableLoved, loved);
-
-      for (final artist in favoriteArtists) {
-        final key = artist['key'] as String?;
-        if (key == null || key.isEmpty) continue;
-        await txn.insert(_tableFavoriteArtists, {
-          'artist_key': key,
-          'artist_json': jsonEncode(artist),
-          'added_at': (artist['addedAt'] as String?) ?? nowIso,
-        }, conflictAlgorithm: ConflictAlgorithm.replace);
-      }
-
-      for (final playlist in playlists) {
-        final id = playlist['id'] as String?;
-        if (id == null || id.isEmpty) continue;
-        final createdAt = (playlist['createdAt'] as String?) ?? nowIso;
-        final updatedAt = (playlist['updatedAt'] as String?) ?? createdAt;
-        await txn.insert(_tablePlaylists, {
-          'id': id,
-          'name': (playlist['name'] as String?) ?? '',
-          'cover_image_path': playlist['coverImagePath'] as String?,
-          'created_at': createdAt,
-          'updated_at': updatedAt,
-        }, conflictAlgorithm: ConflictAlgorithm.replace);
-
-        final tracksRaw = playlist['tracks'];
-        if (tracksRaw is! List) continue;
-        for (final trackEntry in tracksRaw.whereType<Map<Object?, Object?>>()) {
-          final entry = Map<String, dynamic>.from(trackEntry);
-          final key = entry['key'] as String?;
-          final track = entry['track'];
-          if (key == null || key.isEmpty || track is! Map) continue;
-          await txn.insert(_tablePlaylistTracks, {
-            'playlist_id': id,
-            'track_key': key,
-            'track_json': jsonEncode(track),
-            'added_at': (entry['addedAt'] as String?) ?? nowIso,
-          }, conflictAlgorithm: ConflictAlgorithm.replace);
+      var batch = txn.batch();
+      var pending = 0;
+      for (final command in commands) {
+        final table = switch (command['kind']) {
+          'wishlist' => _tableWishlist,
+          'loved' => _tableLoved,
+          'artist' => _tableFavoriteArtists,
+          'playlist' => _tablePlaylists,
+          'playlist_track' => _tablePlaylistTracks,
+          _ => throw const FormatException('Unknown collection command'),
+        };
+        batch.insert(
+          table,
+          command['values'] as Map<String, dynamic>,
+          conflictAlgorithm: ConflictAlgorithm.replace,
+        );
+        if (++pending == 500) {
+          await batch.commit(noResult: true);
+          batch = txn.batch();
+          pending = 0;
         }
       }
+      if (pending > 0) await batch.commit(noResult: true);
     });
 
     _log.i('Restored collections from backup');

rust_backend/crates/mobile/src/native_backup.rs

@@ -1,6 +1,7 @@
 //! History codecs and ZIP payloads stay on the native worker. Backup category
 //! selection, archive layout and presentation remain owned by Flutter.
 mod archive;
+mod collections;
 mod legacy;
 
 use crate::native_library::dart_date;
@@ -105,6 +106,7 @@ pub(crate) fn execute(request: &Value, check: Check<'_>) -> Result<Value, String
     match required(request, "operation")? {
         "backup_history_export" => export(request, check),
         "backup_history_import" => import(request, check),
+        "backup_collections_import" => collections::import(request, check),
         "backup_split_legacy" => legacy::split(request, check),
         "backup_archive_write" => archive::write(request, check),
         "backup_archive_read" => archive::read(request, check),

rust_backend/crates/mobile/src/native_backup/collections.rs

@@ -0,0 +1,221 @@
+//! Dart prepares its exact JSON/date codec on a worker. SQLite stages ordered
+//! commands locally, then publishes all collections in one short transaction.
+use super::{Check, MAX_HISTORY_BYTES, MAX_ROW_BYTES, err, required};
+use rusqlite::{Connection, OpenFlags, TransactionBehavior, params_from_iter};
+use serde_json::{Value, json};
+use std::{
+    fs::File,
+    io::{BufRead, BufReader, Read},
+    path::Path,
+    time::Duration,
+};
+
+struct Table {
+    kind: &'static str,
+    name: &'static str,
+    columns: &'static [&'static str],
+    keys: &'static str,
+}
+
+const TABLES: &[Table] = &[
+    Table {
+        kind: "wishlist",
+        name: "wishlist_tracks",
+        columns: &["track_key", "track_json", "added_at"],
+        keys: "track_key",
+    },
+    Table {
+        kind: "loved",
+        name: "loved_tracks",
+        columns: &["track_key", "track_json", "added_at"],
+        keys: "track_key",
+    },
+    Table {
+        kind: "artist",
+        name: "favorite_artists",
+        columns: &["artist_key", "artist_json", "added_at"],
+        keys: "artist_key",
+    },
+    Table {
+        kind: "playlist",
+        name: "playlists",
+        columns: &["id", "name", "cover_image_path", "created_at", "updated_at"],
+        keys: "id",
+    },
+    Table {
+        kind: "playlist_track",
+        name: "playlist_tracks",
+        columns: &["playlist_id", "track_key", "track_json", "added_at"],
+        keys: "playlist_id,track_key",
+    },
+];
+
+fn stage_batch(db: &mut Connection, rows: &[Value], check: Check<'_>) -> Result<(), String> {
+    let transaction = db
+        .transaction()
+        .map_err(|e| err("stage collection transaction", e))?;
+    for row in rows {
+        check()?;
+        let kind = required(row, "kind")?;
+        let table = TABLES
+            .iter()
+            .find(|table| table.kind == kind)
+            .ok_or("Unknown collection restore command")?;
+        let values = row["values"]
+            .as_object()
+            .ok_or("Collection restore values must be an object")?;
+        let fields = table
+            .columns
+            .iter()
+            .map(|column| match values.get(*column) {
+                Some(Value::String(value)) => Ok(rusqlite::types::Value::Text(value.clone())),
+                None | Some(Value::Null) if *column == "cover_image_path" => {
+                    Ok(rusqlite::types::Value::Null)
+                }
+                _ => Err(format!("Collection restore requires string {column}")),
+            })
+            .collect::<Result<Vec<_>, _>>()?;
+        // REPLACE on a playlist in the original schema cascades old members.
+        // Temporary staging has no FK triggers, so retain that behavior here.
+        if table.kind == "playlist" {
+            transaction
+                .prepare_cached("DELETE FROM restore_playlist_tracks WHERE playlist_id=?")
+                .map_err(|e| err("prepare collection member reset", e))?
+                .execute([&fields[0]])
+                .map_err(|e| err("stage collection member reset", e))?;
+        }
+        let statement = format!(
+            "INSERT OR REPLACE INTO restore_{}({}) VALUES({})",
+            table.name,
+            table.columns.join(","),
+            vec!["?"; table.columns.len()].join(","),
+        );
+        transaction
+            .prepare_cached(&statement)
+            .map_err(|e| err("prepare collection restore", e))?
+            .execute(params_from_iter(fields))
+            .map_err(|e| err("stage collection restore", e))?;
+    }
+    check()?;
+    transaction
+        .commit()
+        .map_err(|e| err("commit collection staging", e))
+}
+
+pub(super) fn import(request: &Value, check: Check<'_>) -> Result<Value, String> {
+    check()?;
+    let path = required(request, "collections_path")?;
+    if !Path::new(path).is_absolute()
+        || Path::new(path).file_name().and_then(|name| name.to_str())
+            != Some("library_collections.db")
+    {
+        return Err("Collection restore requires existing library_collections.db".into());
+    }
+    let mut db = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_WRITE)
+        .map_err(|e| err("open collection restore", e))?;
+    db.busy_timeout(Duration::from_secs(5))
+        .map_err(|e| err("collection restore timeout", e))?;
+    let version: i64 = db
+        .pragma_query_value(None, "user_version", |row| row.get(0))
+        .map_err(|e| err("collection restore schema", e))?;
+    if version != 2 {
+        return Err(format!(
+            "Collection schema v{version} is incompatible with native v2"
+        ));
+    }
+    db.execute_batch(
+        "PRAGMA foreign_keys=ON; PRAGMA recursive_triggers=ON; PRAGMA synchronous=NORMAL;",
+    )
+    .map_err(|e| err("configure collection restore", e))?;
+    for table in TABLES {
+        db.execute_batch(&format!(
+            "CREATE TEMP TABLE restore_{0} AS SELECT {1} FROM main.{0} WHERE 0;
+             CREATE UNIQUE INDEX restore_{0}_keys ON restore_{0}({2});",
+            table.name,
+            table.columns.join(","),
+            table.keys,
+        ))
+        .map_err(|e| err("create collection restore staging", e))?;
+    }
+
+    let input_path = required(request, "ndjson_path")?;
+    let input = File::open(input_path).map_err(|e| err("open collection commands", e))?;
+    let metadata = input
+        .metadata()
+        .map_err(|e| err("inspect collection commands", e))?;
+    if !metadata.is_file() || metadata.len() > MAX_HISTORY_BYTES {
+        return Err("Collection commands exceed size limit or are not a file".into());
+    }
+    let expected = request["expected_count"]
+        .as_u64()
+        .ok_or("Collection restore requires expected_count")?;
+    let mut reader = BufReader::new(input);
+    let mut line = Vec::new();
+    let mut batch = Vec::with_capacity(500);
+    let mut count = 0_u64;
+    loop {
+        check()?;
+        line.clear();
+        let length = reader
+            .by_ref()
+            .take(MAX_ROW_BYTES + 1)
+            .read_until(b'\n', &mut line)
+            .map_err(|e| err("read collection command", e))?;
+        if length == 0 {
+            break;
+        }
+        if length as u64 > MAX_ROW_BYTES {
+            return Err("Collection restore row exceeds size limit".into());
+        }
+        batch.push(serde_json::from_slice(&line).map_err(|e| err("decode collection command", e))?);
+        count += 1;
+        if batch.len() == 500 {
+            stage_batch(&mut db, &batch, check)?;
+            batch.clear();
+        }
+    }
+    if !batch.is_empty() {
+        stage_batch(&mut db, &batch, check)?;
+    }
+    if count != expected {
+        return Err(format!(
+            "Collection restore expected {expected} commands, found {count}"
+        ));
+    }
+    check()?;
+    let transaction = db
+        .transaction_with_behavior(TransactionBehavior::Immediate)
+        .map_err(|e| err("begin collection publication", e))?;
+    for name in [
+        "playlist_tracks",
+        "playlists",
+        "wishlist_tracks",
+        "loved_tracks",
+        "favorite_artists",
+    ] {
+        transaction
+            .execute(&format!("DELETE FROM main.{name}"), [])
+            .map_err(|e| err("replace collection", e))?;
+        check()?;
+    }
+    for table in TABLES {
+        transaction
+            .execute(
+                &format!(
+                    "INSERT INTO main.{0}({1}) SELECT {1} FROM restore_{0} ORDER BY rowid",
+                    table.name,
+                    table.columns.join(","),
+                ),
+                [],
+            )
+            .map_err(|e| err("publish collection restore", e))?;
+        check()?;
+    }
+    transaction
+        .commit()
+        .map_err(|e| err("commit collection restore", e))?;
+    Ok(json!({"committed": true, "command_count": count}))
+}
+
+#[cfg(test)]
+mod tests;

rust_backend/crates/mobile/src/native_backup/collections/tests.rs

@@ -0,0 +1,218 @@
+use super::*;
+use std::{cell::Cell, io::Write};
+use tempfile::TempDir;
+
+struct Fixture {
+    _directory: TempDir,
+    db: Connection,
+    request: Value,
+}
+
+impl Fixture {
+    fn new() -> Self {
+        let directory = tempfile::tempdir().unwrap();
+        let path = directory.path().join("library_collections.db");
+        let db = Connection::open(&path).unwrap();
+        db.execute_batch(
+            "PRAGMA user_version=2; PRAGMA journal_mode=WAL; PRAGMA foreign_keys=ON;
+             CREATE TABLE wishlist_tracks(track_key TEXT PRIMARY KEY,track_json TEXT NOT NULL,added_at TEXT NOT NULL);
+             CREATE TABLE loved_tracks(track_key TEXT PRIMARY KEY,track_json TEXT NOT NULL,added_at TEXT NOT NULL);
+             CREATE TABLE favorite_artists(artist_key TEXT PRIMARY KEY,artist_json TEXT NOT NULL,added_at TEXT NOT NULL);
+             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);
+             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);
+             INSERT INTO wishlist_tracks VALUES('kept','{}','old');
+             CREATE TABLE unrelated(value TEXT); INSERT INTO unrelated VALUES('kept');",
+        ).unwrap();
+        let fixture: Value = serde_json::from_str(include_str!(
+            "../../../../../../test/fixtures/collection_restore_contract.json"
+        ))
+        .unwrap();
+        let commands = fixture["commands"].as_array().unwrap();
+        let input = directory.path().join("commands.ndjson");
+        let mut file = File::create(&input).unwrap();
+        for command in commands {
+            writeln!(file, "{command}").unwrap();
+        }
+        let request = json!({
+            "collections_path":path,"ndjson_path":input,"expected_count":commands.len()
+        });
+        Self {
+            _directory: directory,
+            db,
+            request,
+        }
+    }
+
+    fn keys(&self) -> Vec<String> {
+        self.db
+            .prepare("SELECT track_key FROM wishlist_tracks ORDER BY rowid")
+            .unwrap()
+            .query_map([], |row| row.get(0))
+            .unwrap()
+            .collect::<Result<_, _>>()
+            .unwrap()
+    }
+}
+
+#[test]
+fn dart_codec_duplicates_dates_artwork_and_cascades_survive_native_restore() {
+    let fixture = Fixture::new();
+    let result = import(&fixture.request, &|| Ok(())).unwrap();
+    assert_eq!(result["committed"], true);
+    assert_eq!(result["command_count"], 9);
+    assert_eq!(fixture.keys(), ["example:1"]);
+    let playlist: (String, String, String, String) = fixture
+        .db
+        .query_row(
+            "SELECT name,cover_image_path,created_at,updated_at FROM playlists WHERE id='p'",
+            [],
+            |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
+        )
+        .unwrap();
+    assert_eq!(
+        playlist,
+        (
+            "最新".into(),
+            "/covers/a.jpg".into(),
+            "2026-09-01".into(),
+            "2026-09-01".into()
+        )
+    );
+    let members: Vec<(String, String, String)> = fixture
+        .db
+        .prepare("SELECT track_key,track_json,added_at FROM playlist_tracks ORDER BY rowid")
+        .unwrap()
+        .query_map([], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)))
+        .unwrap()
+        .collect::<Result<_, _>>()
+        .unwrap();
+    assert_eq!(
+        members,
+        [(
+            "example:1".into(),
+            "{\"id\":\"last\"}".into(),
+            "2026-10-03".into()
+        )]
+    );
+    assert_eq!(
+        fixture
+            .db
+            .query_row("SELECT value FROM unrelated", [], |row| row
+                .get::<_, String>(0))
+            .unwrap(),
+        "kept"
+    );
+    assert_eq!(
+        fixture
+            .db
+            .query_row("SELECT artist_json FROM favorite_artists", [], |row| row
+                .get::<_, String>(
+                0
+            ))
+            .unwrap(),
+        "{\"key\":\"example:artist\",\"name\":\"İstanbul\",\"extra\":true}"
+    );
+}
+
+#[test]
+fn cancellation_at_every_precommit_checkpoint_preserves_live_collections() {
+    let fixture = Fixture::new();
+    let calls = Cell::new(0);
+    import(&fixture.request, &|| {
+        calls.set(calls.get() + 1);
+        Ok(())
+    })
+    .unwrap();
+    for cancel_at in 1..=calls.get() {
+        let fixture = Fixture::new();
+        let current = Cell::new(0);
+        let result = import(&fixture.request, &|| {
+            current.set(current.get() + 1);
+            if current.get() == cancel_at {
+                Err("cancelled".into())
+            } else {
+                Ok(())
+            }
+        });
+        assert!(result.is_err(), "checkpoint {cancel_at}");
+        assert_eq!(fixture.keys(), ["kept"], "checkpoint {cancel_at}");
+    }
+}
+
+#[test]
+fn truncated_malformed_unknown_and_orphaned_commands_never_wipe_live_rows() {
+    for mutation in 0..5 {
+        let mut fixture = Fixture::new();
+        let path = fixture.request["ndjson_path"].as_str().unwrap();
+        match mutation {
+            0 => fixture.request["expected_count"] = json!(10),
+            1 => {
+                std::fs::write(path, "{broken\n").unwrap();
+            }
+            2 => {
+                std::fs::write(path, "{\"kind\":\"unknown\",\"values\":{}}\n").unwrap();
+                fixture.request["expected_count"] = json!(1);
+            }
+            3 => {
+                std::fs::write(
+                    path,
+                    "{\"kind\":\"wishlist\",\"values\":{\"track_key\":null}}\n",
+                )
+                .unwrap();
+                fixture.request["expected_count"] = json!(1);
+            }
+            _ => {
+                std::fs::write(path, "{\"kind\":\"playlist_track\",\"values\":{\"playlist_id\":\"missing\",\"track_key\":\"x\",\"track_json\":\"{}\",\"added_at\":\"now\"}}\n").unwrap();
+                fixture.request["expected_count"] = json!(1);
+            }
+        }
+        assert!(import(&fixture.request, &|| Ok(())).is_err());
+        assert_eq!(fixture.keys(), ["kept"]);
+    }
+}
+
+#[test]
+fn empty_restore_clears_categories_and_wrong_schema_is_rejected() {
+    let mut fixture = Fixture::new();
+    fixture.db.execute_batch("PRAGMA user_version=3").unwrap();
+    assert!(import(&fixture.request, &|| Ok(())).is_err());
+    assert_eq!(fixture.keys(), ["kept"]);
+    fixture.db.execute_batch("PRAGMA user_version=2").unwrap();
+    std::fs::write(fixture.request["ndjson_path"].as_str().unwrap(), "").unwrap();
+    fixture.request["expected_count"] = json!(0);
+    assert_eq!(
+        import(&fixture.request, &|| Ok(())).unwrap()["committed"],
+        true
+    );
+    assert!(fixture.keys().is_empty());
+}
+
+#[test]
+fn staging_does_not_hold_main_wal_writer_and_publication_is_atomic() {
+    let fixture = Fixture::new();
+    let calls = Cell::new(0);
+    let changed = Cell::new(false);
+    import(&fixture.request, &|| {
+        calls.set(calls.get() + 1);
+        // The staging transaction has already inserted TEMP rows here.
+        if calls.get() == 16 {
+            changed.set(true);
+            let writer =
+                Connection::open(fixture.request["collections_path"].as_str().unwrap()).unwrap();
+            writer
+                .execute("UPDATE unrelated SET value='external writer'", [])
+                .unwrap();
+        }
+        Ok(())
+    })
+    .unwrap();
+    assert!(changed.get());
+    assert_eq!(
+        fixture
+            .db
+            .query_row("SELECT value FROM unrelated", [], |row| row
+                .get::<_, String>(0))
+            .unwrap(),
+        "external writer"
+    );
+}

test/collection_restore_codec_test.dart

@@ -0,0 +1,86 @@
+import 'dart:convert';
+import 'dart:io';
+
+import 'package:flutter_test/flutter_test.dart';
+import 'package:spotiflac_android/services/collection_restore_codec.dart';
+
+void main() {
+  test(
+    'worker and streamed commands preserve the existing collection codec',
+    () async {
+      final fixture =
+          jsonDecode(
+                File(
+                  'test/fixtures/collection_restore_contract.json',
+                ).readAsStringSync(),
+              )
+              as Map<String, dynamic>;
+      final input = fixture['input'] as Map<String, dynamic>;
+      final now = fixture['now'] as String;
+      expect(
+        collectionRestoreCommands(input, now).toList(),
+        fixture['commands'],
+      );
+      expect(
+        await prepareCollectionRestoreCommands(input, now),
+        fixture['commands'],
+      );
+      final temporary = await Directory.systemTemp.createTemp(
+        'collection-codec-',
+      );
+      addTearDown(() => temporary.delete(recursive: true));
+      final path = '${temporary.path}/commands.ndjson';
+      expect(await writeCollectionRestoreCommands(path, input, now), 9);
+      expect(
+        (await File(path).readAsLines()).map(jsonDecode).toList(),
+        fixture['commands'],
+      );
+    },
+  );
+
+  test('empty or omitted categories produce an empty replacement', () async {
+    expect(await prepareCollectionRestoreCommands({}, 'now'), isEmpty);
+    expect(
+      await prepareCollectionRestoreCommands({
+        'wishlist': false,
+        'loved': <String, dynamic>{},
+        'playlists': [
+          null,
+          {'id': ''},
+        ],
+        'favoriteArtists': [
+          {'key': ''},
+        ],
+      }, 'now'),
+      isEmpty,
+    );
+  });
+
+  test(
+    'invalid typed fields still reject the restore before publication',
+    () async {
+      for (final input in <Map<String, dynamic>>[
+        {
+          'wishlist': [
+            {'key': 7, 'track': <String, dynamic>{}},
+          ],
+        },
+        {
+          'playlists': [
+            {'id': 'p', 'coverImagePath': false},
+          ],
+        },
+        {
+          'favoriteArtists': [
+            {'key': 'a', 'addedAt': 1},
+          ],
+        },
+      ]) {
+        await expectLater(
+          prepareCollectionRestoreCommands(input, 'now'),
+          throwsA(isA<TypeError>()),
+        );
+      }
+    },
+  );
+}

test/fixtures/collection_restore_contract.json

@@ -0,0 +1,32 @@
+{
+  "now": "2026-10-07T05:00:00.000",
+  "input": {
+    "wishlist": [
+      {"key":"example:1","track":{"id":"1","name":"音楽 🎵","extra":[1,true,null]},"addedAt":"2026-10-01"},
+      null,
+      {"key":"","track":{}},
+      {"key":"skipped","track":null}
+    ],
+    "loved": [{"key":"example:2","track":{"id":"2"}}],
+    "favoriteArtists": [{"key":"example:artist","name":"İstanbul","extra":true}],
+    "playlists": [
+      {"id":"p","name":"old","tracks":[{"key":"old","track":{"id":"old"}}]},
+      {"id":"p","name":"最新","coverImagePath":"/covers/a.jpg","createdAt":"2026-09-01","tracks":[
+        {"key":"example:1","track":{"id":"first"},"addedAt":"2026-10-02"},
+        {"key":"example:1","track":{"id":"last"},"addedAt":"2026-10-03"}
+      ]},
+      {"id":"empty"}
+    ]
+  },
+  "commands": [
+    {"kind":"wishlist","values":{"track_key":"example:1","track_json":"{\"id\":\"1\",\"name\":\"音楽 🎵\",\"extra\":[1,true,null]}","added_at":"2026-10-01"}},
+    {"kind":"loved","values":{"track_key":"example:2","track_json":"{\"id\":\"2\"}","added_at":"2026-10-07T05:00:00.000"}},
+    {"kind":"artist","values":{"artist_key":"example:artist","artist_json":"{\"key\":\"example:artist\",\"name\":\"İstanbul\",\"extra\":true}","added_at":"2026-10-07T05:00:00.000"}},
+    {"kind":"playlist","values":{"id":"p","name":"old","cover_image_path":null,"created_at":"2026-10-07T05:00:00.000","updated_at":"2026-10-07T05:00:00.000"}},
+    {"kind":"playlist_track","values":{"playlist_id":"p","track_key":"old","track_json":"{\"id\":\"old\"}","added_at":"2026-10-07T05:00:00.000"}},
+    {"kind":"playlist","values":{"id":"p","name":"最新","cover_image_path":"/covers/a.jpg","created_at":"2026-09-01","updated_at":"2026-09-01"}},
+    {"kind":"playlist_track","values":{"playlist_id":"p","track_key":"example:1","track_json":"{\"id\":\"first\"}","added_at":"2026-10-02"}},
+    {"kind":"playlist_track","values":{"playlist_id":"p","track_key":"example:1","track_json":"{\"id\":\"last\"}","added_at":"2026-10-03"}},
+    {"kind":"playlist","values":{"id":"empty","name":"","cover_image_path":null,"created_at":"2026-10-07T05:00:00.000","updated_at":"2026-10-07T05:00:00.000"}}
+  ]
+}