diff --git a/mobile/lib/domain/services/background_worker.service.dart b/mobile/lib/domain/services/background_worker.service.dart index 0c3aa65107..5a502b99be 100644 --- a/mobile/lib/domain/services/background_worker.service.dart +++ b/mobile/lib/domain/services/background_worker.service.dart @@ -6,6 +6,7 @@ import 'package:background_downloader/background_downloader.dart'; import 'package:flutter/material.dart'; import 'package:hooks_riverpod/hooks_riverpod.dart'; import 'package:immich_data/data_controller.dart'; +import 'package:immich_data/store/store.dart' as data_store; import 'package:immich_mobile/constants/constants.dart'; import 'package:immich_mobile/domain/services/hash.service.dart'; import 'package:immich_mobile/domain/services/local_sync.service.dart'; @@ -20,7 +21,6 @@ import 'package:immich_mobile/providers/api.provider.dart'; import 'package:immich_mobile/providers/backup/drift_backup.provider.dart'; import 'package:immich_mobile/providers/infrastructure/album.provider.dart'; import 'package:immich_mobile/providers/infrastructure/asset.provider.dart'; -import 'package:immich_mobile/providers/infrastructure/data_store.dart' as data_store; import 'package:immich_mobile/providers/infrastructure/platform.provider.dart'; import 'package:immich_mobile/providers/infrastructure/sync.provider.dart'; import 'package:immich_mobile/providers/user.provider.dart'; diff --git a/mobile/lib/main.dart b/mobile/lib/main.dart index 9f6099a807..a62e61093e 100644 --- a/mobile/lib/main.dart +++ b/mobile/lib/main.dart @@ -12,6 +12,7 @@ import 'package:flutter/services.dart'; import 'package:flutter_displaymode/flutter_displaymode.dart'; import 'package:flutter_local_notifications/flutter_local_notifications.dart'; import 'package:hooks_riverpod/hooks_riverpod.dart'; +import 'package:immich_data/store/store.dart'; import 'package:immich_mobile/constants/constants.dart'; import 'package:immich_mobile/constants/locales.dart'; import 'package:immich_mobile/domain/services/background_worker.service.dart'; @@ -25,7 +26,6 @@ import 'package:immich_mobile/platform/background_worker_lock_api.g.dart'; import 'package:immich_mobile/providers/api.provider.dart'; import 'package:immich_mobile/providers/app_life_cycle.provider.dart'; import 'package:immich_mobile/providers/asset_viewer/share_intent_upload.provider.dart'; -import 'package:immich_mobile/providers/infrastructure/data_store.dart'; import 'package:immich_mobile/providers/infrastructure/platform.provider.dart'; import 'package:immich_mobile/providers/infrastructure/settings.provider.dart'; import 'package:immich_mobile/providers/locale_provider.dart'; diff --git a/mobile/lib/mixins/stream_notifier.mixin.dart b/mobile/lib/mixins/stream_notifier.mixin.dart deleted file mode 100644 index a068c46346..0000000000 --- a/mobile/lib/mixins/stream_notifier.mixin.dart +++ /dev/null @@ -1,37 +0,0 @@ -import 'dart:async'; - -import 'package:hooks_riverpod/hooks_riverpod.dart'; - -/// Converts a [Stream] into an [AsyncNotifier] -/// -/// The [build] method MUST call [buildFromStream] in order to properly subscribe -mixin StreamNotifierMixin { - set state(AsyncValue value); - - /// Forwards data into the notifier from [stream]. Future will stay open until [stream] completes - /// - /// Must be called in [build] - Future buildFromStream( - Ref> ref, - Stream stream, { - required T Function(Object error, StackTrace stack) onError, - }) { - final completer = Completer(); - - void apply(T value) { - if (completer.isCompleted) { - state = AsyncData(value); - } else { - completer.complete(value); - } - } - - final subscription = stream.listen( - apply, - onError: (Object error, StackTrace stack) => apply(onError(error, stack)), - ); - ref.onDispose(subscription.cancel); - - return completer.future; - } -} diff --git a/mobile/lib/presentation/pages/drift_activities.page.dart b/mobile/lib/presentation/pages/drift_activities.page.dart index effed4c3ef..3ea7025688 100644 --- a/mobile/lib/presentation/pages/drift_activities.page.dart +++ b/mobile/lib/presentation/pages/drift_activities.page.dart @@ -4,13 +4,13 @@ import 'package:auto_route/auto_route.dart'; import 'package:flutter/material.dart'; import 'package:flutter_hooks/flutter_hooks.dart' hide Store; import 'package:hooks_riverpod/hooks_riverpod.dart'; +import 'package:immich_data/store/store.dart'; import 'package:immich_mobile/domain/models/album/album.model.dart'; import 'package:immich_mobile/extensions/asyncvalue_extensions.dart'; import 'package:immich_mobile/extensions/build_context_extensions.dart'; import 'package:immich_mobile/presentation/widgets/action_buttons/like_activity_action_button.widget.dart'; import 'package:immich_mobile/presentation/widgets/album/drift_activity_text_field.dart'; import 'package:immich_mobile/providers/infrastructure/current_album.provider.dart'; -import 'package:immich_mobile/store/activity.dart'; import 'package:immich_mobile/widgets/activities/comment_bubble.dart'; @RoutePage() @@ -23,8 +23,7 @@ class DriftActivitiesPage extends HookConsumerWidget { @override Widget build(BuildContext context, WidgetRef ref) { - final activityNotifier = ref.read(albumActivityProvider((album.id, assetId)).notifier); - final activities = ref.watch(albumActivityProvider((album.id, assetId))); + final activities = ref.watch(Store.activity.list(album.id, assetId: assetId)); final listViewScrollController = useScrollController(); Future scrollToBottom() { @@ -36,7 +35,7 @@ class DriftActivitiesPage extends HookConsumerWidget { } Future onAddComment(String comment) async { - await activityNotifier.addComment(comment); + await ref.read(Store.activity).addComment(album.id, comment, assetId: assetId); unawaited(scrollToBottom()); } diff --git a/mobile/lib/presentation/widgets/action_buttons/like_activity_action_button.widget.dart b/mobile/lib/presentation/widgets/action_buttons/like_activity_action_button.widget.dart index 63c1181854..4ca5e1b9f4 100644 --- a/mobile/lib/presentation/widgets/action_buttons/like_activity_action_button.widget.dart +++ b/mobile/lib/presentation/widgets/action_buttons/like_activity_action_button.widget.dart @@ -3,13 +3,13 @@ import 'package:easy_localization/easy_localization.dart'; import 'package:flutter/material.dart'; import 'package:hooks_riverpod/hooks_riverpod.dart'; import 'package:immich_data/model/activity.dart'; +import 'package:immich_data/store/store.dart'; import 'package:immich_mobile/domain/models/asset/base_asset.model.dart'; import 'package:immich_mobile/extensions/translate_extensions.dart'; import 'package:immich_mobile/presentation/widgets/action_buttons/base_action_button.widget.dart'; import 'package:immich_mobile/providers/asset_viewer/asset_viewer.provider.dart'; import 'package:immich_mobile/providers/infrastructure/current_album.provider.dart'; import 'package:immich_mobile/providers/user.provider.dart'; -import 'package:immich_mobile/store/activity.dart'; class LikeActivityActionButton extends ConsumerWidget { const LikeActivityActionButton({super.key, this.iconOnly = false, this.menuItem = false}); @@ -23,7 +23,7 @@ class LikeActivityActionButton extends ConsumerWidget { final asset = ref.watch(assetViewerProvider.select((s) => s.currentAsset)) as RemoteAsset?; final user = ref.watch(currentUserProvider); - final activities = ref.watch(albumActivityProvider((album?.id ?? "", asset?.id))); + final activities = ref.watch(Store.activity.list(album?.id ?? "", assetId: asset?.id)); Future onTap(Activity? liked) async { if (user == null) { @@ -31,9 +31,9 @@ class LikeActivityActionButton extends ConsumerWidget { } if (liked != null) { - await ref.read(albumActivityProvider((album?.id ?? "", asset?.id)).notifier).removeActivity(liked.id); + await ref.read(Store.activity).remove(album?.id ?? "", liked.id); } else { - await ref.read(albumActivityProvider((album?.id ?? "", asset?.id)).notifier).addLike(); + await ref.read(Store.activity).addLike(album?.id ?? "", assetId: asset?.id); } } diff --git a/mobile/lib/presentation/widgets/asset_viewer/viewer_top_app_bar.widget.dart b/mobile/lib/presentation/widgets/asset_viewer/viewer_top_app_bar.widget.dart index b85b415521..0b43a9da5e 100644 --- a/mobile/lib/presentation/widgets/asset_viewer/viewer_top_app_bar.widget.dart +++ b/mobile/lib/presentation/widgets/asset_viewer/viewer_top_app_bar.widget.dart @@ -4,6 +4,7 @@ import 'package:auto_route/auto_route.dart'; import 'package:easy_localization/easy_localization.dart'; import 'package:flutter/material.dart'; import 'package:hooks_riverpod/hooks_riverpod.dart'; +import 'package:immich_data/store/store.dart'; import 'package:immich_mobile/domain/models/asset/base_asset.model.dart'; import 'package:immich_mobile/extensions/build_context_extensions.dart'; import 'package:immich_mobile/presentation/actions/action.widget.dart'; @@ -16,7 +17,6 @@ import 'package:immich_mobile/providers/infrastructure/current_album.provider.da import 'package:immich_mobile/providers/infrastructure/readonly_mode.provider.dart'; import 'package:immich_mobile/providers/routes.provider.dart'; import 'package:immich_mobile/routing/router.dart'; -import 'package:immich_mobile/store/activity.dart'; import 'package:immich_mobile/utils/timezone.dart'; import 'package:immich_ui/immich_ui.dart'; @@ -38,7 +38,7 @@ class ViewerTopAppBar extends ConsumerWidget implements PreferredSizeWidget { final showingDetails = ref.watch(assetViewerProvider.select((state) => state.showingDetails)); if (album != null && album.isActivityEnabled && album.isShared && asset is RemoteAsset) { - ref.watch(albumActivityProvider((album.id, asset.id))); + ref.watch(Store.activity.list(album.id, assetId: asset.id)); } final showingControls = ref.watch(assetViewerProvider.select((s) => s.showingControls)); diff --git a/mobile/lib/presentation/widgets/people/person_edit_birthday_modal.widget.dart b/mobile/lib/presentation/widgets/people/person_edit_birthday_modal.widget.dart index 5951958666..6afbe91ea6 100644 --- a/mobile/lib/presentation/widgets/people/person_edit_birthday_modal.widget.dart +++ b/mobile/lib/presentation/widgets/people/person_edit_birthday_modal.widget.dart @@ -3,9 +3,9 @@ import 'package:flutter/material.dart'; import 'package:fluttertoast/fluttertoast.dart'; import 'package:hooks_riverpod/hooks_riverpod.dart'; import 'package:immich_data/model/person.dart'; +import 'package:immich_data/store/store.dart'; import 'package:immich_mobile/extensions/build_context_extensions.dart'; import 'package:immich_mobile/extensions/translate_extensions.dart'; -import 'package:immich_mobile/providers/infrastructure/data_store.dart'; import 'package:immich_mobile/providers/infrastructure/people.provider.dart'; import 'package:immich_mobile/utils/debug_print.dart'; import 'package:immich_mobile/widgets/common/immich_toast.dart'; diff --git a/mobile/lib/presentation/widgets/people/person_edit_name_modal.widget.dart b/mobile/lib/presentation/widgets/people/person_edit_name_modal.widget.dart index 792a72c086..8d913e0d4c 100644 --- a/mobile/lib/presentation/widgets/people/person_edit_name_modal.widget.dart +++ b/mobile/lib/presentation/widgets/people/person_edit_name_modal.widget.dart @@ -3,9 +3,9 @@ import 'package:flutter/material.dart'; import 'package:fluttertoast/fluttertoast.dart'; import 'package:hooks_riverpod/hooks_riverpod.dart'; import 'package:immich_data/model/person.dart'; +import 'package:immich_data/store/store.dart'; import 'package:immich_mobile/extensions/build_context_extensions.dart'; import 'package:immich_mobile/extensions/translate_extensions.dart'; -import 'package:immich_mobile/providers/infrastructure/data_store.dart'; import 'package:immich_mobile/providers/infrastructure/people.provider.dart'; import 'package:immich_mobile/utils/debug_print.dart'; import 'package:immich_mobile/widgets/common/immich_toast.dart'; diff --git a/mobile/lib/providers/infrastructure/db.provider.dart b/mobile/lib/providers/infrastructure/db.provider.dart index 373c76878f..4eb2d6f807 100644 --- a/mobile/lib/providers/infrastructure/db.provider.dart +++ b/mobile/lib/providers/infrastructure/db.provider.dart @@ -1,4 +1,4 @@ -import 'package:immich_mobile/providers/infrastructure/data_store.dart'; +import 'package:immich_data/store/store.dart'; /// Direct database access for the repositories that have not yet moved into `immich_data` // TODO(rewrite): Remove this provider once all repositories have migrated to `immich_data` diff --git a/mobile/lib/providers/infrastructure/people.provider.dart b/mobile/lib/providers/infrastructure/people.provider.dart index dfb4a2f95a..14de9ec886 100644 --- a/mobile/lib/providers/infrastructure/people.provider.dart +++ b/mobile/lib/providers/infrastructure/people.provider.dart @@ -1,6 +1,6 @@ import 'package:hooks_riverpod/hooks_riverpod.dart'; import 'package:immich_data/model/person.dart'; -import 'package:immich_mobile/providers/infrastructure/data_store.dart'; +import 'package:immich_data/store/store.dart'; import 'package:immich_mobile/providers/infrastructure/user_metadata.provider.dart'; final driftPeopleAssetProvider = FutureProvider.family, String>((ref, assetId) async { diff --git a/mobile/lib/providers/search/people.provider.dart b/mobile/lib/providers/search/people.provider.dart index 296b708043..ecae711c67 100644 --- a/mobile/lib/providers/search/people.provider.dart +++ b/mobile/lib/providers/search/people.provider.dart @@ -1,6 +1,6 @@ import 'package:hooks_riverpod/hooks_riverpod.dart'; import 'package:immich_data/model/person.dart'; -import 'package:immich_mobile/providers/infrastructure/data_store.dart'; +import 'package:immich_data/store/store.dart'; import 'package:logging/logging.dart'; final getAllPeopleProvider = FutureProvider.autoDispose>((ref) async { diff --git a/mobile/lib/services/deep_link.service.dart b/mobile/lib/services/deep_link.service.dart index 34d2a6a895..e69d3c995e 100644 --- a/mobile/lib/services/deep_link.service.dart +++ b/mobile/lib/services/deep_link.service.dart @@ -1,6 +1,7 @@ import 'package:auto_route/auto_route.dart'; import 'package:hooks_riverpod/hooks_riverpod.dart'; import 'package:immich_data/store/person.dart'; +import 'package:immich_data/store/store.dart'; import 'package:immich_mobile/domain/models/memory.model.dart'; import 'package:immich_mobile/domain/models/user.model.dart'; import 'package:immich_mobile/domain/services/asset.service.dart' as beta_asset_service; @@ -10,7 +11,6 @@ import 'package:immich_mobile/domain/services/timeline.service.dart'; import 'package:immich_mobile/presentation/widgets/asset_viewer/asset_viewer.page.dart'; import 'package:immich_mobile/providers/infrastructure/album.provider.dart'; import 'package:immich_mobile/providers/infrastructure/asset.provider.dart' as beta_asset_provider; -import 'package:immich_mobile/providers/infrastructure/data_store.dart'; import 'package:immich_mobile/providers/infrastructure/memory.provider.dart'; import 'package:immich_mobile/providers/infrastructure/timeline.provider.dart'; import 'package:immich_mobile/providers/user.provider.dart'; diff --git a/mobile/lib/store/activity.dart b/mobile/lib/store/activity.dart deleted file mode 100644 index 5c2fe21229..0000000000 --- a/mobile/lib/store/activity.dart +++ /dev/null @@ -1,61 +0,0 @@ -import 'dart:async'; - -import 'package:hooks_riverpod/hooks_riverpod.dart'; -import 'package:immich_data/model/activity.dart'; -import 'package:immich_mobile/mixins/error_logger.mixin.dart'; -import 'package:immich_mobile/mixins/stream_notifier.mixin.dart'; -import 'package:immich_mobile/providers/infrastructure/data_store.dart'; -import 'package:logging/logging.dart'; - -/// Activities associated with a `albumId`, `assetId` pair -final albumActivityProvider = AsyncNotifierProvider.autoDispose - .family<_AlbumActivity, List, (String albumId, String? assetId)>(_AlbumActivity.new); - -class _AlbumActivity extends AutoDisposeFamilyAsyncNotifier, (String albumId, String? assetId)> - with ErrorLoggerMixin, StreamNotifierMixin> { - @override - final Logger logger = Logger("ActivityService"); - - late String albumId; - late String? assetId; - - @override - Future> build((String albumId, String? assetId) args) { - albumId = args.$1; - assetId = args.$2; - - return buildFromStream( - ref, - // TODO(rewrite): `force: true` matches the previous behavior of a HTTP request on mount - ref.watch(Store.activities).getAll(albumId, assetId: assetId, force: true), - onError: (error, stack) { - logger.severe("Failed to get all activities for album $albumId", error, stack); - return const []; - }, - ); - } - - Future removeActivity(String id) async { - await logError( - () async { - await ref.read(Store.activities).remove(albumId, id); - }, - defaultValue: null, - errorMessage: "Failed to delete activity", - ); - } - - Future addLike() async { - await guardError( - () => ref.read(Store.activities).addLike(albumId, assetId: assetId), - errorMessage: "Failed to create like for album $albumId", - ); - } - - Future addComment(String comment) async { - await guardError( - () => ref.read(Store.activities).addComment(albumId, comment, assetId: assetId), - errorMessage: "Failed to create comment for album $albumId", - ); - } -} diff --git a/mobile/lib/utils/isolate.dart b/mobile/lib/utils/isolate.dart index 95dfc40077..82ca6d5e10 100644 --- a/mobile/lib/utils/isolate.dart +++ b/mobile/lib/utils/isolate.dart @@ -3,11 +3,11 @@ import 'dart:ui'; import 'package:flutter/services.dart'; import 'package:hooks_riverpod/hooks_riverpod.dart'; +import 'package:immich_data/store/store.dart' as data_store; import 'package:immich_mobile/domain/services/log.service.dart'; import 'package:immich_mobile/entities/store.entity.dart'; import 'package:immich_mobile/providers/api.provider.dart'; import 'package:immich_mobile/providers/infrastructure/cancel.provider.dart'; -import 'package:immich_mobile/providers/infrastructure/data_store.dart' as data_store; import 'package:immich_mobile/utils/bootstrap.dart'; import 'package:immich_mobile/wm_executor.dart'; import 'package:logging/logging.dart'; diff --git a/mobile/lib/widgets/activities/comment_bubble.dart b/mobile/lib/widgets/activities/comment_bubble.dart index 94ecaa729c..23f47559d1 100644 --- a/mobile/lib/widgets/activities/comment_bubble.dart +++ b/mobile/lib/widgets/activities/comment_bubble.dart @@ -2,6 +2,7 @@ import 'package:auto_route/auto_route.dart'; import 'package:flutter/material.dart'; import 'package:hooks_riverpod/hooks_riverpod.dart'; import 'package:immich_data/model/activity.dart'; +import 'package:immich_data/store/store.dart'; import 'package:immich_mobile/domain/services/timeline.service.dart'; import 'package:immich_mobile/extensions/build_context_extensions.dart'; import 'package:immich_mobile/extensions/datetime_extensions.dart'; @@ -12,7 +13,6 @@ import 'package:immich_mobile/providers/infrastructure/current_album.provider.da import 'package:immich_mobile/providers/infrastructure/timeline.provider.dart'; import 'package:immich_mobile/providers/user.provider.dart'; import 'package:immich_mobile/routing/router.dart'; -import 'package:immich_mobile/store/activity.dart'; import 'package:immich_mobile/widgets/activities/dismissible_activity.dart'; import 'package:immich_mobile/widgets/common/user_circle_avatar.dart'; @@ -32,10 +32,6 @@ class CommentBubble extends ConsumerWidget { final isLike = activity.type == ActivityType.like; final bgColor = isOwn ? context.colorScheme.primaryContainer : context.colorScheme.surfaceContainer; - final activityNotifier = ref.read( - albumActivityProvider((album.id, isAssetActivity ? activity.assetId : null)).notifier, - ); - Future openAssetViewer() async { final asset = await ref.read(assetServiceProvider).getRemoteAsset(activity.assetId!); if (asset == null || !context.mounted) { @@ -120,7 +116,7 @@ class CommentBubble extends ConsumerWidget { final List contentChildren = [thumbnail, likes, commentBubble].whereType().toList(); return DismissibleActivity( - onDismiss: canDelete ? (id) async => await activityNotifier.removeActivity(id) : null, + onDismiss: canDelete ? (id) async => await ref.read(Store.activity).remove(album.id, id) : null, activity.id, Align( alignment: isOwn ? Alignment.centerRight : Alignment.centerLeft, diff --git a/mobile/packages/data/lib/data_controller.dart b/mobile/packages/data/lib/data_controller.dart index 461eab254f..51261ed509 100644 --- a/mobile/packages/data/lib/data_controller.dart +++ b/mobile/packages/data/lib/data_controller.dart @@ -2,13 +2,21 @@ import 'package:flutter/foundation.dart'; import 'package:immich_data/db/database.dart'; import 'package:immich_data/db/logger.dart'; import 'package:immich_data/db/person.dart'; -import 'package:immich_data/server/activity.dart'; import 'package:immich_data/server/person.dart'; -import 'package:immich_data/store/activity.dart'; import 'package:immich_data/store/person.dart'; import 'package:openapi/api.dart'; +import 'package:riverpod/riverpod.dart'; import 'package:sqlite3/common.dart'; +/// The [DataController] backing this container's store +/// +/// Must be overridden with a constructed instance (`Store.overrideWithValue`) +final dataControllerProvider = Provider( + (ref) => throw UnimplementedError( + "dataControllerProvider must be overridden in the isolate's ProviderContainer before use", + ), +); + /// Controls all data access. Serves request against the HTTP API and the Drift DB class DataController { final Drift _db; @@ -64,9 +72,8 @@ class DataController { PersonApiRepository(PeopleApi(_apiClient)), ); - // This is optional and not lazy so we only call `dispose` if necessary - ActivityService? _activities; - ActivityService get activities => _activities ??= ActivityService(ActivityApiRepository(ActivitiesApi(_apiClient))); + /// The authenticated HTTP client. Internal: only for the store's server repository providers + ApiClient get apiClient => _apiClient; /// Direct database access for the logic that has not yet moved into this package // TODO(rewrite): Remove once all repositories have migrated into this package @@ -77,9 +84,6 @@ class DataController { DriftLogger get logDb => _logDb; Future close() async { - await _activities?.dispose(); - _activities = null; - await _db.close(); // Close after the primary DB to ensure all logs are captured diff --git a/mobile/packages/data/lib/store/activity.dart b/mobile/packages/data/lib/store/activity.dart index 71a0573cfd..6b8f4b98a5 100644 --- a/mobile/packages/data/lib/store/activity.dart +++ b/mobile/packages/data/lib/store/activity.dart @@ -1,43 +1,108 @@ +import 'package:immich_data/data_controller.dart'; import 'package:immich_data/model/activity.dart'; import 'package:immich_data/server/activity.dart'; import 'package:immich_data/server/errors.dart'; -import 'package:immich_data/store/util/stream_cache.dart'; +import 'package:immich_data/store/util/slice.dart'; +import 'package:logging/logging.dart'; import 'package:meta/meta.dart'; +import 'package:openapi/api.dart'; +import 'package:riverpod/riverpod.dart'; + +final _log = Logger("ActivityStore"); + +@visibleForTesting +final activityApiProvider = Provider( + (ref) => ActivityApiRepository(ActivitiesApi(ref.watch(dataControllerProvider.select((c) => c.apiClient)))), +); /// Activities (comments and likes) on shared albums and their assets /// /// State is fetched over HTTP and mutated and cached in memory // TODO(agg23): This should not be called simply "Activity" -class ActivityService { - final ActivityApiRepository _api; - - late final StreamCache<(String albumId, String? assetId), List> _cache = StreamCache( - fetch: (scope) => _api.getAll(scope.$1, assetId: scope.$2), +extension type const ActivityStore._(Provider _provider) implements Provider { + static final _slice = Slice, ActivityScope>( + commands: (ref, bus) => ActivityMutations._(ref.watch(activityApiProvider), bus), + fetch: _fetch, + apply: _apply, ); @internal - ActivityService(this._api); + static final ActivityStore instance = ActivityStore._(_slice.commands); - /// All activities for an album specified by [albumId], or all activities for a specific asset within that album. - /// Providing [force] will make a new HTTP request on stream open (legacy behavior) + /// All activities for an album specified by [albumId], or all activities for a specific asset within that album /// /// **NOTE:** This is currently reactive only to in memory mutation, not live HTTP changes - Stream> getAll(String albumId, {String? assetId, bool force = false}) { - return _cache.get((albumId, assetId), force: force); + SliceQuery, ActivityEvent, ActivityScope> list(String albumId, {String? assetId}) => + _slice.query((albumId, assetId)); + + static Future> _fetch(Ref>> ref, ActivityScope scope) async { + try { + return await ref.read(activityApiProvider).getAll(scope.$1, assetId: scope.$2); + } catch (error, stack) { + _log.severe("Failed to get all activities for album ${scope.$1}", error, stack); + return const []; + } } + static List _apply(List current, ActivityEvent event, ActivityScope scope) => switch (event) { + ActivityUpserted(:final albumId, :final activity) => + // If there is a list for our album, we update it no matter what + // If there is a list for our specific asset, we also update that + albumId == scope.$1 && (scope.$2 == null || scope.$2 == activity.assetId) ? _upsert(current, activity) : current, + ActivityRemoved(:final albumId, :final activityId) => albumId == scope.$1 ? _remove(current, activityId) : current, + }; + + static List _upsert(List activities, Activity activity) { + final index = activities.indexWhere((a) => a.id == activity.id); + + if (index == -1) { + // Insert new item + return [...activities, activity]; + } else if (activities[index] == activity) { + // No change + return activities; + } else { + // Update existing + return [...activities]..[index] = activity; + } + } + + static List _remove(List activities, String activityId) => + activities.where((a) => a.id != activityId).toList(); +} + +/// Mutations for activities. Each completed mutation is published as an [ActivityEvent] +/// +/// Failures are logged with context and rethrown +// TODO(agg23): This should not be called simply "Activity" +class ActivityMutations { + final ActivityApiRepository _api; + final EventBus _bus; + + const ActivityMutations._(this._api, this._bus); + /// Add a comment to an album or asset. Providing [assetId] will add to the corresponding asset, otherwise the comment will be added to the album Future addComment(String albumId, String comment, {String? assetId}) async { - final activity = await _api.create(albumId, ActivityType.comment, assetId: assetId, comment: comment); - _upsert(albumId, activity); - return activity; + try { + final activity = await _api.create(albumId, ActivityType.comment, assetId: assetId, comment: comment); + _bus.publish(ActivityUpserted(albumId, activity)); + return activity; + } catch (error, stack) { + _log.severe("Failed to create comment for album $albumId", error, stack); + rethrow; + } } /// Add a like to an album or asset Providing [assetId] will add to the corresponding asset, otherwise the like will be added to the album Future addLike(String albumId, {String? assetId}) async { - final activity = await _api.create(albumId, ActivityType.like, assetId: assetId); - _upsert(albumId, activity); - return activity; + try { + final activity = await _api.create(albumId, ActivityType.like, assetId: assetId); + _bus.publish(ActivityUpserted(albumId, activity)); + return activity; + } catch (error, stack) { + _log.severe("Failed to create like for album $albumId", error, stack); + rethrow; + } } /// Remove an activity by its [activityId] @@ -46,42 +111,36 @@ class ActivityService { await _api.delete(activityId); } on NoResponseDtoError { // TODO(agg23): This error should not be thrown at all + } catch (error, stack) { + _log.severe("Failed to delete activity", error, stack); + rethrow; } - // Only drop on "success" (including the broken NoResponseDtoError above) - _drop(albumId, activityId); - } - - /// Terminate all streams and dispose of the cache - Future dispose() { - return _cache.dispose(); - } - - /// Apply an activity upsert to the in memory cache - void _upsert(String albumId, Activity activity) { - _cache.update( - // If there is a list for our album, we update it no matter what - // If there is a list for our specific asset, we also update that - (scope) => scope.$1 == albumId && (scope.$2 == null || scope.$2 == activity.assetId), - (activities) { - final index = activities.indexWhere((a) => a.id == activity.id); - - if (index == -1) { - // Insert new item - return [...activities, activity]; - } else if (activities[index] == activity) { - // No change - return activities; - } else { - // Update existing - return [...activities]..[index] = activity; - } - }, - ); - } - - /// Drop an activity from any cache entry that may contain it - void _drop(String albumId, String activityId) { - _cache.update((scope) => scope.$1 == albumId, (activities) => activities.where((a) => a.id != activityId).toList()); + // Only publish on success (including the broken NoResponseDtoError) + _bus.publish(ActivityRemoved(albumId, activityId)); } } + +/// The `albumId` and optional `assetId` pair an activity list is scoped to +typedef ActivityScope = (String albumId, String? assetId); + +/// A completed mutation to the activities of an album +sealed class ActivityEvent { + const ActivityEvent(); +} + +/// Created/updated [activity] within the album [albumId] +final class ActivityUpserted extends ActivityEvent { + final String albumId; + final Activity activity; + + const ActivityUpserted(this.albumId, this.activity); +} + +/// Deleted [activityId] from album [albumId] +final class ActivityRemoved extends ActivityEvent { + final String albumId; + final String activityId; + + const ActivityRemoved(this.albumId, this.activityId); +} diff --git a/mobile/lib/providers/infrastructure/data_store.dart b/mobile/packages/data/lib/store/store.dart similarity index 55% rename from mobile/lib/providers/infrastructure/data_store.dart rename to mobile/packages/data/lib/store/store.dart index 05d2190ee6..f17c0bebe5 100644 --- a/mobile/lib/providers/infrastructure/data_store.dart +++ b/mobile/packages/data/lib/store/store.dart @@ -1,17 +1,16 @@ -import 'package:hooks_riverpod/hooks_riverpod.dart'; import 'package:immich_data/data_controller.dart'; - -// TODO(rewrite): Rename file once `store.provider.dart` is migrated +import 'package:immich_data/store/activity.dart'; +import 'package:riverpod/riverpod.dart'; /// Global data layer, providing access to Drift and HTTP APIs, scoped by entity // TODO(rewrite): Possibly codegen? abstract final class Store { static Override overrideWithValue(DataController dataController) => - _dataControllerProvider.overrideWithValue(dataController); + dataControllerProvider.overrideWithValue(dataController); static final people = _store((c) => c.people); - static final activities = _store((c) => c.activities); + static final activity = ActivityStore.instance; /// Direct database access for the repositories that have not yet moved into `immich_data` // TODO(rewrite): Remove this provider once all repositories have migrated to `immich_data` @@ -19,11 +18,5 @@ abstract final class Store { // ----- Internal ----- - static ProviderListenable _store(T Function(DataController) get) => _dataControllerProvider.select(get); - - static final _dataControllerProvider = Provider( - (ref) => throw UnimplementedError( - "dataControllerProvider must be overridden in the isolate's ProviderContainer before use", - ), - ); + static ProviderListenable _store(T Function(DataController) get) => dataControllerProvider.select(get); } diff --git a/mobile/packages/data/lib/store/util/slice.dart b/mobile/packages/data/lib/store/util/slice.dart new file mode 100644 index 0000000000..6f4d6f7999 --- /dev/null +++ b/mobile/packages/data/lib/store/util/slice.dart @@ -0,0 +1,79 @@ +import 'dart:async'; + +import 'package:riverpod/riverpod.dart'; + +/// A broadcast channel for a single store [Slice]'s mutation events +class EventBus { + final _controller = StreamController.broadcast(sync: true); + + /// Send [event] to all current listeners + void publish(E event) => _controller.add(event); + + Stream get _stream => _controller.stream; + + Future dispose() => _controller.close(); +} + +/// The read notifier for one read scope (i.e. `getAll()`) of a [Slice]. Fetches once on build, then applies received mutation events and rebroadcasts them to Riverpod +class SliceNotifier extends AutoDisposeFamilyAsyncNotifier { + late final Provider> _bus; + late final Future Function(Ref> ref, Arg arg) _fetch; + late final T Function(T current, E event, Arg arg) _apply; + + @override + Future build(Arg arg) { + // On first build, run `_fetch` function to receive initial state + // Subscribe to event stream to keep our state current + final subscription = ref.watch(_bus)._stream.listen((event) { + final current = state.valueOrNull; + if (current == null) { + return; + } + + final result = _apply(current, event, arg); + if (!identical(result, current)) { + state = AsyncData(result); + } + }); + ref.onDispose(subscription.cancel); + + return _fetch(ref, arg); + } +} + +/// The provider of one read [Slice] scope, returned by the slice's named read accessors (`getAll()`) +typedef SliceQuery = AutoDisposeFamilyAsyncNotifierProvider, T, Arg>; + +/// Constructs a set of [Provider]'s mapping to commands and data subscriptions of the same in-memory store +/// +/// - [commands] - A function projecting a command [Provider]. The methods exposed by this provider will be mapped to be top level methods on [this] (`ref.watch(Store.x).doMutation()`) +/// - [fetch] - A function that provides initial state for all data subscriptions within this [Provider] +/// - [apply] - A function that applies events to the current in-memory store value. Identity must be preserved if no changes/updates are intended +class Slice { + Slice({ + required S Function(Ref ref, EventBus bus) commands, + required Future Function(Ref> ref, Arg arg) fetch, + required T Function(T current, E event, Arg arg) apply, + }) { + // The bus's lifetime is tied to the owning ProviderContainer + final bus = Provider>((ref) { + final bus = EventBus(); + ref.onDispose(() => unawaited(bus.dispose())); + return bus; + }); + + this.commands = Provider((ref) => commands(ref, ref.watch(bus))); + query = AsyncNotifierProvider.autoDispose.family, T, Arg>( + () => SliceNotifier() + .._bus = bus + .._fetch = fetch + .._apply = apply, + ); + } + + /// The slice's command [Provider] + late final Provider commands; + + /// The slice's scoped read providers + late final AutoDisposeAsyncNotifierProviderFamily, T, Arg> query; +} diff --git a/mobile/packages/data/lib/store/util/stream_cache.dart b/mobile/packages/data/lib/store/util/stream_cache.dart deleted file mode 100644 index f3064ccb67..0000000000 --- a/mobile/packages/data/lib/store/util/stream_cache.dart +++ /dev/null @@ -1,144 +0,0 @@ -import 'dart:async'; - -/// A in memory cache for streams that are backed by emphemeral data (typically HTTP servers). -/// Values are fetched on demand and disposed of when the last subscriber disconnects -class StreamCache { - final Future Function(K key) _fetch; - - final Map> _entries = {}; - final Map> _inflightFetches = {}; - - StreamCache({required this._fetch}); - - /// A live view of the value corresponding to [key]. [force] skips the cached emission and always fetches - // TODO(rewrite): Remove force; it exists only for old Flutter views that expect 0 data on first subscribe - Stream get(K key, {bool force = false}) { - // Reprents this call's subscription - late final StreamController localStreamController; - // Reprents the source of all updates for this key, shared between individual `get` callers - StreamSubscription? cacheStreamController; - - localStreamController = StreamController( - onListen: () { - final entry = _entries.putIfAbsent(key, _CacheEntry.new).ref(); - - // Push stream updates from the upstream to the local controller - cacheStreamController = entry.updateStreamController.stream.listen( - localStreamController.add, - onDone: () => unawaited(localStreamController.close()), - ); - - final currentData = entry.value; - if (currentData != null && !force) { - // If we have a cached value at call time, and we're not using legacy `force` behavior, immediately emit that value - localStreamController.add(currentData); - } - - if (currentData == null || force) { - // No data/force request, request new data from upstream - unawaited( - _fetchDeduped( - key, - ).catchError((Object error, StackTrace stack) => localStreamController.addError(error, stack)), - ); - } - }, - onCancel: () async { - await cacheStreamController?.cancel(); - - final entry = _entries[key]; - if (entry == null) { - return; - } - - if (entry.deref().refCount <= 0) { - _entries.remove(key); - } - }, - ); - - return localStreamController.stream; - } - - /// Apply [transform] to all values corresponding to keys matching [predicate] - /// - /// Returning the same value (identity) from [transform] will not emit a new value - void update(bool Function(K key) predicate, V Function(V value) transform) { - for (final MapEntry(:key, value: entry) in _entries.entries) { - if (!predicate(key)) { - continue; - } - - final currentValue = entry.value; - if (currentValue == null) { - continue; - } - - final newValue = transform(currentValue); - if (!identical(newValue, currentValue)) { - entry.value = newValue; - } - } - } - - /// Close all streams and delete caches - Future dispose() async { - _inflightFetches.clear(); - - final oldEntries = [..._entries.values]; - _entries.clear(); - - await Future.wait(oldEntries.map((entry) => entry.updateStreamController.close())); - } - - /// Fetch a single value matching the provided [key]. If a matching fetch is already in progress, await that fetch - Future _fetchDeduped(K key) { - return _inflightFetches.putIfAbsent(key, () async { - try { - final data = await _fetch(key); - final entry = _entries[key]; - - if (entry != null) { - entry.value = data; - } - } finally { - unawaited(_inflightFetches.remove(key)); - } - }); - } -} - -/// A streaming in memory cache entry -class _CacheEntry { - V? _value; - int refCount = 0; - - final StreamController updateStreamController = StreamController.broadcast(); - - /// The current value in the cache - V? get value => _value; - - /// Set the current cache value, sending it to all subscribers - set value(V newValue) { - _value = newValue; - updateStreamController.add(newValue); - } - - /// Increments the reference count of the entry - _CacheEntry ref() { - refCount += 1; - return this; - } - - /// Decrements the reference count of the entry. If there are no active references, closes the update stream - _CacheEntry deref() { - if (refCount == 0) { - unawaited(updateStreamController.close()); - - return this; - } - - refCount -= 1; - return this; - } -} diff --git a/mobile/packages/data/pubspec.yaml b/mobile/packages/data/pubspec.yaml index ceb9e608bf..6b46df438c 100644 --- a/mobile/packages/data/pubspec.yaml +++ b/mobile/packages/data/pubspec.yaml @@ -17,6 +17,7 @@ dependencies: path: ../../generated/openapi path: ^1.9.1 path_provider: ^2.1.5 + riverpod: ^2.6.1 sqlite3: ^3.3.2 sqlite3_connection_pool: ^0.2.6 sqlite_async: 0.14.2 diff --git a/mobile/packages/data/test/store/activity_test.dart b/mobile/packages/data/test/store/activity_test.dart new file mode 100644 index 0000000000..86efb01605 --- /dev/null +++ b/mobile/packages/data/test/store/activity_test.dart @@ -0,0 +1,176 @@ +import 'package:flutter_test/flutter_test.dart'; +import 'package:immich_data/model/activity.dart'; +import 'package:immich_data/model/user/user.dart'; +import 'package:immich_data/server/activity.dart'; +import 'package:immich_data/server/errors.dart'; +import 'package:immich_data/store/activity.dart'; +import 'package:immich_data/store/store.dart'; +import 'package:mocktail/mocktail.dart'; +import 'package:riverpod/riverpod.dart'; + +class MockActivityApiRepository extends Mock implements ActivityApiRepository {} + +void main() { + late ProviderContainer container; + late MockActivityApiRepository api; + + const albumId = 'album-1'; + const assetId = 'asset-1'; + + final user = UserDto(id: 'user-1', email: 'user@test.com', name: 'User', profileChangedAt: DateTime.utc(2025)); + + Activity activity(String id, {String? assetId, ActivityType type = ActivityType.comment, String? comment}) => + Activity(id: id, assetId: assetId, comment: comment, createdAt: DateTime.utc(2025), type: type, user: user); + + final albumScoped = Store.activity.list(albumId); + final assetScoped = Store.activity.list(albumId, assetId: assetId); + + Future pumpBothScopes() async { + container.listen(albumScoped, (_, _) {}); + container.listen(assetScoped, (_, _) {}); + await container.read(albumScoped.future); + await container.read(assetScoped.future); + } + + List idsIn(ProviderListenable>> scope) => + container.read(scope).requireValue.map((a) => a.id).toList(); + + void stubFetch({required List albumScoped, required List assetScoped}) { + when(() => api.getAll(albumId, assetId: null)).thenAnswer((_) async => albumScoped); + when(() => api.getAll(albumId, assetId: assetId)).thenAnswer((_) async => assetScoped); + } + + setUp(() { + api = MockActivityApiRepository(); + container = ProviderContainer(overrides: [activityApiProvider.overrideWithValue(api)]); + addTearDown(container.dispose); + }); + + group('list', () { + test('album/asset scopes fetch and expose independent views', () async { + stubFetch( + albumScoped: [ + activity('c1'), + activity('c2', assetId: assetId), + ], + assetScoped: [activity('c2', assetId: assetId)], + ); + + await pumpBothScopes(); + + expect(idsIn(albumScoped), ['c1', 'c2']); + expect(idsIn(assetScoped), ['c2']); + }); + + test('fetch failures become an empty list', () async { + when(() => api.getAll(albumId, assetId: null)).thenAnswer((_) => Future.error(Exception('network down'))); + + container.listen(albumScoped, (_, _) {}); + + expect(await container.read(albumScoped.future), isEmpty); + }); + + test('an unwatched scope is disposed and refetches on the next watch', () async { + stubFetch(albumScoped: [], assetScoped: []); + + final subscription = container.listen(albumScoped, (_, _) {}); + await container.read(albumScoped.future); + + subscription.close(); + await pumpEventQueue(); + + container.listen(albumScoped, (_, _) {}); + await container.read(albumScoped.future); + + verify(() => api.getAll(albumId, assetId: null)).called(2); + }); + }); + + group('mutations', () { + test('addLike on an asset patches both the asset and album scopes', () async { + stubFetch(albumScoped: [], assetScoped: []); + final like = activity('l1', assetId: assetId, type: ActivityType.like); + when(() => api.create(albumId, ActivityType.like, assetId: assetId)).thenAnswer((_) async => like); + await pumpBothScopes(); + + await container.read(Store.activity).addLike(albumId, assetId: assetId); + await pumpEventQueue(); + + expect(idsIn(albumScoped), ['l1']); + expect(idsIn(assetScoped), ['l1']); + }); + + test('addComment on the album alone does not touch the asset scope', () async { + stubFetch(albumScoped: [], assetScoped: []); + final comment = activity('c1', comment: 'nice'); + when( + () => api.create(albumId, ActivityType.comment, assetId: null, comment: 'nice'), + ).thenAnswer((_) async => comment); + await pumpBothScopes(); + + await container.read(Store.activity).addComment(albumId, 'nice'); + await pumpEventQueue(); + + expect(idsIn(albumScoped), ['c1']); + expect(idsIn(assetScoped), isEmpty); + }); + + test('remove drops the activity from every scope, treating NoResponseDtoError as success', () async { + final doomed = activity('c1', assetId: assetId); + stubFetch(albumScoped: [doomed, activity('c2')], assetScoped: [doomed]); + // `checkNull` throws on every successful delete because the API returns no body + when(() => api.delete('c1')).thenAnswer((_) => Future.error(const NoResponseDtoError())); + await pumpBothScopes(); + + await container.read(Store.activity).remove(albumId, 'c1'); + await pumpEventQueue(); + + expect(idsIn(albumScoped), ['c2']); + expect(idsIn(assetScoped), isEmpty); + }); + + test('mutations in other albums do not touch this scope', () async { + stubFetch(albumScoped: [], assetScoped: []); + final other = activity('x1'); + when( + () => api.create('album-2', ActivityType.comment, assetId: null, comment: 'hi'), + ).thenAnswer((_) async => other); + await pumpBothScopes(); + + await container.read(Store.activity).addComment('album-2', 'hi'); + await pumpEventQueue(); + + expect(idsIn(albumScoped), isEmpty); + expect(idsIn(assetScoped), isEmpty); + }); + }); + + group('failed mutations', () { + test('failed addLike rethrows and changes nothing', () async { + stubFetch(albumScoped: [], assetScoped: []); + when( + () => api.create(albumId, ActivityType.like, assetId: assetId), + ).thenAnswer((_) => Future.error(Exception('rejected'))); + await pumpBothScopes(); + + await expectLater(container.read(Store.activity).addLike(albumId, assetId: assetId), throwsException); + await pumpEventQueue(); + + expect(idsIn(albumScoped), isEmpty); + expect(idsIn(assetScoped), isEmpty); + }); + + test('failed remove rethrows and keeps the activity', () async { + final kept = activity('c1', assetId: assetId); + stubFetch(albumScoped: [kept], assetScoped: [kept]); + when(() => api.delete('c1')).thenAnswer((_) => Future.error(Exception('rejected'))); + await pumpBothScopes(); + + await expectLater(container.read(Store.activity).remove(albumId, 'c1'), throwsException); + await pumpEventQueue(); + + expect(idsIn(albumScoped), ['c1']); + expect(idsIn(assetScoped), ['c1']); + }); + }); +} diff --git a/mobile/packages/data/test/store/util/slice_test.dart b/mobile/packages/data/test/store/util/slice_test.dart new file mode 100644 index 0000000000..1c91e99b65 --- /dev/null +++ b/mobile/packages/data/test/store/util/slice_test.dart @@ -0,0 +1,94 @@ +import 'dart:async'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:immich_data/store/util/slice.dart'; +import 'package:riverpod/riverpod.dart'; + +class _TestCommands { + final EventBus bus; + + const _TestCommands(this.bus); +} + +void main() { + late ProviderContainer container; + late Slice<_TestCommands, int, List, String> slice; + + // The slice under test delegates to these per-test hooks + late Future> Function(String arg) onFetch; + late List Function(List current, int event) onApply; + + void publish(int event) => container.read(slice.commands).bus.publish(event); + + setUp(() { + onFetch = (_) async => [0]; + onApply = (current, event) => [...current, event]; + + slice = Slice( + commands: (ref, bus) => _TestCommands(bus), + fetch: (ref, arg) => onFetch(arg), + apply: (current, event, arg) => onApply(current, event), + ); + container = ProviderContainer(); + addTearDown(container.dispose); + }); + + test('exposes the fetched value', () async { + container.listen(slice.query('a'), (_, _) {}); + + expect(await container.read(slice.query('a').future), [0]); + }); + + test('applies published events to the current state', () async { + container.listen(slice.query('a'), (_, _) {}); + await container.read(slice.query('a').future); + + publish(1); + publish(2); + await pumpEventQueue(); + + expect(container.read(slice.query('a')).requireValue, [0, 1, 2]); + }); + + test('every live argument receives each event', () async { + onFetch = (arg) async => [arg.length]; + + container.listen(slice.query('a'), (_, _) {}); + container.listen(slice.query('bb'), (_, _) {}); + await container.read(slice.query('a').future); + await container.read(slice.query('bb').future); + + publish(9); + await pumpEventQueue(); + + expect(container.read(slice.query('a')).requireValue, [1, 9]); + expect(container.read(slice.query('bb')).requireValue, [2, 9]); + }); + + test('an identical return from apply publishes no new state', () async { + onApply = (current, _) => current; + + var notifications = 0; + container.listen(slice.query('a'), (_, _) => notifications++); + await container.read(slice.query('a').future); + final settled = notifications; + + publish(1); + await pumpEventQueue(); + + expect(notifications, settled); + }); + + test('events arriving before the initial fetch completes are dropped', () async { + final firstFetch = Completer>(); + onFetch = (_) => firstFetch.future; + + container.listen(slice.query('a'), (_, _) {}); + await pumpEventQueue(); + + publish(1); + firstFetch.complete([0]); + + expect(await container.read(slice.query('a').future), [0]); + }); +} diff --git a/mobile/packages/data/test/store/util/stream_cache_test.dart b/mobile/packages/data/test/store/util/stream_cache_test.dart deleted file mode 100644 index 7a7dac25e5..0000000000 --- a/mobile/packages/data/test/store/util/stream_cache_test.dart +++ /dev/null @@ -1,148 +0,0 @@ -import 'dart:async'; - -import 'package:flutter_test/flutter_test.dart'; -import 'package:immich_data/store/util/stream_cache.dart'; - -void main() { - late int fetches; - late StreamCache> cache; - - setUp(() { - fetches = 0; - cache = StreamCache(fetch: (key) async => ['$key-v${++fetches}']); - addTearDown(cache.dispose); - }); - - /// Subscribes and collects every emission - List> subscribe(String key, {bool force = false}) { - final emissions = >[]; - final subscription = cache.get(key, force: force).listen(emissions.add); - addTearDown(subscription.cancel); - return emissions; - } - - test('fetches and emits on first subscribe', () async { - await expectLater(cache.get('a'), emits(['a-v1'])); - }); - - test('a second subscriber is served from the cache without a fetch', () async { - subscribe('a'); - await pumpEventQueue(); - - await expectLater(cache.get('a'), emits(['a-v1'])); - - expect(fetches, 1); - }); - - test('force skips the cached emission and always fetches', () async { - subscribe('a'); - await pumpEventQueue(); - - await expectLater(cache.get('a', force: true), emits(['a-v2'])); - - expect(fetches, 2); - }); - - test('keys are cached independently', () async { - await expectLater(cache.get('a'), emits(['a-v1'])); - await expectLater(cache.get('b'), emits(['b-v2'])); - }); - - test('concurrent first subscribers share a single fetch', () async { - final completer = Completer>(); - cache = StreamCache( - fetch: (key) { - fetches++; - return completer.future; - }, - ); - addTearDown(cache.dispose); - final first = subscribe('a'); - final second = subscribe('a'); - - completer.complete(['shared']); - await pumpEventQueue(); - - expect(first, [ - ['shared'], - ]); - expect(first, second); - expect(fetches, 1); - }); - - test('a failed fetch is a stream error and the stream survives', () async { - var fail = true; - cache = StreamCache(fetch: (key) => fail ? Future.error(StateError('down')) : Future.value(['recovered'])); - addTearDown(cache.dispose); - - final events = []; - final subscription = cache.get('a').listen(events.add, onError: (Object error) => events.add('error')); - addTearDown(subscription.cancel); - await pumpEventQueue(); - expect(events, ['error']); - - fail = false; - subscribe('a', force: true); - await pumpEventQueue(); - - expect(events, [ - 'error', - ['recovered'], - ]); - }); - - test('the cache is dropped when the last subscriber cancels', () async { - final subscription = cache.get('a').listen((_) {}); - await pumpEventQueue(); - await subscription.cancel(); - - await expectLater(cache.get('a'), emits(['a-v2'])); - - expect(fetches, 2); - }); - - group('update', () { - test('transforms and publishes every matching cached value', () async { - final a = subscribe('a'); - final b = subscribe('b'); - final c = subscribe('c'); - await pumpEventQueue(); - - cache.update((key) => key != 'c', (value) => [...value, 'patched']); - await pumpEventQueue(); - - expect(a.last, ['a-v1', 'patched']); - expect(b.last, ['b-v2', 'patched']); - expect(c.last, ['c-v3']); - }); - - test('publishes nothing when the transform returns the identical value', () async { - final a = subscribe('a'); - await pumpEventQueue(); - - cache.update((_) => true, (value) => value); - await pumpEventQueue(); - - expect(a, hasLength(1)); - }); - - test('does not touch keys that have no cached value yet', () { - cache.update((_) => true, (value) => [...value, 'patched']); - // No cached values exist; nothing to transform and nothing thrown. - expect(fetches, 0); - }); - }); - - test('dispose ends every subscriber stream', () async { - var done = false; - final subscription = cache.get('a').listen((_) {}); - subscription.onDone(() => done = true); - addTearDown(subscription.cancel); - await pumpEventQueue(); - - await cache.dispose(); - await pumpEventQueue(); - - expect(done, isTrue); - }); -} diff --git a/mobile/test/providers/activity_provider_test.dart b/mobile/test/providers/activity_provider_test.dart deleted file mode 100644 index c2e8c99207..0000000000 --- a/mobile/test/providers/activity_provider_test.dart +++ /dev/null @@ -1,144 +0,0 @@ -import 'dart:async'; - -import 'package:flutter_test/flutter_test.dart'; -import 'package:hooks_riverpod/hooks_riverpod.dart'; -import 'package:immich_data/data_controller.dart'; -import 'package:immich_data/model/activity.dart'; -import 'package:immich_data/model/user/user.dart'; -import 'package:immich_data/store/activity.dart'; -import 'package:immich_mobile/providers/infrastructure/data_store.dart'; -import 'package:immich_mobile/store/activity.dart'; -import 'package:mocktail/mocktail.dart'; - -class MockDataController extends Mock implements DataController {} - -class MockActivityService extends Mock implements ActivityService {} - -void main() { - late ProviderContainer container; - late MockActivityService service; - late StreamController> albumScopedStream; - late StreamController> assetScopedStream; - - const albumId = 'album-1'; - const assetId = 'asset-1'; - const albumScoped = (albumId, null); - const assetScoped = (albumId, assetId); - - final user = UserDto(id: 'user-1', email: 'user@test.com', name: 'User', profileChangedAt: DateTime.utc(2025)); - - Activity activity(String id, {String? assetId, ActivityType type = ActivityType.comment, String? comment}) => - Activity(id: id, assetId: assetId, comment: comment, createdAt: DateTime.utc(2025), type: type, user: user); - - Future pumpBothScopes() async { - container.listen(albumActivityProvider(assetScoped), (_, _) {}); - container.listen(albumActivityProvider(albumScoped), (_, _) {}); - albumScopedStream.add([]); - assetScopedStream.add([]); - await container.read(albumActivityProvider(assetScoped).future); - await container.read(albumActivityProvider(albumScoped).future); - } - - List idsIn((String, String?) scope) => - container.read(albumActivityProvider(scope)).requireValue.map((a) => a.id).toList(); - - setUp(() { - service = MockActivityService(); - albumScopedStream = StreamController>.broadcast(); - assetScopedStream = StreamController>.broadcast(); - addTearDown(albumScopedStream.close); - addTearDown(assetScopedStream.close); - when(() => service.getAll(albumId, assetId: null, force: true)).thenAnswer((_) => albumScopedStream.stream); - when(() => service.getAll(albumId, assetId: assetId, force: true)).thenAnswer((_) => assetScopedStream.stream); - - final controller = MockDataController(); - when(() => controller.activities).thenReturn(service); - - container = ProviderContainer(overrides: [Store.overrideWithValue(controller)]); - addTearDown(container.dispose); - }); - - group('build', () { - test('album/asset scopes get their own views', () async { - container.listen(albumActivityProvider(assetScoped), (_, _) {}); - container.listen(albumActivityProvider(albumScoped), (_, _) {}); - albumScopedStream.add([activity('c1'), activity('c2', assetId: assetId)]); - assetScopedStream.add([activity('c2', assetId: assetId)]); - - expect(await container.read(albumActivityProvider(albumScoped).future), hasLength(2)); - expect(await container.read(albumActivityProvider(assetScoped).future), hasLength(1)); - }); - - test('errors become an empty list', () async { - container.listen(albumActivityProvider(albumScoped), (_, _) {}); - albumScopedStream.addError(Exception('network down')); - - expect(await container.read(albumActivityProvider(albumScoped).future), isEmpty); - }); - }); - - test('new events get pushed to Riverpod', () async { - await pumpBothScopes(); - - final like = activity('l1', assetId: assetId, type: ActivityType.like); - albumScopedStream.add([like]); - assetScopedStream.add([like]); - await pumpEventQueue(); - - expect(idsIn(albumScoped), ['l1']); - expect(idsIn(assetScoped), ['l1']); - }); - - group('mutations', () { - test('addLike calls API', () async { - final like = activity('l1', assetId: assetId, type: ActivityType.like); - when(() => service.addLike(albumId, assetId: assetId)).thenAnswer((_) async => like); - await pumpBothScopes(); - - await container.read(albumActivityProvider(assetScoped).notifier).addLike(); - - verify(() => service.addLike(albumId, assetId: assetId)).called(1); - }); - - test('addComment calls API', () async { - final comment = activity('c1', assetId: assetId, comment: 'nice'); - when(() => service.addComment(albumId, 'nice', assetId: assetId)).thenAnswer((_) async => comment); - await pumpBothScopes(); - - await container.read(albumActivityProvider(assetScoped).notifier).addComment('nice'); - - verify(() => service.addComment(albumId, 'nice', assetId: assetId)).called(1); - }); - - test('removeActivity calls API', () async { - when(() => service.remove(albumId, 'c1')).thenAnswer((_) async {}); - await pumpBothScopes(); - - await container.read(albumActivityProvider(assetScoped).notifier).removeActivity('c1'); - - verify(() => service.remove(albumId, 'c1')).called(1); - }); - }); - - group('failed mutations', () { - test('addLike logs error', () async { - when(() => service.addLike(albumId, assetId: assetId)).thenAnswer((_) => Future.error(Exception('rejected'))); - await pumpBothScopes(); - - await container.read(albumActivityProvider(assetScoped).notifier).addLike(); - - expect(idsIn(assetScoped), isEmpty); - expect(idsIn(albumScoped), isEmpty); - }); - - test('removeActivity logs error', () async { - when(() => service.remove(albumId, 'c1')).thenAnswer((_) => Future.error(Exception('rejected'))); - await pumpBothScopes(); - - await container.read(albumActivityProvider(assetScoped).notifier).removeActivity('c1'); - - expect(idsIn(assetScoped), isEmpty); - expect(idsIn(albumScoped), isEmpty); - }); - }); -}