diff --git a/NextcloudTalk/Chat/NCChatController.swift b/NextcloudTalk/Chat/NCChatController.swift index c716aec29..e974c8840 100644 --- a/NextcloudTalk/Chat/NCChatController.swift +++ b/NextcloudTalk/Chat/NCChatController.swift @@ -34,8 +34,17 @@ public class NCChatController: NSObject { private var getHistoryTask: URLSessionDataTask? private var pullMessagesTask: URLSessionDataTask? - private enum ChatRelayState { - case inactive, active, catchingUp + enum ChatRelayState { + // We are polling the chat API, the relay does not process messages yet + case inactive + // The chat is up to date, but our signaling session did not join the room yet, so nothing + // would be relayed to us: keep polling until the join is acked + case waitingForJoin + // The relay took over, polling stopped + case active + // The relay is active, but a message could not be rendered from its payload, so we are + // fetching the missing messages over the chat API + case catchingUp } private var chatRelayState: ChatRelayState = .inactive @@ -58,7 +67,9 @@ public class NCChatController: NSObject { super.init() - setupChatRelay() + let signalingController = NCSettingsController.sharedInstance().externalSignalingController(forAccountId: account.accountId) + setupChatRelay(with: signalingController) + AllocationTracker.shared.addAllocation("NCChatController") } @@ -71,7 +82,9 @@ public class NCChatController: NSObject { super.init() - setupChatRelay() + let signalingController = NCSettingsController.sharedInstance().externalSignalingController(forAccountId: account.accountId) + setupChatRelay(with: signalingController) + AllocationTracker.shared.addAllocation("NCChatController") } @@ -425,14 +438,23 @@ public class NCChatController: NSObject { // MARK: - External Signaling / Chat Relay - private func setupChatRelay() { - guard let signalingController = NCSettingsController.sharedInstance().externalSignalingController(forAccountId: account.accountId), - signalingController.hasChatRelay else { return } + // The signaling server only sends us the events of the room our session joined, so this is the + // condition for letting the relay take over from polling the chat API. + private var canChatRelayTakeOverPolling: Bool { + guard let externalSignalingController, externalSignalingController.hasChatRelay else { return false } + + return externalSignalingController.joinedRoomToken == room.token + } + + private func setupChatRelay(with signalingController: NCExternalSignalingController?) { + guard let signalingController, signalingController.hasChatRelay else { return } + externalSignalingController = signalingController chatRelayMessagesQueue = DispatchQueue(label: "chat.relay.message.queue") NotificationCenter.default.addObserver(self, selector: #selector(didReceiveChatMessageFromExternalSignaling(_:)), name: .extSignalingDidReceiveChatMessage, object: signalingController) NotificationCenter.default.addObserver(self, selector: #selector(didRequestChatRefreshFromExternalSignaling(_:)), name: .extSignalingDidRequestChatRefresh, object: signalingController) NotificationCenter.default.addObserver(self, selector: #selector(didReconnectExternalSignaling(_:)), name: .extSignalingDidReconnect, object: signalingController) + NotificationCenter.default.addObserver(self, selector: #selector(didJoinRoomOnExternalSignaling(_:)), name: .extSignalingDidJoinRoom, object: signalingController) } @objc private func didReceiveChatMessageFromExternalSignaling(_ notification: Notification) { @@ -463,6 +485,63 @@ public class NCChatController: NSObject { } } + // The signaling server only sends us the events of the room our session joined, so the relay can + // take over from polling as soon as the join is acked. Poll once more without a timeout instead of + // waiting for the running long poll to time out, that poll then arms the relay on its 304. + @objc private func didJoinRoomOnExternalSignaling(_ notification: Notification) { + guard let roomToken = notification.userInfo?["roomToken"] as? String, roomToken == room.token else { return } + guard !stopChatMessagesPoll else { return } + + chatRelayMessagesQueue?.async { + guard self.chatRelayState == .waitingForJoin else { return } + + // We are not waiting for the join anymore, we are waiting for the poll below to confirm + // that the chat is up to date. Also keeps a second join from polling twice. + self.chatRelayState = .inactive + + DispatchQueue.main.async { + let lastChatBlock = self.chatBlocksForRoomOrThread().last + self.startReceivingChatMessages(fromMessagesId: lastChatBlock?.newestMessageId ?? 0, withTimeout: false) + } + } + } + + // Called when the chat is up to date and the server supports the chat relay. The relay only + // delivers messages of the room our signaling session joined, so handing polling over to it before + // the join is acked would stop polling while nothing is relayed to us yet, and every message posted + // in that window would be lost for good: the next relayed message advances the chat block past the + // gap, so it is never requested again. Keep polling until we are in the room instead. + private func handOverPollingToChatRelay(fromMessagesId messageId: Int) { + if canChatRelayTakeOverPolling { + print("Chat is up to date, now processing new messages from the chat relay") + startProcessingChatRelayMessages() + return + } + + print("Chat is up to date, but we did not join the room on the signaling server yet, keep polling") + + chatRelayMessagesQueue?.async { + self.chatRelayState = .waitingForJoin + + // The join might have been acked while we were switching queues, in that case poll once + // without a timeout so the relay is armed right away instead of after the long poll. + let hasJoinedMeanwhile = self.canChatRelayTakeOverPolling + + DispatchQueue.main.async { + self.startReceivingChatMessages(fromMessagesId: messageId, withTimeout: !hasJoinedMeanwhile) + } + } + } + + // Puts the relay back into its initial state after a poll ended without arming it. Without this, a + // catch-up that doesn't repoll (brute-force protection, blocked chat, thread not found) would leave + // the relay in `.catchingUp` forever, buffering messages that are never flushed. + private func resetChatRelayState() { + chatRelayMessagesQueue?.async { + self.chatRelayState = .inactive + } + } + private func startProcessingChatRelayMessages() { chatRelayMessagesQueue?.async { self.chatRelayState = .active @@ -1147,18 +1226,21 @@ public class NCChatController: NSObject { if let error { if self.isChatBeingBlocked(statusCode) { + self.resetChatRelayState() self.notifyChatIsBlocked() return } if statusCode == 404 { NCLog.log("Thread not found error: \(error.description)") + self.resetChatRelayState() NotificationCenter.default.post(name: .NCChatControllerDidReceiveThreadNotFound, object: self, userInfo: nil) return } if statusCode == 429 { NCLog.log("Brute-force protected, received 429 while receiving messages. No further polling.") + self.resetChatRelayState() return } @@ -1180,6 +1262,7 @@ public class NCChatController: NSObject { // When we receive a "history_cleared" message, we don't continue here, as otherwise // we would request new messages, but instead, we need to request the initial history again if message?.systemMessage == "history_cleared" { + self.resetChatRelayState() return } } @@ -1194,9 +1277,8 @@ public class NCChatController: NSObject { let chatIsUpToDate = statusCode == 304 let lastChatBlock = self.chatBlocksForRoomOrThread().last - if chatIsUpToDate, let extSignaling = self.externalSignalingController, extSignaling.hasChatRelay { - print("Chat is up to date, now processing new messages from the chat relay") - self.startProcessingChatRelayMessages() + if chatIsUpToDate, self.externalSignalingController?.hasChatRelay == true { + self.handOverPollingToChatRelay(fromMessagesId: lastChatBlock?.newestMessageId ?? 0) return } @@ -1436,4 +1518,21 @@ extension NCChatController { // triggerChatRelayCatchUpForTesting() actually schedules the restart on the main queue, mirroring // a catch-up that fires while the user is still in the room (just before they leave). func markChatRelayActiveForTesting() { chatRelayState = .active } + + // Puts the relay state machine into .waitingForJoin — the state the controller is in while it + // keeps polling because our signaling session did not join the room yet. + func markChatRelayWaitingForJoinForTesting() { chatRelayState = .waitingForJoin } + + var chatRelayStateForTesting: ChatRelayState { chatRelayState } + + // The condition the chat relay is gated on (see canChatRelayTakeOverPolling). + var canChatRelayTakeOverPollingForTesting: Bool { canChatRelayTakeOverPolling } + + // Waits until everything queued on the relay queue ran, so tests don't have to guess timings. + func waitForChatRelayQueueForTesting() { chatRelayMessagesQueue?.sync {} } + + // There is no signaling controller configured for the fake account, so tests pass in their own. + func setupChatRelayForTesting(with signalingController: NCExternalSignalingController) { + setupChatRelay(with: signalingController) + } } diff --git a/NextcloudTalk/WebRTC/NCExternalSignalingController.swift b/NextcloudTalk/WebRTC/NCExternalSignalingController.swift index e8a31d251..92ad357bd 100644 --- a/NextcloudTalk/WebRTC/NCExternalSignalingController.swift +++ b/NextcloudTalk/WebRTC/NCExternalSignalingController.swift @@ -27,6 +27,7 @@ extension Notification.Name { static let extSignalingDidReceiveChatMessage = Notification.Name(rawValue: "NCExternalSignalingControllerDidReceiveChatMessageNotification") static let extSignalingDidRequestChatRefresh = Notification.Name(rawValue: "NCExternalSignalingControllerDidRequestChatRefreshNotification") static let extSignalingDidReconnect = Notification.Name(rawValue: "NCExternalSignalingControllerDidReconnectNotification") + static let extSignalingDidJoinRoom = Notification.Name(rawValue: "NCExternalSignalingControllerDidJoinRoomNotification") } public typealias SendMessageCompletionBlock = (_ task: URLSessionWebSocketTask?, _ status: NCExternalSignalingSendMessageStatus) -> Void @@ -43,6 +44,10 @@ public enum NCExternalSignalingSendMessageStatus { public var currentRoom: String? + // The room the signaling server acked us into. Unlike `currentRoom` this is cleared on every + // reconnect, so it answers "will the server relay this room's chat to our session right now?" + public private(set) var joinedRoomToken: String? + public private(set) var account: TalkAccount public private(set) var disconnected: Bool = true public private(set) var hasMCU: Bool = false @@ -210,6 +215,7 @@ public enum NCExternalSignalingSendMessageStatus { self.webSocket?.cancel() self.webSocket = nil self.helloResponseReceived = false + self.joinedRoomToken = nil self.helloMessage?.ignoreCompletionBlock() self.helloMessage = nil self.disconnected = true @@ -340,6 +346,15 @@ public enum NCExternalSignalingSendMessageStatus { let sessionChanged = self.sessionId != newSessionId self.sessionId = newSessionId + if sessionChanged { + // The new session did not join any room yet, the re-join below takes care of that + self.joinedRoomToken = nil + } else { + // The session was resumed, so the server kept us in the room we joined before and replays + // the messages we missed while being disconnected + self.setJoinedRoomToken(self.currentRoom) + } + guard let serverDict = helloDict["server"] as? [AnyHashable: Any], let serverFeatures = serverDict["features"] as? [String], let serverVersion = serverDict["version"] as? String @@ -402,8 +417,10 @@ public enum NCExternalSignalingSendMessageStatus { let roomId = roomDict["roomid"] as? String else { return } - // If we are aware that we were in this room before, we should treat this as a success + // If we are aware that we were in this room before, we should treat this as a success. + // No room message follows in this case, so we have to set the joined room ourselves. if currentRoom == roomId { + self.setJoinedRoomToken(roomId) self.executeCompletionBlock(forMessageId: messageId, withStatus: .success) return } @@ -462,6 +479,7 @@ public enum NCExternalSignalingSendMessageStatus { func leaveRoom(withRoomId roomId: String) { if self.currentRoom == roomId { self.currentRoom = nil + self.joinedRoomToken = nil self.joinRoom(withRoomId: "", withSessionId: "", withFederation: nil, withCompletionBlock: nil) } else { print("External signaling: Not leaving because it's not the room we joined") @@ -556,11 +574,25 @@ public enum NCExternalSignalingSendMessageStatus { self.currentRoom = newRoomId.isEmpty ? nil : newRoomId } + // Set outside the check above: after a reconnect we re-join the same room, so `currentRoom` is + // unchanged, but this is exactly the moment we are part of the room on the signaling server again. + self.setJoinedRoomToken(newRoomId.isEmpty ? nil : newRoomId) + if let messageId = messageDict["id"] as? String { self.executeCompletionBlock(forMessageId: messageId, withStatus: .success) } } + // Notifies about the room we are joined to on the signaling server, so components relying on the + // server pushing room events to us (e.g. the chat relay) know when they can start doing so. + private func setJoinedRoomToken(_ roomToken: String?) { + self.joinedRoomToken = roomToken + + guard let roomToken else { return } + + NotificationCenter.default.post(name: .extSignalingDidJoinRoom, object: self, userInfo: ["roomToken": roomToken]) + } + func eventMessageReceived(eventDict: [AnyHashable: Any]) { let eventTarget = eventDict["target"] as? String diff --git a/NextcloudTalkTests/Unit/Chat/UnitChatViewControllerTest.swift b/NextcloudTalkTests/Unit/Chat/UnitChatViewControllerTest.swift index 569d41bac..f841fcbee 100644 --- a/NextcloudTalkTests/Unit/Chat/UnitChatViewControllerTest.swift +++ b/NextcloudTalkTests/Unit/Chat/UnitChatViewControllerTest.swift @@ -196,6 +196,111 @@ final class UnitChatViewControllerTest: TestBaseRealm { "A chat-relay catch-up scheduled before stop() must not resume polling once stop() has run") } + // MARK: - Chat relay + + private func inertSignalingController(withChatRelay hasChatRelay: Bool = true) -> NCExternalSignalingController { + let account = NCDatabaseManager.sharedInstance().activeAccount() + let signalingController = NCExternalSignalingController(account: account, serverUrl: TestConstants.server, ticket: "fakeTicket") + + // Make the controller inert: without a websocket the delegate callbacks of the failing + // connection attempt are ignored, so nothing reconnects underneath the assertions. + signalingController.disconnect() + + if hasChatRelay { + let helloMessage: [AnyHashable: Any] = [ + "type": "hello", + "id": "1", + "hello": [ + "sessionid": "session-1", + "server": ["version": "2.0.0", "features": ["chat-relay"]] + ] + ] + + signalingController.helloResponseReceived(messageDict: helloMessage) + } + + return signalingController + } + + private func drainMainQueue() { + let exp = expectation(description: "\(#function)\(#line)") + DispatchQueue.main.async { exp.fulfill() } + waitForExpectations(timeout: TestConstants.timeoutShort, handler: nil) + } + + func testChatRelayOnlyTakesOverPollingOnceTheRoomIsJoinedOnTheSignalingServer() throws { + let room = addRoom(withToken: "relayGateRoom") + let chatController = NCChatController(for: room)! + let signalingController = inertSignalingController() + + chatController.setupChatRelayForTesting(with: signalingController) + + // The server supports the chat relay, but our signaling session is not in the room yet, so + // the relay would not deliver anything to us. Handing over to it here loses every message + // posted until the join is acked, since polling stops for good. + XCTAssertFalse(chatController.canChatRelayTakeOverPollingForTesting) + + // Being in *another* room is not enough either + signalingController.roomMessageReceived(messageDict: ["type": "room", "room": ["roomid": "someOtherRoom"]]) + XCTAssertFalse(chatController.canChatRelayTakeOverPollingForTesting) + + signalingController.roomMessageReceived(messageDict: ["type": "room", "room": ["roomid": room.token]]) + XCTAssertTrue(chatController.canChatRelayTakeOverPollingForTesting) + + // We left the room again (or the connection dropped) + signalingController.resetWebSocket() + XCTAssertFalse(chatController.canChatRelayTakeOverPollingForTesting) + + chatController.stop() + drainMainQueue() + } + + func testJoiningTheRoomRepollsWhileWaitingForTheJoin() throws { + let room = addRoom(withToken: "relayJoinRoom") + let chatController = NCChatController(for: room)! + let signalingController = inertSignalingController() + + chatController.setupChatRelayForTesting(with: signalingController) + + // The chat is up to date, but we kept polling because we were not in the room yet + chatController.markChatRelayWaitingForJoinForTesting() + + // The join of another room must not make us repoll + signalingController.roomMessageReceived(messageDict: ["type": "room", "room": ["roomid": "someOtherRoom"]]) + chatController.waitForChatRelayQueueForTesting() + drainMainQueue() + XCTAssertEqual(chatController.chatRelayStateForTesting, .waitingForJoin) + + // Our room was acked: repoll right away instead of waiting for the running long poll to time + // out, so the relay is armed by that poll's 304 + signalingController.roomMessageReceived(messageDict: ["type": "room", "room": ["roomid": room.token]]) + chatController.waitForChatRelayQueueForTesting() + drainMainQueue() + XCTAssertEqual(chatController.chatRelayStateForTesting, .inactive) + + chatController.stop() + drainMainQueue() + } + + func testJoiningTheRoomDoesNotRepollWhenTheRelayIsAlreadyActive() throws { + let room = addRoom(withToken: "relayActiveRoom") + let chatController = NCChatController(for: room)! + let signalingController = inertSignalingController() + + chatController.setupChatRelayForTesting(with: signalingController) + chatController.markChatRelayActiveForTesting() + + // A re-join while the relay is already processing messages must not restart polling + signalingController.roomMessageReceived(messageDict: ["type": "room", "room": ["roomid": room.token]]) + chatController.waitForChatRelayQueueForTesting() + drainMainQueue() + + XCTAssertEqual(chatController.chatRelayStateForTesting, .active) + + chatController.stop() + drainMainQueue() + } + func testContentInsetAdjustsForOverlayViews() throws { let activeAccount = NCDatabaseManager.sharedInstance().activeAccount() let room = NCRoom() diff --git a/NextcloudTalkTests/Unit/UnitExternalSignalingControllerTest.swift b/NextcloudTalkTests/Unit/UnitExternalSignalingControllerTest.swift new file mode 100644 index 000000000..350afdab2 --- /dev/null +++ b/NextcloudTalkTests/Unit/UnitExternalSignalingControllerTest.swift @@ -0,0 +1,149 @@ +// +// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors +// SPDX-License-Identifier: GPL-3.0-or-later +// + +import XCTest +@testable import NextcloudTalk + +// Covers `joinedRoomToken`, the state the chat relay is gated on: the signaling server only sends +// us the events of the room our session actually joined, so arming the relay on anything else +// (a server capability, `currentRoom`) stops the chat API polling while nothing is relayed to us. +final class UnitExternalSignalingControllerTest: TestBaseRealm { + + private var signalingController: NCExternalSignalingController! + + override func setUpWithError() throws { + try super.setUpWithError() + + let account = NCDatabaseManager.sharedInstance().activeAccount() + signalingController = NCExternalSignalingController(account: account, serverUrl: TestConstants.server, ticket: "fakeTicket") + + // Make the controller inert: without a websocket the delegate callbacks of the failing + // connection attempt are ignored, so nothing reconnects underneath the assertions. + signalingController.disconnect() + drainMainQueue() + } + + // MARK: - Helper + + private func drainMainQueue() { + let exp = expectation(description: "\(#function)\(#line)") + DispatchQueue.main.async { exp.fulfill() } + waitForExpectations(timeout: TestConstants.timeoutShort, handler: nil) + } + + private func helloMessage(withSessionId sessionId: String) -> [AnyHashable: Any] { + return [ + "type": "hello", + "id": "1", + "hello": [ + "sessionid": sessionId, + "resumeid": "fakeResumeId", + "server": [ + "version": "2.0.0", + "features": ["mcu", "chat-relay"] + ] + ] + ] + } + + private func roomMessage(withRoomToken roomToken: String) -> [AnyHashable: Any] { + return ["type": "room", "room": ["roomid": roomToken]] + } + + // MARK: - Tests + + func testJoinedRoomTokenIsOnlySetOnceTheRoomIsAcked() throws { + XCTAssertNil(signalingController.joinedRoomToken) + + // Knowing the chat relay is supported does not mean we are in any room yet + signalingController.helloResponseReceived(messageDict: helloMessage(withSessionId: "session-1")) + XCTAssertTrue(signalingController.hasChatRelay) + XCTAssertNil(signalingController.joinedRoomToken) + + signalingController.roomMessageReceived(messageDict: roomMessage(withRoomToken: "joinedToken")) + XCTAssertEqual(signalingController.joinedRoomToken, "joinedToken") + } + + func testJoinedRoomTokenIsPostedAsNotification() throws { + expectation(forNotification: .extSignalingDidJoinRoom, object: signalingController) { notification in + return notification.userInfo?["roomToken"] as? String == "joinedToken" + } + + signalingController.roomMessageReceived(messageDict: roomMessage(withRoomToken: "joinedToken")) + + waitForExpectations(timeout: TestConstants.timeoutShort, handler: nil) + } + + func testLeavingTheRoomClearsTheJoinedRoomToken() throws { + signalingController.roomMessageReceived(messageDict: roomMessage(withRoomToken: "joinedToken")) + XCTAssertEqual(signalingController.joinedRoomToken, "joinedToken") + + // The server acks leaving a room with an empty roomid + signalingController.roomMessageReceived(messageDict: roomMessage(withRoomToken: "")) + XCTAssertNil(signalingController.joinedRoomToken) + } + + func testDisconnectingClearsTheJoinedRoomToken() throws { + signalingController.roomMessageReceived(messageDict: roomMessage(withRoomToken: "joinedToken")) + XCTAssertEqual(signalingController.joinedRoomToken, "joinedToken") + + // Unlike `currentRoom`, which survives a reconnect on purpose so we can re-join, the joined + // room has to be cleared: the new connection is in no room until the server acks the re-join + signalingController.resetWebSocket() + XCTAssertNil(signalingController.joinedRoomToken) + XCTAssertEqual(signalingController.currentRoom, "joinedToken") + } + + func testNewSessionIsNotConsideredJoinedUntilItRejoined() throws { + signalingController.helloResponseReceived(messageDict: helloMessage(withSessionId: "session-1")) + signalingController.roomMessageReceived(messageDict: roomMessage(withRoomToken: "joinedToken")) + XCTAssertEqual(signalingController.joinedRoomToken, "joinedToken") + + // We could not resume the session, so the server created a new one which is in no room yet + signalingController.helloResponseReceived(messageDict: helloMessage(withSessionId: "session-2")) + XCTAssertNil(signalingController.joinedRoomToken) + + // ... until the re-join is acked + signalingController.roomMessageReceived(messageDict: roomMessage(withRoomToken: "joinedToken")) + XCTAssertEqual(signalingController.joinedRoomToken, "joinedToken") + + drainMainQueue() + } + + func testResumedSessionIsStillConsideredJoined() throws { + signalingController.helloResponseReceived(messageDict: helloMessage(withSessionId: "session-1")) + signalingController.roomMessageReceived(messageDict: roomMessage(withRoomToken: "joinedToken")) + + signalingController.resetWebSocket() + XCTAssertNil(signalingController.joinedRoomToken) + + // The session was resumed (same session id), so the server kept us in the room and replays + // the messages we missed while being disconnected. No re-join and no room ack follows here. + signalingController.helloResponseReceived(messageDict: helloMessage(withSessionId: "session-1")) + XCTAssertEqual(signalingController.joinedRoomToken, "joinedToken") + + drainMainQueue() + } + + func testAlreadyJoinedErrorIsTreatedAsJoined() throws { + signalingController.roomMessageReceived(messageDict: roomMessage(withRoomToken: "joinedToken")) + signalingController.resetWebSocket() + XCTAssertNil(signalingController.joinedRoomToken) + + // The server tells us we are still in the room. No room message follows in this case, so + // without handling it here the relay would never be armed again for this room. + let errorMessage: [AnyHashable: Any] = [ + "type": "error", + "id": "2", + "error": [ + "code": "already_joined", + "details": ["room": ["roomid": "joinedToken"]] + ] + ] + + signalingController.errorResponseReceived(messageDict: errorMessage) + XCTAssertEqual(signalingController.joinedRoomToken, "joinedToken") + } +}