helpy4/immich
1
0
Fork 0
mirror of https://github.com/immich-app/immich synced 2026-08-15 13:03:57 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge 0ffccc9d7b into 7fa887bc8c

Browse source
This commit is contained in:
Adam Gastineau 2026-08-07 10:00:24 -07:00 • committed by GitHub
commit 7a4ccc7266
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
25 changed files with 498 additions and 631 deletions
Download patch file Download diff file Expand all files Collapse all files

2
mobile/lib/domain/services/background_worker.service.dart
View file

@ -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';

2
mobile/lib/main.dart
View file

@ -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';

37
mobile/lib/mixins/stream_notifier.mixin.dart
View file

@ -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<T> {
set state(AsyncValue<T> value);
/// Forwards data into the notifier from [stream]. Future will stay open until [stream] completes
///
/// Must be called in [build]
Future<T> buildFromStream(
Ref<AsyncValue<T>> ref,
Stream<T> stream, {
required T Function(Object error, StackTrace stack) onError,
}) {
final completer = Completer<T>();
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;
}
}

7
mobile/lib/presentation/pages/drift_activities.page.dart
View file

@ -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<void> scrollToBottom() {
@ -36,7 +35,7 @@ class DriftActivitiesPage extends HookConsumerWidget {
}
Future<void> onAddComment(String comment) async {
await activityNotifier.addComment(comment);
await ref.read(Store.activity).addComment(album.id, comment, assetId: assetId);
unawaited(scrollToBottom());
}

8
mobile/lib/presentation/widgets/action_buttons/like_activity_action_button.widget.dart
View file

@ -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<void> 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);
}
}

4
mobile/lib/presentation/widgets/asset_viewer/viewer_top_app_bar.widget.dart
View file

@ -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));

2
mobile/lib/presentation/widgets/people/person_edit_birthday_modal.widget.dart
View file

@ -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';

2
mobile/lib/presentation/widgets/people/person_edit_name_modal.widget.dart
View file

@ -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';

2
mobile/lib/providers/infrastructure/db.provider.dart
View file

