From a5e5afd9b161b64edb00d997866b5f400d0075cd Mon Sep 17 00:00:00 2001 From: bwees Date: Wed, 25 Feb 2026 11:32:58 -0600 Subject: [PATCH] fix: websocket handling --- .../domain/services/sync_stream.service.dart | 77 ++++++++----------- mobile/lib/domain/utils/background_sync.dart | 8 +- .../repositories/sync_stream.repository.dart | 23 ++++++ mobile/lib/providers/websocket.provider.dart | 2 +- 4 files changed, 58 insertions(+), 52 deletions(-) diff --git a/mobile/lib/domain/services/sync_stream.service.dart b/mobile/lib/domain/services/sync_stream.service.dart index 9693044212..9769c2eeec 100644 --- a/mobile/lib/domain/services/sync_stream.service.dart +++ b/mobile/lib/domain/services/sync_stream.service.dart @@ -340,60 +340,43 @@ class SyncStreamService { } } - Future handleWsAssetEditReadyV1Batch(List batchData) async { - if (batchData.isEmpty) return; - - _logger.info('Processing batch of ${batchData.length} AssetEditReadyV1 events'); - - final List assets = []; - final List assetEdits = []; + Future handleWsAssetEditReadyV1(dynamic data) async { + _logger.info('Processing AssetEditReadyV1 event'); try { - for (final data in batchData) { - if (data is! Map) { - continue; - } - - final payload = data; - final assetData = payload['asset']; - final editData = payload['edit']; - - if (assetData == null) { - continue; - } - - final asset = SyncAssetV1.fromJson(assetData); - - if (asset != null) { - assets.add(asset); - } - - // Edits are only send on v2.6.0+ - if (editData != null) { - final edits = (editData as List) - .map((e) => SyncAssetEditV1.fromJson(e)) - .whereType() - .toList(); - - assetEdits.addAll(edits); - } + if (data is! Map) { + throw ArgumentError("Invalid data format for AssetEditReadyV1 event"); } - if (assets.isNotEmpty) { - await _syncStreamRepository.updateAssetsV1(assets, debugLabel: 'websocket-edit'); + final payload = data; - // TODO: How do we want to handle this? - // edits that are sent replace previous edits, so we delete existing ones first - // await _syncStreamRepository.deleteAssetEditsV1( - // assets.map((asset) => SyncAssetEditDeleteV1(editId: asset.id)).toList(), - // debugLabel: 'websocket-edit', - // ); - await _syncStreamRepository.updateAssetEditsV1(assetEdits, debugLabel: 'websocket-edit'); - - _logger.info('Successfully processed ${assets.length} edited assets'); + if (payload['asset'] == null) { + throw ArgumentError("Missing 'asset' field in AssetEditReadyV1 event data"); } + + final asset = SyncAssetV1.fromJson(payload['asset']); + if (asset == null) { + throw ArgumentError("Failed to parse 'asset' field in AssetEditReadyV1 event data"); + } + + List assetEdits = []; + + // Edits are only send on v2.6.0+ + if (payload['edit'] != null && payload['edit'] is List) { + assetEdits = (payload['edit'] as List) + .map((e) => SyncAssetEditV1.fromJson(e)) + .whereType() + .toList(); + } + + await _syncStreamRepository.updateAssetsV1([asset], debugLabel: 'websocket-edit'); + await _syncStreamRepository.replaceAssetEditsV1(asset.id, assetEdits, debugLabel: 'websocket-edit'); + + _logger.info( + 'Successfully processed AssetEditReadyV1 event for asset ${asset.id} with ${assetEdits.length} edits', + ); } catch (error, stackTrace) { - _logger.severe("Error processing AssetEditReadyV1 websocket batch events", error, stackTrace); + _logger.severe("Error processing AssetEditReadyV1 websocket event", error, stackTrace); } } diff --git a/mobile/lib/domain/utils/background_sync.dart b/mobile/lib/domain/utils/background_sync.dart index 6840bae595..7c9b6ae061 100644 --- a/mobile/lib/domain/utils/background_sync.dart +++ b/mobile/lib/domain/utils/background_sync.dart @@ -196,11 +196,11 @@ class BackgroundSyncManager { }); } - Future syncWebsocketEditBatch(List batchData) { + Future syncWebsocketEdit(dynamic data) { if (_syncWebsocketTask != null) { return _syncWebsocketTask!.future; } - _syncWebsocketTask = _handleWsAssetEditReadyV1Batch(batchData); + _syncWebsocketTask = _handleWsAssetEditReadyV1(data); return _syncWebsocketTask!.whenComplete(() { _syncWebsocketTask = null; }); @@ -242,7 +242,7 @@ Cancelable _handleWsAssetUploadReadyV1Batch(List batchData) => ru debugLabel: 'websocket-batch', ); -Cancelable _handleWsAssetEditReadyV1Batch(List batchData) => runInIsolateGentle( - computation: (ref) => ref.read(syncStreamServiceProvider).handleWsAssetEditReadyV1Batch(batchData), +Cancelable _handleWsAssetEditReadyV1(dynamic data) => runInIsolateGentle( + computation: (ref) => ref.read(syncStreamServiceProvider).handleWsAssetEditReadyV1(data), debugLabel: 'websocket-edit', ); diff --git a/mobile/lib/infrastructure/repositories/sync_stream.repository.dart b/mobile/lib/infrastructure/repositories/sync_stream.repository.dart index 0dd5ec1147..4319ee63cf 100644 --- a/mobile/lib/infrastructure/repositories/sync_stream.repository.dart +++ b/mobile/lib/infrastructure/repositories/sync_stream.repository.dart @@ -346,6 +346,29 @@ class SyncStreamRepository extends DriftDatabaseRepository { } } + Future replaceAssetEditsV1(String assetId, Iterable data, {String debugLabel = 'user'}) async { + try { + await _db.batch((batch) { + batch.deleteWhere(_db.assetEditEntity, (row) => row.assetId.equals(assetId)); + + for (final edit in data) { + final companion = AssetEditEntityCompanion( + id: Value(edit.id), + assetId: Value(edit.assetId), + action: Value(edit.action.toAssetEditAction()), + parameters: Value(edit.parameters as Map), + sequence: Value(edit.sequence), + ); + + batch.insert(_db.assetEditEntity, companion); + } + }); + } catch (error, stack) { + _logger.severe('Error: replaceAssetEditsV1 - $debugLabel', error, stack); + rethrow; + } + } + Future deleteAssetEditsV1(Iterable data, {String debugLabel = 'user'}) async { try { await _db.batch((batch) { diff --git a/mobile/lib/providers/websocket.provider.dart b/mobile/lib/providers/websocket.provider.dart index f9473ce440..131f6fb1d3 100644 --- a/mobile/lib/providers/websocket.provider.dart +++ b/mobile/lib/providers/websocket.provider.dart @@ -319,7 +319,7 @@ class WebsocketNotifier extends StateNotifier { } void _handleSyncAssetEditReady(dynamic data) { - unawaited(_ref.read(backgroundSyncProvider).syncWebsocketEditBatch([data])); + unawaited(_ref.read(backgroundSyncProvider).syncWebsocketEdit(data)); } void _processBatchedAssetUploadReady() {