fix: watch asset delete/archive in search (#31343)

* fix: watch asset delete/archive in search

* fix: restore via toast

* chore: format

* refactor: simplify search asset pruning

* chore: implement suggestions

* chore: more suggestions

* fix: mounted check

(cherry picked from commit f7b8ae5da370f947223b006618e5b4e5fb42fd33)

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

@@ -41,6 +41,23 @@ class RemoteAssetRepository extends DatabaseAccessor<Drift> with $RemoteAssetRep
     return _assetSelectable(id).watchSingleOrNull();
   }
 
+  Stream<Set<String>> watchDeletedAssetIds(Iterable<String> ids, AssetVisibility visibility) {
+    final idList = ids.toList(growable: false);
+    if (idList.isEmpty) {
+      return Stream.value(const <String>{});
+    }
+
+    final query = _db.remoteAssetEntity.selectOnly()
+      ..addColumns([_db.remoteAssetEntity.id])
+      ..where(
+        _db.remoteAssetEntity.id.isIn(idList) &
+            (_db.remoteAssetEntity.deletedAt.isNotNull() |
+                _db.remoteAssetEntity.visibility.equalsValue(visibility).not()),
+      );
+
+    return query.map((row) => row.read(_db.remoteAssetEntity.id)!).watch().map((rows) => rows.toSet());
+  }
+
   Future<RemoteAsset?> get(String id) {
     return _assetSelectable(id).getSingleOrNull();
   }

mobile/lib/presentation/pages/search/paginated_search.provider.dart

@@ -3,7 +3,9 @@ import 'dart:async';
 import 'package:hooks_riverpod/hooks_riverpod.dart';
 import 'package:immich_mobile/domain/models/asset/base_asset.model.dart';
 import 'package:immich_mobile/domain/services/search.service.dart';
+import 'package:immich_mobile/infrastructure/repositories/remote_asset.repository.dart';
 import 'package:immich_mobile/models/search/search_filter.model.dart';
+import 'package:immich_mobile/providers/infrastructure/db.provider.dart';
 import 'package:immich_mobile/providers/infrastructure/search.provider.dart';
 
 final searchPreFilterProvider = NotifierProvider<SearchFilterProvider, SearchFilter?>(SearchFilterProvider.new);
