rework fetching trashed assets data on native side

optimize handling trashed assets in local sync service
refactor code

mobile/android/app/src/main/kotlin/app/alextran/immich/sync/Messages.g.kt

@@ -296,7 +296,7 @@ private open class MessagesPigeonCodec : StandardMessageCodec() {
 /** Generated interface from Pigeon that represents a handler of messages from Flutter. */
 interface NativeSyncApi {
   fun shouldFullSync(): Boolean
-  fun getMediaChanges(isTrashed: Boolean): SyncDelta
+  fun getMediaChanges(): SyncDelta
   fun checkpointSync()
   fun clearSyncCheckpoint()
   fun getAssetIdsForAlbum(albumId: String): List<String>
@@ -305,7 +305,7 @@ interface NativeSyncApi {
   fun getAssetsForAlbum(albumId: String, updatedTimeCond: Long?): List<PlatformAsset>
   fun hashAssets(assetIds: List<String>, allowNetworkAccess: Boolean, callback: (Result<List<HashResult>>) -> Unit)
   fun cancelHashing()
-  fun getTrashedAssetsForAlbum(albumId: String): List<PlatformAsset>
+  fun getTrashedAssets(albumIds: List<String>, sinceLastCheckpoint: Boolean): Map<String, List<PlatformAsset>>
 
   companion object {
     /** The codec used by NativeSyncApi. */
@@ -335,11 +335,9 @@ interface NativeSyncApi {
       run {
         val channel = BasicMessageChannel<Any?>(binaryMessenger, "dev.flutter.pigeon.immich_mobile.NativeSyncApi.getMediaChanges$separatedMessageChannelSuffix", codec, taskQueue)
         if (api != null) {
-          channel.setMessageHandler { message, reply ->
+          channel.setMessageHandler { _, reply ->
-            val args = message as List<Any?>
-            val isTrashedArg = args[0] as Boolean
             val wrapped: List<Any?> = try {
-              listOf(api.getMediaChanges(isTrashedArg))
+              listOf(api.getMediaChanges())
             } catch (exception: Throwable) {
               MessagesPigeonUtils.wrapError(exception)
             }
@@ -487,13 +485,14 @@ interface NativeSyncApi {
         }
       }
       run {
-        val channel = BasicMessageChannel<Any?>(binaryMessenger, "dev.flutter.pigeon.immich_mobile.NativeSyncApi.getTrashedAssetsForAlbum$separatedMessageChannelSuffix", codec, taskQueue)
+        val channel = BasicMessageChannel<Any?>(binaryMessenger, "dev.flutter.pigeon.immich_mobile.NativeSyncApi.getTrashedAssets$separatedMessageChannelSuffix", codec, taskQueue)
         if (api != null) {
           channel.setMessageHandler { message, reply ->
             val args = message as List<Any?>
-            val albumIdArg = args[0] as String
+            val albumIdsArg = args[0] as List<String>
+            val sinceLastCheckpointArg = args[1] as Boolean
             val wrapped: List<Any?> = try {
-              listOf(api.getTrashedAssetsForAlbum(albumIdArg))
+              listOf(api.getTrashedAssets(albumIdsArg, sinceLastCheckpointArg))
             } catch (exception: Throwable) {
               MessagesPigeonUtils.wrapError(exception)
             }

mobile/android/app/src/main/kotlin/app/alextran/immich/sync/MessagesImpl26.kt

@@ -18,13 +18,14 @@ class NativeSyncApiImpl26(context: Context) : NativeSyncApiImplBase(context), Na
     // No-op for Android 10 and below
   }
 
-  override fun getMediaChanges(isTrashed: Boolean): SyncDelta {
+  override fun getMediaChanges(): SyncDelta {
     throw IllegalStateException("Method not supported on this Android version.")
   }
 
-  override fun getTrashedAssetsForAlbum(
+  override fun getTrashedAssets(
-    albumId: String,
+    albumIds: List<String>,
-  ): List<PlatformAsset> {
+    sinceLastCheckpoint: Boolean
+  ): Map<String, List<PlatformAsset>> {
     throw IllegalStateException("Method not supported on this Android version.")
   }
 }

mobile/android/app/src/main/kotlin/app/alextran/immich/sync/MessagesImpl30.kt

@@ -1,23 +1,13 @@
 package app.alextran.immich.sync
 
 import android.content.ContentResolver
-import android.content.ContentUris
 import android.content.Context
-import android.net.Uri
 import android.os.Build
 import android.os.Bundle
 import android.provider.MediaStore
 import androidx.annotation.RequiresApi
 import androidx.annotation.RequiresExtension
-import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.async
-import kotlinx.coroutines.awaitAll
-import kotlinx.coroutines.ensureActive
-import kotlinx.coroutines.launch
-import kotlinx.coroutines.sync.withPermit
 import kotlinx.serialization.json.Json
-import kotlin.coroutines.cancellation.CancellationException
 
 @RequiresApi(Build.VERSION_CODES.Q)
 @RequiresExtension(extension = Build.VERSION_CODES.R, version = 1)
@@ -59,7 +49,7 @@ class NativeSyncApiImpl30(context: Context) : NativeSyncApiImplBase(context), Na
     }
   }
 
-  override fun getMediaChanges(isTrashed: Boolean): SyncDelta {
+  override fun getMediaChanges(): SyncDelta {
     val genMap = getSavedGenerationMap()
     val currentVolumes = MediaStore.getExternalVolumeNames(ctx)
     val changed = mutableListOf<PlatformAsset>()
@@ -83,17 +73,8 @@ class NativeSyncApiImpl30(context: Context) : NativeSyncApiImplBase(context), Na
         storedGen.toString(),
         storedGen.toString()
       )
-      val cursor = if (isTrashed) {
+
-        val queryArgs = Bundle().apply {
+      getAssets(getCursor(volume, selection, selectionArgs)).forEach {
-          putString(ContentResolver.QUERY_ARG_SQL_SELECTION, selection)
-          putStringArray(ContentResolver.QUERY_ARG_SQL_SELECTION_ARGS, selectionArgs)
-          putInt(MediaStore.QUERY_ARG_MATCH_TRASHED, MediaStore.MATCH_ONLY)
-        }
-        getCursor(volume, queryArgs)
-      } else {
-        getCursor(volume, selection, selectionArgs)
-      }
-      getAssets(cursor).forEach {
         when (it) {
           is AssetResult.ValidAsset -> {
             changed.add(it.asset)
@@ -108,26 +89,81 @@ class NativeSyncApiImpl30(context: Context) : NativeSyncApiImplBase(context), Na
     return SyncDelta(hasChanges, changed, deleted, assetAlbums)
   }
 
-  override fun getTrashedAssetsForAlbum(
+//  override fun getTrashedAssetsForAlbum(
-    albumId: String
+//    albumId: String
-  ): List<PlatformAsset> {
+//  ): List<PlatformAsset> {
-    val trashed = mutableListOf<PlatformAsset>()
+//    val trashed = mutableListOf<PlatformAsset>()
+//    val volumes = MediaStore.getExternalVolumeNames(ctx)
+//
+//    val selection = "$BUCKET_SELECTION AND $MEDIA_SELECTION"
+//    val selectionArgs = mutableListOf(albumId, *MEDIA_SELECTION_ARGS)
+//
+//    for (volume in volumes) {
+//      val cursor = getCursor(volume, Bundle().apply {
+//        putString(ContentResolver.QUERY_ARG_SQL_SELECTION, selection)
+//        putStringArray(ContentResolver.QUERY_ARG_SQL_SELECTION_ARGS, selectionArgs.toTypedArray())
+//        putInt(MediaStore.QUERY_ARG_MATCH_TRASHED, MediaStore.MATCH_ONLY)
+//      })
+//      getAssets(cursor).forEach { res ->
+//        if (res is AssetResult.ValidAsset) trashed += res.asset
+//      }
+//    }
+//
+//    return trashed
+//  }
+
+  override fun getTrashedAssets(
+    albumIds: List<String>,
+    sinceLastCheckpoint: Boolean
+  ): Map<String, List<PlatformAsset>> {
+    if (albumIds.isEmpty()) return emptyMap()
+
+    val result = LinkedHashMap<String, MutableList<PlatformAsset>>(albumIds.size)
     val volumes = MediaStore.getExternalVolumeNames(ctx)
 
-    val selection = "$BUCKET_SELECTION AND $MEDIA_SELECTION"
+    val placeholders = albumIds.joinToString(",") { "?" }
-    val selectionArgs = mutableListOf(albumId, *MEDIA_SELECTION_ARGS)
+    val bucketIn = "(${MediaStore.Files.FileColumns.BUCKET_ID} IN ($placeholders))"
+    val baseSelection = "$bucketIn AND $MEDIA_SELECTION"
+
+    val baseSelectionArgs = ArrayList<String>(albumIds.size + MEDIA_SELECTION_ARGS.size).apply {
+      addAll(albumIds)
+      addAll(MEDIA_SELECTION_ARGS)
+    }
+
+    val genMap = if (sinceLastCheckpoint) getSavedGenerationMap() else emptyMap()
 
     for (volume in volumes) {
-      val cursor = getCursor(volume, Bundle().apply {
+      var selection = baseSelection
+      val selectionArgs = ArrayList<String>(baseSelectionArgs.size + if (sinceLastCheckpoint) 2 else 0).apply {
+        addAll(baseSelectionArgs)
+      }
+
+      if (sinceLastCheckpoint) {
+        val currentGen = MediaStore.getGeneration(ctx, volume)
+        val storedGen = genMap[volume] ?: 0L
+        if (currentGen <= storedGen) {
+          continue
+        }
+        selection += " AND (${MediaStore.MediaColumns.GENERATION_MODIFIED} > ? OR ${MediaStore.MediaColumns.GENERATION_ADDED} > ?)"
+        selectionArgs += storedGen.toString()
+        selectionArgs += storedGen.toString()
+      }
+
+      val queryArgs = Bundle().apply {
         putString(ContentResolver.QUERY_ARG_SQL_SELECTION, selection)
         putStringArray(ContentResolver.QUERY_ARG_SQL_SELECTION_ARGS, selectionArgs.toTypedArray())
         putInt(MediaStore.QUERY_ARG_MATCH_TRASHED, MediaStore.MATCH_ONLY)
-      })
+      }
-      getAssets(cursor).forEach { res ->
+
-        if (res is AssetResult.ValidAsset) trashed += res.asset
+      getCursor(volume, queryArgs).use { cursor ->
+        getAssets(cursor).forEach { res ->
+          if (res is AssetResult.ValidAsset) {
+            result.getOrPut(res.albumId) { mutableListOf() }.add(res.asset)
+          }
+        }
       }
     }
 
-    return trashed
+    return result.mapValues { it.value.toList() }
   }
 }

mobile/ios/Runner/Sync/Messages.g.swift

@@ -355,7 +355,7 @@ class MessagesPigeonCodec: FlutterStandardMessageCodec, @unchecked Sendable {
 /// Generated protocol from Pigeon that represents a handler of messages from Flutter.
 protocol NativeSyncApi {
   func shouldFullSync() throws -> Bool
-  func getMediaChanges(isTrashed: Bool) throws -> SyncDelta
+  func getMediaChanges() throws -> SyncDelta
   func checkpointSync() throws
   func clearSyncCheckpoint() throws
   func getAssetIdsForAlbum(albumId: String) throws -> [String]
@@ -364,7 +364,7 @@ protocol NativeSyncApi {
   func getAssetsForAlbum(albumId: String, updatedTimeCond: Int64?) throws -> [PlatformAsset]
   func hashAssets(assetIds: [String], allowNetworkAccess: Bool, completion: @escaping (Result<[HashResult], Error>) -> Void)
   func cancelHashing() throws
-  func getTrashedAssetsForAlbum(albumId: String) throws -> [PlatformAsset]
+  func getTrashedAssets(albumIds: [String], sinceLastCheckpoint: Bool) throws -> [String: [PlatformAsset]]
 }
 
 /// Generated setup class from Pigeon to handle messages through the `binaryMessenger`.
@@ -395,11 +395,9 @@ class NativeSyncApiSetup {
       ? FlutterBasicMessageChannel(name: "dev.flutter.pigeon.immich_mobile.NativeSyncApi.getMediaChanges\(channelSuffix)", binaryMessenger: binaryMessenger, codec: codec)
       : FlutterBasicMessageChannel(name: "dev.flutter.pigeon.immich_mobile.NativeSyncApi.getMediaChanges\(channelSuffix)", binaryMessenger: binaryMessenger, codec: codec, taskQueue: taskQueue)
     if let api = api {
-      getMediaChangesChannel.setMessageHandler { message, reply in
+      getMediaChangesChannel.setMessageHandler { _, reply in
-        let args = message as! [Any?]
-        let isTrashedArg = args[0] as! Bool
         do {
-          let result = try api.getMediaChanges(isTrashed: isTrashedArg)
+          let result = try api.getMediaChanges()
           reply(wrapResult(result))
         } catch {
           reply(wrapError(error))
@@ -535,22 +533,23 @@ class NativeSyncApiSetup {
     } else {
       cancelHashingChannel.setMessageHandler(nil)
     }
-    let getTrashedAssetsForAlbumChannel = taskQueue == nil
+    let getTrashedAssetsChannel = taskQueue == nil
-      ? FlutterBasicMessageChannel(name: "dev.flutter.pigeon.immich_mobile.NativeSyncApi.getTrashedAssetsForAlbum\(channelSuffix)", binaryMessenger: binaryMessenger, codec: codec)
+      ? FlutterBasicMessageChannel(name: "dev.flutter.pigeon.immich_mobile.NativeSyncApi.getTrashedAssets\(channelSuffix)", binaryMessenger: binaryMessenger, codec: codec)
-      : FlutterBasicMessageChannel(name: "dev.flutter.pigeon.immich_mobile.NativeSyncApi.getTrashedAssetsForAlbum\(channelSuffix)", binaryMessenger: binaryMessenger, codec: codec, taskQueue: taskQueue)
+      : FlutterBasicMessageChannel(name: "dev.flutter.pigeon.immich_mobile.NativeSyncApi.getTrashedAssets\(channelSuffix)", binaryMessenger: binaryMessenger, codec: codec, taskQueue: taskQueue)
     if let api = api {
-      getTrashedAssetsForAlbumChannel.setMessageHandler { message, reply in
+      getTrashedAssetsChannel.setMessageHandler { message, reply in
         let args = message as! [Any?]
-        let albumIdArg = args[0] as! String
+        let albumIdsArg = args[0] as! [String]
+        let sinceLastCheckpointArg = args[1] as! Bool
         do {
-          let result = try api.getTrashedAssetsForAlbum(albumId: albumIdArg)
+          let result = try api.getTrashedAssets(albumIds: albumIdsArg, sinceLastCheckpoint: sinceLastCheckpointArg)
           reply(wrapResult(result))
         } catch {
           reply(wrapError(error))
         }
       }
     } else {
-      getTrashedAssetsForAlbumChannel.setMessageHandler(nil)
+      getTrashedAssetsChannel.setMessageHandler(nil)
     }
   }
 }

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

@@ -39,9 +39,7 @@ class LocalSyncService {
       }
 
       if (CurrentPlatform.isAndroid) {
-        final delta = await _nativeSyncApi.getMediaChanges(isTrashed: true);
+        await _syncTrashedAssets(sinceLastCheckpoint: true);
-        _log.fine("Delta updated in trash: ${delta.updates.length - delta.updates.length}");
-        await _applyTrashDelta(delta);
       }
 
       final delta = await _nativeSyncApi.getMediaChanges();
@@ -98,10 +96,6 @@ class LocalSyncService {
     try {
       final Stopwatch stopwatch = Stopwatch()..start();
 
-      if (CurrentPlatform.isAndroid) {
-        await _syncDeviceTrashSnapshot();
-      }
-
       final deviceAlbums = await _nativeSyncApi.getAlbums();
       final dbAlbums = await _localAlbumRepository.getAll(sortBy: {SortLocalAlbumsBy.id});
 
@@ -113,6 +107,9 @@ class LocalSyncService {
         onlyFirst: removeAlbum,
         onlySecond: addAlbum,
       );
+      if (CurrentPlatform.isAndroid) {
+        await _syncTrashedAssets(sinceLastCheckpoint: false);
+      }
 
       await _nativeSyncApi.checkpointSync();
       stopwatch.stop();
@@ -293,39 +290,37 @@ class LocalSyncService {
     return a.name == b.name && a.assetCount == b.assetCount && a.updatedAt.isAtSameMomentAs(b.updatedAt);
   }
 
-  Future<void> _applyTrashDelta(SyncDelta delta) async {
+  Future<void> _syncTrashedAssets({required bool sinceLastCheckpoint}) async {
-    final trashUpdates = delta.updates;
-    if (trashUpdates.isEmpty) {
-      return Future.value();
-    }
-    final trashedAssets = delta.toTrashedAssets();
-    _log.info("updateLocalTrashChanges trashedAssets: ${trashedAssets.map((e) => e.asset.id)}");
-    await _trashedLocalAssetRepository.applyDelta(trashedAssets);
-    await _applyRemoteRestoreToLocal();
-  }
-
-  Future<void> _syncDeviceTrashSnapshot() async {
     final backupAlbums = await _localAlbumRepository.getBackupAlbums();
     if (backupAlbums.isEmpty) {
-      _log.info("syncDeviceTrashSnapshot, No backup albums found");
+      _log.info("syncTrashedAssets, No local backup albums found");
       return;
     }
-    for (final album in backupAlbums) {
+    final albumIds = backupAlbums.map((e) => e.id).toList();
-      _log.info("syncDeviceTrashSnapshot prepare, album: ${album.id}/${album.name}");
+    final trashedAssetMap = await _nativeSyncApi.getTrashedAssets(
-      final trashedPlatformAssets = await _nativeSyncApi.getTrashedAssetsForAlbum(album.id);
+      albumIds: albumIds,
-      final trashedAssets = trashedPlatformAssets.toTrashedAssets(album.id);
+      sinceLastCheckpoint: sinceLastCheckpoint,
-      await _trashedLocalAssetRepository.applyTrashSnapshot(trashedAssets);
+    );
+    if (trashedAssetMap.isEmpty) {
+      _log.info("syncTrashedAssets, No trashed assets found ${sinceLastCheckpoint ? "since Last Checkpoint" : ""}");
+    }
+    final trashedAssets = trashedAssetMap.cast<String, List<Object?>>().entries.expand(
+      (entry) => entry.value.cast<PlatformAsset>().toTrashedAssets(entry.key),
+    );
+
+    _log.fine("syncTrashedAssets, trashedAssets: ${trashedAssets.map((e) => e.asset.id)}");
+    if (sinceLastCheckpoint) {
+      await _trashedLocalAssetRepository.applyDelta(trashedAssets);
+    } else {
+      await _trashedLocalAssetRepository.applySnapshot(trashedAssets);
     }
-    await _applyRemoteRestoreToLocal();
-  }
 
-  Future<void> _applyRemoteRestoreToLocal() async {
     final remoteAssetsToRestore = await _trashedLocalAssetRepository.getToRestore();
     if (remoteAssetsToRestore.isNotEmpty) {
       final restoredIds = await _localFilesManager.restoreAssetsFromTrash(remoteAssetsToRestore);
       await _trashedLocalAssetRepository.applyRestoredAssets(restoredIds);
     } else {
-      _log.info("No remote assets found for restoration");
+      _log.info("syncTrashedAssets, No remote assets found for restoration");
     }
   }
 }

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

@@ -59,7 +59,7 @@ class DriftTrashedLocalAssetRepository extends DriftDatabaseRepository {
   /// Applies resulted snapshot of trashed assets:
   /// - upserts incoming rows
   /// - deletes rows that are not present in the snapshot
-  Future<void> applyTrashSnapshot(Iterable<TrashedAsset> trashedAssets) async {
+  Future<void> applySnapshot(Iterable<TrashedAsset> trashedAssets) async {
     if (trashedAssets.isEmpty) {
       await _db.delete(_db.trashedLocalAssetEntity).go();
       return;

mobile/lib/platform/native_sync_api.g.dart

@@ -326,7 +326,7 @@ class NativeSyncApi {
     }
   }
 
-  Future<SyncDelta> getMediaChanges({bool isTrashed = false}) async {
+  Future<SyncDelta> getMediaChanges() async {
     final String pigeonVar_channelName =
         'dev.flutter.pigeon.immich_mobile.NativeSyncApi.getMediaChanges$pigeonVar_messageChannelSuffix';
     final BasicMessageChannel<Object?> pigeonVar_channel = BasicMessageChannel<Object?>(
@@ -334,7 +334,7 @@ class NativeSyncApi {
       pigeonChannelCodec,
       binaryMessenger: pigeonVar_binaryMessenger,
     );
-    final Future<Object?> pigeonVar_sendFuture = pigeonVar_channel.send(<Object?>[isTrashed]);
+    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);
@@ -563,15 +563,18 @@ class NativeSyncApi {
     }
   }
 
-  Future<List<PlatformAsset>> getTrashedAssetsForAlbum(String albumId) async {
+  Future<Map<String, List<PlatformAsset>>> getTrashedAssets({
+    required List<String> albumIds,
+    required bool sinceLastCheckpoint,
+  }) async {
     final String pigeonVar_channelName =
-        'dev.flutter.pigeon.immich_mobile.NativeSyncApi.getTrashedAssetsForAlbum$pigeonVar_messageChannelSuffix';
+        'dev.flutter.pigeon.immich_mobile.NativeSyncApi.getTrashedAssets$pigeonVar_messageChannelSuffix';
     final BasicMessageChannel<Object?> pigeonVar_channel = BasicMessageChannel<Object?>(
       pigeonVar_channelName,
       pigeonChannelCodec,
       binaryMessenger: pigeonVar_binaryMessenger,
     );
-    final Future<Object?> pigeonVar_sendFuture = pigeonVar_channel.send(<Object?>[albumId]);
+    final Future<Object?> pigeonVar_sendFuture = pigeonVar_channel.send(<Object?>[albumIds, sinceLastCheckpoint]);
     final List<Object?>? pigeonVar_replyList = await pigeonVar_sendFuture as List<Object?>?;
     if (pigeonVar_replyList == null) {
       throw _createConnectionError(pigeonVar_channelName);
@@ -587,7 +590,7 @@ class NativeSyncApi {
         message: 'Host platform returned null value for non-null return value.',
       );
     } else {
-      return (pigeonVar_replyList[0] as List<Object?>?)!.cast<PlatformAsset>();
+      return (pigeonVar_replyList[0] as Map<Object?, Object?>?)!.cast<String, List<PlatformAsset>>();
     }
   }
 }

mobile/pigeon/native_sync_api.dart

@@ -88,7 +88,7 @@ abstract class NativeSyncApi {
   bool shouldFullSync();
 
   @TaskQueue(type: TaskQueueType.serialBackgroundThread)
-  SyncDelta getMediaChanges({bool isTrashed = false});
+  SyncDelta getMediaChanges();
 
   void checkpointSync();
 
@@ -113,5 +113,8 @@ abstract class NativeSyncApi {
   void cancelHashing();
 
   @TaskQueue(type: TaskQueueType.serialBackgroundThread)
-  List<PlatformAsset> getTrashedAssetsForAlbum(String albumId);
+  Map<String, List<PlatformAsset>> getTrashedAssets({
+    required List<String> albumIds,
+    required bool sinceLastCheckpoint,
+  });
 }