Add activity batch update event for updating state with activity batch operations

CHANGELOG.md

@@ -5,6 +5,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 # Upcoming
 
 ### ✅ Added
+- Update local state when using activity batch operations in `FeedsClient` [#52](https://github.com/GetStream/stream-feeds-swift/pull/52)
 - Keep `FeedData.ownCapabilities` up to date in every model when handling web-socket events [#51](https://github.com/GetStream/stream-feeds-swift/pull/51)
 
 # [0.4.0](https://github.com/GetStream/stream-feeds-swift/releases/tag/0.4.0)

Sources/StreamFeeds/Extensions/Array+Extensions.swift

@@ -42,9 +42,15 @@ extension Array {
     ///
     /// - Parameter ids: Ids of elements to remove.
     mutating func remove(byIds ids: [Element.ID]) where Element: Identifiable {
+        remove(byIds: Set(ids))
+    }
+
+    /// Removes elements from the non-sorted array based on ID.
+    ///
+    /// - Parameter ids: Ids of elements to remove.
+    mutating func remove(byIds ids: Set<Element.ID>) where Element: Identifiable {
         guard !ids.isEmpty else { return }
-        let lookup = Set(ids)
-        removeAll(where: { lookup.contains($0.id) })
+        removeAll(where: { ids.contains($0.id) })
     }
 
     /// Replaces an element from the non-sorted array based on its ID.

Sources/StreamFeeds/FeedsClient.swift

@@ -385,14 +385,20 @@ public final class FeedsClient: Sendable {
         ActivityReactionList(query: query, client: self)
     }
 
-    /// Adds a new activity to the specified feeds.
+    // MARK: - Activity Batch Operations
+
+    /// Adds a new activity to one or multiple feeds.
     ///
-    /// - Parameter request: The request containing the activity data to add
+    /// - Parameter request: The request containing the activity data to add and the list of feeds
     /// - Returns: A response containing the created activity
     /// - Throws: `APIError` if the network request fails or the server returns an error
     @discardableResult
-    public func addActivity(request: AddActivityRequest) async throws -> AddActivityResponse {
-        try await apiClient.addActivity(addActivityRequest: request)
+    public func addActivity(request: AddActivityRequest) async throws -> ActivityData {
+        let activityData = try await activitiesRepository.addActivity(request: request)
+        for feed in request.feeds.map(FeedId.init) {
+            await stateLayerEventPublisher.sendEvent(.activityAdded(activityData, feed))
+        }
+        return activityData
     }
 
     /// Upserts (inserts or updates) multiple activities.
@@ -402,7 +408,16 @@ public final class FeedsClient: Sendable {
     /// - Throws: `APIError` if the network request fails or the server returns an error
     @discardableResult
     public func upsertActivities(_ activities: [ActivityRequest]) async throws -> [ActivityData] {
-        try await activitiesRepository.upsertActivities(activities)
+        let activities = try await activitiesRepository.upsertActivities(activities)
+        let updates = activities.reduce(into: ModelUpdates<ActivityData>()) { partialResult, activityData in
+            if activityData.updatedAt.timeIntervalSince(activityData.createdAt) > 0.1 || activityData.editedAt != nil {
+                partialResult.updated.append(activityData)
+            } else {
+                partialResult.added.append(activityData)
+            }
+        }
+        await stateLayerEventPublisher.sendEvent(.activityBatchUpdate(updates))
+        return activities
     }
 
     /// Deletes multiple activities from the specified feeds.
@@ -411,8 +426,10 @@ public final class FeedsClient: Sendable {
     /// - Returns: A response confirming the deletion of activities
     /// - Throws: `APIError` if the network request fails or the server returns an error
     @discardableResult
-    public func deleteActivities(request: DeleteActivitiesRequest) async throws -> DeleteActivitiesResponse {
-        try await apiClient.deleteActivities(deleteActivitiesRequest: request)
+    public func deleteActivities(request: DeleteActivitiesRequest) async throws -> Set<String> {
+        let response = try await apiClient.deleteActivities(deleteActivitiesRequest: request)
+        await stateLayerEventPublisher.sendEvent(.activityBatchUpdate(.init(added: [], removedIds: response.deletedIds, updated: [])))
+        return Set(response.deletedIds)
     }
 
     // MARK: - Bookmark Lists

Sources/StreamFeeds/Models/ModelUpdates.swift

@@ -5,7 +5,13 @@
 import Foundation
 
 public struct ModelUpdates<Model>: Sendable where Model: Sendable {
-    public let added: [Model]
-    public let removedIds: [String]
-    public let updated: [Model]
+    public internal(set) var added: [Model]
+    public internal(set) var removedIds: Set<String>
+    public internal(set) var updated: [Model]
+}
+
+extension ModelUpdates {
+    init(added: [Model] = [], removedIds: [String] = [], updated: [Model] = []) {
+        self.init(added: added, removedIds: Set(removedIds), updated: updated)
+    }
 }

Sources/StreamFeeds/StateLayer/ActivityState.swift

@@ -54,6 +54,12 @@ extension ActivityState {
             case .activityUpdated(let activityData, let eventFeedId):
                 guard activityData.id == activityId, eventFeedId == feed else { return }
                 await self?.setActivity(activityData)
+            case .activityBatchUpdate(let updates):
+                if let updated = updates.updated.first(where: { $0.id == activityId }) {
+                    await self?.setActivity(updated)
+                } else if updates.removedIds.contains(activityId) {
+                    await self?.setActivity(nil)
+                }
             case .activityReactionAdded(let reactionData, let activityData, let eventFeedId):
                 guard activityData.id == activityId, eventFeedId == feed else { return }
                 await self?.access { state in

Sources/StreamFeeds/StateLayer/EventHandling/Middlewares/OwnCapabilitiesStateLayerEventMiddleware.swift

@@ -95,7 +95,7 @@ final class OwnCapabilitiesStateLayerEventMiddleware: StateLayerEventMiddleware
 
 private extension StateLayerEvent {
     var ownCapabilities: [FeedId: Set<FeedOwnCapability>] {
-        guard let feedDatas else { return [:] }
+        guard let feedDatas, !feedDatas.isEmpty else { return [:] }
         return feedDatas.reduce(into: [FeedId: Set<FeedOwnCapability>](), { all, feedData in
             guard let capabilities = feedData.ownCapabilities, !capabilities.isEmpty else { return }
             all[feedData.feed] = capabilities
@@ -108,6 +108,8 @@ private extension StateLayerEvent {
             return activityData.currentFeed.map { [$0] }
         case .activityUpdated(let activityData, _):
             return activityData.currentFeed.map { [$0] }
+        case .activityBatchUpdate(let updates):
+            return (updates.added + updates.updated).compactMap(\.currentFeed)
         case .activityReactionAdded(_, let activityData, _):
             return activityData.currentFeed.map { [$0] }
         case .activityReactionDeleted(_, let activityData, _):

Sources/StreamFeeds/StateLayer/EventHandling/StateLayerEvent.swift

@@ -10,6 +10,7 @@ enum StateLayerEvent: Sendable {
     case activityAdded(ActivityData, FeedId)
     case activityDeleted(String, FeedId)
     case activityUpdated(ActivityData, FeedId)
+    case activityBatchUpdate(ModelUpdates<ActivityData>)
 
     case activityReactionAdded(FeedsReactionData, ActivityData, FeedId)
     case activityReactionDeleted(FeedsReactionData, ActivityData, FeedId)

Sources/StreamFeeds/StateLayer/FeedState.swift

@@ -91,13 +91,15 @@ import StreamCore
 
 extension FeedState {
     private func subscribe(to publisher: StateLayerEventPublisher) {
+        let matchesActivityQuery: @Sendable (ActivityData) -> Bool = { [feedQuery] activity in
+            guard let filter = feedQuery.activityFilter else { return true }
+            return filter.matches(activity)
+        }
         eventSubscription = publisher.subscribe { [weak self, currentUserId, feed, feedQuery] event in
             switch event {
             case .activityAdded(let activityData, let eventFeedId):
                 guard feed == eventFeedId else { return }
-                if let filter = feedQuery.activityFilter, !filter.matches(activityData) {
-                    return
-                }
+                guard matchesActivityQuery(activityData) else { return }
                 await self?.access { $0.activities.sortedInsert(activityData, sorting: $0.activitiesSorting) }
             case .activityDeleted(let activityId, let eventFeedId):
                 guard feed == eventFeedId else { return }
@@ -108,6 +110,24 @@ extension FeedState {
             case .activityUpdated(let activityData, let eventFeedId):
                 guard feed == eventFeedId else { return }
                 await self?.updateActivity(activityData)
+            case .activityBatchUpdate(let updates):
+                let added = updates.added.filter { $0.feeds.contains(feed.rawValue) }.filter(matchesActivityQuery)
+                let updated = updates.updated.filter { $0.feeds.contains(feed.rawValue) }.filter(matchesActivityQuery)
+                let removedIds = updates.removedIds
+                guard !added.isEmpty || !updated.isEmpty || !removedIds.isEmpty else { return }
+                await self?.access { state in
+                    let sorting = state.activitiesSorting
+                    if !added.isEmpty {
+                        state.activities = state.activities.sortedMerge(added.sorted(by: sorting.areInIncreasingOrder()), sorting: sorting)
+                    }
+                    if !updated.isEmpty {
+                        state.activities = state.activities.sortedMerge(updated.sorted(by: sorting.areInIncreasingOrder()), sorting: sorting)
+                    }
+                    if !removedIds.isEmpty {
+                        state.activities.removeAll(where: { removedIds.contains($0.id) })
+                        state.pinnedActivities.removeAll(where: { removedIds.contains($0.activity.id) })
+                    }
+                }
             case .activityReactionAdded(let reactionData, let activityData, let eventFeedId):
                 guard feed == eventFeedId else { return }
                 await self?.access { state in

Sources/StreamFeeds/StateLayer/PaginatedLists/ActivityCommentListState.swift

@@ -133,6 +133,11 @@ extension ActivityCommentListState {
                 await self?.access { state in
                     state.comments.removeAll()
                 }
+            case .activityBatchUpdate(let updates):
+                guard updates.removedIds.contains(query.objectId) else { return }
+                await self?.access { state in
+                    state.comments.removeAll()
+                }
             case .commentAdded(let commentData, _, _):
                 guard query.objectId == commentData.objectId, query.objectType == commentData.objectType else { return }
                 await self?.access { state in

Sources/StreamFeeds/StateLayer/PaginatedLists/ActivityListState.swift

@@ -100,6 +100,23 @@ extension ActivityListState {
                 await self?.access { state in
                     state.activities.removeAll { $0.id == activityId }
                 }
+            case .activityBatchUpdate(let updates):
+                let added = updates.added.filter(matchesQuery)
+                let updated = updates.updated.filter(matchesQuery)
+                let removedIds = updates.removedIds
+                guard !added.isEmpty || !updated.isEmpty || !removedIds.isEmpty else { return }
+                await self?.access { state in
+                    let sorting = state.activitiesSorting
+                    if !added.isEmpty {
+                        state.activities = state.activities.sortedMerge(added.sorted(by: sorting.areInIncreasingOrder()), sorting: sorting)
+                    }
+                    if !updated.isEmpty {
+                        state.activities = state.activities.sortedMerge(updated.sorted(by: sorting.areInIncreasingOrder()), sorting: sorting)
+                    }
+                    if !removedIds.isEmpty {
+                        state.activities.removeAll(where: { removedIds.contains($0.id) })
+                    }
+                }
             case .activityReactionAdded(let reactionData, let activityData, _):
                 guard matchesQuery(activityData) else { return }
                 await self?.access { state in

Tests/StreamFeedsTests/FeedsClient_Test.swift

@@ -168,7 +168,7 @@ struct FeedsClient_Test {
         )
 
         let response = try await client.addActivity(request: activityRequest)
-        #expect(response.activity.id == mockResponse.activity.id)
+        #expect(response.id == mockResponse.activity.id)
     }
 
     @Test func upsertActivities() async throws {
@@ -201,7 +201,7 @@ struct FeedsClient_Test {
         )
 
         let response = try await client.deleteActivities(request: deleteRequest)
-        #expect(response.deletedIds == mockResponse.deletedIds)
+        #expect(response == Set(mockResponse.deletedIds))
     }
 
     // MARK: - Properties Tests

Tests/StreamFeedsTests/StateLayer/ActivityCommentList_Tests.swift

@@ -370,6 +370,40 @@ struct ActivityCommentList_Tests {
         #expect(commentsAfterDeletion.isEmpty) // All comments should be removed
     }
 
+    @Test func activityBatchUpdateEventUpdatesState() async throws {
+        let client = defaultClient(
+            comments: [
+                .dummy(id: "comment-1", objectId: Self.activityId, objectType: "activity"),
+                .dummy(id: "comment-2", objectId: Self.activityId, objectType: "activity")
+            ],
+            additionalPayloads: [
+                DeleteActivitiesResponse.dummy(
+                    deletedIds: ["different-activity"]
+                ),
+                DeleteActivitiesResponse.dummy(
+                    deletedIds: [Self.activityId]
+                )
+            ]
+        )
+        let commentList = client.activityCommentList(
+            for: ActivityCommentsQuery(objectId: Self.activityId, objectType: "activity")
+        )
+        try await commentList.get()
+        await #expect(commentList.state.comments.map(\.id) == ["comment-1", "comment-2"])
+
+        // Send batch update with unrelated activity removed - should not affect comments
+        _ = try await client.deleteActivities(
+            request: DeleteActivitiesRequest(ids: ["different-activity"])
+        )
+        await #expect(commentList.state.comments.map(\.id) == ["comment-1", "comment-2"])
+
+        // Send batch update with matching activity removed - should clear all comments
+        _ = try await client.deleteActivities(
+            request: DeleteActivitiesRequest(ids: [Self.activityId])
+        )
+        await #expect(commentList.state.comments.isEmpty)
+    }
+
     // MARK: -
 
     private func defaultClient(