feat: new mobile asset sync

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

@@ -4,4 +4,6 @@ abstract interface class ISyncApiRepository {
   Future<void> ack(String data);
 
   Stream<List<SyncEvent>> watchUserSyncEvent();
+
+  Stream<List<SyncEvent>> watchAssetSyncEvent();
 }

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

@@ -43,6 +43,28 @@ class SyncStreamService {
     });
   }
 
+  void syncAssets() {
+    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) {
+          final data = event.data as SyncAssetDeleteV1;
+
+          // debugPrint("Asset delete: $data");
+          // await _syncApiRepository.ack(event.ack);
+        }
+      }
+    });
+  }
+
   Future<void> dispose() async {
     await _userSyncSubscription?.cancel();
   }

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

@@ -19,6 +19,14 @@ class SyncApiRepository implements ISyncApiRepository {
     );
   }
 
+  @override
+  Stream<List<SyncEvent>> watchAssetSyncEvent() {
+    return _getSyncStream(
+      SyncStreamDto(types: [SyncRequestType.assetsV1]),
+      methodName: 'watchAssetSyncEvent',
+    );
+  }
+
   @override
   Future<void> ack(String data) {
     return _api.syncApi.sendSyncAck(SyncAckSetDto(acks: [data]));
@@ -26,8 +34,10 @@ class SyncApiRepository implements ISyncApiRepository {
 
   Stream<List<SyncEvent>> _getSyncStream(
     SyncStreamDto dto, {
-    int batchSize = 5000,
+    int batchSize = 20000,
+    String methodName = '',
   }) async* {
+    final stopwatch = Stopwatch()..start();
     final client = http.Client();
     final endpoint = "${_api.apiClient.basePath}/sync/stream";
 
@@ -77,6 +87,9 @@ class SyncApiRepository implements ISyncApiRepository {
         yield await compute(_parseSyncResponse, lines);
       }
       client.close();
+      debugPrint(
+        "[_getSyncStream] [$methodName] Sync stream took ${stopwatch.elapsedMilliseconds}ms",
+      );
     }
   }
 }
@@ -84,6 +97,8 @@ class SyncApiRepository implements ISyncApiRepository {
 const _kResponseMap = <SyncEntityType, Function(dynamic)>{
   SyncEntityType.userV1: SyncUserV1.fromJson,
   SyncEntityType.userDeleteV1: SyncUserDeleteV1.fromJson,
+  SyncEntityType.assetV1: SyncAssetV1.fromJson,
+  SyncEntityType.assetDeleteV1: SyncAssetDeleteV1.fromJson,
 };
 
 // Need to be outside of the class to be able to use compute

mobile/lib/widgets/common/immich_app_bar.dart

@@ -10,6 +10,7 @@ import 'package:immich_mobile/models/backup/backup_state.model.dart';
 import 'package:immich_mobile/models/server_info/server_info.model.dart';
 import 'package:immich_mobile/providers/backup/backup.provider.dart';
 import 'package:immich_mobile/providers/immich_logo_provider.dart';
+import 'package:immich_mobile/providers/infrastructure/sync_stream.provider.dart';
 import 'package:immich_mobile/providers/server_info.provider.dart';
 import 'package:immich_mobile/routing/router.dart';
 import 'package:immich_mobile/widgets/common/app_bar_dialog/app_bar_dialog.dart';
@@ -185,6 +186,12 @@ class ImmichAppBar extends ConsumerWidget implements PreferredSizeWidget {
         },
       ),
       actions: [
+        IconButton(
+          onPressed: () {
+            ref.read(syncStreamServiceProvider).syncAssets();
+          },
+          icon: const Icon(Icons.sync),
+        ),
         if (actions != null)
           ...actions!.map(
             (action) => Padding(