fix: android background backups (#21795)

* upload using dart client

* add connectivity api

* respect backup network setting

* comment as to why we need to wait for setForegroundAsync call

* log assets skipped due to network constraint

* dynamic spawning -> false

---------

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

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

@@ -1,10 +1,15 @@
 import 'dart:async';
+import 'dart:io';
 import 'dart:ui';
 
 import 'package:background_downloader/background_downloader.dart';
+import 'package:cancellation_token_http/http.dart';
 import 'package:flutter/material.dart';
 import 'package:hooks_riverpod/hooks_riverpod.dart';
 import 'package:immich_mobile/constants/constants.dart';
+import 'package:immich_mobile/extensions/network_capability_extensions.dart';
+import 'package:immich_mobile/extensions/translate_extensions.dart';
+import 'package:immich_mobile/generated/intl_keys.g.dart';
 import 'package:immich_mobile/infrastructure/repositories/db.repository.dart';
 import 'package:immich_mobile/infrastructure/repositories/logger_db.repository.dart';
 import 'package:immich_mobile/platform/background_worker_api.g.dart';
@@ -13,11 +18,13 @@ import 'package:immich_mobile/providers/background_sync.provider.dart';
 import 'package:immich_mobile/providers/backup/drift_backup.provider.dart';
 import 'package:immich_mobile/providers/db.provider.dart';
 import 'package:immich_mobile/providers/infrastructure/db.provider.dart';
+import 'package:immich_mobile/providers/infrastructure/platform.provider.dart';
 import 'package:immich_mobile/providers/user.provider.dart';
 import 'package:immich_mobile/repositories/file_media.repository.dart';
 import 'package:immich_mobile/services/app_settings.service.dart';
 import 'package:immich_mobile/services/auth.service.dart';
 import 'package:immich_mobile/services/localization.service.dart';
+import 'package:immich_mobile/services/server_info.service.dart';
 import 'package:immich_mobile/services/upload.service.dart';
 import 'package:immich_mobile/utils/bootstrap.dart';
 import 'package:immich_mobile/utils/http_ssl_options.dart';
@@ -42,6 +49,7 @@ class BackgroundWorkerBgService extends BackgroundWorkerFlutterApi {
   final Drift _drift;
   final DriftLogger _driftLogger;
   final BackgroundWorkerBgHostApi _backgroundHostApi;
+  final CancellationToken _cancellationToken = CancellationToken();
   final Logger _logger = Logger('BackgroundWorkerBgService');
 
   bool _isCleanedUp = false;
@@ -87,6 +95,13 @@ class BackgroundWorkerBgService extends BackgroundWorkerFlutterApi {
 
       configureFileDownloaderNotifications();
 
+      if (Platform.isAndroid) {
+        await _backgroundHostApi.showNotification(
+          IntlKeys.uploading_media.t(),
+          IntlKeys.backup_background_service_in_progress_notification.t(),
+        );
+      }
+
       // Notify the host that the background worker service has been initialized and is ready to use
       _backgroundHostApi.onInitialized();
     } catch (error, stack) {
@@ -102,7 +117,7 @@ class BackgroundWorkerBgService extends BackgroundWorkerFlutterApi {
       final sw = Stopwatch()..start();
 
       await _syncAssets(hashTimeout: Duration(minutes: _isBackupEnabled ? 3 : 6));
-      await _handleBackup(processBulk: false);
+      await _handleBackup();
 
       sw.stop();
       _logger.info("Android background processing completed in ${sw.elapsed.inSeconds}s");
@@ -155,9 +170,13 @@ class BackgroundWorkerBgService extends BackgroundWorkerFlutterApi {
 
     try {
       _isCleanedUp = true;
+      _cancellationToken.cancel();
       _logger.info("Cleaning up background worker");
       final cleanupFutures = [
-        workerManager.dispose(),
+        workerManager.dispose().catchError((_) async {
+          // Discard any errors on the dispose call
+          return;
+        }),
         _drift.close(),
         _driftLogger.close(),
         _ref.read(backgroundSyncProvider).cancel(),
@@ -175,7 +194,7 @@ class BackgroundWorkerBgService extends BackgroundWorkerFlutterApi {
     }
   }
 
-  Future<void> _handleBackup({bool processBulk = true}) async {
+  Future<void> _handleBackup() async {
     if (!_isBackupEnabled || _isCleanedUp) {
       _logger.info("[_handleBackup 1] Backup is disabled. Skipping backup routine");
       return;
@@ -189,19 +208,22 @@ class BackgroundWorkerBgService extends BackgroundWorkerFlutterApi {
       return;
     }
 
-    if (processBulk) {
-      _logger.info("[_handleBackup 4] Resume backup from background");
+    _logger.info("[_handleBackup 4] Resume backup from background");
+    if (Platform.isIOS) {
       return _ref.read(driftBackupProvider.notifier).handleBackupResume(currentUser.id);
     }
 
-    final activeTask = await _ref.read(uploadServiceProvider).getActiveTasks(currentUser.id);
-    if (activeTask.isNotEmpty) {
-      _logger.info("[_handleBackup 5] Resuming backup for active tasks from background");
-      await _ref.read(uploadServiceProvider).resumeBackup();
-    } else {
-      _logger.info("[_handleBackup 6] Starting serial backup for new tasks from background");
-      await _ref.read(uploadServiceProvider).startBackupSerial(currentUser.id);
+    final canPing = await _ref.read(serverInfoServiceProvider).ping();
+    if (!canPing) {
+      _logger.warning("[_handleBackup 5] Server is not reachable. Skipping backup from background");
+      return;
     }
+
+    final networkCapabilities = await _ref.read(connectivityApiProvider).getCapabilities();
+
+    return _ref
+        .read(uploadServiceProvider)
+        .startBackupWithHttpClient(currentUser.id, networkCapabilities.hasWifi, _cancellationToken);
   }
 
   Future<void> _syncAssets({Duration? hashTimeout}) async {

mobile/lib/extensions/network_capability_extensions.dart

@@ -0,0 +1,8 @@
+import 'package:immich_mobile/platform/connectivity_api.g.dart';
+
+extension NetworkCapabilitiesGetters on List<NetworkCapability> {
+  bool get hasCellular => contains(NetworkCapability.cellular);
+  bool get hasWifi => contains(NetworkCapability.wifi);
+  bool get hasVpn => contains(NetworkCapability.vpn);
+  bool get isUnmetered => contains(NetworkCapability.unmetered);
+}

mobile/lib/main.dart

@@ -46,7 +46,7 @@ void main() async {
   await Bootstrap.initDomain(isar, drift, logDb);
   await initApp();
   // Warm-up isolate pool for worker manager
-  await workerManager.init(dynamicSpawning: true);
+  await workerManager.init(dynamicSpawning: false);
   await migrateDatabaseIfNeeded(isar, drift);
   HttpSSLOptions.apply();
 

mobile/lib/platform/background_worker_api.g.dart

@@ -142,6 +142,29 @@ class BackgroundWorkerBgHostApi {
     }
   }
 
+  Future<void> showNotification(String title, String content) async {
+    final String pigeonVar_channelName =
+        'dev.flutter.pigeon.immich_mobile.BackgroundWorkerBgHostApi.showNotification$pigeonVar_messageChannelSuffix';
+    final BasicMessageChannel<Object?> pigeonVar_channel = BasicMessageChannel<Object?>(
+      pigeonVar_channelName,
+      pigeonChannelCodec,
+      binaryMessenger: pigeonVar_binaryMessenger,
+    );
+    final Future<Object?> pigeonVar_sendFuture = pigeonVar_channel.send(<Object?>[title, content]);
+    final List<Object?>? pigeonVar_replyList = await pigeonVar_sendFuture as List<Object?>?;
+    if (pigeonVar_replyList == null) {
+      throw _createConnectionError(pigeonVar_channelName);
+    } else if (pigeonVar_replyList.length > 1) {
+      throw PlatformException(
+        code: pigeonVar_replyList[0]! as String,
+        message: pigeonVar_replyList[1] as String?,
+        details: pigeonVar_replyList[2],
+      );
+    } else {
+      return;
+    }
+  }
+
   Future<void> close() async {
     final String pigeonVar_channelName =
         'dev.flutter.pigeon.immich_mobile.BackgroundWorkerBgHostApi.close$pigeonVar_messageChannelSuffix';

mobile/lib/platform/connectivity_api.g.dart

@@ -0,0 +1,87 @@
+// Autogenerated from Pigeon (v26.0.0), do not edit directly.
+// See also: https://pub.dev/packages/pigeon
+// ignore_for_file: public_member_api_docs, non_constant_identifier_names, avoid_as, unused_import, unnecessary_parenthesis, prefer_null_aware_operators, omit_local_variable_types, unused_shown_name, unnecessary_import, no_leading_underscores_for_local_identifiers
+
+import 'dart:async';
+import 'dart:typed_data' show Float64List, Int32List, Int64List, Uint8List;
+
+import 'package:flutter/foundation.dart' show ReadBuffer, WriteBuffer;
+import 'package:flutter/services.dart';
+
+PlatformException _createConnectionError(String channelName) {
+  return PlatformException(
+    code: 'channel-error',
+    message: 'Unable to establish connection on channel: "$channelName".',
+  );
+}
+
+enum NetworkCapability { cellular, wifi, vpn, unmetered }
+
+class _PigeonCodec extends StandardMessageCodec {
+  const _PigeonCodec();
+  @override
+  void writeValue(WriteBuffer buffer, Object? value) {
+    if (value is int) {
+      buffer.putUint8(4);
+      buffer.putInt64(value);
+    } else if (value is NetworkCapability) {
+      buffer.putUint8(129);
+      writeValue(buffer, value.index);
+    } else {
+      super.writeValue(buffer, value);
+    }
+  }
+
+  @override
+  Object? readValueOfType(int type, ReadBuffer buffer) {
+    switch (type) {
+      case 129:
+        final int? value = readValue(buffer) as int?;
+        return value == null ? null : NetworkCapability.values[value];
+      default:
+        return super.readValueOfType(type, buffer);
+    }
+  }
+}
+
+class ConnectivityApi {
+  /// Constructor for [ConnectivityApi].  The [binaryMessenger] named argument is
+  /// available for dependency injection.  If it is left null, the default
+  /// BinaryMessenger will be used which routes to the host platform.
+  ConnectivityApi({BinaryMessenger? binaryMessenger, String messageChannelSuffix = ''})
+    : pigeonVar_binaryMessenger = binaryMessenger,
+      pigeonVar_messageChannelSuffix = messageChannelSuffix.isNotEmpty ? '.$messageChannelSuffix' : '';
+  final BinaryMessenger? pigeonVar_binaryMessenger;
+
+  static const MessageCodec<Object?> pigeonChannelCodec = _PigeonCodec();
+
+  final String pigeonVar_messageChannelSuffix;
+
+  Future<List<NetworkCapability>> getCapabilities() async {
+    final String pigeonVar_channelName =
+        'dev.flutter.pigeon.immich_mobile.ConnectivityApi.getCapabilities$pigeonVar_messageChannelSuffix';
+    final BasicMessageChannel<Object?> pigeonVar_channel = BasicMessageChannel<Object?>(
+      pigeonVar_channelName,
+      pigeonChannelCodec,
+      binaryMessenger: pigeonVar_binaryMessenger,
+    );
+    final Future<Object?> pigeonVar_sendFuture = pigeonVar_channel.send(null);
+    final List<Object?>? pigeonVar_replyList = await pigeonVar_sendFuture as List<Object?>?;
+    if (pigeonVar_replyList == null) {
+      throw _createConnectionError(pigeonVar_channelName);
+    } else if (pigeonVar_replyList.length > 1) {
+      throw PlatformException(
+        code: pigeonVar_replyList[0]! as String,
+        message: pigeonVar_replyList[1] as String?,
+        details: pigeonVar_replyList[2],
+      );
+    } else if (pigeonVar_replyList[0] == null) {
+      throw PlatformException(
+        code: 'null-error',
+        message: 'Host platform returned null value for non-null return value.',
+      );
+    } else {
+      return (pigeonVar_replyList[0] as List<Object?>?)!.cast<NetworkCapability>();
+    }
+  }
+}

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

@@ -1,6 +1,7 @@
 import 'package:hooks_riverpod/hooks_riverpod.dart';
 import 'package:immich_mobile/domain/services/background_worker.service.dart';
 import 'package:immich_mobile/platform/background_worker_api.g.dart';
+import 'package:immich_mobile/platform/connectivity_api.g.dart';
 import 'package:immich_mobile/platform/native_sync_api.g.dart';
 import 'package:immich_mobile/platform/thumbnail_api.g.dart';
 
@@ -8,4 +9,6 @@ final backgroundWorkerFgServiceProvider = Provider((_) => BackgroundWorkerFgServ
 
 final nativeSyncApiProvider = Provider<NativeSyncApi>((_) => NativeSyncApi());
 
+final connectivityApiProvider = Provider<ConnectivityApi>((_) => ConnectivityApi());
+
 final thumbnailApi = ThumbnailApi();

mobile/lib/repositories/upload.repository.dart

@@ -1,7 +1,21 @@
+import 'dart:convert';
+import 'dart:io';
+
 import 'package:background_downloader/background_downloader.dart';
+import 'package:cancellation_token_http/http.dart';
 import 'package:flutter/material.dart';
 import 'package:hooks_riverpod/hooks_riverpod.dart';
 import 'package:immich_mobile/constants/constants.dart';
+import 'package:immich_mobile/domain/models/store.model.dart';
+import 'package:immich_mobile/entities/store.entity.dart';
+import 'package:logging/logging.dart';
+
+class UploadTaskWithFile {
+  final File file;
+  final UploadTask task;
+
+  const UploadTaskWithFile({required this.file, required this.task});
+}
 
 final uploadRepositoryProvider = Provider((ref) => UploadRepository());
 
@@ -31,7 +45,7 @@ class UploadRepository {
     return FileDownloader().enqueue(task);
   }
 
-  Future<void> enqueueBackgroundAll(List<UploadTask> tasks) {
+  Future<List<bool>> enqueueBackgroundAll(List<UploadTask> tasks) {
     return FileDownloader().enqueueAll(tasks);
   }
 
@@ -74,4 +88,53 @@ class UploadRepository {
       Paused: ${pausedTasks.length}
     """);
   }
+
+  Future<void> backupWithDartClient(Iterable<UploadTaskWithFile> tasks, CancellationToken cancelToken) async {
+    final httpClient = Client();
+    final String savedEndpoint = Store.get(StoreKey.serverEndpoint);
+
+    Logger logger = Logger('UploadRepository');
+    for (final candidate in tasks) {
+      if (cancelToken.isCancelled) {
+        logger.warning("Backup was cancelled by the user");
+        break;
+      }
+
+      try {
+        final fileStream = candidate.file.openRead();
+        final assetRawUploadData = MultipartFile(
+          "assetData",
+          fileStream,
+          candidate.file.lengthSync(),
+          filename: candidate.task.filename,
+        );
+
+        final baseRequest = MultipartRequest('POST', Uri.parse('$savedEndpoint/assets'));
+
+        baseRequest.headers.addAll(candidate.task.headers);
+        baseRequest.fields.addAll(candidate.task.fields);
+        baseRequest.files.add(assetRawUploadData);
+
+        final response = await httpClient.send(baseRequest, cancellationToken: cancelToken);
+
+        final responseBody = jsonDecode(await response.stream.bytesToString());
+
+        if (![200, 201].contains(response.statusCode)) {
+          final error = responseBody;
+
+          logger.warning(
+            "Error(${error['statusCode']}) uploading ${candidate.task.filename} | Created on ${candidate.task.fields["fileCreatedAt"]} | ${error['error']}",
+          );
+
+          continue;
+        }
+      } on CancelledException {
+        logger.warning("Backup was cancelled by the user");
+        break;
+      } catch (error, stackTrace) {
+        logger.warning("Error backup asset: ${error.toString()}: $stackTrace");
+        continue;
+      }
+    }
+  }
 }

mobile/lib/services/server_info.service.dart

@@ -14,6 +14,15 @@ class ServerInfoService {
 
   const ServerInfoService(this._apiService);
 
+  Future<bool> ping() async {
+    try {
+      await _apiService.serverInfoApi.pingServer().timeout(const Duration(seconds: 5));
+      return true;
+    } catch (e) {
+      return false;
+    }
+  }
+
   Future<ServerDiskInfo?> getDiskInfo() async {
     try {
       final dto = await _apiService.serverInfoApi.getStorage();

mobile/lib/services/upload.service.dart

@@ -3,6 +3,7 @@ import 'dart:convert';
 import 'dart:io';
 
 import 'package:background_downloader/background_downloader.dart';
+import 'package:cancellation_token_http/http.dart';
 import 'package:flutter/material.dart';
 import 'package:hooks_riverpod/hooks_riverpod.dart';
 import 'package:immich_mobile/constants/constants.dart';
@@ -19,6 +20,7 @@ import 'package:immich_mobile/providers/infrastructure/storage.provider.dart';
 import 'package:immich_mobile/repositories/upload.repository.dart';
 import 'package:immich_mobile/services/api.service.dart';
 import 'package:immich_mobile/services/app_settings.service.dart';
+import 'package:logging/logging.dart';
 import 'package:path/path.dart' as p;
 
 final uploadServiceProvider = Provider((ref) {
@@ -51,6 +53,7 @@ class UploadService {
   final StorageRepository _storageRepository;
   final DriftLocalAssetRepository _localAssetRepository;
   final AppSettingsService _appSettingsService;
+  final Logger _logger = Logger('UploadService');
 
   final StreamController<TaskStatusUpdate> _taskStatusController = StreamController<TaskStatusUpdate>.broadcast();
   final StreamController<TaskProgressUpdate> _taskProgressController = StreamController<TaskProgressUpdate>.broadcast();
@@ -78,7 +81,7 @@ class UploadService {
     _taskProgressController.close();
   }
 
-  Future<void> enqueueTasks(List<UploadTask> tasks) {
+  Future<List<bool>> enqueueTasks(List<UploadTask> tasks) {
     return _uploadRepository.enqueueBackgroundAll(tasks);
   }
 
@@ -138,7 +141,6 @@ class UploadService {
       }
 
       final batch = candidates.skip(i).take(batchSize).toList();
-
       List<UploadTask> tasks = [];
       for (final asset in batch) {
         final task = await _getUploadTask(asset);
@@ -156,9 +158,7 @@ class UploadService {
     }
   }
 
-  // Enqueue All does not work from the background on Android yet. This method is a temporary workaround
-  // that enqueues tasks one by one.
-  Future<void> startBackupSerial(String userId) async {
+  Future<void> startBackupWithHttpClient(String userId, bool hasWifi, CancellationToken token) async {
     await _storageRepository.clearCache();
 
     shouldAbortQueuingTasks = false;
@@ -168,14 +168,29 @@ class UploadService {
       return;
     }
 
-    for (final asset in candidates) {
-      if (shouldAbortQueuingTasks) {
+    const batchSize = 100;
+    for (int i = 0; i < candidates.length; i += batchSize) {
+      if (shouldAbortQueuingTasks || token.isCancelled) {
         break;
       }
 
-      final task = await _getUploadTask(asset);
-      if (task != null) {
-        await _uploadRepository.enqueueBackground(task);
+      final batch = candidates.skip(i).take(batchSize).toList();
+      List<UploadTaskWithFile> tasks = [];
+      for (final asset in batch) {
+        final requireWifi = _shouldRequireWiFi(asset);
+        if (requireWifi && !hasWifi) {
+          _logger.warning('Skipping upload for ${asset.id} because it requires WiFi');
+          continue;
+        }
+
+        final task = await _getUploadTaskWithFile(asset);
+        if (task != null) {
+          tasks.add(task);
+        }
+      }
+
+      if (tasks.isNotEmpty && !shouldAbortQueuingTasks) {
+        await _uploadRepository.backupWithDartClient(tasks, token);
       }
     }
   }
@@ -242,6 +257,42 @@ class UploadService {
     }
   }
 
+  Future<UploadTaskWithFile?> _getUploadTaskWithFile(LocalAsset asset) async {
+    final entity = await _storageRepository.getAssetEntityForAsset(asset);
+    if (entity == null) {
+      return null;
+    }
+
+    final file = await _storageRepository.getFileForAsset(asset.id);
+    if (file == null) {
+      return null;
+    }
+
+    final originalFileName = entity.isLivePhoto ? p.setExtension(asset.name, p.extension(file.path)) : asset.name;
+
+    String metadata = UploadTaskMetadata(
+      localAssetId: asset.id,
+      isLivePhotos: entity.isLivePhoto,
+      livePhotoVideoId: '',
+    ).toJson();
+
+    return UploadTaskWithFile(
+      file: file,
+      task: await buildUploadTask(
+        file,
+        createdAt: asset.createdAt,
+        modifiedAt: asset.updatedAt,
+        originalFileName: originalFileName,
+        deviceAssetId: asset.id,
+        metadata: metadata,
+        group: "group",
+        priority: 0,
+        isFavorite: asset.isFavorite,
+        requiresWiFi: false,
+      ),
+    );
+  }
+
   Future<UploadTask?> _getUploadTask(LocalAsset asset, {String group = kBackupGroup, int? priority}) async {
     final entity = await _storageRepository.getAssetEntityForAsset(asset);
     if (entity == null) {