refactor: Change read receipts API to user-centric [String: Long] and add fetchReadReceipts

Author: jguz-pubnub

PubNubSwiftChatSDK/PubNubSwiftChatSDK.docc/Channel.md

@@ -10,6 +10,8 @@
 - ``streamUpdatesOn(channels:callback:)``
 - ``streamReadReceipts()``
 - ``streamReadReceipts(callback:)``
+- ``fetchReadReceipts(limit:page:filter:sort:)``
+- ``fetchReadReceipts(limit:page:filter:sort:completion:)``
 - ``streamMessageReports()``
 - ``streamMessageReports(callback:)``
 - ``streamPresence()``

PubNubSwiftChatSDK/PubNubSwiftChatSDK.docc/ChannelImpl.md

@@ -10,6 +10,8 @@
 - ``streamUpdatesOn(channels:callback:)``
 - ``streamReadReceipts()``
 - ``streamReadReceipts(callback:)``
+- ``fetchReadReceipts(limit:page:filter:sort:)``
+- ``fetchReadReceipts(limit:page:filter:sort:completion:)``
 - ``streamMessageReports()``
 - ``streamMessageReports(callback:)``
 - ``streamPresence()``

Snippets/MessageSnippets.swift

@@ -1270,8 +1270,8 @@ func streamReadReceiptsAsyncStream() {
   Task {
     if let channel = try await chat.getChannel(channelId: "support") {
       for await readReceipt in channel.streamReadReceipts() {
-        readReceipt.forEach { (messageTimetoken, users) in
-          debugPrint("Message timetoken: \(messageTimetoken) was read by users: \(users)")
+        readReceipt.forEach { (userId, timetoken) in
+          debugPrint("User \(userId) has read up to timetoken: \(timetoken)")
         }
       }
     } else {
@@ -1288,8 +1288,29 @@ func streamReadReceiptsClosure() {
   // Assumes a "ChannelImpl" reference named "channel"
   autoCloseable = channel.streamReadReceipts { receipts in
     print("Read Receipts Received:")
-    receipts.forEach { (messageTimetoken, users) in
-      print("Message Timetoken: \(messageTimetoken) was read by users: \(users)")
+    receipts.forEach { (userId, timetoken) in
+      print("User \(userId) has read up to timetoken: \(timetoken)")
+    }
+  }
+  // snippet.end
+}
+
+// MARK: - Fetch Read Receipts
+
+func fetchReadReceipts() {
+  // snippet.messages.readReceipts.fetch
+  // Assumes a "ChatImpl" reference named "chat"
+  Task {
+    if let channel = try await chat.getChannel(channelId: "support") {
+      let result = try await channel.fetchReadReceipts()
+      // Dictionary mapping user IDs to the timetoken they have read up to
+      let receipts = result.receipts
+      // Use for subsequent call: channel.fetchReadReceipts(page: nextPage)
+      let nextPage = result.page
+
+      receipts.forEach { (userId, timetoken) in
+        print("User \(userId) has read up to timetoken: \(timetoken)")
+      }
     }
   }
   // snippet.end

Sources/Entities/BaseChannel.swift

@@ -479,21 +479,41 @@ final class BaseChannel<C: PubNubChat.Channel_, M: PubNubChat.Message>: Channel
     )
   }
 
-  func streamReadReceipts(callback: @escaping (([Timetoken: [String]]) -> Void)) -> AutoCloseable {
+  func streamReadReceipts(callback: @escaping (([String: Timetoken]) -> Void)) -> AutoCloseable {
     AutoCloseableImpl(
       channel.streamReadReceipts { [weak self] in
         if self != nil {
-          callback(
-            $0.reduce(into: [Timetoken: [String]]()) { res, currentItem in
-              res[Timetoken(currentItem.key.uint64Value)] = currentItem.value
-            }
-          )
+          callback($0.mapValues {
+            Timetoken($0.uint64Value)
+          })
         }
       },
       owner: self
     )
   }
 
+  func fetchReadReceipts(
+    limit: Int?,
+    page: PubNubHashedPage?,
+    filter: String?,
+    sort: [PubNub.MembershipSortField],
+    completion: ((Swift.Result<(receipts: [String: Timetoken], page: PubNubHashedPage?), Error>) -> Void)?
+  ) {
+    getMembers(limit: limit, page: page, filter: filter, sort: sort) { result in
+      switch result {
+      case let .success((memberships, page)):
+        let receipts = memberships.reduce(into: [String: Timetoken]()) { result, membership in
+          if let timetoken = membership.lastReadMessageTimetoken {
+            result[membership.user.id] = timetoken
+          }
+        }
+        completion?(.success((receipts: receipts, page: page)))
+      case let .failure(error):
+        completion?(.failure(error))
+      }
+    }
+  }
+
   func getFiles(
     limit: Int,
     next: String?,

Sources/Entities/Channel+AsyncAwait.swift

@@ -303,7 +303,7 @@ public extension Channel {
   ///   - limit: Number of objects to return in response
   ///   - page: Object used for pagination to define which previous or next result page you want to fetch
   ///   - filter: Expression used to filter the results. Returns only these members whose properties satisfy the given expression
-  ///   - sort: A collection to specify the sort order. Available options are id, name, and updated
+  ///   - sort: A collection to specify the sort order
   /// - Returns: A `Tuple` containing an array of the members of the channel, and the next pagination `PubNubHashedPage` (if one exists)
   func getMembers(
     limit: Int? = nil,
@@ -525,7 +525,7 @@ public extension Channel {
   }
 
   /// Lets you get a read confirmation status for messages you published on a channel.
-  func streamReadReceipts() -> AsyncStream<[Timetoken: [String]]> {
+  func streamReadReceipts() -> AsyncStream<[String: Timetoken]> {
     AsyncStream { continuation in
       let autoCloseable = streamReadReceipts {
         continuation.yield($0)
@@ -536,6 +536,32 @@ public extension Channel {
     }
   }
 
+  /// Fetches the read receipts for members of this channel.
+  ///
+  /// - Parameters:
+  ///   - limit: Number of objects to return in response
+  ///   - page: Object used for pagination to define which previous or next result page you want to fetch
+  ///   - filter: Expression used to filter the results. Returns only these members whose properties satisfy the given expression
+  ///   - sort: A collection to specify the sort order
+  /// - Returns: A `Tuple` containing a dictionary mapping user IDs to the timetoken they have read up to, and the next pagination `PubNubHashedPage` (if one exists)
+  func fetchReadReceipts(
+    limit: Int? = nil,
+    page: PubNubHashedPage? = nil,
+    filter: String? = nil,
+    sort: [PubNub.MembershipSortField] = []
+  ) async throws -> (receipts: [String: Timetoken], page: PubNubHashedPage?) {
+    try await withCheckedThrowingContinuation { continuation in
+      fetchReadReceipts(limit: limit, page: page, filter: filter, sort: sort) {
+        switch $0 {
+        case let .success(result):
+          continuation.resume(returning: result)
+        case let .failure(error):
+          continuation.resume(throwing: error)
+        }
+      }
+    }
+  }
+
   /// Returns all files attached to messages on a given channel.
   ///
   /// - Parameters:

Sources/Entities/Channel.swift

@@ -273,7 +273,7 @@ public protocol Channel: CustomStringConvertible {
   ///   - limit: Number of objects to return in response
   ///   - page: Object used for pagination to define which previous or next result page you want to fetch
   ///   - filter: Expression used to filter the results. Returns only these members whose properties satisfy the given expression
-  ///   - sort: A collection to specify the sort order. Available options are id, name, and updated
+  ///   - sort: A collection to specify the sort order
   ///   - completion: The async `Result` of the method call
   ///     - **Success**: A `Tuple` containing an array of the members of the channel, and the next pagination `PubNubHashedPage` (if one exists)
   ///     - **Failure**: An `Error` describing the failure
@@ -428,12 +428,30 @@ public protocol Channel: CustomStringConvertible {
   /// - Important: Keep a strong reference to the returned ``AutoCloseable`` object as long as you want to receive updates. If ``AutoCloseable`` is deallocated,
   /// the stream will be canceled, and no further items will be produced. You can also stop receiving updates manually by calling ``AutoCloseable/close()``.
   ///
-  /// - Parameter callback: Defines the custom behavior to be executed when receiving a read confirmation status on the joined channel.
+  /// - Parameter callback: Defines the custom behavior to be executed when receiving a read confirmation status on the joined channel. The callback receives a dictionary mapping user IDs to the timetoken they have read up to.
   /// - Returns: AutoCloseable Interface you can call to stop listening for message read receipts and clean up resources by invoking the close() method
   func streamReadReceipts(
-    callback: @escaping (([Timetoken: [String]]) -> Void)
+    callback: @escaping (([String: Timetoken]) -> Void)
   ) -> AutoCloseable
 
+  /// Fetches the read receipts for members of this channel.
+  ///
+  /// - Parameters:
+  ///   - limit: Number of objects to return in response
+  ///   - page: Object used for pagination to define which previous or next result page you want to fetch
+  ///   - filter: Expression used to filter the results. Returns only these members whose properties satisfy the given expression
+  ///   - sort: A collection to specify the sort order
+  ///   - completion: The async `Result` of the method call
+  ///     - **Success**: A `Tuple` containing a dictionary mapping user IDs to the timetoken they have read up to, and the next pagination `PubNubHashedPage` (if one exists)
+  ///     - **Failure**: An `Error` describing the failure
+  func fetchReadReceipts(
+    limit: Int?,
+    page: PubNubHashedPage?,
+    filter: String?,
+    sort: [PubNub.MembershipSortField],
+    completion: ((Swift.Result<(receipts: [String: Timetoken], page: PubNubHashedPage?), Error>) -> Void)?
+  )
+
   /// Returns all files attached to messages on a given channel.
   ///
   /// - Parameters:

Sources/Entities/ChannelImpl.swift

@@ -354,12 +354,28 @@ extension ChannelImpl: Channel {
     )
   }
 
-  public func streamReadReceipts(callback: @escaping (([Timetoken: [String]]) -> Void)) -> AutoCloseable {
+  public func streamReadReceipts(callback: @escaping (([String: Timetoken]) -> Void)) -> AutoCloseable {
     target.streamReadReceipts(
       callback: callback
     )
   }
 
+  public func fetchReadReceipts(
+    limit: Int? = nil,
+    page: PubNubHashedPage? = nil,
+    filter: String? = nil,
+    sort: [PubNub.MembershipSortField] = [],
+    completion: ((Swift.Result<(receipts: [String: Timetoken], page: PubNubHashedPage?), Error>) -> Void)? = nil
+  ) {
+    target.fetchReadReceipts(
+      limit: limit,
+      page: page,
+      filter: filter,
+      sort: sort,
+      completion: completion
+    )
+  }
+
   public func getFiles(
     limit: Int = 100,
     next: String? = nil,

Sources/Entities/ThreadChannelImpl.swift

@@ -391,12 +391,28 @@ extension ThreadChannelImpl: ThreadChannel {
     )
   }
 
-  public func streamReadReceipts(callback: @escaping (([Timetoken: [String]]) -> Void)) -> AutoCloseable {
+  public func streamReadReceipts(callback: @escaping (([String: Timetoken]) -> Void)) -> AutoCloseable {
     target.streamReadReceipts(
       callback: callback
     )
   }
 
+  public func fetchReadReceipts(
+    limit: Int? = nil,
+    page: PubNubHashedPage? = nil,
+    filter: String? = nil,
+    sort: [PubNub.MembershipSortField] = [],
+    completion: ((Swift.Result<(receipts: [String: Timetoken], page: PubNubHashedPage?), Error>) -> Void)? = nil
+  ) {
+    target.fetchReadReceipts(
+      limit: limit,
+      page: page,
+      filter: filter,
+      sort: sort,
+      completion: completion
+    )
+  }
+
   public func getFiles(
     limit: Int = 100,
     next: String? = nil,

Tests/AsyncAwait/ChannelAsyncIntegrationTests.swift

@@ -507,10 +507,8 @@ class ChannelAsyncIntegrationTests: BaseAsyncIntegrationTestCase {
 
     let task = Task {
       for await readReceipt in channel.streamReadReceipts() {
-        XCTAssertEqual(readReceipt[timetoken]?.count, 1)
-        XCTAssertEqual(readReceipt[timetoken]?.first, currentUserId)
-        XCTAssertEqual(readReceipt[secondTimetoken]?.count, 1)
-        XCTAssertEqual(readReceipt[secondTimetoken]?.first, anotherUserId)
+        XCTAssertEqual(readReceipt[currentUserId], timetoken)
+        XCTAssertEqual(readReceipt[anotherUserId], secondTimetoken)
         expectation.fulfill()
       }
     }
@@ -523,6 +521,29 @@ class ChannelAsyncIntegrationTests: BaseAsyncIntegrationTestCase {
     }
   }
 
+  func testChannelAsync_FetchReadReceipts() async throws {
+    let anotherUser = try await chat.createUser(
+      user: UserImpl(chat: chat, id: randomString())
+    )
+    let membership = try await channel.invite(user: chat.currentUser)
+    try await Task.sleep(nanoseconds: 1_000_000_000)
+    let anotherMembership = try await channel.invite(user: anotherUser)
+
+    let timetoken = try XCTUnwrap(membership.lastReadMessageTimetoken)
+    let secondTimetoken = try XCTUnwrap(anotherMembership.lastReadMessageTimetoken)
+    let currentUserId = chat.currentUser.id
+    let anotherUserId = anotherUser.id
+
+    let (receipts, _) = try await channel.fetchReadReceipts()
+
+    XCTAssertEqual(receipts[currentUserId], timetoken)
+    XCTAssertEqual(receipts[anotherUserId], secondTimetoken)
+
+    addTeardownBlock { [unowned self] in
+      _ = try? await chat.deleteUser(id: anotherUserId)
+    }
+  }
+
   func testChannelAsync_GetFiles() async throws {
     let fileUrlSession = URLSession(
       configuration: URLSessionConfiguration.default,

Tests/ChannelIntegrationTests.swift

@@ -768,10 +768,8 @@ class ChannelIntegrationTests: BaseClosureIntegrationTestCase {
     let anotherUserId = anotherUser.id
 
     let closeable = channel.streamReadReceipts {
-      XCTAssertEqual($0[timetoken]?.count, 1)
-      XCTAssertEqual($0[timetoken]?.first, currentUserId)
-      XCTAssertEqual($0[secondTimetoken]?.count, 1)
-      XCTAssertEqual($0[secondTimetoken]?.first, anotherUserId)
+      XCTAssertEqual($0[currentUserId], timetoken)
+      XCTAssertEqual($0[anotherUserId], secondTimetoken)
       expectation.fulfill()
     }