Add caching for own capabilities

Author: renefloor

packages/stream_feeds/lib/src/generated/api/model/own_capabilities_batch_response.dart

@@ -23,7 +23,7 @@ class OwnCapabilitiesBatchResponse with _$OwnCapabilitiesBatchResponse {
   });
 
   @override
-  final Map<String, List<String>> capabilities;
+  final Map<String, List<FeedOwnCapability>> capabilities;
 
   @override
   final String duration;

packages/stream_feeds/lib/src/generated/api/model/own_capabilities_batch_response.freezed.dart

@@ -0,0 +1,56 @@
+import 'dart:async';
+
+import '../../stream_feeds.dart' as api;
+import '../../stream_feeds.dart';
+import '../utils/batcher.dart';
+
+class CapabilitiesRepository {
+  CapabilitiesRepository({required api.DefaultApi api}) : _api = api;
+
+  final api.DefaultApi _api;
+  final Map<String, List<FeedOwnCapability>> _capabilities = {};
+
+  late final Batcher<String, Result<Map<String, List<FeedOwnCapability>>>>
+      _fetchBatcher = Batcher(
+    action: (feeds) => fetchCapabilities(feeds: feeds),
+  );
+
+  Future<Result<Map<String, List<FeedOwnCapability>>>> fetchCapabilities({
+    required List<String> feeds,
+  }) async {
+    final result = await _api.ownCapabilitiesBatch(
+      ownCapabilitiesBatchRequest: api.OwnCapabilitiesBatchRequest(
+        feeds: feeds,
+      ),
+    );
+
+    return result.map((response) {
+      _mergeWithCache(response.capabilities);
+      return response.capabilities;
+    });
+  }
+
+  void addCapabilities(String feed, List<FeedOwnCapability> capabilities) {
+    _capabilities[feed] = capabilities;
+  }
+
+  Future<List<FeedOwnCapability>?> getCapabilities(String feed) async {
+    return _capabilities[feed] ??
+        (await _fetchBatchedFeedCapabilities(feed)).getOrNull();
+  }
+
+  void dispose() {
+    _fetchBatcher.dispose();
+  }
+
+  Future<Result<List<FeedOwnCapability>>> _fetchBatchedFeedCapabilities(
+    String feed,
+  ) async {
+    final capabilities = await _fetchBatcher.add(feed);
+    return capabilities.map((capabilities) => capabilities[feed] ?? []);
+  }
+
+  void _mergeWithCache(Map<String, List<FeedOwnCapability>> capabilities) {
+    _capabilities.addAll(capabilities);
+  }
+}

packages/stream_feeds/lib/src/generated/api/model/own_capabilities_batch_response.g.dart

@@ -11,6 +11,7 @@ import '../../models/feed_id.dart';
 import '../../models/feeds_reaction_data.dart';
 import '../../models/follow_data.dart';
 import '../../models/mark_activity_data.dart';
+import '../../repository/capabilities_repository.dart';
 import '../feed_state.dart';
 
 import 'state_event_handler.dart';
