feat(mobile): Implement adaptive parallel upload pipeline with intelligent throttling

mobile/android/app/src/main/kotlin/app/alextran/immich/BackgroundServicePlugin.kt

@@ -194,10 +194,59 @@ class BackgroundServicePlugin : FlutterPlugin, MethodChannel.MethodCallHandler,
 
       "manageMediaPermission" -> requestManageMediaPermission(result)
 
+      // App restart for backup recovery (Level 3 recovery)
+      "restartApp" -> {
+        restartApp(result)
+      }
+
       else -> result.notImplemented()
     }
   }
 
+  /**
+   * Restart the app for backup recovery.
+   * This is used as a last resort when memory issues cannot be resolved.
+   */
+  private fun restartApp(result: Result) {
+    val ctx = context
+    val activity = activityBinding?.activity
+
+    if (ctx == null || activity == null) {
+      result.error("RESTART_ERROR", "Context or activity not available", null)
+      return
+    }
+
+    try {
+      val packageManager = ctx.packageManager
+      val intent = packageManager.getLaunchIntentForPackage(ctx.packageName)
+
+      if (intent == null) {
+        result.error("RESTART_ERROR", "Could not get launch intent", null)
+        return
+      }
+
+      intent.addFlags(Intent.FLAG_ACTIVITY_CLEAR_TOP)
+      intent.addFlags(Intent.FLAG_ACTIVITY_NEW_TASK)
+      intent.addFlags(Intent.FLAG_ACTIVITY_CLEAR_TASK)
+
+      // Add extra to indicate this is a restart for backup recovery
+      intent.putExtra("backup_recovery_restart", true)
+
+      ctx.startActivity(intent)
+
+      // Give the new activity a moment to start
+      android.os.Handler(android.os.Looper.getMainLooper()).postDelayed({
+        activity.finishAffinity()
+        Runtime.getRuntime().exit(0)
+      }, 500)
+
+      result.success(true)
+    } catch (e: Exception) {
+      Log.e(TAG, "Failed to restart app: ${e.message}", e)
+      result.error("RESTART_ERROR", "Failed to restart app: ${e.message}", null)
+    }
+  }
+
   private fun hasManageMediaPermission(): Boolean {
     return if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.S) {
       MediaStore.canManageMedia(context!!);

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

@@ -11,6 +11,28 @@ import 'package:logging/logging.dart';
 
 const String _kHashCancelledCode = "HASH_CANCELLED";
 
+/// Information about a completed hash batch
+class HashBatchResult {
+  /// Number of assets successfully hashed in this batch
+  final int hashedCount;
+  /// Total assets hashed so far
+  final int totalHashedSoFar;
+  /// Total assets remaining to hash
+  final int remainingToHash;
+  /// IDs of the assets that were just hashed
+  final List<String> hashedAssetIds;
+
+  const HashBatchResult({
+    required this.hashedCount,
+    required this.totalHashedSoFar,
+    required this.remainingToHash,
+    required this.hashedAssetIds,
+  });
+}
+
+/// Callback type for when a batch of assets has been hashed
+typedef OnHashBatchComplete = void Function(HashBatchResult result);
+
 class HashService {
   final int _batchSize;
   final DriftLocalAlbumRepository _localAlbumRepository;
@@ -19,6 +41,13 @@ class HashService {
   final NativeSyncApi _nativeSyncApi;
   final bool Function()? _cancelChecker;
   final _log = Logger('HashService');
+
+  /// Callback that fires when each batch completes - enables parallel upload
+  OnHashBatchComplete? onBatchComplete;
+
+  /// Track total hashed for progress reporting
+  int _totalHashedSoFar = 0;
+  int _totalToHash = 0;
 
   HashService({
     required DriftLocalAlbumRepository localAlbumRepository,
@@ -27,6 +56,7 @@ class HashService {
     required NativeSyncApi nativeSyncApi,
     bool Function()? cancelChecker,
     int? batchSize,
+    this.onBatchComplete,
   }) : _localAlbumRepository = localAlbumRepository,
        _localAssetRepository = localAssetRepository,
        _trashedLocalAssetRepository = trashedLocalAssetRepository,
@@ -35,13 +65,35 @@ class HashService {
        _batchSize = batchSize ?? kBatchHashFileLimit;
 
   bool get isCancelled => _cancelChecker?.call() ?? false;
+
+  /// Sets the callback for batch completion - enables pipeline mode
+  void setOnBatchComplete(OnHashBatchComplete? callback) {
+    onBatchComplete = callback;
+  }
 
   Future<void> hashAssets() async {
+    return hashAssetsWithCallback(onBatchComplete: onBatchComplete);
+  }
+
+  /// Hash assets with an optional callback that fires after each batch.
+  /// This enables the parallel pipeline where uploads can start immediately
+  /// after each batch is hashed, rather than waiting for all hashing to complete.
+  Future<void> hashAssetsWithCallback({OnHashBatchComplete? onBatchComplete}) async {
     _log.info("Starting hashing of assets");
     final Stopwatch stopwatch = Stopwatch()..start();
+    _totalHashedSoFar = 0;
+
     try {
       // Sorted by backupSelection followed by isCloud
       final localAlbums = await _localAlbumRepository.getBackupAlbums();
+
+      // Calculate total assets to hash for progress reporting
+      _totalToHash = 0;
+      for (final album in localAlbums) {
+        final assetsToHash = await _localAlbumRepository.getAssetsToHash(album.id);
+        _totalToHash += assetsToHash.length;
+      }
+      _log.info("Total assets to hash: $_totalToHash");
 
       for (final album in localAlbums) {
         if (isCancelled) {
@@ -51,15 +103,15 @@ class HashService {
 
         final assetsToHash = await _localAlbumRepository.getAssetsToHash(album.id);
         if (assetsToHash.isNotEmpty) {
-          await _hashAssets(album, assetsToHash);
+          await _hashAssets(album, assetsToHash, onBatchComplete: onBatchComplete);
         }
       }
       if (CurrentPlatform.isAndroid && localAlbums.isNotEmpty) {
         final backupAlbumIds = localAlbums.map((e) => e.id);
         final trashedToHash = await _trashedLocalAssetRepository.getAssetsToHash(backupAlbumIds);
         if (trashedToHash.isNotEmpty) {
           final pseudoAlbum = LocalAlbum(id: '-pseudoAlbum', name: 'Trash', updatedAt: DateTime.now());
-          await _hashAssets(pseudoAlbum, trashedToHash, isTrashed: true);
+          await _hashAssets(pseudoAlbum, trashedToHash, isTrashed: true, onBatchComplete: onBatchComplete);
         }
       }
     } on PlatformException catch (e) {
@@ -78,7 +130,12 @@ class HashService {
   /// Processes a list of [LocalAsset]s, storing their hash and updating the assets in the DB
   /// with hash for those that were successfully hashed. Hashes are looked up in a table
   /// [LocalAssetHashEntity] by local id. Only missing entries are newly hashed and added to the DB.
-  Future<void> _hashAssets(LocalAlbum album, List<LocalAsset> assetsToHash, {bool isTrashed = false}) async {
+  Future<void> _hashAssets(
+    LocalAlbum album, 
+    List<LocalAsset> assetsToHash, {
+    bool isTrashed = false,
+    OnHashBatchComplete? onBatchComplete,
+  }) async {
     final toHash = <String, LocalAsset>{};
 
     for (final asset in assetsToHash) {
@@ -89,23 +146,29 @@ class HashService {
 
       toHash[asset.id] = asset;
       if (toHash.length == _batchSize) {
-        await _processBatch(album, toHash, isTrashed);
+        await _processBatch(album, toHash, isTrashed, onBatchComplete: onBatchComplete);
         toHash.clear();
       }
     }
 
-    await _processBatch(album, toHash, isTrashed);
+    await _processBatch(album, toHash, isTrashed, onBatchComplete: onBatchComplete);
   }
 
   /// Processes a batch of assets.
-  Future<void> _processBatch(LocalAlbum album, Map<String, LocalAsset> toHash, bool isTrashed) async {
+  Future<void> _processBatch(
+    LocalAlbum album, 
+    Map<String, LocalAsset> toHash, 
+    bool isTrashed, {
+    OnHashBatchComplete? onBatchComplete,
+  }) async {
     if (toHash.isEmpty) {
       return;
     }
 
     _log.fine("Hashing ${toHash.length} files");
 
     final hashed = <String, String>{};
+    final hashedIds = <String>[];
     final hashResults = await _nativeSyncApi.hashAssets(
       toHash.keys.toList(),
       allowNetworkAccess: album.backupSelection == BackupSelection.selected,
@@ -124,6 +187,7 @@ class HashService {
       final hashResult = hashResults[i];
       if (hashResult.hash != null) {
         hashed[hashResult.assetId] = hashResult.hash!;
+        hashedIds.add(hashResult.assetId);
       } else {
         final asset = toHash[hashResult.assetId];
         _log.warning(
@@ -138,5 +202,20 @@ class HashService {
     } else {
       await _localAssetRepository.updateHashes(hashed);
     }
+
+    // Update progress and notify callback
+    _totalHashedSoFar += hashed.length;
+    final remaining = _totalToHash - _totalHashedSoFar;
+
+    // Fire callback to enable parallel uploads
+    if (onBatchComplete != null && hashed.isNotEmpty) {
+      _log.info("Batch complete: ${hashed.length} hashed, $_totalHashedSoFar total, $remaining remaining");
+      onBatchComplete(HashBatchResult(
+        hashedCount: hashed.length,
+        totalHashedSoFar: _totalHashedSoFar,
+        remainingToHash: remaining,
+        hashedAssetIds: hashedIds,
+      ));
+    }
   }
 }

mobile/lib/domain/utils/background_sync.dart

@@ -1,5 +1,6 @@
 import 'dart:async';
 
+import 'package:immich_mobile/domain/services/hash.service.dart';
 import 'package:immich_mobile/domain/utils/sync_linked_album.dart';
 import 'package:immich_mobile/providers/infrastructure/sync.provider.dart';
 import 'package:immich_mobile/utils/isolate.dart';
@@ -8,6 +9,7 @@ import 'package:worker_manager/worker_manager.dart';
 typedef SyncCallback = void Function();
 typedef SyncCallbackWithResult<T> = void Function(T result);
 typedef SyncErrorCallback = void Function(String error);
+typedef HashBatchCallback = void Function(HashBatchResult result);
 
 class BackgroundSyncManager {
   final SyncCallback? onRemoteSyncStart;
@@ -145,6 +147,42 @@ class BackgroundSyncManager {
         });
   }
 
+  /// Hash assets with a callback that fires after each batch completes.
+  /// This enables the parallel pipeline where uploads can start immediately
+  /// after each batch is hashed.
+  /// 
+  /// Note: The [onBatchComplete] callback is called from the isolate, so
+  /// it may need to use ports for communication with the main isolate.
+  Future<void> hashAssetsWithCallback({
+    HashBatchCallback? onBatchComplete,
+  }) {
+    if (_hashTask != null) {
+      return _hashTask!.future;
+    }
+
+    onHashingStart?.call();
+
+    _hashTask = runInIsolateGentle(
+      computation: (ref) {
+        final hashService = ref.read(hashServiceProvider);
+        return hashService.hashAssetsWithCallback(
+          onBatchComplete: onBatchComplete,
+        );
+      },
+      debugLabel: 'hash-assets-pipeline',
+    );
+
+    return _hashTask!
+        .whenComplete(() {
+          onHashingComplete?.call();
+          _hashTask = null;
+        })
+        .catchError((error) {
+          onHashingError?.call(error.toString());
+          _hashTask = null;
+        });
+  }
+
   Future<bool> syncRemote() {
     if (_syncTask != null) {
       return _syncTask!.future.then((result) => result ?? false).catchError((_) => false);

mobile/lib/infrastructure/repositories/storage.repository.dart

@@ -81,6 +81,26 @@ class StorageRepository {
     }
     return entity;
   }
+
+  /// Check if an asset is available locally (not cloud-only)
+  /// Returns true if the file exists on device, false if it needs to be downloaded from cloud
+  Future<bool> isAssetLocallyAvailable(String assetId) async {
+    final log = Logger('StorageRepository');
+
+    try {
+      final entity = await AssetEntity.fromId(assetId);
+      if (entity == null) {
+        return false;
+      }
+
+      // Check if file is locally available (not just in cloud)
+      final isLocal = await entity.isLocallyAvailable(isOrigin: true);
+      return isLocal;
+    } catch (error, stackTrace) {
+      log.warning("Error checking local availability for asset $assetId", error, stackTrace);
+      return false;
+    }
+  }
 
   Future<void> clearCache() async {
     final log = Logger('StorageRepository');