permissionlesstech/bitchat
1
0
Fork 0
mirror of https://github.com/permissionlesstech/bitchat.git synced 2026-07-26 16:45:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix live DM handoff across transport aliases

Browse Source
This commit is contained in:
jack
2026-07-25 23:49:42 +02:00
committed by jack
parent c056a46299
commit f6abd10fb5
6 changed files with 558 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bitchat/Services/Courier/MessageOutboxStore.swift
+61 -9
View File
@@ -110,6 +110,10 @@ final class MessageOutboxStore {
/// Delivery/read acknowledgments received before a deferred cold-load
/// reveals the durable queue. Applied to every merge before persistence.
private var pendingRemovalMessageIDs = Set<String>()
/// Peer-scoped acknowledgments received before a deferred cold-load
/// reveals the durable queue. Unlike the legacy global tombstones above,
/// these must not remove a colliding message ID queued for another peer.
private var pendingScopedRemovalMessageIDs: [PeerID: Set<String>] = [:]
private var recoveryHandler: (@MainActor (Snapshot) -> Void)?
/// Recovery loaded durable state that MessageRouter has not merged yet.
/// While true, router saves must union with `cachedSnapshot` instead of
@@ -195,7 +199,7 @@ final class MessageOutboxStore {
? (pendingSnapshot ?? [:])
: Self.merge(durable, pendingSnapshot ?? [:]))
diskState = .loaded
if pendingSnapshot != nil || !pendingRemovalMessageIDs.isEmpty {
if pendingSnapshot != nil || hasPendingRemovalsLocked {
if persistSnapshotAndClearRemovalsLocked(cachedSnapshot) {
pendingSnapshot = nil
pendingSnapshotIsAuthoritative = false
@@ -212,7 +216,7 @@ final class MessageOutboxStore {
case .missing:
cachedSnapshot = applyingPendingRemovalsLocked(pendingSnapshot ?? [:])
diskState = .loaded
if pendingSnapshot != nil || !pendingRemovalMessageIDs.isEmpty {
if pendingSnapshot != nil || hasPendingRemovalsLocked {
if persistSnapshotAndClearRemovalsLocked(cachedSnapshot) {
pendingSnapshot = nil
pendingSnapshotIsAuthoritative = false
@@ -235,7 +239,7 @@ final class MessageOutboxStore {
diskState = .loaded
cachedSnapshot = applyingPendingRemovalsLocked(pendingSnapshot ?? [:])
SecureLogger.error("Failed to decode encrypted outbox: \(error)", category: .session)
if pendingSnapshot != nil || !pendingRemovalMessageIDs.isEmpty {
if pendingSnapshot != nil || hasPendingRemovalsLocked {
if persistSnapshotAndClearRemovalsLocked(cachedSnapshot) {
pendingSnapshot = nil
pendingSnapshotIsAuthoritative = false
@@ -442,6 +446,25 @@ final class MessageOutboxStore {
lock.unlock()
}
/// Records an ack for only the supplied peer aliases. This preserves
/// another recipient's queued entry when message IDs happen to collide,
/// including while the durable snapshot is hidden by protected data.
func recordRemoval(messageID: String, for peerIDs: Set<PeerID>) {
guard !peerIDs.isEmpty else { return }
lock.lock()
for peerID in peerIDs {
pendingScopedRemovalMessageIDs[peerID, default: []].insert(messageID)
}
cachedSnapshot = Self.removing(pendingScopedRemovalMessageIDs, from: cachedSnapshot)
unseenRecoveredSnapshot = Self.removing(pendingScopedRemovalMessageIDs, from: unseenRecoveredSnapshot)
recoveryRouterSnapshot = Self.removing(pendingScopedRemovalMessageIDs, from: recoveryRouterSnapshot)
if let pendingSnapshot {
self.pendingSnapshot = Self.removing(pendingScopedRemovalMessageIDs, from: pendingSnapshot)
}
lock.unlock()
}
/// Retries a deferred protected-data load. The returned snapshot includes
/// both durable messages and any messages queued during the locked wake.
@discardableResult
@@ -478,7 +501,7 @@ final class MessageOutboxStore {
: (pendingSnapshotIsAuthoritative ? known : Self.merge(durable, known)))
cachedSnapshot = merged
diskState = .loaded
if (pendingSnapshot == nil && pendingRemovalMessageIDs.isEmpty) ||
if (pendingSnapshot == nil && !hasPendingRemovalsLocked) ||
persistSnapshotAndClearRemovalsLocked(merged) {
pendingSnapshot = nil
pendingSnapshotIsAuthoritative = false
@@ -507,7 +530,7 @@ final class MessageOutboxStore {
: known)
cachedSnapshot = merged
diskState = .loaded
if (pendingSnapshot == nil && pendingRemovalMessageIDs.isEmpty) ||
if (pendingSnapshot == nil && !hasPendingRemovalsLocked) ||
persistSnapshotAndClearRemovalsLocked(merged) {
pendingSnapshot = nil
pendingSnapshotIsAuthoritative = false
@@ -539,7 +562,7 @@ final class MessageOutboxStore {
: known)
cachedSnapshot = merged
diskState = .loaded
if (pendingSnapshot == nil && pendingRemovalMessageIDs.isEmpty) ||
if (pendingSnapshot == nil && !hasPendingRemovalsLocked) ||
persistSnapshotAndClearRemovalsLocked(merged) {
pendingSnapshot = nil
pendingSnapshotIsAuthoritative = false
@@ -578,6 +601,7 @@ final class MessageOutboxStore {
pendingSnapshot = nil
pendingSnapshotIsAuthoritative = false
pendingRemovalMessageIDs.removeAll()
pendingScopedRemovalMessageIDs.removeAll()
recoveryDeliveryPending = false
unseenRecoveryPendingPersistence = false
unseenRecoveredSnapshot = [:]
@@ -742,12 +766,21 @@ final class MessageOutboxStore {
private func persistSnapshotAndClearRemovalsLocked(_ snapshot: Snapshot) -> Bool {
guard persistSnapshotLocked(snapshot) else { return false }
pendingRemovalMessageIDs.removeAll()
pendingScopedRemovalMessageIDs.removeAll()
return true
}
/// Must be called with `lock` held.
private func applyingPendingRemovalsLocked(_ snapshot: Snapshot) -> Snapshot {
Self.removing(pendingRemovalMessageIDs, from: snapshot)
Self.removing(
pendingScopedRemovalMessageIDs,
from: Self.removing(pendingRemovalMessageIDs, from: snapshot)
)
}
/// Must be read with `lock` held.
private var hasPendingRemovalsLocked: Bool {
!pendingRemovalMessageIDs.isEmpty || !pendingScopedRemovalMessageIDs.isEmpty
}
private static func removing(_ messageIDs: Set<String>, from snapshot: Snapshot) -> Snapshot {
@@ -760,9 +793,28 @@ final class MessageOutboxStore {
return filtered
}
private static func removing(
_ messageIDsByPeer: [PeerID: Set<String>],
from snapshot: Snapshot
) -> Snapshot {
guard !messageIDsByPeer.isEmpty else { return snapshot }
var filtered = snapshot
for (peerID, messageIDs) in messageIDsByPeer {
guard !messageIDs.isEmpty, let queue = filtered[peerID] else { continue }
let remaining = queue.filter { !messageIDs.contains($0.messageID) }
filtered[peerID] = remaining.isEmpty ? nil : remaining
}
return filtered
}
private static func excludingKnownMessages(from durable: Snapshot, known: Snapshot) -> Snapshot {
let knownIDs = Set(known.values.flatMap { $0.map(\.messageID) })
return removing(knownIDs, from: durable)
var unseen: Snapshot = [:]
for (peerID, durableQueue) in durable {
let knownIDs = Set(known[peerID]?.map(\.messageID) ?? [])
let remaining = durableQueue.filter { !knownIDs.contains($0.messageID) }
if !remaining.isEmpty { unseen[peerID] = remaining }
}
return unseen
}
private static func merge(_ durable: Snapshot, _ pending: Snapshot) -> Snapshot {
bitchat/Services/MessageRouter.swift
+26
View File
@@ -385,6 +385,32 @@ final class MessageRouter {
persistOutbox()
}
/// A delivery or read ack authenticated to one account confirms receipt
/// only for that account's transport aliases. A colliding message ID
/// queued for another peer must remain retained.
@discardableResult
func markDelivered(_ messageID: String, for peerAliases: [PeerID]) -> Bool {
let peerIDs = Set(peerAliases)
guard !peerIDs.isEmpty else { return false }
var cleared = false
for peerID in peerIDs {
guard let queue = outbox[peerID] else { continue }
let filtered = queue.filter { $0.messageID != messageID }
guard filtered.count != queue.count else { continue }
outbox[peerID] = filtered.isEmpty ? nil : filtered
cleared = true
}
// Preserve the scoped ack even when protected data hides the durable
// queue during a cold launch.
outboxStore?.recordRemoval(messageID: messageID, for: peerIDs)
if cleared {
metrics?.record(.outboxDelivered)
}
persistOutbox()
return cleared
}
private func enqueue(_ message: QueuedMessage, for peerID: PeerID) {
var message = message
var queue = outbox[peerID] ?? []
bitchat/ViewModels/ChatPrivateConversationCoordinator.swift
+109 -13
View File
@@ -85,6 +85,11 @@ protocol ChatPrivateConversationContext: AnyObject {
func routePrivateMessage(_ content: String, to peerID: PeerID, recipientNickname: String, messageID: String)
@discardableResult
func routeReadReceipt(_ receipt: ReadReceipt, to peerID: PeerID) -> Bool
/// Confirms an authenticated delivery/read acknowledgement so the router
/// stops retaining the original private message for resend. The peer
/// aliases scope the removal to the authenticated sender.
@discardableResult
func markMessageDelivered(_ messageID: String, for peerIDs: [PeerID]) -> Bool
func sendMeshReadReceipt(_ receipt: ReadReceipt, to peerID: PeerID)
@discardableResult
func sendGeohashPrivateMessage(
@@ -172,6 +177,11 @@ extension ChatViewModel: ChatPrivateConversationContext {
messageRouter.sendReadReceipt(receipt, to: peerID)
}
@discardableResult
func markMessageDelivered(_ messageID: String, for peerIDs: [PeerID]) -> Bool {
messageRouter.markDelivered(messageID, for: peerIDs)
}
func routeFavoriteNotification(to peerID: PeerID, isFavorite: Bool) {
messageRouter.sendFavoriteNotification(to: peerID, isFavorite: isFavorite)
}
@@ -269,6 +279,60 @@ final class ChatPrivateConversationCoordinator {
return true
}
/// Account DMs can arrive under the authenticated peer's full Noise key
/// while an existing mesh conversation is keyed by its derived short ID.
/// These are the only aliases we may safely join: the short ID is derived
/// directly from the authenticated key, rather than guessed from a
/// nickname or found by scanning unrelated chats.
private func accountConversationAliases(for peerID: PeerID) -> [PeerID] {
guard peerID.noiseKey != nil else { return [peerID] }
let shortPeerID = peerID.toShort()
return shortPeerID == peerID ? [peerID] : [peerID, shortPeerID]
}
/// Keeps a connected account DM on its short routing ID and an offline DM
/// on its stable Noise-key ID, folding the other authenticated alias into
/// it and handing an open sheet across without closing it.
private func consolidateAccountConversationAliases(for peerID: PeerID) -> PeerID {
let aliases = accountConversationAliases(for: peerID)
guard aliases.count > 1 else { return peerID }
let shortPeerID = peerID.toShort()
let targetPeerID = context.isPeerConnected(shortPeerID) ? shortPeerID : peerID
let sourcePeerIDs = aliases.filter { $0 != targetPeerID }
for sourcePeerID in sourcePeerIDs where !context.privateMessages(for: sourcePeerID).isEmpty {
context.migratePrivateChat(from: sourcePeerID, to: targetPeerID)
// ConversationStore deliberately preserves message values during a
// generic migration, including its destination-wins rule for
// duplicate IDs. Rewrite the resulting canonical copies so later
// read-receipt scans compare against the new routing key without
// replacing a newer destination value with the source snapshot.
let canonicalMessages = context.privateMessages(for: targetPeerID)
for message in canonicalMessages where message.senderPeerID == sourcePeerID {
context.upsertPrivateMessage(
BitchatMessage(
id: message.id,
sender: message.sender,
content: message.content,
timestamp: message.timestamp,
isRelay: message.isRelay,
originalSender: message.originalSender,
isPrivate: message.isPrivate,
recipientNickname: message.recipientNickname,
senderPeerID: targetPeerID,
mentions: message.mentions,
deliveryStatus: message.deliveryStatus,
isBridged: message.isBridged
),
in: targetPeerID
)
}
}
context.handOffSelectedPrivateChat(from: sourcePeerIDs, to: targetPeerID)
return targetPeerID
}
func sendPrivateMessage(_ content: String, to peerID: PeerID) {
guard !content.isEmpty else { return }
@@ -491,6 +555,8 @@ final class ChatPrivateConversationCoordinator {
return
}
let conversationPeerID = consolidateAccountConversationAliases(for: convKey)
if context.privateChatsContainMessage(withID: messageId) { return }
let message = BitchatMessage(
@@ -501,18 +567,18 @@ final class ChatPrivateConversationCoordinator {
isRelay: false,
isPrivate: true,
recipientNickname: context.nickname,
senderPeerID: convKey,
senderPeerID: conversationPeerID,
deliveryStatus: .delivered(to: context.nickname, at: Date())
)
context.appendPrivateMessage(message, to: convKey)
context.appendPrivateMessage(message, to: conversationPeerID)
let isViewing = context.selectedPrivateChatPeer == convKey
let isViewing = context.selectedPrivateChatPeer == conversationPeerID
let wasReadBefore = context.sentReadReceipts.contains(messageId)
let isRecentMessage = Date().timeIntervalSince(messageTimestamp) < 30
let shouldMarkUnread = !wasReadBefore && !isViewing && isRecentMessage
if shouldMarkUnread {
context.markPrivateChatUnread(convKey)
context.markPrivateChatUnread(conversationPeerID)
}
if isViewing {
@@ -520,7 +586,7 @@ final class ChatPrivateConversationCoordinator {
}
if !isViewing && shouldMarkUnread {
context.notifyPrivateMessage(from: senderName, message: pm.content, peerID: convKey)
context.notifyPrivateMessage(from: senderName, message: pm.content, peerID: conversationPeerID)
}
context.notifyUIChanged()
@@ -529,17 +595,32 @@ final class ChatPrivateConversationCoordinator {
func handleDelivered(_ payload: NoisePayload, senderPubkey: String, convKey: PeerID) {
guard let messageID = String(data: payload.data, encoding: .utf8) else { return }
if context.privateChat(convKey, containsMessageWithID: messageID) {
context.setPrivateDeliveryStatus(
let aliases = accountConversationAliases(for: convKey)
let clearedRetainedMessage = convKey.noiseKey != nil
? context.markMessageDelivered(messageID, for: aliases)
: false
let hasConversationMessage = aliases.contains {
context.privateChat($0, containsMessageWithID: messageID)
}
if hasConversationMessage {
let conversationPeerID = consolidateAccountConversationAliases(for: convKey)
let didChange = context.setPrivateDeliveryStatus(
.delivered(to: context.displayNameForNostrPubkey(senderPubkey), at: Date()),
forMessageID: messageID,
peerID: convKey
peerID: conversationPeerID
)
context.notifyUIChanged()
if didChange {
context.notifyUIChanged()
}
SecureLogger.info(
"GeoDM: recv DELIVERED for mid=\(messageID.prefix(8))… from=\(senderPubkey.prefix(8))…",
category: .session
)
} else if clearedRetainedMessage {
SecureLogger.debug(
"GeoDM: recv DELIVERED for cleared mid=\(messageID.prefix(8))… from=\(senderPubkey.prefix(8))…",
category: .session
)
} else {
// A stale ack for a message this device no longer tracks (dropped
// outbox entry, cleared chat, or a peer re-acking after losing our
@@ -551,14 +632,29 @@ final class ChatPrivateConversationCoordinator {
func handleReadReceipt(_ payload: NoisePayload, senderPubkey: String, convKey: PeerID) {
guard let messageID = String(data: payload.data, encoding: .utf8) else { return }
if context.privateChat(convKey, containsMessageWithID: messageID) {
context.setPrivateDeliveryStatus(
let aliases = accountConversationAliases(for: convKey)
let clearedRetainedMessage = convKey.noiseKey != nil
? context.markMessageDelivered(messageID, for: aliases)
: false
let hasConversationMessage = aliases.contains {
context.privateChat($0, containsMessageWithID: messageID)
}
if hasConversationMessage {
let conversationPeerID = consolidateAccountConversationAliases(for: convKey)
let didChange = context.setPrivateDeliveryStatus(
.read(by: context.displayNameForNostrPubkey(senderPubkey), at: Date()),
forMessageID: messageID,
peerID: convKey
peerID: conversationPeerID
)
context.notifyUIChanged()
if didChange {
context.notifyUIChanged()
}
SecureLogger.info("GeoDM: recv READ for mid=\(messageID.prefix(8))… from=\(senderPubkey.prefix(8))…", category: .session)
} else if clearedRetainedMessage {
SecureLogger.debug(
"GeoDM: recv READ for cleared mid=\(messageID.prefix(8))… from=\(senderPubkey.prefix(8))…",
category: .session
)
} else {
SecureLogger.warning("GeoDM: read ack for unknown mid=\(messageID.prefix(8))… conv=\(convKey)", category: .session)
}