fix: reset sqlite on beta migration (#20735)

reset sync stream on migration

Co-authored-by: shenlong-tanwen <139912620+shalong-tanwen@users.noreply.github.com>

mobile/lib/domain/models/store.model.dart

@@ -77,7 +77,9 @@ enum StoreKey<T> {
   enableBackup<bool>._(1003),
   useWifiForUploadVideos<bool>._(1004),
   useWifiForUploadPhotos<bool>._(1005),
-  needBetaMigration<bool>._(1006);
+  needBetaMigration<bool>._(1006),
+  // TODO: Remove this after patching open-api
+  shouldResetSync<bool>._(1007);
 
   const StoreKey._(this.id);
   final int id;

mobile/lib/domain/services/sync_stream.service.dart

@@ -29,6 +29,7 @@ class SyncStreamService {
     bool shouldReset = false;
     await _syncApiRepository.streamChanges(_handleEvents, onReset: () => shouldReset = true);
     if (shouldReset) {
+      _logger.info("Resetting sync state as requested by server");
       await _syncApiRepository.streamChanges(_handleEvents);
     }
   }

mobile/lib/infrastructure/repositories/db.repository.dart

@@ -69,6 +69,29 @@ class Drift extends $Drift implements IDatabaseRepository {
   Drift([QueryExecutor? executor])
     : super(executor ?? driftDatabase(name: 'immich', native: const DriftNativeOptions(shareAcrossIsolates: true)));
 
+  Future<void> reset() async {
+    // https://github.com/simolus3/drift/commit/bd80a46264b6dd833ef4fd87fffc03f5a832ab41#diff-3f879e03b4a35779344ef16170b9353608dd9c42385f5402ec6035aac4dd8a04R76-R94
+    await exclusively(() async {
+      // https://stackoverflow.com/a/65743498/25690041
+      await customStatement('PRAGMA writable_schema = 1;');
+      await customStatement('DELETE FROM sqlite_master;');
+      await customStatement('VACUUM;');
+      await customStatement('PRAGMA writable_schema = 0;');
+      await customStatement('PRAGMA integrity_check');
+
+      await customStatement('PRAGMA user_version = 0');
+      await beforeOpen(
+        // ignore: invalid_use_of_internal_member
+        resolvedEngine.executor,
+        OpeningDetails(null, schemaVersion),
+      );
+      await customStatement('PRAGMA user_version = $schemaVersion');
+
+      // Refresh all stream queries
+      notifyUpdates({for (final table in allTables) TableUpdate.onTable(table)});
+    });
+  }
+
   @override
   int get schemaVersion => 10;
 

mobile/lib/infrastructure/repositories/sync_api.repository.dart

@@ -3,7 +3,9 @@ import 'dart:convert';
 
 import 'package:http/http.dart' as http;
 import 'package:immich_mobile/constants/constants.dart';
+import 'package:immich_mobile/domain/models/store.model.dart';
 import 'package:immich_mobile/domain/models/sync_event.model.dart';
+import 'package:immich_mobile/entities/store.entity.dart';
 import 'package:immich_mobile/services/api.service.dart';
 import 'package:logging/logging.dart';
 import 'package:openapi/api.dart';
@@ -33,6 +35,7 @@ class SyncApiRepository {
     await _api.applyToParams([], headerParams);
     headers.addAll(headerParams);
 
+    final shouldReset = Store.get(StoreKey.shouldResetSync, false);
     final request = http.Request('POST', Uri.parse(endpoint));
     request.headers.addAll(headers);
     request.body = jsonEncode(
@@ -58,6 +61,7 @@ class SyncApiRepository {
           SyncRequestType.peopleV1,
           SyncRequestType.assetFacesV1,
         ],
+        reset: shouldReset,
       ).toJson(),
     );
 
@@ -81,6 +85,9 @@ class SyncApiRepository {
         throw ApiException(response.statusCode, 'Failed to get sync stream: $errorBody');
       }
 
+      // Reset after successful stream start
+      await Store.put(StoreKey.shouldResetSync, false);
+
       await for (final chunk in response.stream.transform(utf8.decoder)) {
         if (shouldAbort) {
           break;

mobile/lib/pages/common/change_experience.page.dart

@@ -91,6 +91,8 @@ class _ChangeExperiencePageState extends ConsumerState<ChangeExperiencePage> {
       ref.read(websocketProvider.notifier).stopListenToOldEvents();
       ref.read(websocketProvider.notifier).startListeningToBetaEvents();
 
+      await ref.read(driftProvider).reset();
+      await Store.put(StoreKey.shouldResetSync, true);
       final permission = await ref.read(galleryPermissionNotifier.notifier).requestGalleryPermission();
 
       if (permission.isGranted) {

mobile/lib/utils/migration.dart

@@ -21,13 +21,14 @@ import 'package:immich_mobile/infrastructure/entities/store.entity.dart';
 import 'package:immich_mobile/infrastructure/entities/store.entity.drift.dart';
 import 'package:immich_mobile/infrastructure/entities/user.entity.dart';
 import 'package:immich_mobile/infrastructure/repositories/db.repository.dart';
+import 'package:immich_mobile/infrastructure/repositories/sync_stream.repository.dart';
+import 'package:immich_mobile/utils/debug_print.dart';
 import 'package:immich_mobile/utils/diff.dart';
 import 'package:isar/isar.dart';
 // ignore: import_rule_photo_manager
 import 'package:photo_manager/photo_manager.dart';
-import 'package:immich_mobile/utils/debug_print.dart';
 
-const int targetVersion = 15;
+const int targetVersion = 16;
 
 Future<void> migrateDatabaseIfNeeded(Isar db, Drift drift) async {
   final hasVersion = Store.tryGet(StoreKey.version) != null;
@@ -76,11 +77,16 @@ Future<void> migrateDatabaseIfNeeded(Isar db, Drift drift) async {
       await Store.put(StoreKey.needBetaMigration, false);
       await Store.put(StoreKey.betaTimeline, true);
     } else {
-      await resetDriftDatabase(drift);
+      await drift.reset();
       await Store.put(StoreKey.needBetaMigration, true);
     }
   }
 
+  if (version < 16) {
+    await SyncStreamRepository(drift).reset();
+    await Store.put(StoreKey.shouldResetSync, true);
+  }
+
   if (targetVersion >= 12) {
     await Store.put(StoreKey.version, targetVersion);
     return;
@@ -298,27 +304,3 @@ class _DeviceAsset {
 
   const _DeviceAsset({required this.assetId, this.hash, this.dateTime});
 }
-
-Future<void> resetDriftDatabase(Drift drift) async {
-  // https://github.com/simolus3/drift/commit/bd80a46264b6dd833ef4fd87fffc03f5a832ab41#diff-3f879e03b4a35779344ef16170b9353608dd9c42385f5402ec6035aac4dd8a04R76-R94
-  final database = drift.attachedDatabase;
-  await database.exclusively(() async {
-    // https://stackoverflow.com/a/65743498/25690041
-    await database.customStatement('PRAGMA writable_schema = 1;');
-    await database.customStatement('DELETE FROM sqlite_master;');
-    await database.customStatement('VACUUM;');
-    await database.customStatement('PRAGMA writable_schema = 0;');
-    await database.customStatement('PRAGMA integrity_check');
-
-    await database.customStatement('PRAGMA user_version = 0');
-    await database.beforeOpen(
-      // ignore: invalid_use_of_internal_member
-      database.resolvedEngine.executor,
-      OpeningDetails(null, database.schemaVersion),
-    );
-    await database.customStatement('PRAGMA user_version = ${database.schemaVersion}');
-
-    // Refresh all stream queries
-    database.notifyUpdates({for (final table in database.allTables) TableUpdate.onTable(table)});
-  });
-}

mobile/lib/widgets/settings/beta_sync_settings/sync_status_and_actions.dart

@@ -5,13 +5,12 @@ import 'package:hooks_riverpod/hooks_riverpod.dart';
 import 'package:immich_mobile/extensions/build_context_extensions.dart';
 import 'package:immich_mobile/extensions/translate_extensions.dart';
 import 'package:immich_mobile/providers/background_sync.provider.dart';
-import 'package:immich_mobile/providers/infrastructure/asset.provider.dart';
 import 'package:immich_mobile/providers/infrastructure/album.provider.dart';
+import 'package:immich_mobile/providers/infrastructure/asset.provider.dart';
 import 'package:immich_mobile/providers/infrastructure/db.provider.dart';
 import 'package:immich_mobile/providers/infrastructure/memory.provider.dart';
 import 'package:immich_mobile/providers/infrastructure/storage.provider.dart';
 import 'package:immich_mobile/providers/sync_status.provider.dart';
-import 'package:immich_mobile/utils/migration.dart';
 import 'package:immich_mobile/widgets/settings/beta_sync_settings/entity_count_tile.dart';
 import 'package:path/path.dart' as path;
 import 'package:path_provider/path_provider.dart';
@@ -83,7 +82,7 @@ class SyncStatusAndActions extends HookConsumerWidget {
               ),
               TextButton(
                 onPressed: () async {
-                  await resetDriftDatabase(ref.read(driftProvider));
+                  await ref.read(driftProvider).reset();
                   context.pop();
                   context.scaffoldMessenger.showSnackBar(
                     SnackBar(content: Text("reset_sqlite_success".t(context: context))),

mobile/test/infrastructure/repositories/sync_api_repository_test.dart

@@ -4,12 +4,15 @@ import 'dart:convert';
 import 'package:flutter_test/flutter_test.dart';
 import 'package:http/http.dart' as http;
 import 'package:immich_mobile/domain/models/sync_event.model.dart';
+import 'package:immich_mobile/domain/services/store.service.dart';
+import 'package:immich_mobile/infrastructure/repositories/store.repository.dart';
 import 'package:immich_mobile/infrastructure/repositories/sync_api.repository.dart';
 import 'package:mocktail/mocktail.dart';
 import 'package:openapi/api.dart';
 
 import '../../api.mocks.dart';
 import '../../service.mocks.dart';
+import '../../test_utils.dart';
 
 class MockHttpClient extends Mock implements http.Client {}
 
@@ -33,6 +36,10 @@ void main() {
   late StreamController<List<int>> responseStreamController;
   late int testBatchSize = 3;
 
+  setUpAll(() async {
+    await StoreService.init(storeRepository: IsarStoreRepository(await TestUtils.initIsar()));
+  });
+
   setUp(() {
     mockApiService = MockApiService();
     mockApiClient = MockApiClient();