@ -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`

2
mobile/lib/providers/infrastructure/people.provider.dart
View file

@ -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<List<Person>, String>((ref, assetId) async {

2
mobile/lib/providers/search/people.provider.dart
View file

@ -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<List<PersonDto>>((ref) async {

2
mobile/lib/services/deep_link.service.dart
View file

@ -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';

61
mobile/lib/store/activity.dart
View file

@ -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<Activity>, (String albumId, String? assetId)>(_AlbumActivity.new);
class _AlbumActivity extends AutoDisposeFamilyAsyncNotifier<List<Activity>, (String albumId, String? assetId)>
with ErrorLoggerMixin, StreamNotifierMixin<List<Activity>> {
@override
final Logger logger = Logger("ActivityService");
late String albumId;
late String? assetId;
@override
Future<List<Activity>> 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<void> removeActivity(String id) async {
await logError(
() async {
await ref.read(Store.activities).remove(albumId, id);
},
defaultValue: null,
errorMessage: "Failed to delete activity",
);
}
Future<void> addLike() async {
await guardError(
() => ref.read(Store.activities).addLike(albumId, assetId: assetId),
errorMessage: "Failed to create like for album $albumId",
);
}
Future<void> addComment(String comment) async {
await guardError(
() => ref.read(Store.activities).addComment(albumId, comment, assetId: assetId),
errorMessage: "Failed to create comment for album $albumId",
);
}
}

2
mobile/lib/utils/isolate.dart
View file

@ -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';

8
mobile/lib/widgets/activities/comment_bubble.dart
View file

@ -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<void> 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<Widget> contentChildren = [thumbnail, likes, commentBubble].whereType<Widget>().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,

20
mobile/packages/data/lib/data_controller.dart
View file

@ -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<DataController>(
(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<void> close() async {
await _activities?.dispose();
_activities = null;
await _db.close();
// Close after the primary DB to ensure all logs are captured

163
mobile/packages/data/lib/store/activity.dart
View file

@ -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<ActivityApiRepository>(
(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<Activity>> _cache = StreamCache(
fetch: (scope) => _api.getAll(scope.$1, assetId: scope.$2),
extension type const ActivityStore._(Provider<ActivityMutations> _provider) implements Provider<ActivityMutations> {
static final _slice = Slice<ActivityMutations, ActivityEvent, List<Activity>, 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<List<Activity>> getAll(String albumId, {String? assetId, bool force = false}) {
return _cache.get((albumId, assetId), force: force);
SliceQuery<List<Activity>, ActivityEvent, ActivityScope> list(String albumId, {String? assetId}) =>
_slice.query((albumId, assetId));
static Future<List<Activity>> _fetch(Ref<AsyncValue<List<Activity>>> 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<Activity> _apply(List<Activity> 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<Activity> _upsert(List<Activity> 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<Activity> _remove(List<Activity> 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<ActivityEvent> _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<Activity> 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<Activity> 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<void> 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);
}

17
mobile/lib/providers/infrastructure/data_store.dart → mobile/packages/data/lib/store/store.dart
View file

@ -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<T> _store<T>(T Function(DataController) get) => _dataControllerProvider.select(get);
static final _dataControllerProvider = Provider<DataController>(
(ref) => throw UnimplementedError(
"dataControllerProvider must be overridden in the isolate's ProviderContainer before use",
),
);
static ProviderListenable<T> _store<T>(T Function(DataController) get) => dataControllerProvider.select(get);
}

79
mobile/packages/data/lib/store/util/slice.dart Normal file
View file

@ -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<E> {
final _controller = StreamController<E>.broadcast(sync: true);
/// Send [event] to all current listeners
void publish(E event) => _controller.add(event);
Stream<E> get _stream => _controller.stream;
Future<void> 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<T, E, Arg> extends AutoDisposeFamilyAsyncNotifier<T, Arg> {
late final Provider<EventBus<E>> _bus;
late final Future<T> Function(Ref<AsyncValue<T>> ref, Arg arg) _fetch;
late final T Function(T current, E event, Arg arg) _apply;
@override
Future<T> 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<T, E, Arg> = AutoDisposeFamilyAsyncNotifierProvider<SliceNotifier<T, E, Arg>, 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<S, E, T, Arg> {
Slice({
required S Function(Ref<S> ref, EventBus<E> bus) commands,
required Future<T> Function(Ref<AsyncValue<T>> 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<EventBus<E>>((ref) {
final bus = EventBus<E>();
ref.onDispose(() => unawaited(bus.dispose()));
return bus;
});
this.commands = Provider<S>((ref) => commands(ref, ref.watch(bus)));
query = AsyncNotifierProvider.autoDispose.family<SliceNotifier<T, E, Arg>, T, Arg>(
() => SliceNotifier<T, E, Arg>()
.._bus = bus
.._fetch = fetch
.._apply = apply,
);
}
/// The slice's command [Provider]
late final Provider<S> commands;
/// The slice's scoped read providers
late final AutoDisposeAsyncNotifierProviderFamily<SliceNotifier<T, E, Arg>, T, Arg> query;
}

144
mobile/packages/data/lib/store/util/stream_cache.dart
View file

@ -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<K, V extends Object> {
final Future<V> Function(K key) _fetch;
final Map<K, _CacheEntry<V>> _entries = {};
final Map<K, Future<void>> _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<V> get(K key, {bool force = false}) {
// Reprents this call's subscription
late final StreamController<V> localStreamController;
// Reprents the source of all updates for this key, shared between individual `get` callers
StreamSubscription<V>? 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<void> 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<void> _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> {
V? _value;
int refCount = 0;
final StreamController<V> 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<V> ref() {
refCount += 1;
return this;
}
/// Decrements the reference count of the entry. If there are no active references, closes the update stream
_CacheEntry<V> deref() {
if (refCount == 0) {
unawaited(updateStreamController.close());
return this;
}
refCount -= 1;
return this;
}
}

1
mobile/packages/data/pubspec.yaml
View file

@ -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

176
mobile/packages/data/test/store/activity_test.dart Normal file
View file

@ -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<void> pumpBothScopes() async {
container.listen(albumScoped, (_, _) {});
container.listen(assetScoped, (_, _) {});
await container.read(albumScoped.future);
await container.read(assetScoped.future);
}
List<String> idsIn(ProviderListenable<AsyncValue<List<Activity>>> scope) =>
container.read(scope).requireValue.map((a) => a.id).toList();
void stubFetch({required List<Activity> albumScoped, required List<Activity> 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']);
});
});
}

94
mobile/packages/data/test/store/util/slice_test.dart Normal file
View file

@ -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<int> bus;
const _TestCommands(this.bus);
}
void main() {
late ProviderContainer container;
late Slice<_TestCommands, int, List<int>, String> slice;
// The slice under test delegates to these per-test hooks
late Future<List<int>> Function(String arg) onFetch;
late List<int> Function(List<int> 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<List<int>>();
onFetch = (_) => firstFetch.future;
container.listen(slice.query('a'), (_, _) {});
await pumpEventQueue();
publish(1);
firstFetch.complete([0]);
expect(await container.read(slice.query('a').future), [0]);
});
}

148
mobile/packages/data/test/store/util/stream_cache_test.dart
View file

@ -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<String, List<String>> cache;
setUp(() {
fetches = 0;
cache = StreamCache(fetch: (key) async => ['$key-v${++fetches}']);
addTearDown(cache.dispose);
});
/// Subscribes and collects every emission
List<List<String>> subscribe(String key, {bool force = false}) {
final emissions = <List<String>>[];
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<List<String>>();
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 = <Object>[];
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);
});
}

144
mobile/test/providers/activity_provider_test.dart
View file

@ -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<List<Activity>> albumScopedStream;
late StreamController<List<Activity>> 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<void> 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<String> idsIn((String, String?) scope) =>
container.read(albumActivityProvider(scope)).requireValue.map((a) => a.id).toList();
setUp(() {
service = MockActivityService();
albumScopedStream = StreamController<List<Activity>>.broadcast();
assetScopedStream = StreamController<List<Activity>>.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);
});
});
}