Merge pull request #89 from ably-labs/ECO-4945

[ECO-4945] Spec complete for Ephemeral Room Reactions

Example/AblyChatExample/ContentView.swift

@@ -1,3 +1,4 @@
+import Ably
 import AblyChat
 import SwiftUI
 
@@ -11,11 +12,24 @@ struct ContentView: View {
         let screenHeight = UIScreen.main.bounds.height
     #endif
 
-    @State private var chatClient = MockChatClient(
+    // Can be replaced with your own room ID
+    private let roomID = "DemoRoomID"
+
+    // Set mode to `.live` if you wish to connect to actual instances of the Chat client in either Prod or Sandbox environments. Setting the mode to `.mock` will use the `MockChatClient`, and therefore simulate all features of the Chat app.
+    private let mode = Environment.mock
+    private enum Environment {
+        case mock
+        case live
+    }
+
+    @State private var mockChatClient = MockChatClient(
         realtime: MockRealtime.create(),
         clientOptions: ClientOptions()
     )
 
+    private let liveRealtime: ARTRealtime
+    @State private var liveChatClient: DefaultChatClient
+
     @State private var title = "Room"
     @State private var messages = [BasicListItem]()
     @State private var reactions: [Reaction] = []
@@ -24,8 +38,19 @@ struct ContentView: View {
     @State private var occupancyInfo = "Connections: 0"
     @State private var statusInfo = ""
 
+    // You only need to set `options.key` and `options.clientId` if your mode is set to `.live`. Otherwise, you can ignore this.
+    init() {
+        let options = ARTClientOptions()
+        options.key = ""
+        options.clientId = ""
+        liveRealtime = ARTRealtime(options: options)
+
+        _liveChatClient = State(initialValue: DefaultChatClient(realtime: liveRealtime, clientOptions: .init()))
+    }
+
     private func room() async throws -> Room {
-        try await chatClient.rooms.get(roomID: "Demo", options: .init())
+        let chosenChatClient: ChatClient = (mode == .mock) ? mockChatClient : liveChatClient
+        return try await chosenChatClient.rooms.get(roomID: roomID, options: .init(reactions: .init()))
     }
 
     private var sendTitle: String {
@@ -99,18 +124,24 @@ struct ContentView: View {
             }
         }
         .tryTask { try await setDefaultTitle() }
+        .tryTask { try await attachRoom() }
         .tryTask { try await showMessages() }
         .tryTask { try await showReactions() }
-        .tryTask { try await showPresence() }
-        .tryTask { try await showTypings() }
-        .tryTask { try await showOccupancy() }
-        .tryTask { try await showRoomStatus() }
+        .tryTask {
+            // NOTE: As we implement more features, move them out of the `if mode == .mock` block and into the main block just above.
+            if mode == .mock {
+                try await showPresence()
+                try await showTypings()
+                try await showOccupancy()
+                try await showRoomStatus()
+            }
+        }
     }
 
     func sendButtonAction() {
         if newMessage.isEmpty {
             Task {
-                try await sendReaction(type: ReactionType.like.rawValue)
+                try await sendReaction(type: ReactionType.like.emoji)
             }
         } else {
             Task {
@@ -123,16 +154,30 @@ struct ContentView: View {
         title = try await "\(room().roomID)"
     }
 
+    func attachRoom() async throws {
+        try await room().attach()
+    }
+
     func showMessages() async throws {
-        for await message in try await room().messages.subscribe(bufferingPolicy: .unbounded) {
+        let messagesSubscription = try await room().messages.subscribe(bufferingPolicy: .unbounded)
+        let previousMessages = try await messagesSubscription.getPreviousMessages(params: .init())
+
+        for message in previousMessages.items {
+            withAnimation {
+                messages.append(BasicListItem(id: message.timeserial, title: message.clientID, text: message.text))
+            }
+        }
+
+        for await message in messagesSubscription {
             withAnimation {
                 messages.insert(BasicListItem(id: message.timeserial, title: message.clientID, text: message.text), at: 0)
             }
         }
     }
 
     func showReactions() async throws {
-        for await reaction in try await room().reactions.subscribe(bufferingPolicy: .unbounded) {
+        let reactionSubscription = try await room().reactions.subscribe(bufferingPolicy: .unbounded)
+        for await reaction in reactionSubscription {
             withAnimation {
                 showReaction(reaction.displayedText)
             }

Example/AblyChatExample/Mocks/Misc.swift

@@ -88,6 +88,6 @@ enum ReactionType: String, CaseIterable {
 
 extension Reaction {
     var displayedText: String {
-        ReactionType(rawValue: type)?.emoji ?? ReactionType.idk.emoji
+        type
     }
 }

Example/AblyChatExample/Mocks/MockClients.swift

@@ -69,7 +69,7 @@ actor MockRoom: Room {
     private var mockSubscriptions: [MockSubscription<RoomStatusChange>] = []
 
     func attach() async throws {
-        fatalError("Not yet implemented")
+        print("Mock client attached to room with roomID: \(roomID)")
     }
 
     func detach() async throws {
@@ -165,7 +165,7 @@ actor MockRoomReactions: RoomReactions {
     private func createSubscription() -> MockSubscription<Reaction> {
         let subscription = MockSubscription<Reaction>(randomElement: {
             Reaction(
-                type: ReactionType.allCases.randomElement()!.rawValue,
+                type: ReactionType.allCases.randomElement()!.emoji,
                 metadata: [:],
                 headers: [:],
                 createdAt: Date(),

Sources/AblyChat/DefaultMessages.swift

@@ -75,7 +75,7 @@ internal final class DefaultMessages: Messages, EmitsDiscontinuities {
                 }
 
                 let metadata = data["metadata"] as? Metadata
-                let headers = try message.extras?.toJSON()["headers"] as? Headers
+                let headers = extras["headers"] as? Headers
 
                 let message = Message(
                     timeserial: timeserial,
@@ -205,27 +205,33 @@ internal final class DefaultMessages: Messages, EmitsDiscontinuities {
 
         // (CHA-M5b) If a subscription is added when the underlying realtime channel is in any other state, then its subscription point becomes the attachSerial at the the point of channel attachment.
         return try await withCheckedThrowingContinuation { continuation in
+            // avoids multiple invocations of the continuation
+            var nillableContinuation: CheckedContinuation<TimeserialString, any Error>? = continuation
+
             channel.on { [weak self] stateChange in
                 guard let self else {
                     return
                 }
+
                 switch stateChange.current {
                 case .attached:
                     // Handle successful attachment
                     if let attachSerial = channel.properties.attachSerial {
-                        continuation.resume(returning: attachSerial)
+                        nillableContinuation?.resume(returning: attachSerial)
                     } else {
-                        continuation.resume(throwing: ARTErrorInfo.create(withCode: 40000, status: 400, message: "Channel is attached, but attachSerial is not defined"))
+                        nillableContinuation?.resume(throwing: ARTErrorInfo.create(withCode: 40000, status: 400, message: "Channel is attached, but attachSerial is not defined"))
                     }
+                    nillableContinuation = nil
                 case .failed, .suspended:
                     // TODO: Revisit as part of https://github.com/ably-labs/ably-chat-swift/issues/32
-                    continuation.resume(
+                    nillableContinuation?.resume(
                         throwing: ARTErrorInfo.create(
                             withCode: ErrorCode.messagesAttachmentFailed.rawValue,
                             status: ErrorCode.messagesAttachmentFailed.statusCode,
                             message: "Channel failed to attach"
                         )
                     )
+                    nillableContinuation = nil
                 default:
                     break
                 }

Sources/AblyChat/DefaultRoomReactions.swift

@@ -0,0 +1,87 @@
+import Ably
+
+// TODO: This class errors with "Task-isolated value of type '() async throws -> ()' passed as a strongly transferred parameter; later accesses could race". Adding @MainActor fixes this, revisit as part of https://github.com/ably-labs/ably-chat-swift/issues/83
+@MainActor
+internal final class DefaultRoomReactions: RoomReactions, EmitsDiscontinuities {
+    private let roomID: String
+    public let featureChannel: FeatureChannel
+    private let logger: InternalLogger
+    private let clientID: String
+
+    internal nonisolated var channel: any RealtimeChannelProtocol {
+        featureChannel.channel
+    }
+
+    internal init(featureChannel: FeatureChannel, clientID: String, roomID: String, logger: InternalLogger) {
+        self.roomID = roomID
+        self.featureChannel = featureChannel
+        self.logger = logger
+        self.clientID = clientID
+    }
+
+    // (CHA-ER3) Ephemeral room reactions are sent to Ably via the Realtime connection via a send method.
+    // (CHA-ER3a) Reactions are sent on the channel using a message in a particular format - see spec for format.
+    internal func send(params: SendReactionParams) async throws {
+        let extras = ["headers": params.headers ?? [:]] as ARTJsonCompatible
+        channel.publish(RoomReactionEvents.reaction.rawValue, data: params.asQueryItems(), extras: extras)
+    }
+
+    // (CHA-ER4) A user may subscribe to reaction events in Realtime.
+    // (CHA-ER4a) A user may provide a listener to subscribe to reaction events. This operation must have no side-effects in relation to room or underlying status. When a realtime message with name roomReaction is received, this message is converted into a reaction object and emitted to subscribers.
+    internal func subscribe(bufferingPolicy: BufferingPolicy) async -> Subscription<Reaction> {
+        let subscription = Subscription<Reaction>(bufferingPolicy: bufferingPolicy)
+
+        // (CHA-ER4c) Realtime events with an unknown name shall be silently discarded.
+        channel.subscribe(RoomReactionEvents.reaction.rawValue) { [clientID, logger] message in
+            Task {
+                do {
+                    guard let data = message.data as? [String: Any],
+                          let reactionType = data["type"] as? String
+                    else {
+                        throw ARTErrorInfo.create(withCode: 50000, status: 500, message: "Received incoming message without data or text")
+                    }
+
+                    guard let messageClientID = message.clientId else {
+                        throw ARTErrorInfo.create(withCode: 50000, status: 500, message: "Received incoming message without clientId")
+                    }
+
+                    guard let timestamp = message.timestamp else {
+                        throw ARTErrorInfo.create(withCode: 50000, status: 500, message: "Received incoming message without timestamp")
+                    }
+
+                    guard let extras = try message.extras?.toJSON() else {
+                        throw ARTErrorInfo.create(withCode: 50000, status: 500, message: "Received incoming message without extras")
+                    }
+
+                    let metadata = data["metadata"] as? Metadata
+                    let headers = extras["headers"] as? Headers
+
+                    // (CHA-ER4d) Realtime events that are malformed (unknown fields should be ignored) shall not be emitted to listeners.
+                    let reaction = Reaction(
+                        type: reactionType,
+                        metadata: metadata ?? .init(),
+                        headers: headers ?? .init(),
+                        createdAt: timestamp,
+                        clientID: messageClientID,
+                        isSelf: messageClientID == clientID
+                    )
+
+                    subscription.emit(reaction)
+                } catch {
+                    logger.log(message: "Error processing incoming reaction message: \(error)", level: .error)
+                }
+            }
+        }
+
+        return subscription
+    }
+
+    // (CHA-ER5) Users may subscribe to discontinuity events to know when there’s been a break in reactions that they need to resolve. Their listener will be called when a discontinuity event is triggered from the room lifecycle.
+    internal func subscribeToDiscontinuities() async -> Subscription<ARTErrorInfo> {
+        await featureChannel.subscribeToDiscontinuities()
+    }
+
+    private enum RoomReactionsError: Error {
+        case noReferenceToSelf
+    }
+}

Sources/AblyChat/Events.swift

@@ -1,3 +1,7 @@
 internal enum MessageEvent: String {
     case created = "message.created"
 }
+
+internal enum RoomReactionEvents: String {
+    case reaction = "roomReaction"
+}

Sources/AblyChat/Reaction.swift

@@ -3,6 +3,7 @@ import Foundation
 public typealias ReactionHeaders = Headers
 public typealias ReactionMetadata = Metadata
 
+// (CHA-ER2) A Reaction corresponds to a single reaction in a chat room. This is analogous to a single user-specified message on an Ably channel (NOTE: not a ProtocolMessage).
 public struct Reaction: Sendable {
     public var type: String
     public var metadata: ReactionMetadata

Sources/AblyChat/Room.swift

@@ -61,6 +61,7 @@ internal actor DefaultRoom<LifecycleManagerFactory: RoomLifecycleManagerFactory>
     private let chatAPI: ChatAPI
 
     public nonisolated let messages: any Messages
+    private let _reactions: (any RoomReactions)?
 
     // Exposed for testing.
     private nonisolated let realtime: RealtimeClient
@@ -90,16 +91,25 @@ internal actor DefaultRoom<LifecycleManagerFactory: RoomLifecycleManagerFactory>
             logger: logger
         )
 
+        // TODO: Address force unwrapping of `channels` within feature initialisation below: https://github.com/ably-labs/ably-chat-swift/issues/105
+
         messages = await DefaultMessages(
             featureChannel: featureChannels[.messages]!,
             chatAPI: chatAPI,
             roomID: roomID,
             clientID: clientId
         )
+
+        _reactions = options.reactions != nil ? await DefaultRoomReactions(
+            featureChannel: featureChannels[.reactions]!,
+            clientID: clientId,
+            roomID: roomID,
+            logger: logger
+        ) : nil
     }
 
     private static func createFeatureChannels(roomID: String, realtime: RealtimeClient) -> [RoomFeature: DefaultFeatureChannel] {
-        .init(uniqueKeysWithValues: [RoomFeature.messages].map { feature in
+        .init(uniqueKeysWithValues: [RoomFeature.messages, RoomFeature.reactions].map { feature in
             let channel = realtime.getChannel(feature.channelNameForRoomID(roomID))
             let contributor = DefaultRoomLifecycleContributor(channel: .init(underlyingChannel: channel), feature: feature)
 
@@ -112,7 +122,11 @@ internal actor DefaultRoom<LifecycleManagerFactory: RoomLifecycleManagerFactory>
     }
 
     public nonisolated var reactions: any RoomReactions {
-        fatalError("Not yet implemented")
+        guard let _reactions else {
+            fatalError("Reactions are not enabled for this room")
+        }
+
+        return _reactions
     }
 
     public nonisolated var typing: any Typing {

Sources/AblyChat/RoomFeature.swift

@@ -17,7 +17,10 @@ internal enum RoomFeature {
         case .messages:
             // (CHA-M1) Chat messages for a Room are sent on a corresponding realtime channel <roomId>::$chat::$chatMessages. For example, if your room id is my-room then the messages channel will be my-room::$chat::$chatMessages.
             "chatMessages"
-        case .typing, .reactions, .presence, .occupancy:
+        case .reactions:
+            // (CHA-ER1) Reactions for a Room are sent on a corresponding realtime channel <roomId>::$chat::$reactions. For example, if your room id is my-room then the reactions channel will be my-room::$chat::$reactions.
+            "reactions"
+        case .typing, .presence, .occupancy:
             // We’ll add these, with reference to the relevant spec points, as we implement these features
             fatalError("Don’t know channel name suffix for room feature \(self)")
         }

Sources/AblyChat/RoomReactions.swift

@@ -17,3 +17,13 @@ public struct SendReactionParams: Sendable {
         self.headers = headers
     }
 }
+
+internal extension SendReactionParams {
+    // Same as `ARTDataQuery.asQueryItems` from ably-cocoa.
+    func asQueryItems() -> [String: String] {
+        var dict: [String: String] = [:]
+        dict["type"] = "\(type)"
+        dict["metadata"] = "\(metadata ?? [:])"
+        return dict
+    }
+}

Sources/AblyChat/Subscription.swift

@@ -71,6 +71,16 @@ public struct Subscription<Element: Sendable>: Sendable, AsyncSequence {
         }
     }
 
+    // TODO: https://github.com/ably-labs/ably-chat-swift/issues/36 Revisit how we want to unsubscribe to fulfil CHA-M4b & CHA-ER4b. I think exposing this publicly for all Subscription types is suitable.
+    public func finish() {
+        switch mode {
+        case let .default(_, continuation):
+            continuation.finish()
+        case .mockAsyncSequence:
+            fatalError("`finish` cannot be called on a Subscription that was created using init(mockAsyncSequence:)")
+        }
+    }
+
     public struct AsyncIterator: AsyncIteratorProtocol {
         fileprivate enum Mode {
             case `default`(iterator: AsyncStream<Element>.AsyncIterator)