feat: new mobile asset sync

mobile/lib/domain/interfaces/sync_api.interface.dart

@@ -6,4 +6,6 @@ abstract interface class ISyncApiRepository {
   Stream<List<SyncEvent>> watchUserSyncEvent();
 
   Stream<List<SyncEvent>> watchAssetSyncEvent();
+
+  Stream<List<SyncEvent>> watchExifSyncEvent();
 }

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

@@ -10,62 +10,117 @@ class SyncStreamService {
   SyncStreamService(this._syncApiRepository);
 
   StreamSubscription? _userSyncSubscription;
+  StreamSubscription? _assetSyncSubscription;
+  StreamSubscription? _exifSyncSubscription;
+
+  bool _isUserSyncing = false;
+  bool _isAssetSyncing = false;
+  bool _isExifSyncing = false;
 
   void syncUsers() {
-    _userSyncSubscription =
+    if (_isUserSyncing) {
-        _syncApiRepository.watchUserSyncEvent().listen((events) async {
+      return;
-      for (final event in events) {
+    }
-        if (event.data is SyncUserV1) {
-          final data = event.data as SyncUserV1;
-          debugPrint("User Update: $data");
 
-          // final user = await _userRepository.get(data.id);
+    _isUserSyncing = true;
+    _userSyncSubscription = _syncApiRepository.watchUserSyncEvent().listen(
+      (events) async {
+        for (final event in events) {
+          if (event.data is SyncUserV1) {
+            final data = event.data as SyncUserV1;
+            debugPrint("User Update: $data");
 
-          // if (user == null) {
+            // await _syncApiRepository.ack(event.ack);
-          //   continue;
+          }
-          // }
 
-          // user.name = data.name;
+          if (event.data is SyncUserDeleteV1) {
-          // user.email = data.email;
+            final data = event.data as SyncUserDeleteV1;
-          // user.updatedAt = DateTime.now();
 
-          // await _userRepository.update(user);
+            debugPrint("User delete: $data");
-          // await _syncApiRepository.ack(event.ack);
+            // await _syncApiRepository.ack(event.ack);
+          }
         }
-
+        await _syncApiRepository.ack(events.last.ack);
-        if (event.data is SyncUserDeleteV1) {
+      },
-          final data = event.data as SyncUserDeleteV1;
+      onDone: () {
-
+        _isUserSyncing = false;
-          debugPrint("User delete: $data");
+      },
-          // await _syncApiRepository.ack(event.ack);
+      onError: (_) {
-        }
+        _isUserSyncing = false;
-      }
+      },
-    });
+    );
   }
 
   void syncAssets() {
+    if (_isAssetSyncing) {
+      debugPrint("Asset syncing already in progress");
+      return;
+    }
+    _isAssetSyncing = true;
     int eventCount = 0;
-    _syncApiRepository.watchAssetSyncEvent().listen((events) async {
-      eventCount += events.length;
-      debugPrint("Asset events: $eventCount");
-      for (final event in events) {
-        if (event.data is SyncAssetV1) {
-          final data = event.data as SyncAssetV1;
-          // debugPrint("Asset Update: $data");
-          // await _syncApiRepository.ack(event.ack);
-        }
 
-        if (event.data is SyncAssetDeleteV1) {
+    _assetSyncSubscription = _syncApiRepository.watchAssetSyncEvent().listen(
-          final data = event.data as SyncAssetDeleteV1;
+      (events) async {
+        eventCount += events.length;
+        debugPrint("Asset events: $eventCount");
+        for (final event in events) {
+          if (event.data is SyncAssetV1) {
+            // final data = event.data as SyncAssetV1;
+            // await _syncApiRepository.ack(event.ack);
+          }
 
-          // debugPrint("Asset delete: $data");
+          if (event.data is SyncAssetDeleteV1) {
-          // await _syncApiRepository.ack(event.ack);
+            // final data = event.data as SyncAssetDeleteV1;
+
+            // debugPrint("Asset delete: $data");
+            // await _syncApiRepository.ack(event.ack);
+          }
         }
-      }
+        await _syncApiRepository.ack(events.last.ack);
-    });
+      },
+      onDone: () {
+        _isAssetSyncing = false;
+      },
+      onError: (_) {
+        _isAssetSyncing = false;
+      },
+    );
+  }
+
+  void syncExif() {
+    if (_isExifSyncing) {
+      debugPrint("EXIF syncing already in progress");
+      return;
+    }
+
+    _isExifSyncing = true;
+    int eventCount = 0;
+
+    _exifSyncSubscription = _syncApiRepository.watchExifSyncEvent().listen(
+      (events) async {
+        eventCount += events.length;
+        debugPrint("exif events: $eventCount");
+        for (final event in events) {
+          if (event.data is SyncAssetExifV1) {
+            // final data = event.data as SyncAssetExifV1;
+
+            // await _syncApiRepository.ack(event.ack);
+          }
+        }
+        await _syncApiRepository.ack(events.last.ack);
+      },
+      onDone: () {
+        _isExifSyncing = false;
+      },
+      onError: (_) {
+        _isExifSyncing = false;
+      },
+    );
   }
 
   Future<void> dispose() async {
     await _userSyncSubscription?.cancel();
+    await _assetSyncSubscription?.cancel();
+    await _exifSyncSubscription?.cancel();
   }
 }

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

@@ -16,17 +16,33 @@ class SyncApiRepository implements ISyncApiRepository {
   Stream<List<SyncEvent>> watchUserSyncEvent() {
     return _getSyncStream(
       SyncStreamDto(types: [SyncRequestType.usersV1]),
+      methodName: 'watchUserSyncEvent',
     );
   }
 
   @override
   Stream<List<SyncEvent>> watchAssetSyncEvent() {
     return _getSyncStream(
-      SyncStreamDto(types: [SyncRequestType.assetsV1]),
+      SyncStreamDto(
+        types: [SyncRequestType.assetsV1, SyncRequestType.partnerAssetsV1],
+      ),
       methodName: 'watchAssetSyncEvent',
     );
   }
 
+  @override
+  Stream<List<SyncEvent>> watchExifSyncEvent() {
+    return _getSyncStream(
+      SyncStreamDto(
+        types: [
+          SyncRequestType.assetExifsV1,
+          SyncRequestType.partnerAssetExifsV1,
+        ],
+      ),
+      methodName: 'watchExifSyncEvent',
+    );
+  }
+
   @override
   Future<void> ack(String data) {
     return _api.syncApi.sendSyncAck(SyncAckSetDto(acks: [data]));
@@ -34,7 +50,7 @@ class SyncApiRepository implements ISyncApiRepository {
 
   Stream<List<SyncEvent>> _getSyncStream(
     SyncStreamDto dto, {
-    int batchSize = 20000,
+    int batchSize = 10000,
     String methodName = '',
   }) async* {
     final stopwatch = Stopwatch()..start();
@@ -95,17 +111,33 @@ class SyncApiRepository implements ISyncApiRepository {
 }
 
 const _kResponseMap = <SyncEntityType, Function(dynamic)>{
+  /// user
   SyncEntityType.userV1: SyncUserV1.fromJson,
   SyncEntityType.userDeleteV1: SyncUserDeleteV1.fromJson,
+
+  /// partners
+  SyncEntityType.partnerV1: SyncPartnerV1.fromJson,
+  SyncEntityType.partnerDeleteV1: SyncPartnerDeleteV1.fromJson,
+
+  /// assets
   SyncEntityType.assetV1: SyncAssetV1.fromJson,
   SyncEntityType.assetDeleteV1: SyncAssetDeleteV1.fromJson,
+  SyncEntityType.assetExifV1: SyncAssetExifV1.fromJson,
+
+  /// partners' assets
+  SyncEntityType.partnerAssetV1: SyncAssetV1.fromJson,
+  SyncEntityType.partnerAssetDeleteV1: SyncAssetDeleteV1.fromJson,
+  SyncEntityType.partnerAssetExifV1: SyncAssetExifV1.fromJson,
+
+  /// album
 };
 
 // Need to be outside of the class to be able to use compute
 List<SyncEvent> _parseSyncResponse(List<String> lines) {
+  final stopwatch = Stopwatch()..start();
   final List<SyncEvent> data = [];
 
-  for (var line in lines) {
+  for (final line in lines) {
     try {
       final jsonData = jsonDecode(line);
       final type = SyncEntityType.fromJson(jsonData['type'])!;
@@ -113,15 +145,19 @@ List<SyncEvent> _parseSyncResponse(List<String> lines) {
       final ack = jsonData['ack'];
       final converter = _kResponseMap[type];
       if (converter == null) {
-        debugPrint("[_parseSyncReponse] Unknown type $type");
+        debugPrint("[_parseSyncResponse] Unknown type $type");
         continue;
       }
 
       data.add(SyncEvent(data: converter(dataJson), ack: ack));
     } catch (error, stack) {
-      debugPrint("[_parseSyncReponse] Error parsing json $error $stack");
+      debugPrint("[_parseSyncResponse] Error parsing json $error $stack");
     }
   }
 
+  debugPrint(
+    "[_parseSyncResponse] Parsed ${data.length} events in ${stopwatch.elapsedMilliseconds}ms",
+  );
+
   return data;
 }

mobile/lib/widgets/common/immich_app_bar.dart

@@ -188,6 +188,7 @@ class ImmichAppBar extends ConsumerWidget implements PreferredSizeWidget {
       actions: [
         IconButton(
           onPressed: () {
+            ref.read(syncStreamServiceProvider).syncExif();
             ref.read(syncStreamServiceProvider).syncAssets();
           },
           icon: const Icon(Icons.sync),

mobile/lib/widgets/forms/login/login_form.dart

@@ -165,9 +165,13 @@ class LoginForm extends HookConsumerWidget {
     }
 
     populateTestLoginInfo1() {
-      emailController.text = 'testuser@email.com';
+      emailController.text = 'admin@immich.cloud';
-      passwordController.text = 'password';
+      passwordController.text = 'kthxbye';
-      serverEndpointController.text = 'http://10.1.15.216:2283/api';
+      serverEndpointController.text =
+          'https://pr-16658.preview.internal.immich.cloud/';
+      // emailController.text = 'testuser@email.com';
+      // passwordController.text = 'password';
+      // serverEndpointController.text = 'http://10.1.15.216:2283/api';
     }
 
     login() async {