Add StateLayerEventPublisher for centralised change handling

Sources/StreamFeeds/Extensions/Array+Extensions.swift

@@ -13,23 +13,29 @@ extension Array {
     /// Otherwise, the new element will be inserted at the start of the array.
     ///
     /// - Parameter element: The element to insert or replace.
-    mutating func insert(byId element: Element) where Element: Identifiable {
+    /// - Returns: True, if the element was inserted, false if the element exists and was updated.
+    @discardableResult mutating func insert(byId element: Element) -> Bool where Element: Identifiable {
         if let index = firstIndex(where: { $0.id == element.id }) {
             replaceSubrange(index...index, with: CollectionOfOne(element))
+            return false
         } else {
             insert(element, at: startIndex)
+            return true
         }
     }
 
     /// Removes an element from the array based on its ID, optionally searching nested elements.
     ///
     /// - Parameter id: The ID of the element to remove.
     /// - Parameter nesting: Optional key path to search for nested elements.
-    mutating func remove(
+    /// - Returns: True, if the element was removed, false if it did not exist.
+    @discardableResult mutating func remove(
         byId id: Element.ID,
         nesting nestingKeyPath: WritableKeyPath<Element, [Element]?>? = nil
-    ) where Element: Identifiable {
+    ) -> Bool where Element: Identifiable {
+        let count = count
         _unsortedUpdate(ofId: id, nesting: nestingKeyPath, changes: { _ in nil })
+        return count != self.count
     }
 
     /// Removes elements from the non-sorted array based on ID.
@@ -71,7 +77,9 @@ extension Array where Element: Identifiable {
     ///
     /// - Parameter element: The element to insert.
     /// - Parameter sorting: Closure that defines the sorting order.
-    mutating func sortedInsert(_ element: Element, sorting: (Element, Element) -> Bool) {
+    /// - Returns: True, if the element was inserted, false if existed and was updated.
+    @discardableResult mutating func sortedInsert(_ element: Element, sorting: (Element, Element) -> Bool) -> Bool {
+        let count = count
         let insertionIndex = binarySearchInsertionIndex(for: element, sorting: sorting)
         insert(element, at: insertionIndex)
         // Look for duplicates
@@ -89,6 +97,7 @@ extension Array where Element: Identifiable {
             }
             distance += 1
         }
+        return self.count != count
     }
 
     /// Inserts an element into a sorted array using SortField configuration.

Sources/StreamFeeds/FeedsClient.swift

@@ -30,6 +30,7 @@ public final class FeedsClient: Sendable {
 
     let eventsMiddleware = WSEventsMiddleware()
     let eventNotificationCenter: EventNotificationCenter
+    let stateLayerEventPublisher = StateLayerEventPublisher()
 
     let activitiesRepository: ActivitiesRepository
     let bookmarksRepository: BookmarksRepository
@@ -133,6 +134,7 @@ public final class FeedsClient: Sendable {
         moderation = Moderation(apiClient: apiClient)
 
         eventsMiddleware.add(subscriber: self)
+        eventsMiddleware.add(subscriber: stateLayerEventPublisher)
         eventNotificationCenter.add(middlewares: [eventsMiddleware])
     }
 

Sources/StreamFeeds/Models/ActivityData.swift

@@ -51,26 +51,34 @@ public typealias ActivityDataVisibility = ActivityResponse.ActivityResponseVisib
 
 extension ActivityData {
     mutating func addComment(_ comment: CommentData) {
-        comments.insert(byId: comment)
-        commentCount += 1
+        if comments.insert(byId: comment) {
+            commentCount += 1
+        }
     }
 
     mutating func deleteComment(_ comment: CommentData) {
-        commentCount = max(0, commentCount - 1)
-        comments.remove(byId: comment.id)
+        if comments.remove(byId: comment.id) {
+            commentCount = max(0, commentCount - 1)
+        }
     }
 
     mutating func addBookmark(_ bookmark: BookmarkData, currentUserId: String) {
         if bookmark.user.id == currentUserId {
-            ownBookmarks.insert(byId: bookmark)
+            if ownBookmarks.insert(byId: bookmark) {
+                bookmarkCount += 1
+            }
+        } else {
+            bookmarkCount += 1
         }
-        bookmarkCount += 1
     }
 
     mutating func deleteBookmark(_ bookmark: BookmarkData, currentUserId: String) {
-        bookmarkCount = max(0, bookmarkCount - 1)
         if bookmark.user.id == currentUserId {
-            ownBookmarks.remove(byId: bookmark.id)
+            if ownBookmarks.remove(byId: bookmark.id) {
+                bookmarkCount = max(0, bookmarkCount - 1)
+            }
+        } else {
+            bookmarkCount = max(0, bookmarkCount - 1)
         }
     }
 

Sources/StreamFeeds/Models/FeedsReactionData.swift

@@ -28,7 +28,7 @@ extension FeedsReactionData {
         to latestReactions: inout [FeedsReactionData],
         reactionGroups: inout [String: ReactionGroupData]
     ) {
-        latestReactions.insert(byId: reaction)
+        guard latestReactions.insert(byId: reaction) else { return }
         var reactionGroup = reactionGroups[reaction.type] ?? ReactionGroupData(count: 1, firstReactionAt: reaction.createdAt, lastReactionAt: reaction.createdAt)
         reactionGroup.increment(with: reaction.createdAt)
         reactionGroups[reaction.type] = reactionGroup
@@ -39,7 +39,7 @@ extension FeedsReactionData {
         from latestReactions: inout [FeedsReactionData],
         reactionGroups: inout [String: ReactionGroupData]
     ) {
-        latestReactions.remove(byId: reaction.id)
+        guard latestReactions.remove(byId: reaction.id) else { return }
         if var reactionGroup = reactionGroups[reaction.type] {
             reactionGroup.decrement(with: reaction.createdAt)
             if !reactionGroup.isEmpty {

Sources/StreamFeeds/Repositories/ActivitiesRepository.swift

@@ -96,8 +96,13 @@ final class ActivitiesRepository: Sendable {
     // MARK: - Activity Interactions
 
     func pin(_ flag: Bool, activityId: String, in feed: FeedId) async throws -> ActivityData {
-        let response = try await apiClient.pinActivity(feedGroupId: feed.group, feedId: feed.id, activityId: activityId)
-        return response.activity.toModel()
+        if flag {
+            let response = try await apiClient.pinActivity(feedGroupId: feed.group, feedId: feed.id, activityId: activityId)
+            return response.activity.toModel()
+        } else {
+            let response = try await apiClient.unpinActivity(feedGroupId: feed.group, feedId: feed.id, activityId: activityId)
+            return response.activity.toModel()
+        }
     }
 
     func markActivity(feedGroupId: String, feedId: String, request: MarkActivityRequest) async throws {
@@ -116,14 +121,14 @@ final class ActivitiesRepository: Sendable {
 
     // MARK: - Reactions
 
-    func addReaction(activityId: String, request: AddReactionRequest) async throws -> FeedsReactionData {
+    func addReaction(activityId: String, request: AddReactionRequest) async throws -> (reaction: FeedsReactionData, activity: ActivityData) {
         let response = try await apiClient.addReaction(activityId: activityId, addReactionRequest: request)
-        return response.reaction.toModel()
+        return (response.reaction.toModel(), response.activity.toModel())
     }
 
-    func deleteReaction(activityId: String, type: String) async throws -> FeedsReactionData {
+    func deleteReaction(activityId: String, type: String) async throws -> (reaction: FeedsReactionData, activity: ActivityData) {
         let response = try await apiClient.deleteActivityReaction(activityId: activityId, type: type)
-        return response.reaction.toModel()
+        return (response.reaction.toModel(), response.activity.toModel())
     }
 
     // MARK: - Activity Reactions Pagination

Sources/StreamFeeds/StateLayer/Activity.swift

@@ -15,6 +15,7 @@ public final class Activity: Sendable {
     private let activitiesRepository: ActivitiesRepository
     private let commentsRepository: CommentsRepository
     private let pollsRepository: PollsRepository
+    private let eventPublisher: StateLayerEventPublisher
     @MainActor private let stateBuilder: StateBuilder<ActivityState>
 
     /// The unique identifier of this activity.
@@ -36,17 +37,17 @@ public final class Activity: Sendable {
         activityId = id
         activitiesRepository = client.activitiesRepository
         commentsRepository = client.commentsRepository
+        eventPublisher = client.stateLayerEventPublisher
         self.feed = feed
         pollsRepository = client.pollsRepository
         let currentUserId = client.user.id
-        let events = client.eventsMiddleware
-        stateBuilder = StateBuilder { [currentUserId] in
+        stateBuilder = StateBuilder { [currentUserId, eventPublisher] in
             ActivityState(
                 activityId: id,
                 feed: feed,
                 data: data,
                 currentUserId: currentUserId,
-                events: events,
+                eventPublisher: eventPublisher,
                 commentListState: commentList.state
             )
         }
@@ -67,7 +68,7 @@ public final class Activity: Sendable {
         async let activity = activitiesRepository.getActivity(activityId: activityId)
         async let comments = queryComments()
         let (activityData, _) = try await (activity, comments)
-        await state.updateActivity(activityData)
+        await state.setActivity(activityData)
         return activityData
     }
 
@@ -162,7 +163,31 @@ public final class Activity: Sendable {
         return comment
     }
 
-    // MARK: - Comment Reactions
+    // MARK: - Activity and Comment Reactions
+
+    /// Adds a reaction to an activity.
+    ///
+    /// - Parameter request: The request containing the reaction data
+    /// - Returns: The created reaction data
+    /// - Throws: `APIError` if the network request fails or the server returns an error
+    @discardableResult
+    public func addReaction(request: AddReactionRequest) async throws -> FeedsReactionData {
+        let result = try await activitiesRepository.addReaction(activityId: activityId, request: request)
+        await eventPublisher.sendEvent(.activityUpdated(result.activity, feed))
+        return result.reaction
+    }
+
+    /// Removes a reaction from an activity.
+    ///
+    /// - Parameter type: The type of reaction to remove
+    /// - Returns: The removed reaction data
+    /// - Throws: `APIError` if the network request fails or the server returns an error
+    @discardableResult
+    public func deleteReaction(type: String) async throws -> FeedsReactionData {
+        let result = try await activitiesRepository.deleteReaction(activityId: activityId, type: type)
+        await eventPublisher.sendEvent(.activityUpdated(result.activity, feed))
+        return result.reaction
+    }
 
     /// Adds a reaction to a comment.
     ///
@@ -200,7 +225,7 @@ public final class Activity: Sendable {
     /// - Throws: `APIError` if the network request fails or the server returns an error
     public func pin() async throws {
         let activity = try await activitiesRepository.pin(true, activityId: activityId, in: feed)
-        await state.changeHandlers.activityUpdated(activity)
+        await eventPublisher.sendEvent(.activityUpdated(activity, feed))
     }
 
     /// Unpins an activity from the feed.
@@ -209,7 +234,7 @@ public final class Activity: Sendable {
     /// - Throws: `APIError` if the network request fails or the server returns an error
     public func unpin() async throws {
         let activity = try await activitiesRepository.pin(false, activityId: activityId, in: feed)
-        await state.changeHandlers.activityUpdated(activity)
+        await eventPublisher.sendEvent(.activityUpdated(activity, feed))
     }
 
     // MARK: - Polls
@@ -220,9 +245,9 @@ public final class Activity: Sendable {
     /// - Throws: `APIError` if the network request fails or the server returns an error
     @discardableResult
     public func closePoll() async throws -> PollData {
-        let pollData = try await updatePollPartial(request: .init(set: ["is_closed": .bool(true)]))
-        await state.changeHandlers.pollUpdated(pollData)
-        return pollData
+        let poll = try await updatePollPartial(request: .init(set: ["is_closed": .bool(true)]))
+        await eventPublisher.sendEvent(.pollUpdated(poll, feed))
+        return poll
     }
 
     /// Deletes a poll.
@@ -233,7 +258,7 @@ public final class Activity: Sendable {
     public func deletePoll(userId: String? = nil) async throws {
         let pollId = try await pollId()
         try await pollsRepository.deletePoll(pollId: pollId, userId: userId)
-        await state.changeHandlers.pollDeleted(pollId)
+        await eventPublisher.sendEvent(.pollDeleted(pollId, feed))
     }
 
     /// Gets a specific poll by its identifier.
@@ -245,7 +270,7 @@ public final class Activity: Sendable {
     @discardableResult
     public func getPoll(userId: String? = nil) async throws -> PollData {
         let poll = try await pollsRepository.getPoll(pollId: pollId(), userId: userId)
-        await state.changeHandlers.pollUpdated(poll)
+        await eventPublisher.sendEvent(.pollUpdated(poll, feed))
         return poll
     }
 
@@ -259,7 +284,7 @@ public final class Activity: Sendable {
     @discardableResult
     public func updatePollPartial(request: UpdatePollPartialRequest) async throws -> PollData {
         let poll = try await pollsRepository.updatePollPartial(pollId: pollId(), request: request)
-        await state.changeHandlers.pollUpdated(poll)
+        await eventPublisher.sendEvent(.pollUpdated(poll, feed))
         return poll
     }
 
@@ -272,7 +297,7 @@ public final class Activity: Sendable {
     @discardableResult
     public func updatePoll(request: UpdatePollRequest) async throws -> PollData {
         let poll = try await pollsRepository.updatePoll(request: request)
-        await state.changeHandlers.pollUpdated(poll)
+        await eventPublisher.sendEvent(.pollUpdated(poll, feed))
         return poll
     }
 
@@ -287,7 +312,10 @@ public final class Activity: Sendable {
     @discardableResult
     public func createPollOption(request: CreatePollOptionRequest) async throws -> PollOptionData {
         let option = try await pollsRepository.createPollOption(pollId: pollId(), request: request)
-        await state.access { $0.poll?.addOption(option) }
+        if var poll = await state.poll {
+            poll.addOption(option)
+            await eventPublisher.sendEvent(.pollUpdated(poll, feed))
+        }
         return option
     }
 
@@ -302,7 +330,10 @@ public final class Activity: Sendable {
         userId: String? = nil
     ) async throws {
         try await pollsRepository.deletePollOption(pollId: pollId(), optionId: optionId, userId: userId)
-        await state.access { $0.poll?.removeOption(withId: optionId) }
+        if var poll = await state.poll {
+            poll.removeOption(withId: optionId)
+            await eventPublisher.sendEvent(.pollUpdated(poll, feed))
+        }
     }
 
     /// Gets a specific poll option by its identifier.
@@ -317,9 +348,12 @@ public final class Activity: Sendable {
         optionId: String,
         userId: String?
     ) async throws -> PollOptionData {
-        let pollOption = try await pollsRepository.getPollOption(pollId: pollId(), optionId: optionId, userId: userId)
-        await state.access { $0.poll?.updateOption(pollOption) }
-        return pollOption
+        let option = try await pollsRepository.getPollOption(pollId: pollId(), optionId: optionId, userId: userId)
+        if var poll = await state.poll {
+            poll.updateOption(option)
+            await eventPublisher.sendEvent(.pollUpdated(poll, feed))
+        }
+        return option
     }
 
     /// Updates a poll option.
@@ -334,7 +368,10 @@ public final class Activity: Sendable {
             pollId: pollId(),
             request: request
         )
-        await state.access { $0.poll?.updateOption(option) }
+        if var poll = await state.poll {
+            poll.updateOption(option)
+            await eventPublisher.sendEvent(.pollUpdated(poll, feed))
+        }
         return option
     }
 
@@ -350,7 +387,10 @@ public final class Activity: Sendable {
     public func castPollVote(request: CastPollVoteRequest) async throws -> PollVoteData? {
         let vote = try await pollsRepository.castPollVote(activityId: activityId, pollId: pollId(), request: request)
         if let vote {
-            await state.access { $0.poll?.castVote(vote, currentUserId: $0.currentUserId) }
+            if var poll = await state.poll {
+                await poll.castVote(vote, currentUserId: state.currentUserId)
+                await eventPublisher.sendEvent(.pollUpdated(poll, feed))
+            }
         }
         return vote
     }
@@ -374,7 +414,10 @@ public final class Activity: Sendable {
             userId: userId
         )
         if let vote {
-            await state.access { $0.poll?.removeVote(vote, currentUserId: $0.currentUserId) }
+            if var poll = await state.poll {
+                await poll.removeVote(vote, currentUserId: state.currentUserId)
+                await eventPublisher.sendEvent(.pollUpdated(poll, feed))
+            }
         }
         return vote
     }