added .read state

Sources/Shared/Models/AppModel.swift

@@ -113,6 +113,17 @@ final class AppModel: ObservableObject {
     private var pendingRetractions: [String: XMPPService.RetractionEnvelope] = [:]
     private var pendingReactions: [String: XMPPService.ReactionEnvelope] = [:]
     private var correctionReceiptTargets: [String: String] = [:]
+
+    private struct PendingReadMarker {
+        var envelope: ReadStateSync.Envelope
+        var cameFromPeer: Bool
+    }
+    /// Read markers that arrived before the message they reference. MAM
+    /// replays newest stanzas first, so a read-marker is routinely applied
+    /// before its message on a fresh install; the marker waits here until the
+    /// message itself is upserted.
+    private var pendingReadMarkers: [String: PendingReadMarker] = [:]
+    private var lastDisplayedMarkerByConversation: [String: String] = [:]
     private var localChatStateByConversation: [String: ChatTypingState] = [:]
     private var localTypingPauseTasks: [String: Task<Void, Never>] = [:]
     private var remoteTypingExpiryTasks: [String: Task<Void, Never>] = [:]
@@ -730,6 +741,7 @@ final class AppModel: ObservableObject {
         if let index = conversations.firstIndex(where: { $0.jid == normalized }) {
             conversations[index].unreadCount = 0
         }
+        sendDisplayedMarkerIfNeeded(for: normalized)
         if addToRoster {
             rosterContactJIDs.insert(normalized)
             xmpp.addToRoster(jid: normalized, name: name)
@@ -747,6 +759,7 @@ final class AppModel: ObservableObject {
         if let index = conversations.firstIndex(where: { $0.jid == normalized }) {
             conversations[index].unreadCount = 0
         }
+        sendDisplayedMarkerIfNeeded(for: normalized)
         schedulePersist()
         syncWatch()
         if hasMoreOlderHistoryByConversation[normalized] == false,
@@ -2197,6 +2210,25 @@ final class AppModel: ObservableObject {
         case .delivered(let messageID):
             let targetID = correctionReceiptTargets.removeValue(forKey: messageID) ?? messageID
             updateMessage(id: targetID) { $0.delivery = .delivered }
+        case .read(let conversationID, let messageID):
+            // Peer-originated read receipt (XEP-0333 displayed marker): apply
+            // it locally and broadcast it to the user's other devices through
+            // a self-addressed read-marker message.
+            let targetID = correctionReceiptTargets.removeValue(forKey: messageID) ?? messageID
+            applyReadState(
+                ReadStateSync.Envelope(
+                    id: UUID().uuidString,
+                    conversationJID: conversationID,
+                    messageID: targetID,
+                    stanzaID: nil,
+                    timestamp: Date()
+                ),
+                cameFromPeer: true
+            )
+        case .readState(let envelope):
+            // Read-marker from another of the user's devices (Carbons live or
+            // MAM replay): apply without re-broadcasting.
+            applyReadState(envelope, cameFromPeer: false)
         case .omemo(let ready, let fingerprint):
             isOMEMOReady = ready
             ownFingerprint = fingerprint
@@ -2413,6 +2445,112 @@ final class AppModel: ObservableObject {
         schedulePersist()
     }
 
+    // MARK: - Read receipts (XEP-0333) and cross-device read-state sync
+
+    private func applyReadState(_ envelope: ReadStateSync.Envelope, cameFromPeer: Bool) {
+        let conversationID = envelope.conversationJID.lowercased()
+        if let index =
+            messageIndex(originID: envelope.messageID, conversationID: conversationID)
+            ?? messageIndex(referenceID: envelope.messageID, conversationID: conversationID)
+            ?? envelope.stanzaID.flatMap({
+                messageIndex(referenceID: $0, conversationID: conversationID)
+            })
+        {
+            markRead(at: index, cameFromPeer: cameFromPeer)
+            return
+        }
+        // The marker raced ahead of the message (typical for MAM replays,
+        // which arrive newest first). Apply it when the message lands.
+        stashPendingReadMarker(envelope, cameFromPeer: cameFromPeer)
+    }
+
+    /// Promotes a matching outgoing message to `.read` and, when the receipt
+    /// came from the peer (not from another own device), broadcasts the state
+    /// to the user's other devices via Carbons/MAM. Failed messages are never
+    /// promoted by a receipt.
+    private func markRead(at index: Int, cameFromPeer: Bool) {
+        let message = messages[index]
+        guard message.direction == .outgoing,
+            message.delivery != .read,
+            message.delivery != .failed,
+            !message.isRetracted
+        else { return }
+        messages[index].delivery = .read
+        schedulePersist()
+        guard cameFromPeer else { return }
+        xmpp.syncReadState(
+            ReadStateSync.Envelope(
+                id: UUID().uuidString,
+                conversationJID: message.conversationID,
+                messageID: message.originID ?? message.clientID,
+                stanzaID: message.stanzaID,
+                timestamp: Date()
+            ))
+    }
+
+    private func stashPendingReadMarker(_ envelope: ReadStateSync.Envelope, cameFromPeer: Bool) {
+        // The marker may reference the message through its origin-id or its
+        // archived stanza-id; index it under every reference the arriving
+        // message can expose so `applyPendingReadState` always finds it.
+        let references = Set([envelope.messageID, envelope.stanzaID].compactMap { $0 })
+        for reference in references {
+            let key = pendingReadMarkerKey(
+                conversationID: envelope.conversationJID,
+                reference: reference
+            )
+            // A peer marker must keep its broadcast flag: a device-sync
+            // envelope must not silently shadow it.
+            if pendingReadMarkers[key]?.cameFromPeer == true { continue }
+            pendingReadMarkers[key] = PendingReadMarker(envelope: envelope, cameFromPeer: cameFromPeer)
+        }
+        if pendingReadMarkers.count > 200 {
+            if let oldestKey = pendingReadMarkers.min(by: {
+                $0.value.envelope.timestamp < $1.value.envelope.timestamp
+            })?.key {
+                pendingReadMarkers.removeValue(forKey: oldestKey)
+            }
+        }
+    }
+
+    private func pendingReadMarkerKey(conversationID: String, reference: String) -> String {
+        "\(conversationID.lowercased())|\(reference)"
+    }
+
+    /// Applies any read marker that arrived before the message it references.
+    /// Called from `upsertMessage` once the message itself is in the array.
+    private func applyPendingReadState(for message: ChatMessage, at index: Int) {
+        guard message.direction == .outgoing else { return }
+        let references = Set([message.clientID, message.originID, message.stanzaID].compactMap { $0 })
+        for reference in references {
+            let key = pendingReadMarkerKey(
+                conversationID: message.conversationID,
+                reference: reference
+            )
+            guard let pending = pendingReadMarkers.removeValue(forKey: key) else { continue }
+            markRead(at: index, cameFromPeer: pending.cameFromPeer)
+        }
+    }
+
+    /// Sends an XEP-0333 `<displayed/>` marker for the newest incoming 1:1
+    /// message once the chat is opened, so the peer's client can show it as
+    /// read. MUC rooms never get markers (XEP-0333); deduped per message so
+    /// repeated SwiftUI appearances do not resend.
+    private func sendDisplayedMarkerIfNeeded(for conversationID: String) {
+        guard let account else { return }
+        let normalized = conversationID.lowercased()
+        guard normalized != account.normalizedJID else { return }
+        guard let latest = messages
+            .filter({ $0.conversationID == normalized && $0.direction == .incoming })
+            .max(by: { $0.timestamp < $1.timestamp }),
+            !latest.isGroupMessage,
+            !latest.isRetracted,
+            latest.kind != .system
+        else { return }
+        guard lastDisplayedMarkerByConversation[normalized] != latest.clientID else { return }
+        lastDisplayedMarkerByConversation[normalized] = latest.clientID
+        xmpp.sendDisplayedMarker(messageID: latest.clientID, to: normalized)
+    }
+
     private func recordCallHistory(_ entry: CallHistoryEntry) {
         guard let account else { return }
         let direction: ChatMessage.Direction =
@@ -2868,9 +3006,7 @@ final class AppModel: ObservableObject {
         ) {
             let previous = messages[index]
             if previous.isRetracted {
-                if message.delivery == .delivered {
-                    messages[index].delivery = .delivered
-                }
+                messages[index].delivery = previous.delivery.merged(with: message.delivery)
                 updateConversationPreview(for: messages[index], incrementUnread: false)
                 return false
             }
@@ -2880,9 +3016,9 @@ final class AppModel: ObservableObject {
                 merged.mimeType = previous.mimeType ?? message.mimeType
                 merged.duration = previous.duration ?? message.duration
                 merged.byteCount = previous.byteCount ?? message.byteCount
-                if previous.delivery == .delivered {
-                    merged.delivery = .delivered
-                }
+                // MAM/MUC-MAM replays and Carbon copies carry `.sent`; a
+                // locally delivered or read copy must never be downgraded.
+                merged.delivery = previous.delivery.merged(with: message.delivery)
             }
             merged.replyToID = message.replyToID ?? previous.replyToID
             merged.replyToJID = message.replyToJID ?? previous.replyToJID
@@ -2944,6 +3080,7 @@ final class AppModel: ObservableObject {
                         conversationID: merged.conversationID
                     )] = index
             }
+            applyPendingReadState(for: merged, at: index)
             updateConversationPreview(for: merged, incrementUnread: false)
             return false
         }
@@ -2975,6 +3112,7 @@ final class AppModel: ObservableObject {
         let shouldIncrement =
             unreadOverride
             ?? (message.direction == .incoming && selectedConversationID != message.conversationID)
+        applyPendingReadState(for: message, at: insertedIndex)
         updateConversationPreview(for: message, incrementUnread: shouldIncrement)
         return true
     }

Sources/Shared/Models/ChatMessage.swift

@@ -14,6 +14,27 @@ final class ChatMessage {
         case delivered
         case failed
         case read
+
+        /// Monotonic promotion used when the same outgoing message is replayed
+        /// from MAM/MUC-MAM or Carbons: a copy that was already delivered or
+        /// read locally must never be downgraded by an archive replay carrying
+        /// `.sent`. `.failed` is sticky in both directions — an archive
+        /// replay must not silently "fix" a failed send; the user resends it
+        /// explicitly.
+        func merged(with other: Delivery) -> Delivery {
+            guard self != .failed, other != .failed else { return .failed }
+            return other.rank > self.rank ? other : self
+        }
+
+        private var rank: Int {
+            switch self {
+            case .sending: return 0
+            case .sent: return 1
+            case .delivered: return 2
+            case .read: return 3
+            case .failed: return -1
+            }
+        }
     }
 
     enum Security: String, Codable {
@@ -136,7 +157,7 @@ final class ChatMessage {
             && !isGroupMessage
             && direction == .outgoing
             && kind == .text
-            && (delivery == .sent || delivery == .delivered)
+            && (delivery == .sent || delivery == .delivered || delivery == .read)
     }
 
     var isRetracted: Bool {
@@ -158,7 +179,7 @@ final class ChatMessage {
             && !isGroupMessage
             && !isRetracted
             && kind != .system
-            && (delivery == .sent || delivery == .delivered)
+            && (delivery == .sent || delivery == .delivered || delivery == .read)
     }
 
     var replyIdentifier: String? {

Sources/Shared/XMPP/ReadStateSync.swift

@@ -0,0 +1,74 @@
+import Foundation
+import Martin
+
+/// Service messages that sync read receipts between the user's own devices.
+/// When a device marks an outgoing message as read (the peer's XEP-0333
+/// <displayed/> marker arrived there) it sends a message to its own bare JID
+/// carrying a `<read-marker xmlns='https://luma.chat/read-marker'>` payload;
+/// Carbons deliver it to the other devices and MAM archives it, so a fresh
+/// install and every later archive replay converge on the same read state.
+/// Receivers upsert the state on the matching outgoing message, deduplicating
+/// by the marker id.
+///
+/// MUC read state never leaks into the room archive: the marker is a plain
+/// self-addressed message whose payload references the room JID and the
+/// archived stanza-id of the room message. Group conversations are matched
+/// through the MUC-MAM stanza-id, 1:1 chats through the origin-id.
+enum ReadStateSync {
+    static let namespace = "https://luma.chat/read-marker"
+
+    struct Envelope: Equatable, Sendable {
+        /// Unique marker id; matches the message origin-id so Carbons and MAM
+        /// copies deduplicate into a single state application.
+        let id: String
+        /// Bare JID of the conversation holding the read message: a peer or
+        /// a MUC room.
+        let conversationJID: String
+        /// Origin-id (clientID) of the read message on the sending device.
+        let messageID: String
+        /// Archived stanza-id of the read message (1:1 archive or MUC-MAM);
+        /// group messages are matched through it on the receiving devices.
+        let stanzaID: String?
+        let timestamp: Date
+    }
+
+    static func envelope(from message: Message) -> Envelope? {
+        guard let element = message.element.findChild(
+            name: "read-marker",
+            xmlns: namespace
+        ) else {
+            return nil
+        }
+        let value = { (name: String) -> String? in
+            element.findChild(name: name)?.value
+        }
+        guard let rawConversation = value("conversation"), !rawConversation.isEmpty,
+            let rawMessage = value("id"), !rawMessage.isEmpty
+        else { return nil }
+        let rawStanza = value("stanza")
+        return Envelope(
+            id: message.originId ?? message.id ?? UUID().uuidString,
+            conversationJID: rawConversation.lowercased(),
+            messageID: rawMessage,
+            stanzaID: (rawStanza?.isEmpty == false) ? rawStanza : nil,
+            timestamp: value("when").flatMap { timestampFormatter.date(from: $0) } ?? Date()
+        )
+    }
+
+    static func payloadElement(marker: Envelope) -> Element {
+        let element = Element(name: "read-marker", xmlns: namespace)
+        element.addChild(Element(name: "conversation", cdata: marker.conversationJID.lowercased()))
+        element.addChild(Element(name: "id", cdata: marker.messageID))
+        if let stanzaID = marker.stanzaID, !stanzaID.isEmpty {
+            element.addChild(Element(name: "stanza", cdata: stanzaID))
+        }
+        element.addChild(Element(name: "when", cdata: timestampFormatter.string(from: marker.timestamp)))
+        return element
+    }
+
+    private static let timestampFormatter: ISO8601DateFormatter = {
+        let formatter = ISO8601DateFormatter()
+        formatter.formatOptions = [.withInternetDateTime]
+        return formatter
+    }()
+}

Sources/Shared/XMPP/XMPPService.swift

@@ -248,6 +248,7 @@ final class XMPPService {
         case roomInvitation(RoomInvitationEnvelope)
         case avatar(jid: String, data: Data)
         case delivered(messageID: String)
+        case read(conversationJID: String, messageID: String)
         case omemo(ready: Bool, ownFingerprint: String?)
         case archiveBatch([ArchiveMutation])
         case archiveSyncing(Bool)
@@ -256,6 +257,7 @@ final class XMPPService {
         case call(CallSnapshot?)
         case callHistory(CallHistoryEntry)
         case callHistorySync(CallHistorySync.Envelope)
+        case readState(ReadStateSync.Envelope)
         case callError(String)
         case recoverableError(String)
     }
@@ -816,6 +818,41 @@ final class XMPPService {
             .info("sent call-history id=\(entry.id) peer=\(entry.peerJID) status=\(entry.outcome.rawValue)")
     }
 
+    /// Sends a `<read-marker/>` service message to the user's own bare JID
+    /// so Carbons deliver it to the other devices and MAM archives it. The
+    /// message is plaintext and never reaches the chat counterpart; MUC read
+    /// state rides the same self-addressed envelope and never enters the room
+    /// archive.
+    func syncReadState(_ marker: ReadStateSync.Envelope) {
+        guard let client, client.state == .connected(), let account else { return }
+        let message = Message()
+        message.to = JID(account.normalizedJID)
+        message.type = .chat
+        message.id = marker.id
+        message.body = Self.readMarkerBody
+        addOriginID(marker.id, to: message)
+        message.addChild(ReadStateSync.payloadElement(marker: marker))
+        client.context.writer.write(message, writeCompleted: nil)
+        Logger(subsystem: "Luma", category: "read-sync")
+            .info("sent read-marker id=\(marker.id) message=\(marker.messageID) conversation=\(marker.conversationJID)")
+    }
+
+    /// Sends an XEP-0333 `<displayed/>` chat marker so the peer's client can
+    /// show the message as read. Markers are never sent to MUC rooms
+    /// (XEP-0333 forbids them there); the read state of outgoing MUC messages
+    /// only changes when a member's client sends a marker anyway or when
+    /// another of the user's devices broadcasts a read-marker sync message.
+    func sendDisplayedMarker(messageID: String, to peerJID: String) {
+        guard let client, client.state == .connected() else { return }
+        let message = Message()
+        message.to = JID(peerJID)
+        message.type = .chat
+        message.chatMarkers = .displayed(id: messageID)
+        client.context.writer.write(message, writeCompleted: nil)
+    }
+
+    private static let readMarkerBody = "Сообщение прочитано"
+
     /// Detects OMEMO 2 (`urn:xmpp:omemo:2`) payloads, which the pinned
     /// MartinOMEMO version cannot decrypt (it only speaks the legacy
     /// `eu.siacs.conversations.axolotl` namespace).
@@ -2105,6 +2142,17 @@ final class XMPPService {
             }
             .store(in: &cancellables)
 
+        // XEP-0333 chat markers. Martin republishes markers found on live
+        // stanzas, Carbons copies and archived messages through one publisher,
+        // so `<displayed/>` markers the server carbon-copies to the user's
+        // other devices reach this handler there as well.
+        client.module(.chatMarkers).markersPublisher
+            .receive(on: DispatchQueue.main)
+            .sink { [weak self] received in
+                self?.handleChatMarker(received)
+            }
+            .store(in: &cancellables)
+
         // Do not enqueue every archived stanza onto DispatchQueue.main. Martin
         // publishes all results before the final IQ callback, so a locked inbox
         // can receive the whole page on the parser queue and hand it to the
@@ -2182,6 +2230,41 @@ final class XMPPService {
         }
     }
 
+    /// XEP-0333 chat markers: `<displayed/>` marks an outgoing message as read,
+    /// `<received/>` as delivered. MUC markers are best effort — few clients
+    /// send them, but the room stanza-id maps them to the right outgoing
+    /// message.
+    private func handleChatMarker(_ received: ChatMarkersModule.ChatMarkerReceived) {
+        let message = received.message
+        guard message.type != .error else { return }
+        switch received.marker {
+        case .displayed(let id):
+            guard let conversationJID = markerConversationJID(for: message) else { return }
+            eventHandler?(.read(conversationJID: conversationJID, messageID: id))
+        case .received(let id):
+            eventHandler?(.delivered(messageID: id))
+        case .acknowledged:
+            break
+        }
+    }
+
+    /// Bare JID of the conversation a marker references. Peer markers arrive
+    /// with the peer in `from`; our own displayed markers carbon-copied from
+    /// another device carry our JID in `from` and the peer in `to`. Groupchat
+    /// markers reference the room.
+    private func markerConversationJID(for message: Message) -> String? {
+        if message.type == .groupchat {
+            return message.from?.bareJid.stringValue
+        }
+        if let from = message.from?.bareJid {
+            if let client, from == client.userBareJid {
+                return message.to?.bareJid.stringValue ?? from.stringValue
+            }
+            return from.stringValue
+        }
+        return message.to?.bareJid.stringValue
+    }
+
     /// Runs OMEMO decryption off the main actor and resumes with the raw
     /// result. Kept as a thin wrapper so `handle` keeps its existing switch
     /// logic and only the expensive `decode` call leaves the main thread.
@@ -2244,6 +2327,15 @@ final class XMPPService {
             eventHandler?(.callHistorySync(sync))
             return
         }
+        // Read-marker service messages sync read receipts between the user's
+        // devices the same way: Carbons deliver them live, MAM replays them
+        // after reinstalls. They never become ordinary chat messages.
+        if let marker = ReadStateSync.envelope(from: message) {
+            Logger(subsystem: "Luma", category: "read-sync")
+                .info("received read-marker id=\(marker.id) message=\(marker.messageID) conversation=\(marker.conversationJID)")
+            eventHandler?(.readState(marker))
+            return
+        }
         guard message.type != .error,
             message.type != .groupchat,
             let client,