@@ -19,16 +20,28 @@ class FeedEventHandler implements StateEventHandler {
   const FeedEventHandler({
     required this.fid,
     required this.state,
+    required this.capabilitiesRepository,
   });
 
   final FeedId fid;
   final FeedStateNotifier state;
+  final CapabilitiesRepository capabilitiesRepository;
 
   @override
-  void handleEvent(WsEvent event) {
+  Future<void> handleEvent(WsEvent event) async {
     if (event is api.ActivityAddedEvent) {
       if (event.fid != fid.rawValue) return;
-      return state.onActivityAdded(event.activity.toModel());
+      var activity = event.activity.toModel();
+      state.onActivityAdded(activity);
+      
+      final ownCapabilities =
+          await capabilitiesRepository.getCapabilities(fid.rawValue);
+      activity = activity.copyWith(
+        currentFeed: activity.currentFeed?.copyWith(
+          ownCapabilities: ownCapabilities ?? [],
+        ),
+      );
+      state.onActivityUpdated(activity);
     }
 
     if (event is api.ActivityDeletedEvent) {

packages/stream_feeds/lib/src/repository/capabilities_repository.dart

@@ -19,6 +19,7 @@ import '../models/request/activity_add_comment_request.dart';
 import '../models/request/feed_add_activity_request.dart';
 import '../repository/activities_repository.dart';
 import '../repository/bookmarks_repository.dart';
+import '../repository/capabilities_repository.dart';
 import '../repository/comments_repository.dart';
 import '../repository/feeds_repository.dart';
 import '../repository/polls_repository.dart';
@@ -48,6 +49,7 @@ class Feed with Disposable {
     required this.commentsRepository,
     required this.feedsRepository,
     required this.pollsRepository,
+    required this.capabilitiesRepository,
     required this.eventsEmitter,
     required Stream<void> onReconnectEmitter,
   }) {
@@ -66,7 +68,11 @@ class Feed with Disposable {
     );
 
     // Attach event handlers for the feed events
-    final handler = FeedEventHandler(fid: fid, state: _stateNotifier);
+    final handler = FeedEventHandler(
+      fid: fid,
+      state: _stateNotifier,
+      capabilitiesRepository: capabilitiesRepository,
+    );
     _feedSubscriptions.add(eventsEmitter.listen(handler.handleEvent));
 
     // Automatically refetch data on reconnection
@@ -84,6 +90,7 @@ class Feed with Disposable {
   final CommentsRepository commentsRepository;
   final FeedsRepository feedsRepository;
   final PollsRepository pollsRepository;
+  final CapabilitiesRepository capabilitiesRepository;
 
   late final MemberList _memberList;
 
@@ -108,7 +115,23 @@ class Feed with Disposable {
   /// Returns a [Result] containing the [FeedData] or an error.
   Future<Result<FeedData>> getOrCreate() async {
     final result = await feedsRepository.getOrCreateFeed(query);
-    result.onSuccess(_stateNotifier.onQueryFeed);
+    result.onSuccess((feedData) {
+      _stateNotifier.onQueryFeed(feedData);
+
+      // TODO move and also do with fetchMore
+      capabilitiesRepository.addCapabilities(
+        feedData.feed.id,
+        feedData.feed.ownCapabilities,
+      );
+      for (final activity in feedData.activities.items) {
+        if (activity.currentFeed case final feed?) {
+          capabilitiesRepository.addCapabilities(
+            feed.id,
+            feed.ownCapabilities,
+          );
+        }
+      }
+    });
 
     return result.map((feedData) => feedData.feed);
   }

packages/stream_feeds/lib/src/state/event/feed_event_handler.dart

@@ -0,0 +1,73 @@
+import 'dart:async';
+
+class Batcher<T, R> {
+  Batcher({
+    required this.action,
+    this.interval = const Duration(seconds: 2),
+  });
+
+  final Duration interval;
+  final Action<T, R> action;
+  Timer? _timer;
+  Set<T> _itemsToProcess = {};
+  Set<T> _itemsBeingProcessed = {};
+
+  late DateTime _lastRun = DateTime.fromMillisecondsSinceEpoch(0);
+  Completer<R>? _nextActionCompleter;
+  Completer<R>? _ongoingActionCompleter;
+
+  /// Adds an item to the batch and returns the result of the whole batch.
+  Future<R> add(T item) {
+    if (_itemsBeingProcessed.contains(item)) {
+      return _ongoingActionCompleter!.future;
+    }
+
+    _itemsToProcess.add(item);
+    _nextActionCompleter ??= _planBatchFetch();
+    return _nextActionCompleter!.future;
+  }
+
+  void dispose() {
+    _timer?.cancel();
+  }
+
+  Completer<R> _planBatchFetch() {
+    final timeSinceLastRun = DateTime.now().difference(_lastRun);
+
+    final newActionCompleter = Completer<R>();
+    _nextActionCompleter = newActionCompleter;
+    _lastRun = DateTime.now();
+
+    if (timeSinceLastRun >= interval) {
+      _runBatch();
+    } else {
+      _timer = Timer(interval - timeSinceLastRun, _runBatch);
+    }
+
+    return newActionCompleter;
+  }
+
+  void _runBatch() {
+    if (_nextActionCompleter == null) {
+      return;
+    }
+    final completer = _nextActionCompleter!;
+    _ongoingActionCompleter = completer;
+    _itemsBeingProcessed = _itemsToProcess;
+
+    final currentAction = action(_itemsToProcess.toList());
+    completer.complete(currentAction);
+    currentAction.whenComplete(
+      () {
+        _ongoingActionCompleter = null;
+        _itemsBeingProcessed = {};
+      },
+    );
+
+    _nextActionCompleter = null;
+    _itemsToProcess = {};
+    _timer = null;
+  }
+}
+
+typedef Action<T, R> = Future<R> Function(List<T> items);