feat: user sync stream

mobile/analysis_options.yaml

@@ -35,6 +35,7 @@ linter:
 analyzer:
   exclude:
     - openapi/**
+    - build/**
     - lib/generated_plugin_registrant.dart
     - lib/**/*.g.dart
     - lib/**/*.drift.dart
@@ -92,6 +93,9 @@ custom_lint:
       allowed:
         # required / wanted
         - lib/repositories/*_api.repository.dart
+        - lib/domain/models/sync_event.model.dart
+        - lib/{domain,infrastructure}/**/sync_stream.*
+        - lib/{domain,infrastructure}/**/sync_api.*
         - lib/infrastructure/repositories/*_api.repository.dart
         - lib/infrastructure/utils/*.converter.dart
         # acceptable exceptions for the time being
@@ -144,7 +148,9 @@ dart_code_metrics:
     - avoid-global-state
     - avoid-inverted-boolean-checks
     - avoid-late-final-reassignment
-    - avoid-local-functions
+    - avoid-local-functions:
+        exclude:
+          - test/**.dart
     - avoid-negated-conditions
     - avoid-nested-streams-and-futures
     - avoid-referencing-subclasses

mobile/devtools_options.yaml

@@ -1,3 +1,4 @@
 description: This file stores settings for Dart & Flutter DevTools.
 documentation: https://docs.flutter.dev/tools/devtools/extensions#configure-extension-enablement-states
 extensions:
+  - drift: true

mobile/lib/constants/constants.dart

@@ -5,5 +5,9 @@ const double downloadFailed = -2;
 // Number of log entries to retain on app start
 const int kLogTruncateLimit = 250;
 
+// Sync
+const int kSyncEventBatchSize = 5000;
+
+// Hash batch limits
 const int kBatchHashFileLimit = 128;
 const int kBatchHashSizeLimit = 1024 * 1024 * 1024; // 1GB

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

@@ -1,7 +1,8 @@
-import 'package:immich_mobile/domain/models/sync/sync_event.model.dart';
+import 'package:immich_mobile/domain/models/sync_event.model.dart';
+import 'package:openapi/api.dart';
 
 abstract interface class ISyncApiRepository {
-  Future<void> ack(String data);
+  Future<void> ack(List<String> data);
 
-  Stream<List<SyncEvent>> watchUserSyncEvent();
+  Stream<List<SyncEvent>> getSyncEvents(List<SyncRequestType> type);
 }

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

@@ -0,0 +1,10 @@
+import 'package:immich_mobile/domain/interfaces/db.interface.dart';
+import 'package:openapi/api.dart';
+
+abstract interface class ISyncStreamRepository implements IDatabaseRepository {
+  Future<bool> updateUsersV1(Iterable<SyncUserV1> data);
+  Future<bool> deleteUsersV1(Iterable<SyncUserDeleteV1> data);
+
+  Future<bool> updatePartnerV1(Iterable<SyncPartnerV1> data);
+  Future<bool> deletePartnerV1(Iterable<SyncPartnerDeleteV1> data);
+}

mobile/lib/domain/models/sync_event.model.dart

@@ -0,0 +1,13 @@
+import 'package:openapi/api.dart';
+
+class SyncEvent {
+  final SyncEntityType type;
+  // ignore: avoid-dynamic
+  final dynamic data;
+  final String ack;
+
+  const SyncEvent({required this.type, required this.data, required this.ack});
+
+  @override
+  String toString() => 'SyncEvent(type: $type, data: $data, ack: $ack)';
+}

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

@@ -1,49 +1,175 @@
+// ignore_for_file: avoid-passing-async-when-sync-expected
+
 import 'dart:async';
 
-import 'package:flutter/foundation.dart';
+import 'package:collection/collection.dart';
 import 'package:immich_mobile/domain/interfaces/sync_api.interface.dart';
+import 'package:immich_mobile/domain/interfaces/sync_stream.interface.dart';
+import 'package:immich_mobile/domain/utils/cancel.exception.dart';
+import 'package:logging/logging.dart';
 import 'package:openapi/api.dart';
 
 class SyncStreamService {
+  final Logger _logger = Logger('SyncStreamService');
+
   final ISyncApiRepository _syncApiRepository;
+  final ISyncStreamRepository _syncStreamRepository;
+  final Completer<bool>? _cancelCompleter;
+
+  SyncStreamService({
+    required ISyncApiRepository syncApiRepository,
+    required ISyncStreamRepository syncStreamRepository,
+    Completer<bool>? cancelCompleter,
+  })  : _syncApiRepository = syncApiRepository,
+        _syncStreamRepository = syncStreamRepository,
+        _cancelCompleter = cancelCompleter;
 
-  SyncStreamService(this._syncApiRepository);
+  Future<bool> _handleSyncData(
+    SyncEntityType type,
+    // ignore: avoid-dynamic
+    Iterable<dynamic> data,
+  ) async {
+    if (data.isEmpty) {
+      _logger.warning("Received empty sync data for $type");
+      return false;
+    }
 
-  StreamSubscription? _userSyncSubscription;
+    _logger.fine("Processing sync data for $type of length ${data.length}");
 
-  void syncUsers() {
-    _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");
+    try {
+      if (type == SyncEntityType.partnerV1) {
+        return await _syncStreamRepository.updatePartnerV1(data.cast());
+      }
+
+      if (type == SyncEntityType.partnerDeleteV1) {
+        return await _syncStreamRepository.deletePartnerV1(data.cast());
+      }
 
-          // final user = await _userRepository.get(data.id);
+      if (type == SyncEntityType.userV1) {
+        return await _syncStreamRepository.updateUsersV1(data.cast());
+      }
 
-          // if (user == null) {
-          //   continue;
-          // }
+      if (type == SyncEntityType.userDeleteV1) {
+        return await _syncStreamRepository.deleteUsersV1(data.cast());
+      }
+    } catch (error, stack) {
+      _logger.severe("Error processing sync data for $type", error, stack);
+      return false;
+    }
 
-          // user.name = data.name;
-          // user.email = data.email;
-          // user.updatedAt = DateTime.now();
+    _logger.warning("Unknown sync data type: $type");
+    return false;
+  }
 
-          // await _userRepository.update(user);
-          // await _syncApiRepository.ack(event.ack);
+  Future<void> _syncEvent(List<SyncRequestType> types) async {
+    _logger.info("Syncing Events: $types");
+    final streamCompleter = Completer();
+    bool shouldComplete = false;
+    // the onDone callback might fire before the events are processed
+    // the following flag ensures that the onDone callback is not called
+    // before the events are processed and also that events are processed sequentially
+    Completer? mutex;
+    StreamSubscription? subscription;
+    subscription = _syncApiRepository.getSyncEvents(types).listen(
+      (events) async {
+        if (events.isEmpty) {
+          _logger.warning("Received empty sync events");
+          return;
         }
 
-        if (event.data is SyncUserDeleteV1) {
-          final data = event.data as SyncUserDeleteV1;
+        // If previous events are still being processed, wait for them to finish
+        if (mutex != null) {
+          await mutex!.future;
+        }
 
-          debugPrint("User delete: $data");
-          // await _syncApiRepository.ack(event.ack);
+        if (_cancelCompleter?.isCompleted ?? false) {
+          _logger.info("Sync cancelled, stopping stream");
+          subscription?.cancel();
+          if (!streamCompleter.isCompleted) {
+            streamCompleter.completeError(
+              const CancelException(),
+              StackTrace.current,
+            );
+          }
+          return;
         }
-      }
+
+        // Take control of the mutex and process the events
+        mutex = Completer();
+
+        try {
+          final eventsMap = events.groupListsBy((event) => event.type);
+          final Map<SyncEntityType, String> acks = {};
+
+          for (final entry in eventsMap.entries) {
+            if (_cancelCompleter?.isCompleted ?? false) {
+              _logger.info("Sync cancelled, stopping stream");
+              mutex?.complete();
+              mutex = null;
+              if (!streamCompleter.isCompleted) {
+                streamCompleter.completeError(
+                  const CancelException(),
+                  StackTrace.current,
+                );
+              }
+
+              return;
+            }
+
+            final type = entry.key;
+            final data = entry.value;
+
+            if (data.isEmpty) {
+              _logger.warning("Received empty sync events for $type");
+              continue;
+            }
+
+            if (await _handleSyncData(type, data.map((e) => e.data))) {
+              // ignore: avoid-unsafe-collection-methods
+              acks[type] = data.last.ack;
+            } else {
+              _logger.warning("Failed to handle sync events for $type");
+            }
+          }
+
+          if (acks.isNotEmpty) {
+            await _syncApiRepository.ack(acks.values.toList());
+          }
+          _logger.info("$types events processed");
+        } catch (error, stack) {
+          _logger.warning("Error handling sync events", error, stack);
+        } finally {
+          mutex?.complete();
+          mutex = null;
+        }
+
+        if (shouldComplete) {
+          _logger.info("Sync done, completing stream");
+          if (!streamCompleter.isCompleted) streamCompleter.complete();
+        }
+      },
+      onError: (error, stack) {
+        _logger.warning("Error in sync stream for $types", error, stack);
+        // Do not proceed if the stream errors
+        if (!streamCompleter.isCompleted) streamCompleter.complete();
+      },
+      onDone: () {
+        _logger.info("$types stream done");
+        if (mutex == null && !streamCompleter.isCompleted) {
+          streamCompleter.complete();
+        } else {
+          // Marks the stream as done but does not complete the completer
+          // until the events are processed
+          shouldComplete = true;
+        }
+      },
+    );
+    return await streamCompleter.future.whenComplete(() {
+      _logger.info("Sync stream completed");
+      return subscription?.cancel();
     });
   }
 
-  Future<void> dispose() async {
-    await _userSyncSubscription?.cancel();
-  }
+  Future<void> syncUsers() =>
+      _syncEvent([SyncRequestType.usersV1, SyncRequestType.partnersV1]);
 }

mobile/lib/domain/utils/background_sync.dart

@@ -0,0 +1,42 @@
+// ignore_for_file: avoid-passing-async-when-sync-expected
+
+import 'dart:async';
+
+import 'package:async/async.dart';
+import 'package:immich_mobile/providers/infrastructure/cancel.provider.dart';
+import 'package:immich_mobile/providers/infrastructure/sync_stream.provider.dart';
+import 'package:immich_mobile/utils/isolate.dart';
+
+class BackgroundSyncManager {
+  CancelableOperation<void>? _userSyncFuture;
+
+  BackgroundSyncManager();
+
+  Future<void> cancel() async {
+    await _userSyncFuture?.cancel();
+    _userSyncFuture = null;
+  }
+
+  Future<void> syncUsers() async {
+    if (_userSyncFuture != null) {
+      return _userSyncFuture!.valueOrCancellation();
+    }
+
+    if (_userSyncFuture == null) {
+      final isolate = await IsolateManager.spawn(
+        computation: (ref) => ref.read(syncStreamServiceProvider).syncUsers(),
+        onCancel: (ref) => ref.read(cancellationProvider).complete(true),
+      );
+      _userSyncFuture = CancelableOperation.fromFuture(
+        isolate.run(),
+        onCancel: () {
+          isolate.cancel();
+          _userSyncFuture = null;
+        },
+      );
+      return _userSyncFuture!
+          .valueOrCancellation()
+          .then((_) => _userSyncFuture = null);
+    }
+  }
+}

mobile/lib/domain/utils/cancel.exception.dart

@@ -0,0 +1,8 @@
+class CancelException implements Exception {
+  final String message;
+
+  const CancelException([this.message = "Operation was cancelled."]);
+
+  @override
+  String toString() => "CancelException: $message";
+}

mobile/lib/extensions/string_extensions.dart

@@ -1,3 +1,7 @@
+import 'dart:typed_data';
+
+import 'package:uuid/parsing.dart';
+
 extension StringExtension on String {
   String capitalize() {
     return split(" ")
@@ -29,3 +33,8 @@ extension DurationExtension on String {
     return int.parse(this);
   }
 }
+
+extension UUIDExtension on String {
+  Uint8List toUuidByte({bool shouldValidate = false}) =>
+      UuidParsing.parseAsByteList(this, validate: shouldValidate);
+}