feat: websocket background sync (#19888)

* feat: websocket background sync

* batch websocket

* pr feedback

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

@@ -31,6 +31,59 @@ class SyncStreamService {
     return _syncApiRepository.streamChanges(_handleEvents);
   }
 
+  Future<void> handleWsAssetUploadReadyV1Batch(List<dynamic> batchData) async {
+    if (batchData.isEmpty) return;
+
+    _logger.info(
+      'Processing batch of ${batchData.length} AssetUploadReadyV1 events',
+    );
+
+    final List<SyncAssetV1> assets = [];
+    final List<SyncAssetExifV1> exifs = [];
+
+    try {
+      for (final data in batchData) {
+        if (data is! Map<String, dynamic>) {
+          continue;
+        }
+
+        final payload = data;
+        final assetData = payload['asset'];
+        final exifData = payload['exif'];
+
+        if (assetData == null || exifData == null) {
+          continue;
+        }
+
+        final asset = SyncAssetV1.fromJson(assetData);
+        final exif = SyncAssetExifV1.fromJson(exifData);
+
+        if (asset != null && exif != null) {
+          assets.add(asset);
+          exifs.add(exif);
+        }
+      }
+
+      if (assets.isNotEmpty && exifs.isNotEmpty) {
+        await _syncStreamRepository.updateAssetsV1(
+          assets,
+          debugLabel: 'websocket-batch',
+        );
+        await _syncStreamRepository.updateAssetsExifV1(
+          exifs,
+          debugLabel: 'websocket-batch',
+        );
+        _logger.info('Successfully processed ${assets.length} assets in batch');
+      }
+    } catch (error, stackTrace) {
+      _logger.severe(
+        "Error processing AssetUploadReadyV1 websocket batch events",
+        error,
+        stackTrace,
+      );
+    }
+  }
+
   Future<void> _handleEvents(List<SyncEvent> events, Function() abort) async {
     List<SyncEvent> items = [];
     for (final event in events) {

mobile/lib/domain/utils/background_sync.dart

@@ -6,6 +6,7 @@ import 'package:worker_manager/worker_manager.dart';
 
 class BackgroundSyncManager {
   Cancelable<void>? _syncTask;
+  Cancelable<void>? _syncWebsocketTask;
   Cancelable<void>? _deviceAlbumSyncTask;
   Cancelable<void>? _hashTask;
 
@@ -20,6 +21,12 @@ class BackgroundSyncManager {
     _syncTask?.cancel();
     _syncTask = null;
 
+    if (_syncWebsocketTask != null) {
+      futures.add(_syncWebsocketTask!.future);
+    }
+    _syncWebsocketTask?.cancel();
+    _syncWebsocketTask = null;
+
     return Future.wait(futures);
   }
 
@@ -72,4 +79,19 @@ class BackgroundSyncManager {
       _syncTask = null;
     });
   }
+
+  Future<void> syncWebsocketBatch(List<dynamic> batchData) {
+    if (_syncWebsocketTask != null) {
+      return _syncWebsocketTask!.future;
+    }
+
+    _syncWebsocketTask = runInIsolateGentle(
+      computation: (ref) => ref
+          .read(syncStreamServiceProvider)
+          .handleWsAssetUploadReadyV1Batch(batchData),
+    );
+    return _syncWebsocketTask!.whenComplete(() {
+      _syncWebsocketTask = null;
+    });
+  }
 }