cjones/immich
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

merge main

Browse Source 
# Conflicts:
#	mobile/drift_schemas/main/drift_schema_v10.json
#	mobile/lib/infrastructure/repositories/db.repository.dart
#	mobile/lib/infrastructure/repositories/db.repository.drift.dart
#	mobile/lib/infrastructure/repositories/db.repository.steps.dart
#	mobile/test/drift/main/generated/schema_v10.dart
This commit is contained in:
shenlong-tanwen
2025-09-09 01:06:03 +05:30
parent 67fb9b7a2a 59accbf32a
commit 53fc603d91
52 changed files with 2515 additions and 789 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mobile/lib/domain/services/background_worker.service.dart
+10 -5
View File
@@ -169,15 +169,20 @@ class BackgroundWorkerBgService extends BackgroundWorkerFlutterApi {
try {
_isCleanedUp = true;
_logger.info("Cleaning up background worker");
await _ref.read(backgroundSyncProvider).cancel();
await _ref.read(backgroundSyncProvider).cancelLocal();
final cleanupFutures = [
_drift.close(),
_driftLogger.close(),
_ref.read(backgroundSyncProvider).cancel(),
_ref.read(backgroundSyncProvider).cancelLocal(),
];
if (_isar.isOpen) {
await _isar.close();
cleanupFutures.add(_isar.close());
}
await _drift.close();
await _driftLogger.close();
_ref.dispose();
_lockManager.releaseLock();
await Future.wait(cleanupFutures);
_logger.info("Background worker resources cleaned up");
} catch (error, stack) {
debugPrint('Failed to cleanup background worker: $error with stack: $stack');
mobile/lib/domain/services/sync_stream.service.dart
+59 -43
View File
@@ -23,54 +23,17 @@ class SyncStreamService {
bool get isCancelled => _cancelChecker?.call() ?? false;
Future<void> sync() {
Future<void> sync() async {
_logger.info("Remote sync request for user");
// Start the sync stream and handle events
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);
bool shouldReset = false;
await _syncApiRepository.streamChanges(_handleEvents, onReset: () => shouldReset = true);
if (shouldReset) {
await _syncApiRepository.streamChanges(_handleEvents);
}
}
Future<void> _handleEvents(List<SyncEvent> events, Function() abort) async {
Future<void> _handleEvents(List<SyncEvent> events, Function() abort, Function() reset) async {
List<SyncEvent> items = [];
for (final event in events) {
if (isCancelled) {
@@ -83,6 +46,10 @@ class SyncStreamService {
await _processBatch(items);
}
if (event.type == SyncEntityType.syncResetV1) {
reset();
}
items.add(event);
}
@@ -103,6 +70,8 @@ class SyncStreamService {
Future<void> _handleSyncData(SyncEntityType type, Iterable<Object> data) async {
_logger.fine("Processing sync data for $type of length ${data.length}");
switch (type) {
case SyncEntityType.authUserV1:
return _syncStreamRepository.updateAuthUsersV1(data.cast());
case SyncEntityType.userV1:
return _syncStreamRepository.updateUsersV1(data.cast());
case SyncEntityType.userDeleteV1:
@@ -163,6 +132,12 @@ class SyncStreamService {
// to acknowledge that the client has processed all the backfill events
case SyncEntityType.syncAckV1:
return;
// No-op. SyncCompleteV1 is used to signal the completion of the sync process
case SyncEntityType.syncCompleteV1:
return;
// Request to reset the client state. Clear everything related to remote entities
case SyncEntityType.syncResetV1:
return _syncStreamRepository.reset();
case SyncEntityType.memoryV1:
return _syncStreamRepository.updateMemoriesV1(data.cast());
case SyncEntityType.memoryDeleteV1:
@@ -197,4 +172,45 @@ class SyncStreamService {
_logger.warning("Unknown sync data type: $type");
}
}
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);
}
}
}