@@ -32,16 +34,21 @@ class SearchState {
 }
 
 final paginatedSearchProvider = StateNotifierProvider<PaginatedSearchNotifier, SearchState>(
-  (ref) => PaginatedSearchNotifier(ref.watch(searchServiceProvider)),
+  (ref) => PaginatedSearchNotifier(ref.watch(searchServiceProvider), ref.watch(driftProvider).remoteAssetRepository),
 );
 
 class PaginatedSearchNotifier extends StateNotifier<SearchState> {
   final SearchService _searchService;
+  final RemoteAssetRepository _remoteAssetRepository;
   final _assetCountController = StreamController<int>.broadcast();
 
-  PaginatedSearchNotifier(this._searchService) : super(const SearchState());
+  StreamSubscription<Set<String>>? _deletedIdsSubscription;
+  List<BaseAsset> _results = const [];
+  int? _nextPage = 1;
 
-  Stream<int> get assetCount => _assetCountController.stream;
+  PaginatedSearchNotifier(this._searchService, this._remoteAssetRepository) : super(const SearchState());
+
+  Stream<int> get assetCount => _assetCountController.stream.distinct();
 
   Future<void> search(SearchFilter filter) async {
     if (state.nextPage == null || state.isLoading) {
@@ -52,24 +59,47 @@ class PaginatedSearchNotifier extends StateNotifier<SearchState> {
 
     final result = await _searchService.search(filter, state.nextPage!);
 
+    if (!mounted) {
+      return;
+    }
+
     if (result == null) {
       state = SearchState(assets: state.assets, nextPage: state.nextPage);
       return;
     }
 
-    final assets = [...state.assets, ...result.assets];
+    _results = [..._results, ...result.assets];
-    state = SearchState(assets: assets, nextPage: result.nextPage);
+    _nextPage = result.nextPage;
-
+    _watchDeletedIds(filter.display.isArchive ? AssetVisibility.archive : AssetVisibility.timeline);
-    _assetCountController.add(assets.length);
   }
 
   void clear() {
-    state = const SearchState();
+    unawaited(_deletedIdsSubscription?.cancel());
-    _assetCountController.add(0);
+    _deletedIdsSubscription = null;
+    _results = const [];
+    _nextPage = 1;
+    _emit(const {});
+  }
+
+  void _watchDeletedIds(AssetVisibility visibility) {
+    unawaited(_deletedIdsSubscription?.cancel());
+
+    final ids = _results.whereType<RemoteAsset>().map((asset) => asset.id);
+    _deletedIdsSubscription = _remoteAssetRepository.watchDeletedAssetIds(ids, visibility).listen(_emit);
+  }
+
+  void _emit(Set<String> deletedIds) {
+    final visible = deletedIds.isEmpty
+        ? _results
+        : _results.where((asset) => asset is! RemoteAsset || !deletedIds.contains(asset.id)).toList(growable: false);
+
+    state = SearchState(assets: visible, nextPage: _nextPage);
+    _assetCountController.add(visible.length);
   }
 
   @override
   void dispose() {
+    unawaited(_deletedIdsSubscription?.cancel());
     unawaited(_assetCountController.close());
     super.dispose();
   }

mobile/test/medium/repositories/remote_asset_repository_test.dart

@@ -1,4 +1,6 @@
+import 'package:async/async.dart';
 import 'package:flutter_test/flutter_test.dart';
+import 'package:immich_mobile/domain/models/asset/base_asset.model.dart';
 import 'package:immich_mobile/infrastructure/repositories/remote_asset.repository.dart';
 
 import '../repository_context.dart';
@@ -64,4 +66,53 @@ void main() {
       expect(result[0].id, remote.id);
     });
   });
+
+  group('watchDeletedAssetIds', () {
+    late String userId;
+
+    setUp(() async {
+      final user = await ctx.newUser();
+      userId = user.id;
+      await ctx.newAuthUser(id: userId);
+    });
+
+    test('hides trashed and archived assets but keeps matching and unsynced ones', () async {
+      final timeline = await ctx.newRemoteAsset(ownerId: userId);
+      final trashed = await ctx.newRemoteAsset(ownerId: userId, deletedAt: DateTime(2020));
+      final archived = await ctx.newRemoteAsset(ownerId: userId, visibility: AssetVisibility.archive);
+
+      final result = await sut.watchDeletedAssetIds([
+        timeline.id,
+        trashed.id,
+        archived.id,
+        'never-synced',
+      ], AssetVisibility.timeline).first;
+
+      expect(result, {trashed.id, archived.id});
+    });
+
+    test('hides assets not matching the requested visibility for archive searches', () async {
+      final timeline = await ctx.newRemoteAsset(ownerId: userId);
+      final archived = await ctx.newRemoteAsset(ownerId: userId, visibility: AssetVisibility.archive);
+
+      final result = await sut.watchDeletedAssetIds([timeline.id, archived.id], AssetVisibility.archive).first;
+
+      expect(result, {timeline.id});
+    });
+
+    test('emits when an asset is trashed and again when it is restored', () async {
+      final asset = await ctx.newRemoteAsset(ownerId: userId);
+      final queue = StreamQueue(sut.watchDeletedAssetIds([asset.id], AssetVisibility.timeline));
+
+      expect(await queue.next, isEmpty);
+
+      await sut.trash([asset.id]);
+      expect(await queue.next, {asset.id});
+
+      await sut.restoreTrash([asset.id]);
+      expect(await queue.next, isEmpty);
+
+      await queue.cancel();
+    });
+  });
 }