chore: refactor upload service (#20130)

* chore: refactor upload service

* fix: cancel upload queue on logout (#20131)

* fix: cancel upload on logout

* fix: test

---------

Co-authored-by: shenlong-tanwen <139912620+shalong-tanwen@users.noreply.github.com>
Co-authored-by: Alex <alex.tran1502@gmail.com>

---------

Co-authored-by: shenlong <139912620+shenlong-tanwen@users.noreply.github.com>
Co-authored-by: shenlong-tanwen <139912620+shalong-tanwen@users.noreply.github.com>

mobile/lib/providers/asset_viewer/share_intent_upload.provider.dart

@@ -27,8 +27,8 @@ class ShareIntentUploadStateNotifier extends StateNotifier<List<ShareIntentAttac
     this._uploadService,
     this._shareIntentService,
   ) : super([]) {
-    _uploadService.onUploadStatus = _updateUploadStatus;
-    _uploadService.onTaskProgress = _taskProgressCallback;
+    _uploadService.taskStatusStream.listen(_updateUploadStatus);
+    _uploadService.taskProgressStream.listen(_taskProgressCallback);
   }
 
   void init() {

mobile/lib/providers/auth.provider.dart

@@ -13,6 +13,7 @@ import 'package:immich_mobile/providers/infrastructure/user.provider.dart';
 import 'package:immich_mobile/services/api.service.dart';
 import 'package:immich_mobile/services/auth.service.dart';
 import 'package:immich_mobile/services/secure_storage.service.dart';
+import 'package:immich_mobile/services/upload.service.dart';
 import 'package:immich_mobile/services/widget.service.dart';
 import 'package:immich_mobile/utils/hash.dart';
 import 'package:logging/logging.dart';
@@ -23,6 +24,7 @@ final authProvider = StateNotifierProvider<AuthNotifier, AuthState>((ref) {
     ref.watch(authServiceProvider),
     ref.watch(apiServiceProvider),
     ref.watch(userServiceProvider),
+    ref.watch(uploadServiceProvider),
     ref.watch(secureStorageServiceProvider),
     ref.watch(widgetServiceProvider),
   );
@@ -32,6 +34,7 @@ class AuthNotifier extends StateNotifier<AuthState> {
   final AuthService _authService;
   final ApiService _apiService;
   final UserService _userService;
+  final UploadService _uploadService;
   final SecureStorageService _secureStorageService;
   final WidgetService _widgetService;
   final _log = Logger("AuthenticationNotifier");
@@ -42,6 +45,7 @@ class AuthNotifier extends StateNotifier<AuthState> {
     this._authService,
     this._apiService,
     this._userService,
+    this._uploadService,
     this._secureStorageService,
     this._widgetService,
   ) : super(
@@ -83,6 +87,7 @@ class AuthNotifier extends StateNotifier<AuthState> {
       await _widgetService.clearCredentials();
 
       await _authService.logout();
+      await _uploadService.cancelBackup();
     } finally {
       await _cleanUp();
     }

mobile/lib/providers/backup/drift_backup.provider.dart

@@ -8,7 +8,6 @@ import 'package:flutter/widgets.dart';
 import 'package:hooks_riverpod/hooks_riverpod.dart';
 
 import 'package:immich_mobile/constants/constants.dart';
-import 'package:immich_mobile/services/drift_backup.service.dart';
 import 'package:immich_mobile/services/upload.service.dart';
 
 class EnqueueStatus {
@@ -199,14 +198,12 @@ class DriftBackupState {
 
 final driftBackupProvider = StateNotifierProvider<ExpBackupNotifier, DriftBackupState>((ref) {
   return ExpBackupNotifier(
-    ref.watch(driftBackupServiceProvider),
     ref.watch(uploadServiceProvider),
   );
 });
 
 class ExpBackupNotifier extends StateNotifier<DriftBackupState> {
   ExpBackupNotifier(
-    this._backupService,
     this._uploadService,
   ) : super(
           const DriftBackupState(
@@ -225,7 +222,6 @@ class ExpBackupNotifier extends StateNotifier<DriftBackupState> {
     }
   }
 
-  final DriftBackupService _backupService;
   final UploadService _uploadService;
   StreamSubscription<TaskStatusUpdate>? _statusSubscription;
   StreamSubscription<TaskProgressUpdate>? _progressSubscription;
@@ -328,9 +324,9 @@ class ExpBackupNotifier extends StateNotifier<DriftBackupState> {
 
   Future<void> getBackupStatus(String userId) async {
     final [totalCount, backupCount, remainderCount] = await Future.wait([
-      _backupService.getTotalCount(),
-      _backupService.getBackupCount(userId),
-      _backupService.getRemainderCount(userId),
+      _uploadService.getBackupTotalCount(),
+      _uploadService.getBackupFinishedCount(userId),
+      _uploadService.getBackupRemainderCount(userId),
     ]);
 
     state = state.copyWith(
@@ -340,8 +336,8 @@ class ExpBackupNotifier extends StateNotifier<DriftBackupState> {
     );
   }
 
-  Future<void> backup(String userId) {
-    return _backupService.backup(userId, _updateEnqueueCount);
+  Future<void> startBackup(String userId) {
+    return _uploadService.startBackup(userId, _updateEnqueueCount);
   }
 
   void _updateEnqueueCount(EnqueueStatus status) {
@@ -352,22 +348,22 @@ class ExpBackupNotifier extends StateNotifier<DriftBackupState> {
   }
 
   Future<void> cancel() async {
+    debugPrint("Canceling backup tasks...");
     state = state.copyWith(
       enqueueCount: 0,
       enqueueTotalCount: 0,
       isCanceling: true,
     );
 
-    await _backupService.cancel();
+    final activeTaskCount = await _uploadService.cancelBackup();
 
-    // Check if there are any tasks left in the queue
-    final tasks = await FileDownloader().allTasks(group: kBackupGroup);
-
-    debugPrint("Tasks left to cancel: ${tasks.length}");
-
-    if (tasks.isNotEmpty) {
+    if (activeTaskCount > 0) {
+      debugPrint(
+        "$activeTaskCount tasks left, continuing to cancel...",
+      );
       await cancel();
     } else {
+      debugPrint("All tasks canceled successfully.");
       // Clear all upload items when cancellation is complete
       state = state.copyWith(
         isCanceling: false,
@@ -377,14 +373,18 @@ class ExpBackupNotifier extends StateNotifier<DriftBackupState> {
   }
 
   Future<void> handleBackupResume(String userId) async {
-    final tasks = await FileDownloader().allTasks(group: kBackupGroup);
+    debugPrint("handleBackupResume");
+    final tasks = await _uploadService.getActiveTasks(kBackupGroup);
+    debugPrint("Found ${tasks.length} tasks");
+
     if (tasks.isEmpty) {
       // Start a new backup queue
-      await backup(userId);
+      debugPrint("Start a new backup queue");
+      await startBackup(userId);
     }
 
     debugPrint("Tasks to resume: ${tasks.length}");
-    await FileDownloader().start();
+    await _uploadService.resumeBackup();
   }
 
   @override

mobile/lib/providers/infrastructure/action.provider.dart

@@ -5,8 +5,8 @@ import 'package:immich_mobile/providers/infrastructure/asset_viewer/current_asse
 import 'package:immich_mobile/providers/timeline/multiselect.provider.dart';
 import 'package:immich_mobile/providers/user.provider.dart';
 import 'package:immich_mobile/services/action.service.dart';
-import 'package:immich_mobile/services/drift_backup.service.dart';
 import 'package:immich_mobile/services/timeline.service.dart';
+import 'package:immich_mobile/services/upload.service.dart';
 import 'package:logging/logging.dart';
 import 'package:riverpod_annotation/riverpod_annotation.dart';
 
@@ -32,14 +32,14 @@ class ActionResult {
 class ActionNotifier extends Notifier<void> {
   final Logger _logger = Logger('ActionNotifier');
   late ActionService _service;
-  late DriftBackupService _backupService;
+  late UploadService _uploadService;
 
   ActionNotifier() : super();
 
   @override
   void build() {
+    _uploadService = ref.watch(uploadServiceProvider);
     _service = ref.watch(actionServiceProvider);
-    _backupService = ref.watch(driftBackupServiceProvider);
   }
 
   List<String> _getRemoteIdsForSource(ActionSource source) {
@@ -366,7 +366,7 @@ class ActionNotifier extends Notifier<void> {
   Future<ActionResult> upload(ActionSource source) async {
     final assets = _getAssets(source).whereType<LocalAsset>().toList();
     try {
-      await _backupService.manualBackup(assets);
+      await _uploadService.manualBackup(assets);
       return ActionResult(count: assets.length, success: true);
     } catch (error, stack) {
       _logger.severe('Failed manually upload assets', error, stack);