From 65605ded95095b11fc531fd0d52003746e2f6a0b Mon Sep 17 00:00:00 2001 From: TolaMironcenko Date: Tue, 11 Aug 2026 22:00:49 +0700 Subject: [PATCH] mam fixes --- Sources/Shared/Models/AppModel.swift | 94 ++- .../Shared/Models/ArchiveSyncCheckpoint.swift | 35 ++ Sources/Shared/Models/ChatMessage.swift | 5 + Sources/Shared/Persistence/ChatArchive.swift | 11 +- Sources/Shared/UI/ChatView.swift | 28 + Sources/Shared/XMPP/XMPPService.swift | 549 ++++++++++++++++-- 6 files changed, 649 insertions(+), 73 deletions(-) diff --git a/Sources/Shared/Models/AppModel.swift b/Sources/Shared/Models/AppModel.swift index 16f862b..2e03a8f 100644 --- a/Sources/Shared/Models/AppModel.swift +++ b/Sources/Shared/Models/AppModel.swift @@ -42,6 +42,8 @@ final class AppModel: ObservableObject { @Published private(set) var isOMEMOReady = false @Published private(set) var ownFingerprint: String? @Published private(set) var isArchiveSyncing = false + @Published private(set) var isLoadingOlderHistory = false + @Published private(set) var hasMoreOlderHistory = true @Published private(set) var isAuthenticating = false @Published private(set) var isSendingAttachment = false @Published private(set) var isUpdatingAvatar = false @@ -93,6 +95,7 @@ final class AppModel: ObservableObject { private var locallyDeletedMessageIDs: Set = [] private var lastSuccessfulMAMSync: Date? private var lastSuccessfulMAMCursor: String? + private var mamCheckpoints: [MAMArchiveKey: MAMArchiveCheckpoint] = [:] private var pendingRoomPasswords: [String: String] = [:] private var joiningRoomJIDs: Set = [] private var mediaSendActivity = MediaSendActivityTracker() @@ -286,7 +289,8 @@ final class AppModel: ObservableObject { try await xmpp.connect( account: saved, password: password, - archiveCheckpoint: durableArchiveSyncCheckpoint + archiveCheckpoint: durableArchiveSyncCheckpoint, + mamCheckpoints: mamCheckpoints ) await notifications.requestAuthorization() } catch { @@ -324,7 +328,8 @@ final class AppModel: ObservableObject { try await xmpp.connect( account: account, password: password, - archiveCheckpoint: durableArchiveSyncCheckpoint + archiveCheckpoint: durableArchiveSyncCheckpoint, + mamCheckpoints: mamCheckpoints ) try credentials.save(password: password, for: account.normalizedJID) try preferences.save(account) @@ -500,11 +505,41 @@ final class AppModel: ObservableObject { func selectConversation(id: String) { let normalized = id.lowercased() selectedConversationID = normalized - if let index = conversations.firstIndex(where: { $0.id == normalized }) { - conversations[index].unreadCount = 0 +// if let index = conversations.firstIndex(where: { $0.id == normalized }) { +// conversations[index].unreadCount = 0 +// } +// schedulePersist() +// syncWatch() + hasMoreOlderHistory = true + } + + func loadOlderHistoryForSelectedConversation() { + guard !isLoadingOlderHistory, + hasMoreOlderHistory, + let conversation = selectedConversation else { return } + let oldestServerID = selectedMessages + .sorted { lhs, rhs in + if lhs.timestamp != rhs.timestamp { return lhs.timestamp < rhs.timestamp } + return lhs.id < rhs.id + } + .compactMap(\.stanzaID) + .first + + isLoadingOlderHistory = true + xmpp.loadOlderHistory( + conversationJID: conversation.jid, + isGroup: conversation.isGroup, + before: oldestServerID + ) { [weak self] result in + guard let self else { return } + self.isLoadingOlderHistory = false + switch result { + case .success(let hasMore): + self.hasMoreOlderHistory = hasMore + case .failure(let error): + self.errorMessage = error.localizedDescription + } } - schedulePersist() - syncWatch() } func createOrJoinGroup( @@ -1806,6 +1841,12 @@ final class AppModel: ObservableObject { case .archiveSyncCompleted(let checkpoint): lastSuccessfulMAMSync = checkpoint.timestamp lastSuccessfulMAMCursor = checkpoint.cursor + if let account { + mamCheckpoints[.account(account.normalizedJID)] = MAMArchiveCheckpoint( + timestamp: checkpoint.timestamp, + cursor: checkpoint.cursor + ) + } // Синхронно форсируем persist, чтобы checkpoint не потерялся persistTask?.cancel() persistTask = nil @@ -1817,10 +1858,14 @@ final class AppModel: ObservableObject { locallyDeletedMessageIDs: self.locallyDeletedMessageIDs, rosterContactJIDs: self.rosterContactJIDs, lastSuccessfulMAMSync: self.lastSuccessfulMAMSync, - lastSuccessfulMAMCursor: self.lastSuccessfulMAMCursor + lastSuccessfulMAMCursor: self.lastSuccessfulMAMCursor, + mamCheckpoints: self.mamCheckpoints ) try? await archive.save(snapshot) } + case .mucArchiveSyncCompleted(let archiveKey, let checkpoint): + mamCheckpoints[archiveKey] = checkpoint + schedulePersist() case .call(let call): activeCall = call case .callHistory(let entry): @@ -1886,6 +1931,13 @@ final class AppModel: ObservableObject { { return // уже есть это сообщение } + if let originID = envelope.originID, + messages.contains(where: { + $0.conversationID == envelope.peerJID.lowercased() && $0.originID == originID + }) { + mergeServerIdentity(from: envelope, matchingOriginID: originID) + return + } if envelope.isGroupMessage { upsertGroupConversation( jid: envelope.peerJID, @@ -1933,6 +1985,7 @@ final class AppModel: ObservableObject { replyToID: envelope.replyToID, replyToJID: envelope.replyToJID, replyPreview: envelope.replyPreview, + originID: envelope.originID, stanzaID: envelope.stanzaID, senderDisplayName: envelope.senderDisplayName, isGroupMessage: envelope.isGroupMessage @@ -1974,6 +2027,23 @@ final class AppModel: ObservableObject { ) } } + + private func mergeServerIdentity( + from envelope: XMPPService.MessageEnvelope, + matchingOriginID originID: String + ) { + guard let index = messages.firstIndex(where: { + $0.conversationID == envelope.peerJID.lowercased() && $0.originID == originID + }) else { return } + if messages[index].stanzaID == nil { + messages[index].stanzaID = envelope.stanzaID + } + if messages[index].delivery == .sending { + messages[index].delivery = .sent + } + rebuildMessageIndex() + schedulePersist() + } private func recordCallHistory(_ entry: CallHistoryEntry) { guard let account else { return } @@ -2603,6 +2673,13 @@ final class AppModel: ObservableObject { locallyDeletedMessageIDs = snapshot.locallyDeletedMessageIDs lastSuccessfulMAMSync = snapshot.lastSuccessfulMAMSync lastSuccessfulMAMCursor = snapshot.lastSuccessfulMAMCursor + mamCheckpoints = snapshot.mamCheckpoints + if mamCheckpoints.isEmpty, let lastSuccessfulMAMSync { + mamCheckpoints[.account(jid)] = MAMArchiveCheckpoint( + timestamp: lastSuccessfulMAMSync, + cursor: lastSuccessfulMAMCursor + ) + } messages = snapshot.messages.filter { !locallyDeletedMessageIDs.contains( Self.localDeletionKey( @@ -2633,7 +2710,8 @@ final class AppModel: ObservableObject { locallyDeletedMessageIDs: self.locallyDeletedMessageIDs, rosterContactJIDs: self.rosterContactJIDs, lastSuccessfulMAMSync: self.lastSuccessfulMAMSync, - lastSuccessfulMAMCursor: self.lastSuccessfulMAMCursor + lastSuccessfulMAMCursor: self.lastSuccessfulMAMCursor, + mamCheckpoints: self.mamCheckpoints ) try? await archive.save(snapshot) } diff --git a/Sources/Shared/Models/ArchiveSyncCheckpoint.swift b/Sources/Shared/Models/ArchiveSyncCheckpoint.swift index 9401ca4..4eb60c0 100644 --- a/Sources/Shared/Models/ArchiveSyncCheckpoint.swift +++ b/Sources/Shared/Models/ArchiveSyncCheckpoint.swift @@ -1,5 +1,40 @@ import Foundation +/// A MAM UID is only meaningful inside one archive. Account MAM and every MUC +/// therefore get their own namespace and durable checkpoint. +struct MAMArchiveKey: Codable, Hashable, Sendable { + enum Kind: String, Codable, Sendable { + case account + case muc + } + + let kind: Kind + let jid: String + + init(kind: Kind, jid: String) { + self.kind = kind + self.jid = jid.lowercased() + } + + static func account(_ jid: String) -> Self { + Self(kind: .account, jid: jid) + } + + static func muc(_ jid: String) -> Self { + Self(kind: .muc, jid: jid) + } +} + +struct MAMArchiveCheckpoint: Codable, Equatable, Sendable { + let timestamp: Date + let cursor: String? + + init(timestamp: Date, cursor: String? = nil) { + self.timestamp = timestamp + self.cursor = ArchiveSyncCheckpoint.normalizedCursor(cursor) + } +} + /// Durable MAM position. The archive UID is the primary synchronization /// anchor; the timestamp is retained as a compatibility fallback for legacy /// snapshots and servers that have expired an old UID. diff --git a/Sources/Shared/Models/ChatMessage.swift b/Sources/Shared/Models/ChatMessage.swift index 073af59..ee44364 100644 --- a/Sources/Shared/Models/ChatMessage.swift +++ b/Sources/Shared/Models/ChatMessage.swift @@ -61,6 +61,7 @@ struct ChatMessage: Codable, Identifiable, Hashable, Sendable { var replyPreview: String? var forwardedFrom: String? var retractedAt: Date? + var originID: String? var stanzaID: String? var senderDisplayName: String? var isGroupMessage: Bool @@ -89,6 +90,7 @@ struct ChatMessage: Codable, Identifiable, Hashable, Sendable { replyPreview: String? = nil, forwardedFrom: String? = nil, retractedAt: Date? = nil, + originID: String? = nil, stanzaID: String? = nil, senderDisplayName: String? = nil, isGroupMessage: Bool = false, @@ -116,6 +118,7 @@ struct ChatMessage: Codable, Identifiable, Hashable, Sendable { self.replyPreview = replyPreview self.forwardedFrom = forwardedFrom self.retractedAt = retractedAt + self.originID = originID ?? (direction == .outgoing ? id : nil) self.stanzaID = stanzaID self.senderDisplayName = senderDisplayName self.isGroupMessage = isGroupMessage @@ -145,6 +148,7 @@ struct ChatMessage: Codable, Identifiable, Hashable, Sendable { case replyPreview case forwardedFrom case retractedAt + case originID case stanzaID case senderDisplayName case isGroupMessage @@ -175,6 +179,7 @@ struct ChatMessage: Codable, Identifiable, Hashable, Sendable { replyPreview = try values.decodeIfPresent(String.self, forKey: .replyPreview) forwardedFrom = try values.decodeIfPresent(String.self, forKey: .forwardedFrom) retractedAt = try values.decodeIfPresent(Date.self, forKey: .retractedAt) + originID = try values.decodeIfPresent(String.self, forKey: .originID) stanzaID = try values.decodeIfPresent(String.self, forKey: .stanzaID) senderDisplayName = try values.decodeIfPresent(String.self, forKey: .senderDisplayName) isGroupMessage = try values.decodeIfPresent(Bool.self, forKey: .isGroupMessage) ?? false diff --git a/Sources/Shared/Persistence/ChatArchive.swift b/Sources/Shared/Persistence/ChatArchive.swift index 4ffc700..f40698c 100644 --- a/Sources/Shared/Persistence/ChatArchive.swift +++ b/Sources/Shared/Persistence/ChatArchive.swift @@ -2,7 +2,7 @@ import Foundation actor ChatArchive { struct Snapshot: Codable, Sendable { - static let currentSchemaVersion = 5 + static let currentSchemaVersion = 6 var schemaVersion: Int var conversations: [Conversation] @@ -11,6 +11,7 @@ actor ChatArchive { var rosterContactJIDs: Set var lastSuccessfulMAMSync: Date? var lastSuccessfulMAMCursor: String? + var mamCheckpoints: [MAMArchiveKey: MAMArchiveCheckpoint] init( conversations: [Conversation], @@ -18,7 +19,8 @@ actor ChatArchive { locallyDeletedMessageIDs: Set = [], rosterContactJIDs: Set = [], lastSuccessfulMAMSync: Date? = nil, - lastSuccessfulMAMCursor: String? = nil + lastSuccessfulMAMCursor: String? = nil, + mamCheckpoints: [MAMArchiveKey: MAMArchiveCheckpoint] = [:] ) { self.schemaVersion = Self.currentSchemaVersion self.conversations = conversations @@ -29,6 +31,7 @@ actor ChatArchive { self.lastSuccessfulMAMCursor = ArchiveSyncCheckpoint.normalizedCursor( lastSuccessfulMAMCursor ) + self.mamCheckpoints = mamCheckpoints } private enum CodingKeys: String, CodingKey { @@ -39,6 +42,7 @@ actor ChatArchive { case rosterContactJIDs case lastSuccessfulMAMSync case lastSuccessfulMAMCursor + case mamCheckpoints } init(from decoder: Decoder) throws { @@ -61,6 +65,9 @@ actor ChatArchive { lastSuccessfulMAMCursor = ArchiveSyncCheckpoint.normalizedCursor( try values.decodeIfPresent(String.self, forKey: .lastSuccessfulMAMCursor) ) + mamCheckpoints = try values.decodeIfPresent( + [MAMArchiveKey: MAMArchiveCheckpoint].self, forKey: .mamCheckpoints + ) ?? [:] } static let empty = Snapshot(conversations: [], messages: []) diff --git a/Sources/Shared/UI/ChatView.swift b/Sources/Shared/UI/ChatView.swift index 748dd32..722cd93 100644 --- a/Sources/Shared/UI/ChatView.swift +++ b/Sources/Shared/UI/ChatView.swift @@ -44,6 +44,7 @@ struct ChatView: View { @State private var replyThreadSelection: ReplyThreadSelection? @State private var hasCompletedInitialScroll = false @State private var isNearTimelineBottom = true + @State private var historyLoadAnchorID: String? @State private var activeCaptureMode: ComposerCaptureMode? @State private var preparingCaptureMode: ComposerCaptureMode? @State private var captureGestureIsActive = false @@ -237,6 +238,7 @@ struct ChatView: View { .onChange(of: conversation.id) { _, _ in hasCompletedInitialScroll = false isNearTimelineBottom = true + historyLoadAnchorID = nil } .onDisappear { model.endComposerActivity(in: liveConversation) @@ -324,6 +326,23 @@ struct ChatView: View { ZStack(alignment: .bottomTrailing) { ScrollView { LazyVStack(spacing: 5) { + if hasCompletedInitialScroll, + model.hasMoreOlderHistory, + !timelineEntries.isEmpty { + Color.clear + .frame(height: 1) + .accessibilityHidden(true) + .onAppear { + guard !model.isLoadingOlderHistory else { return } + historyLoadAnchorID = timelineEntries.first?.id + model.loadOlderHistoryForSelectedConversation() + } + if model.isLoadingOlderHistory { + ProgressView() + .controlSize(.small) + .padding(.vertical, 6) + } + } if timelineEntries.isEmpty { ContentUnavailableView( emptyChatTitle, @@ -413,6 +432,15 @@ struct ChatView: View { } scrollToBottom(proxy, animated: true) } + .onChange(of: model.isLoadingOlderHistory) { wasLoading, isLoading in + guard wasLoading, !isLoading, + let anchor = historyLoadAnchorID else { return } + Task { @MainActor in + await Task.yield() + proxy.scrollTo(anchor, anchor: .top) + historyLoadAnchorID = nil + } + } .onChange(of: timelineEntries.count) { _, _ in selectedMessageIDs.formIntersection( Set(timelineEntries.map(\.id)) diff --git a/Sources/Shared/XMPP/XMPPService.swift b/Sources/Shared/XMPP/XMPPService.swift index 480062e..99696a3 100644 --- a/Sources/Shared/XMPP/XMPPService.swift +++ b/Sources/Shared/XMPP/XMPPService.swift @@ -17,19 +17,32 @@ private struct BufferedArchivePage { let rejectedSource: Bool } +private enum BufferedLiveDelivery { + case direct(message: Message, timestamp: Date) + case group(message: Message, room: RoomProtocol) +} + /// Martin publishes every MAM result synchronously on its parser queue. Keep /// that burst off the main queue and hand the completed page to XMPPService /// only after the final IQ callback. Message instances are immutable after the /// publisher returns; the lock provides the ownership hand-off. private final class ArchiveStanzaInbox: @unchecked Sendable { + private struct State { + let allowedSources: Set + var stanzas: [BufferedArchiveStanza] = [] + var overflowed = false + var rejectedSource = false + } + private let lock = NSLock() private let maximumCount: Int - private var activeQueryID: String? - private var allowedSources: Set = [] - private var stanzas: [BufferedArchiveStanza] = [] - private var overflowed = false - private var rejectedSource = false - +// private var activeQueryID: String? +// private var allowedSources: Set = [] +// private var stanzas: [BufferedArchiveStanza] = [] +// private var overflowed = false +// private var rejectedSource = false + private var states: [String: State] = [:] + init(maximumCount: Int) { self.maximumCount = maximumCount } @@ -37,11 +50,14 @@ private final class ArchiveStanzaInbox: @unchecked Sendable { func begin(queryID: String, allowedSources: Set) { lock.lock() defer { lock.unlock() } - activeQueryID = queryID - self.allowedSources = Set(allowedSources.map { $0.lowercased() }) - stanzas.removeAll(keepingCapacity: true) - overflowed = false - rejectedSource = false +// activeQueryID = queryID +// self.allowedSources = Set(allowedSources.map { $0.lowercased() }) +// stanzas.removeAll(keepingCapacity: true) +// overflowed = false +// rejectedSource = false + states[queryID] = State( + allowedSources: Set(allowedSources.map { $0.lowercased() }) + ) } func append( @@ -53,53 +69,73 @@ private final class ArchiveStanzaInbox: @unchecked Sendable { ) { lock.lock() defer { lock.unlock() } - guard activeQueryID == queryID else { return } - guard allowedSources.contains(source.stringValue.lowercased()) else { - rejectedSource = true +// guard activeQueryID == queryID else { return } +// guard allowedSources.contains(source.stringValue.lowercased()) else { +// rejectedSource = true + guard var state = states[queryID] else { return } + guard state.allowedSources.contains(source.stringValue.lowercased()) else { + state.rejectedSource = true + states[queryID] = state return } - guard stanzas.count < maximumCount else { - overflowed = true +// guard stanzas.count < maximumCount else { +// overflowed = true + guard state.stanzas.count < maximumCount else { + state.overflowed = true + states[queryID] = state return } - stanzas.append( +// stanzas.append( + state.stanzas.append( BufferedArchiveStanza( queryID: queryID, message: message, timestamp: timestamp, archiveID: archiveID )) - print("MAM source: \(source), allowed: \(allowedSources)") +// print("MAM source: \(source), allowed: \(allowedSources)") + states[queryID] = state } func take(queryID: String) -> BufferedArchivePage { lock.lock() defer { lock.unlock() } - guard activeQueryID == queryID else { +// guard activeQueryID == queryID else { + guard let state = states.removeValue(forKey: queryID) else { return BufferedArchivePage(stanzas: [], overflowed: false, rejectedSource: false) } - let page = BufferedArchivePage( - stanzas: stanzas, - overflowed: overflowed, - rejectedSource: rejectedSource +// let page = BufferedArchivePage( +// stanzas: stanzas, +// overflowed: overflowed, +// rejectedSource: rejectedSource +// ) +// activeQueryID = nil +// allowedSources.removeAll(keepingCapacity: true) +// stanzas.removeAll(keepingCapacity: true) +// overflowed = false +// rejectedSource = false +// return page + return BufferedArchivePage( + stanzas: state.stanzas, + overflowed: state.overflowed, + rejectedSource: state.rejectedSource ) - activeQueryID = nil - allowedSources.removeAll(keepingCapacity: true) - stanzas.removeAll(keepingCapacity: true) - overflowed = false - rejectedSource = false - return page } func cancel(queryID: String? = nil) { lock.lock() defer { lock.unlock() } - if let queryID, activeQueryID != queryID { return } - activeQueryID = nil - allowedSources.removeAll(keepingCapacity: true) - stanzas.removeAll(keepingCapacity: true) - overflowed = false - rejectedSource = false +// if let queryID, activeQueryID != queryID { return } +// activeQueryID = nil +// allowedSources.removeAll(keepingCapacity: true) +// stanzas.removeAll(keepingCapacity: true) +// overflowed = false +// rejectedSource = false + if let queryID { + states.removeValue(forKey: queryID) + } else { + states.removeAll(keepingCapacity: true) + } } } @@ -130,6 +166,7 @@ final class XMPPService { let replyToID: String? let replyToJID: String? let replyPreview: String? + let originID: String? let stanzaID: String? let senderDisplayName: String? let isGroupMessage: Bool @@ -214,6 +251,7 @@ final class XMPPService { case archiveBatch([ArchiveMutation]) case archiveSyncing(Bool) case archiveSyncCompleted(ArchiveSyncCheckpoint) + case mucArchiveSyncCompleted(archive: MAMArchiveKey, checkpoint: MAMArchiveCheckpoint) case call(CallSnapshot?) case callHistory(CallHistoryEntry) case callError(String) @@ -251,6 +289,13 @@ final class XMPPService { private var archiveStanzaBuffer: [BufferedArchiveStanza] = [] private var archiveBufferOverflowed = false private var archiveRejectedSource = false + private var mamCheckpoints: [MAMArchiveKey: MAMArchiveCheckpoint] = [:] + private var pendingMUCCatchups: [MAMArchiveKey] = [] + private var activeMUCCatchup: MAMArchiveKey? + private var mucCatchupHighWatermark: Date? + private var mucCatchupLastCursor: String? + private var mucCursorFallbackArchives: Set = [] + private var delayedLiveByArchive: [MAMArchiveKey: [BufferedLiveDelivery]] = [:] private let archiveStanzaInbox = ArchiveStanzaInbox( maximumCount: ArchiveMessageBatchPolicy.maximumBufferedStanzas ) @@ -301,13 +346,15 @@ final class XMPPService { func connect( account: AccountConfiguration, password: String, - archiveCheckpoint: ArchiveSyncCheckpoint? = nil + archiveCheckpoint: ArchiveSyncCheckpoint? = nil, + mamCheckpoints: [MAMArchiveKey: MAMArchiveCheckpoint] = [:] ) async throws { await disconnect() cancellables.removeAll() archiveSyncStarted = false archiveSyncCompletedForConnection = false archiveSyncCheckpoint = archiveCheckpoint + self.mamCheckpoints = mamCheckpoints archiveSyncQueryStartedAt = nil archivePagination = ArchiveSyncPagination() archiveRetryTask?.cancel() @@ -330,6 +377,11 @@ final class XMPPService { archiveBufferOverflowed = false archiveRejectedSource = false archiveStanzaInbox.cancel() + pendingMUCCatchups.removeAll() + activeMUCCatchup = nil + mucCursorFallbackArchives.removeAll() + mucCatchupHighWatermark = nil + mucCatchupLastCursor = nil setArchiveSyncIndicator(false) avatarRequests.removeAll() observedRoomJIDs.removeAll() @@ -577,6 +629,8 @@ final class XMPPService { room = joinedRoom } observe(room: room) + enqueueMUCCatchup(roomJID: room.jid) + eventHandler?( .roomState( RoomStateEnvelope( @@ -700,6 +754,7 @@ final class XMPPService { throw LumaXMPPError.roomNotJoined } let message = room.createMessage(text: wireBody, id: messageID) + addOriginID(messageID, to: message) message.lastMessageCorrectionId = replacingMessageID if chatStatesEnabled { message.chatState = .active } addReply(replyTo, fallback: replyFallback, to: message) @@ -736,6 +791,7 @@ final class XMPPService { } let message = chat.createMessage(text: wireBody, id: messageID) + addOriginID(messageID, to: message) message.lastMessageCorrectionId = replacingMessageID if chatStatesEnabled { message.chatState = .active } addReply(replyTo, fallback: replyFallback, to: message) @@ -754,6 +810,107 @@ final class XMPPService { return nil } } + + /// Loads one older MAM page for a single conversation. `before` is the + /// oldest server/MAM id currently known by the UI. For direct chats the + /// query is scoped with XEP-0313 `with`; for MUC the IQ is addressed to the + /// room archive itself. + func loadOlderHistory( + conversationJID: String, + isGroup: Bool, + before: String?, + completion: @escaping (Result) -> Void + ) { + guard let client, client.state == .connected() else { + completion(.failure(LumaXMPPError.notConnected)) + return + } + guard !archiveSyncStarted, activeMUCCatchup == nil else { + completion(.success(false)) + return + } + guard let version = client.module(.mam).availableVersions.first else { + completion(.success(false)) + return + } + + let archive = isGroup + ? MAMArchiveKey.muc(conversationJID) + : MAMArchiveKey.account(client.userBareJid.stringValue) + let queryID = UUID().uuidString + archiveStanzaInbox.begin( + queryID: queryID, + allowedSources: allowedMAMSources(for: archive, client: client) + ) + + // Martin 3.2.4's convenience overload accidentally clears `with`. + // Construct MAMQueryForm explicitly so direct-chat history is truly + // scoped to this peer. + let form = MAMQueryForm(version: version) + if !isGroup { + form.with = JID(conversationJID.lowercased()) + } + let componentJID: JID? = isGroup ? JID(conversationJID.lowercased()) : nil + let rsm: RSM.Query = before.map { + RSM.Query(before: $0, max: archivePageSize) + } ?? RSM.Query(lastItems: archivePageSize) + + client.module(.mam).queryItems( + version: version, + componentJid: componentJID, + query: form, + queryId: queryID, + rsm: rsm + ) { [weak self, weak client] result in + DispatchQueue.main.async { + guard let self, let client else { + completion(.success(false)) + return + } + let page = self.archiveStanzaInbox.take(queryID: queryID) + guard !page.overflowed, !page.rejectedSource else { + completion(.failure(LumaXMPPError.connection("Некорректная MAM history page"))) + return + } + + // History is only allowed while catch-up is idle, so the + // legacy mutation accumulator can safely be used as a local + // page collector here. + let savedMutations = self.archivePassMutations + self.archivePassMutations = [] + for stanza in page.stanzas { + if stanza.message.type == .groupchat, + let roomJID = stanza.message.from?.bareJid { + self.handle( + groupMessage: stanza.message, + archivedRoomJID: roomJID, + archivedTimestamp: stanza.timestamp, + archiveID: stanza.archiveID, + isArchived: true + ) + } else { + self.handle( + message: stanza.message, + timestamp: stanza.timestamp, + archiveID: stanza.archiveID, + isArchived: true + ) + } + } + let historyMutations = self.archivePassMutations + self.archivePassMutations = savedMutations + if !historyMutations.isEmpty { + self.eventHandler?(.archiveBatch(historyMutations)) + } + switch result { + case .success(let response): + completion(.success(!response.complete)) + case .failure(let error): + completion(.failure(error)) + } + } + } + } func sendRetraction( to recipient: String, @@ -1271,14 +1428,16 @@ final class XMPPService { client.module(.message).messagesPublisher .receive(on: DispatchQueue.main) .sink { [weak self] incoming in - self?.handle(message: incoming.message, timestamp: Date(), archiveID: nil) +// self?.handle(message: incoming.message, timestamp: Date(), archiveID: nil) + self?.deliverOrDelayDirect(incoming.message, timestamp: Date()) } .store(in: &cancellables) client.module(.muc).messagesPublisher .receive(on: DispatchQueue.main) .sink { [weak self] incoming in - self?.handle(groupMessage: incoming.message, room: incoming.room) +// self?.handle(groupMessage: incoming.message, room: incoming.room) + self?.deliverOrDelayGroup(incoming.message, room: incoming.room) } .store(in: &cancellables) @@ -1299,7 +1458,8 @@ final class XMPPService { client.module(.messageCarbons).carbonsPublisher .receive(on: DispatchQueue.main) .sink { [weak self] carbon in - self?.handle(message: carbon.message, timestamp: Date(), archiveID: nil) +// self?.handle(message: carbon.message, timestamp: Date(), archiveID: nil) + self?.deliverOrDelayDirect(carbon.message, timestamp: Date()) } .store(in: &cancellables) @@ -1354,6 +1514,41 @@ final class XMPPService { } .store(in: &cancellables) } + + private func deliverOrDelayDirect(_ message: Message, timestamp: Date) { + guard let client else { return } + let archive = MAMArchiveKey.account(client.userBareJid.stringValue) + if archiveSyncStarted { + delayedLiveByArchive[archive, default: []].append( + .direct(message: message, timestamp: timestamp) + ) + return + } + handle(message: message, timestamp: timestamp, archiveID: nil) + } + + private func deliverOrDelayGroup(_ message: Message, room: RoomProtocol) { + let archive = MAMArchiveKey.muc(room.jid.stringValue) + if activeMUCCatchup == archive { + delayedLiveByArchive[archive, default: []].append( + .group(message: message, room: room) + ) + return + } + handle(groupMessage: message, room: room) + } + + private func replayDelayedLive(for archive: MAMArchiveKey) { + let delayed = delayedLiveByArchive.removeValue(forKey: archive) ?? [] + for item in delayed { + switch item { + case .direct(let message, let timestamp): + handle(message: message, timestamp: timestamp, archiveID: nil) + case .group(let message, let room): + handle(groupMessage: message, room: room) + } + } + } private func handle(state: XMPPClient.State) { switch state { @@ -1565,6 +1760,7 @@ final class XMPPService { replyToID: reply?.getAttribute("id"), replyToJID: reply?.getAttribute("to"), replyPreview: parsedReply.preview, + originID: message.originId, stanzaID: archiveID, senderDisplayName: nil, isGroupMessage: false, @@ -1572,19 +1768,33 @@ final class XMPPService { ), isArchived: isArchived) } - private func handle(groupMessage message: Message, room: RoomProtocol) { + private func handle( + groupMessage message: Message, + room: RoomProtocol? = nil, + archivedRoomJID: BareJID? = nil, + archivedTimestamp: Date? = nil, + archiveID: String? = nil, + isArchived: Bool = false + ) { guard message.type != .error, let client, - let from = message.from + let from = message.from, + let roomBareJID = room?.jid ?? archivedRoomJID ?? message.from?.bareJid else { return } let nickname = from.resource ?? "Участник" - let outgoing = nickname == room.nickname - let roomJID = room.jid.stringValue.lowercased() +// let outgoing = nickname == room.nickname +// let roomJID = room.jid.stringValue.lowercased() + let disclosedSender = Self.originalSenderJID(in: message) + ?? XMucUserElement.extract(from: message)?.jid?.bareJid + let outgoing = room.map { nickname == $0.nickname } + ?? (disclosedSender == client.userBareJid) + let roomJID = roomBareJID.stringValue.lowercased() // In a MUC the author address used by XEP-0461 is the full occupant // JID (room@service/nickname), not the participant's disclosed real JID. let senderJID = from.stringValue - let stanzaID = Self.roomStanzaID(in: message, roomJID: room.jid) +// let stanzaID = Self.roomStanzaID(in: message, roomJID: room.jid) + let stanzaID = Self.roomStanzaID(in: message, roomJID: roomBareJID) ?? archiveID let id = outgoing ? (message.originId ?? message.id ?? stanzaID ?? UUID().uuidString) @@ -1600,7 +1810,8 @@ final class XMPPService { } else if let disclosed = XMucUserElement.extract(from: message)?.jid?.bareJid { realSender = disclosed roomRealJIDByNickname[roomJID, default: [:]][nickname] = disclosed - } else if let disclosed = room.occupant(nickname: nickname)?.jid?.bareJid { + // } else if let disclosed = room.occupant(nickname: nickname)?.jid?.bareJid { + } else if let disclosed = room?.occupant(nickname: nickname)?.jid?.bareJid { realSender = disclosed roomRealJIDByNickname[roomJID, default: [:]][nickname] = disclosed } else { @@ -1662,7 +1873,8 @@ final class XMPPService { } let payload = contentMessage ?? message - if let state = Self.chatTypingState(payload.chatState ?? message.chatState) { +// if let state = Self.chatTypingState(payload.chatState ?? message.chatState) { + if !isArchived, let state = Self.chatTypingState(payload.chatState ?? message.chatState) { knownChatStatePeers.insert(roomJID) if !outgoing { eventHandler?( @@ -1695,11 +1907,13 @@ final class XMPPService { guard child.name == "reaction" else { return nil } return child.value }, - timestamp: message.delay?.stamp ?? Date(), +// timestamp: message.delay?.stamp ?? Date(), + timestamp: archivedTimestamp ?? message.delay?.stamp ?? Date(), isOutgoing: outgoing, isGroupMessage: true ), - isArchived: false +// isArchived: false + isArchived: isArchived ) return } @@ -1751,14 +1965,17 @@ final class XMPPService { attachmentURL == nil ? visibleBody : Self.displayBody(kind: kind, filename: transportFilename ?? "Вложение") - eventHandler?( - .message( +// eventHandler?( + emitMessage( +// .message( MessageEnvelope( id: id, - peerJID: room.jid.stringValue, +// peerJID: room.jid.stringValue, + peerJID: roomBareJID.stringValue, senderJID: senderJID, body: displayBody, - timestamp: message.delay?.stamp ?? Date(), +// timestamp: message.delay?.stamp ?? Date(), + timestamp: archivedTimestamp ?? message.delay?.stamp ?? Date(), isOutgoing: outgoing, security: security, kind: kind, @@ -1773,11 +1990,14 @@ final class XMPPService { replyToID: reply?.getAttribute("id"), replyToJID: reply?.getAttribute("to"), replyPreview: parsedReply.preview, + originID: message.originId, stanzaID: stanzaID, senderDisplayName: nickname, isGroupMessage: true, - isArchived: false - ))) + // isArchived: false + // ))) + isArchived: isArchived + ), isArchived: isArchived) } private func observe(room: RoomProtocol) { @@ -2009,6 +2229,13 @@ final class XMPPService { fallback.addChild(bodyRange) message.addChild(fallback) } + + private func addOriginID(_ id: String, to message: Message) { + guard message.originId == nil else { return } + let origin = Element(name: "origin-id", xmlns: Self.stanzaIDNamespace) + origin.setAttribute("id", value: id) + message.addChild(origin) + } private static func roomStanzaID(in message: Message, roomJID: BareJID) -> String? { message.children.first { child in @@ -2255,6 +2482,48 @@ final class XMPPService { } return encryptedMessage } + + /// Single Martin-specific MAM entry point. Martin 3.2.4 supports + /// `componentJid:` and writes it to IQ `to`, which is required for MUC MAM. + private func queryMAM( + client: XMPPClient, + archive: MAMArchiveKey, + start: Date?, + queryID: String, + rsm: RSM.Query?, + completion: @escaping (Result) -> Void + ) { + let componentJID: JID? + switch archive.kind { + case .account: + componentJID = nil + case .muc: + componentJID = JID(archive.jid) + } + + client.module(.mam).queryItems( + componentJid: componentJID, + start: start, + queryId: queryID, + rsm: rsm, + completionHandler: completion + ) + } + + private func allowedMAMSources( + for archive: MAMArchiveKey, + client: XMPPClient + ) -> Set { + switch archive.kind { + case .account: + return [ + client.userBareJid.stringValue.lowercased(), + client.userBareJid.domain.lowercased(), + ] + case .muc: + return [archive.jid.lowercased()] + } + } private func startArchiveSyncIfAvailable() { guard let client, @@ -2309,6 +2578,145 @@ final class XMPPService { retry: 0 ) } + + private func enqueueMUCCatchup(roomJID: BareJID) { + let archive = MAMArchiveKey.muc(roomJID.stringValue) + guard activeMUCCatchup != archive, + !pendingMUCCatchups.contains(archive) else { return } + pendingMUCCatchups.append(archive) + startNextMUCCatchupIfPossible() + } + + /// Keep this first version serialized with the account pass. It prevents + /// archive mutations from different namespaces sharing the legacy batch. + /// Patch 0007 makes live delivery archive-aware while catch-up runs. + private func startNextMUCCatchupIfPossible() { + guard activeMUCCatchup == nil, + !archiveSyncStarted, + let client, + client.state == .connected(), + !client.module(.mam).availableVersions.isEmpty, + !pendingMUCCatchups.isEmpty else { return } + + let archive = pendingMUCCatchups.removeFirst() + activeMUCCatchup = archive + mucCatchupHighWatermark = nil + mucCatchupLastCursor = mamCheckpoints[archive]?.cursor + queryMUCArchivePage( + archive: archive, + after: mamCheckpoints[archive]?.cursor, + client: client + ) + } + + private func queryMUCArchivePage( + archive: MAMArchiveKey, + after: String?, + client: XMPPClient + ) { + guard activeMUCCatchup == archive else { return } + let queryID = UUID().uuidString + archiveStanzaInbox.begin( + queryID: queryID, + allowedSources: allowedMAMSources(for: archive, client: client) + ) + + let checkpoint = mamCheckpoints[archive] + let start = after == nil + ? checkpoint?.timestamp.addingTimeInterval(-ArchiveSyncRecoveryPolicy.incrementalOverlap) + : nil + let rsm = checkpoint == nil + ? RSM.Query(lastItems: archiveBootstrapMessageLimit) + : RSM.Query(after: after, max: archivePageSize) + + queryMAM( + client: client, + archive: archive, + start: start, + queryID: queryID, + rsm: rsm + ) { [weak self, weak client] result in + DispatchQueue.main.async { + guard let self, let client, + self.activeMUCCatchup == archive else { return } + let page = self.archiveStanzaInbox.take(queryID: queryID) + guard !page.overflowed, !page.rejectedSource else { + self.finishMUCCatchup(archive: archive, succeeded: false) + return + } + for stanza in page.stanzas { + self.mucCatchupHighWatermark = max( + self.mucCatchupHighWatermark ?? .distantPast, + stanza.timestamp + ) +// guard stanza.message.type == .groupchat, +// let roomJID = stanza.message.from?.bareJid, +// let room = client.module(.muc).roomManager.room( +// for: client, with: roomJID +// ) else { continue } +// self.handle(groupMessage: stanza.message, room: room) + guard stanza.message.type == .groupchat, + let roomJID = stanza.message.from?.bareJid else { continue } + self.handle( + groupMessage: stanza.message, + archivedRoomJID: roomJID, + archivedTimestamp: stanza.timestamp, + archiveID: stanza.archiveID, + isArchived: true + ) + } + + switch result { +// case .failure: +// self.finishMUCCatchup(archive: archive, succeeded: false) + case .failure: + // Same recovery principle as account MAM: a retained local + // UID may have expired on the server. Retry once from the + // durable timestamp overlap, never loop cursor<->time. + if self.mamCheckpoints[archive]?.cursor != nil, + self.mucCursorFallbackArchives.insert(archive).inserted, + let old = self.mamCheckpoints[archive] { + self.mamCheckpoints[archive] = MAMArchiveCheckpoint( + timestamp: old.timestamp, + cursor: nil + ) + self.queryMUCArchivePage( + archive: archive, + after: nil, + client: client + ) + } else { + self.finishMUCCatchup(archive: archive, succeeded: false) + } + case .success(let response): + self.mucCatchupLastCursor = response.rsm?.last ?? self.mucCatchupLastCursor + if !response.complete, let next = response.rsm?.last, next != after { + self.queryMUCArchivePage(archive: archive, after: next, client: client) + } else { + self.finishMUCCatchup(archive: archive, succeeded: true) + } + } + } + } + } + + private func finishMUCCatchup(archive: MAMArchiveKey, succeeded: Bool) { + defer { + activeMUCCatchup = nil + mucCatchupHighWatermark = nil + mucCatchupLastCursor = nil + replayDelayedLive(for: archive) + startNextMUCCatchupIfPossible() + } + guard succeeded else { return } + let checkpoint = MAMArchiveCheckpoint( + timestamp: max(mucCatchupHighWatermark ?? .distantPast, Date()), + cursor: mucCatchupLastCursor + ) + mamCheckpoints[archive] = checkpoint + mucCursorFallbackArchives.remove(archive) + eventHandler?(.mucArchiveSyncCompleted(archive: archive, checkpoint: checkpoint)) + } private func queryArchive(client: XMPPClient, after: String?, retry: Int) { guard archiveSyncStarted, @@ -2327,12 +2735,14 @@ final class XMPPService { archiveStanzaBuffer.removeAll(keepingCapacity: true) archiveBufferOverflowed = false archiveRejectedSource = false + let accountArchive = MAMArchiveKey.account(client.userBareJid.stringValue) archiveStanzaInbox.begin( queryID: queryID, - allowedSources: [ - client.userBareJid.stringValue, - client.userBareJid.domain, - ] +// allowedSources: [ +// client.userBareJid.stringValue, +// client.userBareJid.domain, +// ] + allowedSources: allowedMAMSources(for: accountArchive, client: client) ) scheduleArchiveQueryTimeout(client: client, queryID: queryID) let start = @@ -2346,9 +2756,12 @@ final class XMPPService { archiveIsBootstrapQuery ? RSM.Query(lastItems: archiveBootstrapMessageLimit) : RSM.Query(after: after, max: archivePageSize) - client.module(.mam).queryItems( +// client.module(.mam).queryItems( + queryMAM( + client: client, + archive: accountArchive, start: start, - queryId: queryID, + queryID: queryID, rsm: resultSet ) { [weak self, weak client] result in // XEP-0313 guarantees that the final IQ follows every result. The @@ -2431,13 +2844,21 @@ final class XMPPService { // isArchived: true // ) if stanza.message.type == .groupchat, - let roomJID = stanza.message.from?.bareJid, - let room = client?.module(.muc).roomManager.room(for: client!, with: roomJID) +// let roomJID = stanza.message.from?.bareJid, +// let room = client?.module(.muc).roomManager.room(for: client!, with: roomJID) + let roomJID = stanza.message.from?.bareJid { // Some servers include MUC messages in the account MAM. // Route them through the same group handler instead of // dropping them in the direct-message handler. - handle(groupMessage: stanza.message, room: room) +// handle(groupMessage: stanza.message, room: room) + handle( + groupMessage: stanza.message, + archivedRoomJID: roomJID, + archivedTimestamp: stanza.timestamp, + archiveID: stanza.archiveID, + isArchived: true + ) } else { handle( message: stanza.message, @@ -2779,6 +3200,8 @@ final class XMPPService { archiveResumeAfter = nil archiveRetrySuppressedUntilActivation = false eventHandler?(.archiveSyncCompleted(resolvedCheckpoint)) + replayDelayedLive(for: .account(client.userBareJid.stringValue)) + startNextMUCCatchupIfPossible() return }