feat(llc): merge feed and activity event handlers

packages/stream_feeds/CHANGELOG.md

@@ -1,6 +1,7 @@
 ## unreleased
 - Update follower and following counts on the feed state when receiving follow websocket events.
 - Fix FeedsReactionData id for updating reactions in the feed state.
+- Improve feed and activity state updates for websocket events.
 - Improvement for stories and minor updates to other AggregatedActivity state updates.
 
 ## 0.3.1

packages/stream_feeds/lib/src/client/feeds_client_impl.dart

@@ -383,6 +383,7 @@ class StreamFeedsClientImpl implements StreamFeedsClient {
       query: query,
       commentsRepository: _commentsRepository,
       eventsEmitter: events,
+      currentUserId: user.id,
     );
   }
 

packages/stream_feeds/lib/src/models.dart

@@ -7,7 +7,7 @@ export 'models/feed_member_data.dart';
 export 'models/feed_member_request_data.dart';
 export 'models/feeds_config.dart';
 export 'models/follow_data.dart';
-export 'models/poll_data.dart';
+export 'models/poll_data.dart' show PollData;
 export 'models/poll_option_data.dart';
 export 'models/poll_vote_data.dart';
 export 'models/push_notifications_config.dart';

packages/stream_feeds/lib/src/models/comment_data.dart

@@ -1,6 +1,7 @@
 // ignore_for_file: avoid_redundant_argument_values
 
 import 'package:freezed_annotation/freezed_annotation.dart';
+import 'package:stream_core/stream_core.dart';
 
 import '../generated/api/models.dart';
 import '../state/query/comments_query.dart';
@@ -197,3 +198,99 @@ extension CommentResponseMapper on CommentResponse {
     );
   }
 }
+
+extension CommentDataMutations on CommentData {
+  /// Adds a reaction to the comment, updating the latest reactions, reaction groups, reaction count,
+  /// and own reactions if applicable.
+  ///
+  /// @param reaction The reaction to add.
+  /// @param currentUserId The ID of the current user, used to update own reactions.
+  /// @return A new [CommentData] instance with the updated reaction data.
+  CommentData addReaction(
+    FeedsReactionData reaction,
+    String currentUserId,
+  ) {
+    final updatedOwnReactions = switch (reaction.user.id == currentUserId) {
+      true => ownReactions.upsert(reaction, key: (it) => it.id),
+      false => ownReactions,
+    };
+
+    final updatedLatestReactions = latestReactions.upsert(
+      reaction,
+      key: (reaction) => reaction.id,
+    );
+
+    final reactionGroup = switch (reactionGroups[reaction.type]) {
+      final existingGroup? => existingGroup,
+      _ => ReactionGroupData(
+          count: 1,
+          firstReactionAt: reaction.createdAt,
+          lastReactionAt: reaction.createdAt,
+        ),
+    };
+
+    final updatedReactionGroups = {
+      ...reactionGroups,
+      reaction.type: reactionGroup.increment(reaction.createdAt),
+    };
+
+    final updatedReactionCount = updatedReactionGroups.values.sumOf(
+      (group) => group.count,
+    );
+
+    return copyWith(
+      ownReactions: updatedOwnReactions,
+      latestReactions: updatedLatestReactions,
+      reactionGroups: updatedReactionGroups,
+      reactionCount: updatedReactionCount,
+    );
+  }
+
+  /// Removes a reaction from the comment, updating the latest reactions, reaction groups, reaction
+  /// count, and own reactions if applicable.
+  ///
+  /// @param reaction The reaction to remove.
+  /// @param currentUserId The ID of the current user, used to update own reactions.
+  /// @return A new [CommentData] instance with the updated reaction data.
+  CommentData removeReaction(
+    FeedsReactionData reaction,
+    String currentUserId,
+  ) {
+    final updatedOwnReactions = switch (reaction.user.id == currentUserId) {
+      true => ownReactions.where((it) => it.id != reaction.id).toList(),
+      false => ownReactions,
+    };
+
+    final updatedLatestReactions = latestReactions.where((it) {
+      return it.id != reaction.id;
+    }).toList(growable: false);
+
+    final updatedReactionGroups = {...reactionGroups};
+    final reactionGroup = updatedReactionGroups.remove(reaction.type);
+
+    if (reactionGroup == null) {
+      // If there is no reaction group for this type, just update latest and own reactions.
+      // Note: This is only a hypothetical case, as we should always have a reaction group.
+      return copyWith(
+        latestReactions: updatedLatestReactions,
+        ownReactions: updatedOwnReactions,
+      );
+    }
+
+    final updatedReactionGroup = reactionGroup.decrement(reaction.createdAt);
+    if (updatedReactionGroup.count > 0) {
+      updatedReactionGroups[reaction.type] = updatedReactionGroup;
+    }
+
+    final updatedReactionCount = updatedReactionGroups.values.sumOf(
+      (group) => group.count,
+    );
+
+    return copyWith(
+      ownReactions: updatedOwnReactions,
+      latestReactions: updatedLatestReactions,
+      reactionGroups: updatedReactionGroups,
+      reactionCount: updatedReactionCount,
+    );
+  }
+}

packages/stream_feeds/lib/src/models/feeds_reaction_data.dart

@@ -26,6 +26,8 @@ class FeedsReactionData with _$FeedsReactionData {
   @override
   final String activityId;
 
+  /// The ID of the comment this reaction is associated with.
+  @override
   final String? commentId;
 
   /// The date and time when the reaction was created.

packages/stream_feeds/lib/src/models/poll_data.dart

@@ -175,12 +175,19 @@ extension PollDataMutations on PollData {
     return copyWith(options: updatedOptions);
   }
 
-  PollData castAnswer(PollVoteData answer, String currentUserId) {
-    final updatedLatestAnswers = latestAnswers.let((it) {
+  PollData castAnswer(
+    PollVoteData answer,
+    String currentUserId, {
+    List<PollVoteData>? currentLatestAnswers,
+    List<PollVoteData>? currentOwnVotesAndAnswers,
+  }) {
+    final updatedLatestAnswers =
+        (currentLatestAnswers ?? latestAnswers).let((it) {
       return it.upsert(answer, key: (it) => it.id == answer.id);
     });
 
-    final updatedOwnVotesAndAnswers = ownVotesAndAnswers.let((it) {
+    final updatedOwnVotesAndAnswers =
+        (currentOwnVotesAndAnswers ?? ownVotesAndAnswers).let((it) {
       if (answer.userId != currentUserId) return it;
       return it.upsert(answer, key: (it) => it.id == answer.id);
     });
@@ -190,6 +197,65 @@ extension PollDataMutations on PollData {
       ownVotesAndAnswers: updatedOwnVotesAndAnswers,
     );
   }
+
+  PollData changeVote(
+    PollVoteData vote,
+    String currentUserId, {
+    List<PollVoteData>? currentLatestVotes,
+    List<PollVoteData>? currentOwnVotesAndAnswers,
+  }) {
+    final latestAnswers = currentLatestVotes ?? latestVotes;
+    final ownVotesAndAnswers =
+        (currentOwnVotesAndAnswers ?? this.ownVotesAndAnswers).let((it) {
+      if (vote.userId != currentUserId) return it;
+      return it.upsert(vote, key: (it) => it.id == vote.id);
+    });
+
+    return copyWith(
+      latestAnswers: latestAnswers,
+      ownVotesAndAnswers: ownVotesAndAnswers,
+    );
+  }
+
+  PollData removeAnswer(
+    PollVoteData answer,
+    String currentUserId, {
+    List<PollVoteData>? currentLatestAnswers,
+    List<PollVoteData>? currentOwnVotesAndAnswers,
+  }) {
+    final latestAnswers =
+        (currentLatestAnswers ?? this.latestAnswers).where((it) {
+      return it.id != answer.id;
+    }).toList();
+
+    final ownVotesAndAnswers =
+        (currentOwnVotesAndAnswers ?? this.ownVotesAndAnswers).where((it) {
+      return it.id != answer.id;
+    }).toList();
+
+    return copyWith(
+      latestAnswers: latestAnswers,
+      ownVotesAndAnswers: ownVotesAndAnswers,
+    );
+  }
+
+  PollData removeVote(
+    PollVoteData vote,
+    String currentUserId, {
+    List<PollVoteData>? currentLatestVotes,
+    List<PollVoteData>? currentOwnVotesAndAnswers,
+  }) {
+    final latestAnswers = currentLatestVotes ?? latestVotes;
+    final ownVotesAndAnswers =
+        (currentOwnVotesAndAnswers ?? this.ownVotesAndAnswers).where((it) {
+      return it.id != vote.id;
+    }).toList();
+
+    return copyWith(
+      latestAnswers: latestAnswers,
+      ownVotesAndAnswers: ownVotesAndAnswers,
+    );
+  }
 }
 
 /// Extension function to convert a [PollResponseData] to a [PollData] model.

packages/stream_feeds/lib/src/state/activity.dart

@@ -68,7 +68,11 @@ class Activity with Disposable {
     );
 
     // Attach event handlers for real-time updates
-    final handler = ActivityEventHandler(fid: fid, state: _stateNotifier);
+    final handler = ActivityEventHandler(
+      fid: fid,
+      state: _stateNotifier,
+      capabilitiesRepository: capabilitiesRepository,
+    );
     _eventsSubscription = eventsEmitter.listen(handler.handleEvent);
   }