diff --git a/bitchat/App/ConversationStore.swift b/bitchat/App/ConversationStore.swift index 1952b4d8..4e752b9b 100644 --- a/bitchat/App/ConversationStore.swift +++ b/bitchat/App/ConversationStore.swift @@ -138,6 +138,23 @@ final class Conversation: ObservableObject, Identifiable { return removed } + /// Removes every message matching `predicate`. Returns the removed + /// message IDs (empty when nothing matched). + fileprivate func removeAll(where predicate: (BitchatMessage) -> Bool) -> [String] { + var removedIDs: [String] = [] + messages.removeAll { message in + guard predicate(message) else { return false } + removedIDs.append(message.id) + return true + } + guard !removedIDs.isEmpty else { return [] } + for id in removedIDs { + indexByMessageID.removeValue(forKey: id) + } + reindex(from: 0) + return removedIDs + } + fileprivate func clearMessages() { messages.removeAll() indexByMessageID.removeAll() @@ -346,6 +363,16 @@ final class ConversationStore: ObservableObject { return removed } + /// Removes every message matching `predicate` from a conversation, + /// emitting one `.messageRemoved` per removed message after the + /// conversation is consistent. No-op for unknown conversations. + func removeMessages(from id: ConversationID, where predicate: (BitchatMessage) -> Bool) { + guard let conversation = conversationsByID[id] else { return } + for messageID in conversation.removeAll(where: predicate) { + changes.send(.messageRemoved(id, messageID: messageID)) + } + } + /// Empties a conversation's timeline but keeps the conversation (and /// its unread/selection state) alive. func clear(_ id: ConversationID) { @@ -464,3 +491,26 @@ extension ConversationStore { } } } + +// MARK: - Migration step 3 compatibility (public timeline derived views) + +extension ConversationStore { + /// Removes a message by ID from whichever public (mesh/geohash) + /// conversation contains it — the compat shape of the legacy + /// `PublicTimelineStore.removeMessage(withID:)`. Returns the removed + /// message, if any. + @discardableResult + func removePublicMessage(withID messageID: String) -> BitchatMessage? { + for (id, conversation) in conversationsByID { + switch id { + case .mesh, .geohash: + if conversation.containsMessage(withID: messageID) { + return removeMessage(withID: messageID, from: id) + } + case .direct: + continue + } + } + return nil + } +} diff --git a/bitchat/App/LegacyConversationStoreBridge.swift b/bitchat/App/LegacyConversationStoreBridge.swift index 7ed57f17..cf412298 100644 --- a/bitchat/App/LegacyConversationStoreBridge.swift +++ b/bitchat/App/LegacyConversationStoreBridge.swift @@ -5,18 +5,20 @@ // Migration step 2 adapter (DELETE IN STEP 5, see // docs/CONVERSATION-STORE-DESIGN.md §4). // -// The new `ConversationStore` is the single writer for private (direct) -// message state; the feature models (`PrivateInboxModel`, -// `PrivateConversationModel`, `ConversationUIModel`, `PeerListModel`) still -// read the replace-based `LegacyConversationStore` until step 5. This bridge -// keeps Legacy fed from the new store's `changes` subject: per-message -// changes mark the affected conversation dirty and a `Task.yield`-coalesced -// flush mirrors only the dirty conversations — a burst of N appends costs -// ONE Legacy replace (like the old debounced sync) without the full-dict -// pass. Structural changes (migration/removal) resynchronize immediately. -// Legacy is therefore eventually consistent within one run-loop tick — the -// same visibility the old `$privateChats` sink provided — while the new -// store stays synchronously authoritative. +// The new `ConversationStore` is the single writer for private (direct) AND +// public (mesh/geohash) message state; the feature models +// (`PrivateInboxModel`, `PrivateConversationModel`, `ConversationUIModel`, +// `PeerListModel`, `PublicChatModel`) still read the replace-based +// `LegacyConversationStore` until step 5. This bridge keeps Legacy fed from +// the new store's `changes` subject: per-message changes mark the affected +// conversation dirty and a `Task.yield`-coalesced flush mirrors only the +// dirty conversations — a burst of N appends costs ONE Legacy replace (like +// the old debounced sync) without the full-dict pass. Structural direct +// changes (migration/removal) resynchronize immediately; public removals +// mirror an empty timeline. Legacy is therefore eventually consistent within +// one run-loop tick — the same visibility the old sinks and per-message +// public syncs provided — while the new store stays synchronously +// authoritative. // // This is free and unencumbered software released into the public domain. // For more information, see @@ -58,9 +60,10 @@ final class LegacyConversationStoreBridge { /// `synchronizePrivateChats` full pass — acceptable because it only runs /// on peer-list changes and rare migrations, never per message. func resynchronizeAll() { - // The full pass covers every conversation; pending per-conversation - // work is redundant. - dirtyConversations.removeAll() + // The full pass covers every direct conversation; pending direct + // per-conversation work is redundant. Dirty public conversations + // keep their scheduled mirror. + dirtyConversations = dirtyConversations.filter { !isDirect($0) } legacyStore.synchronizePrivateChats( store.directMessagesByRoutingPeerID(), unreadPeerIDs: store.unreadDirectRoutingPeerIDs(), @@ -92,13 +95,19 @@ private extension LegacyConversationStoreBridge { resynchronizeAll() case .removed(let id): - guard isDirect(id) else { return } - resynchronizeAll() + if isDirect(id) { + resynchronizeAll() + } else { + // Public conversation removed (panic clear): mirror an empty + // timeline immediately so Legacy readers never show stale + // messages. + dirtyConversations.remove(id) + legacyStore.replaceMessages([], for: id) + } } } func markDirty(_ id: ConversationID) { - guard isDirect(id) else { return } dirtyConversations.insert(id) scheduleFlush() } @@ -127,16 +136,21 @@ private extension LegacyConversationStoreBridge { } func mirrorConversation(_ id: ConversationID) { - guard case .direct(let handle) = id, - let conversation = store.conversationsByID[id] else { - // Removed while dirty; the removal already resynchronized. + guard let conversation = store.conversationsByID[id] else { + // Removed while dirty; the removal already resynchronized + // (direct) or mirrored an empty timeline (public). return } - legacyStore.replaceDirectMessages( - conversation.messages, - for: handle.routingPeerID, - identityResolver: identityResolver - ) + switch id { + case .direct(let handle): + legacyStore.replaceDirectMessages( + conversation.messages, + for: handle.routingPeerID, + identityResolver: identityResolver + ) + case .mesh, .geohash: + legacyStore.replaceMessages(conversation.messages, for: id) + } } func isDirect(_ id: ConversationID) -> Bool { diff --git a/bitchat/ViewModels/ChatDeliveryCoordinator.swift b/bitchat/ViewModels/ChatDeliveryCoordinator.swift index e94a5a73..8acf7bec 100644 --- a/bitchat/ViewModels/ChatDeliveryCoordinator.swift +++ b/bitchat/ViewModels/ChatDeliveryCoordinator.swift @@ -12,7 +12,11 @@ import Foundation /// coordinators off their `unowned let viewModel: ChatViewModel` back-refs. @MainActor protocol ChatDeliveryContext: AnyObject { - var messages: [BitchatMessage] { get set } + /// Get-only derived views of the `ConversationStore`. `BitchatMessage` + /// is a reference type, so the public-timeline status patch below writes + /// through the shared message objects; step 4 replaces this with + /// `setDeliveryStatus(for:in:)` store intents. + var messages: [BitchatMessage] { get } var privateChats: [PeerID: [BitchatMessage]] { get } var isStartupPhase: Bool { get } /// Applies a delivery status to a private message by ID (single-writer diff --git a/bitchat/ViewModels/ChatMediaTransferCoordinator.swift b/bitchat/ViewModels/ChatMediaTransferCoordinator.swift index 54ce27ed..d604f8f8 100644 --- a/bitchat/ViewModels/ChatMediaTransferCoordinator.swift +++ b/bitchat/ViewModels/ChatMediaTransferCoordinator.swift @@ -29,9 +29,10 @@ protocol ChatMediaTransferContext: AnyObject { /// Appends a private message via the single-writer store intent. @discardableResult func appendPrivateMessage(_ message: BitchatMessage, to peerID: PeerID) -> Bool - func appendTimelineMessage(_ message: BitchatMessage, to channel: ChannelID) - func refreshVisibleMessages(from channel: ChannelID?) - func trimMessagesIfNeeded() + /// Appends a public message via the single-writer store intent + /// (immediate: outgoing media placeholders must render without batching). + @discardableResult + func appendPublicMessage(_ message: BitchatMessage, to conversationID: ConversationID) -> Bool func removeMessage(withID messageID: String, cleanupFile: Bool) func addSystemMessage(_ content: String) /// Signals that message state changed so observers refresh (e.g. `objectWillChange.send()`). @@ -52,8 +53,7 @@ extension ChatViewModel: ChatMediaTransferContext { // `canSendMediaInCurrentContext`, `selectedPrivateChatPeer`, `nickname`, // `myPeerID`, `activeChannel`, `nicknameForPeer(_:)`, // `currentPublicSender()`, `privateChats`, - // `appendTimelineMessage(_:to:)`, `refreshVisibleMessages(from:)`, - // `trimMessagesIfNeeded()`, `removeMessage(withID:cleanupFile:)`, + // `appendPublicMessage(_:to:)`, `removeMessage(withID:cleanupFile:)`, // `addSystemMessage(_:)`, `notifyUIChanged()`, // `updateMessageDeliveryStatus(_:status:)`, `normalizedContentKey(_:)`, // and `recordContentKey(_:timestamp:)` are shared requirements with the @@ -232,7 +232,6 @@ final class ChatMediaTransferCoordinator { deliveryStatus: .sending ) context.appendPrivateMessage(message, to: peerID) - context.trimMessagesIfNeeded() } else { let (displayName, senderPeerID) = context.currentPublicSender() message = BitchatMessage( @@ -246,9 +245,7 @@ final class ChatMediaTransferCoordinator { senderPeerID: senderPeerID, deliveryStatus: .sending ) - context.appendTimelineMessage(message, to: context.activeChannel) - context.refreshVisibleMessages(from: context.activeChannel) - context.trimMessagesIfNeeded() + context.appendPublicMessage(message, to: ConversationID(channelID: context.activeChannel)) } let key = context.normalizedContentKey(message.content) diff --git a/bitchat/ViewModels/ChatOutgoingCoordinator.swift b/bitchat/ViewModels/ChatOutgoingCoordinator.swift index 5cf5d87b..915ea7fa 100644 --- a/bitchat/ViewModels/ChatOutgoingCoordinator.swift +++ b/bitchat/ViewModels/ChatOutgoingCoordinator.swift @@ -25,9 +25,10 @@ protocol ChatOutgoingContext: AnyObject { // MARK: Public timeline (local echo) func parseMentions(from content: String) -> [String] - func appendTimelineMessage(_ message: BitchatMessage, to channel: ChannelID) - func refreshVisibleMessages(from channel: ChannelID?) - func trimMessagesIfNeeded() + /// Appends a public message via the single-writer store intent + /// (immediate: the local echo must render without batching). + @discardableResult + func appendPublicMessage(_ message: BitchatMessage, to conversationID: ConversationID) -> Bool func addSystemMessage(_ content: String) // MARK: Content dedup @@ -50,8 +51,7 @@ extension ChatViewModel: ChatOutgoingContext { // `nickname`, `myPeerID`, `activeChannel`, `selectedPrivateChatPeer`, // `isTeleported`, `handleCommand(_:)`, `updatePrivateChatPeerIfNeeded()`, // `sendPrivateMessage(_:to:)`, `parseMentions(from:)`, - // `appendTimelineMessage(_:to:)`, `refreshVisibleMessages(from:)`, - // `trimMessagesIfNeeded()`, `addSystemMessage(_:)`, + // `appendPublicMessage(_:to:)`, `addSystemMessage(_:)`, // `normalizedContentKey(_:)`, `recordContentKey(_:timestamp:)`, // `sendMeshMessage(_:mentions:messageID:timestamp:)`, // `sendGeohash(context:)`, and `deriveNostrIdentity(forGeohash:)` are @@ -169,12 +169,10 @@ private extension ChatOutgoingCoordinator { } func appendLocalEcho(_ message: BitchatMessage) { - context.appendTimelineMessage(message, to: context.activeChannel) - context.refreshVisibleMessages(from: context.activeChannel) + context.appendPublicMessage(message, to: ConversationID(channelID: context.activeChannel)) let contentKey = context.normalizedContentKey(message.content) context.recordContentKey(contentKey, timestamp: message.timestamp) - context.trimMessagesIfNeeded() } func routePublicMessage( diff --git a/bitchat/ViewModels/ChatPublicConversationCoordinator.swift b/bitchat/ViewModels/ChatPublicConversationCoordinator.swift index 2cb1ddd9..4a6dab4a 100644 --- a/bitchat/ViewModels/ChatPublicConversationCoordinator.swift +++ b/bitchat/ViewModels/ChatPublicConversationCoordinator.swift @@ -19,8 +19,7 @@ import UIKit /// pipeline. @MainActor protocol ChatPublicConversationContext: AnyObject { - // MARK: Channel & visible timeline state - var messages: [BitchatMessage] { get set } + // MARK: Channel state var activeChannel: ChannelID { get } var currentGeohash: String? { get } var nickname: String { get } @@ -31,25 +30,26 @@ protocol ChatPublicConversationContext: AnyObject { func setPublicBatching(_ isBatching: Bool) /// Signals that message state changed so observers refresh (e.g. `objectWillChange.send()`). func notifyUIChanged() - func trimMessagesIfNeeded() - // MARK: Public timeline store - func timelineMessages(for channel: ChannelID) -> [BitchatMessage] - func appendTimelineMessage(_ message: BitchatMessage, to channel: ChannelID) + // MARK: Public conversation store (single-writer intents) + /// Appends a public message in timestamp order. Returns `false` when a + /// message with the same ID is already in that conversation. + @discardableResult + func appendPublicMessage(_ message: BitchatMessage, to conversationID: ConversationID) -> Bool + /// Appends a geohash message if absent. Returns `true` when stored. + @discardableResult func appendGeohashMessageIfAbsent(_ message: BitchatMessage, toGeohash geohash: String) -> Bool - func removeTimelineMessage(withID id: String) -> BitchatMessage? - func removeGeohashTimelineMessages(in geohash: String, where predicate: (BitchatMessage) -> Bool) - func clearTimeline(for channel: ChannelID) - func timelineGeohashKeys() -> [String] + func publicConversationContainsMessage(withID messageID: String, in conversationID: ConversationID) -> Bool + /// Removes a message by ID from whichever public conversation contains it. + @discardableResult + func removePublicMessage(withID messageID: String) -> BitchatMessage? + /// Removes every matching message from a geohash conversation (block purge). + func removePublicMessages(fromGeohash geohash: String, where predicate: (BitchatMessage) -> Bool) + /// Empties a public conversation's timeline (`/clear`). + func clearPublicConversation(_ conversationID: ConversationID) /// Queues a system message for the next geohash channel visit. func queueGeohashSystemMessage(_ content: String) - // MARK: Conversation stores - func setConversationActiveChannel(_ channel: ChannelID) - func replaceConversationMessages(_ messages: [BitchatMessage], for channelID: ChannelID) - func replaceConversationMessages(_ messages: [BitchatMessage], for conversationID: ConversationID) - func synchronizeConversationSelectionStore() - // MARK: Private chats (block cleanup & message removal) var privateChats: [PeerID: [BitchatMessage]] { get } /// Removes the peer's chat entirely, including unread state @@ -84,7 +84,9 @@ protocol ChatPublicConversationContext: AnyObject { func processActionMessage(_ message: BitchatMessage) -> BitchatMessage func isMessageBlocked(_ message: BitchatMessage) -> Bool func allowPublicMessage(senderKey: String, contentKey: String) -> Bool - func enqueuePublicMessage(_ message: BitchatMessage) + /// Buffers a visible-channel message for the batched (~80 ms) pipeline + /// flush, which commits it to `conversationID` in the store. + func enqueuePublicMessage(_ message: BitchatMessage, to conversationID: ConversationID) func cachedStablePeerID(for shortPeerID: PeerID) -> PeerID? // MARK: Content dedup & formatting @@ -100,55 +102,21 @@ protocol ChatPublicConversationContext: AnyObject { } extension ChatViewModel: ChatPublicConversationContext { - // `messages`, `privateChats`, `unreadPrivateMessages`, `nostrKeyMapping`, + // `privateChats`, `unreadPrivateMessages`, `nostrKeyMapping`, // `nickname`, `activeChannel`, `currentGeohash`, `geoNicknames`, // `myPeerID`, `isTeleported`, `notifyUIChanged()`, // `geoParticipantCount(for:)`, `isNostrBlocked(pubkeyHexLowercased:)`, - // `deriveNostrIdentity(forGeohash:)`, and - // `appendGeohashMessageIfAbsent(_:toGeohash:)` are shared requirements - // with `ChatDeliveryContext` / `ChatPrivateConversationContext` / - // `ChatNostrContext`; their witnesses already exist. The members below - // flatten nested service accesses into intent-named calls. - - func timelineMessages(for channel: ChannelID) -> [BitchatMessage] { - timelineStore.messages(for: channel) - } - - func appendTimelineMessage(_ message: BitchatMessage, to channel: ChannelID) { - timelineStore.append(message, to: channel) - } - - func removeTimelineMessage(withID id: String) -> BitchatMessage? { - timelineStore.removeMessage(withID: id) - } - - func removeGeohashTimelineMessages(in geohash: String, where predicate: (BitchatMessage) -> Bool) { - timelineStore.removeMessages(in: geohash, where: predicate) - } - - func clearTimeline(for channel: ChannelID) { - timelineStore.clear(channel: channel) - } - - func timelineGeohashKeys() -> [String] { - timelineStore.geohashKeys() - } - - func queueGeohashSystemMessage(_ content: String) { - timelineStore.queueGeohashSystemMessage(content) - } - - func setConversationActiveChannel(_ channel: ChannelID) { - conversationStore.setActiveChannel(channel) - } - - func replaceConversationMessages(_ messages: [BitchatMessage], for channelID: ChannelID) { - conversationStore.replaceMessages(messages, for: channelID) - } - - func replaceConversationMessages(_ messages: [BitchatMessage], for conversationID: ConversationID) { - conversationStore.replaceMessages(messages, for: conversationID) - } + // `deriveNostrIdentity(forGeohash:)`, the public conversation store + // intents (`appendPublicMessage(_:to:)`, + // `appendGeohashMessageIfAbsent(_:toGeohash:)`, + // `publicConversationContainsMessage(withID:in:)`, + // `removePublicMessage(withID:)`, + // `removePublicMessages(fromGeohash:where:)`, + // `clearPublicConversation(_:)`, and `queueGeohashSystemMessage(_:)`) + // are shared requirements with `ChatDeliveryContext` / + // `ChatPrivateConversationContext` / `ChatNostrContext` or satisfied by + // existing `ChatViewModel` members. The members below flatten nested + // service accesses into intent-named calls. func visibleGeoPeople() -> [GeoPerson] { participantTracker.getVisiblePeople() @@ -174,8 +142,8 @@ extension ChatViewModel: ChatPublicConversationContext { publicRateLimiter.allow(senderKey: senderKey, contentKey: contentKey) } - func enqueuePublicMessage(_ message: BitchatMessage) { - publicMessagePipeline.enqueue(message) + func enqueuePublicMessage(_ message: BitchatMessage, to conversationID: ConversationID) { + publicMessagePipeline.enqueue(message, to: conversationID) } func normalizedContentKey(_ content: String) -> String { @@ -247,11 +215,7 @@ final class ChatPublicConversationCoordinator: PublicMessagePipelineDelegate { } return false } - context.removeGeohashTimelineMessages(in: gh, where: predicate) - synchronizePublicConversationStore(forGeohash: gh) - if case .location = context.activeChannel { - context.messages.removeAll(where: predicate) - } + context.removePublicMessages(fromGeohash: gh, where: predicate) } let conversationPeerID = PeerID(nostr_: hex) @@ -313,16 +277,7 @@ final class ChatPublicConversationCoordinator: PublicMessagePipelineDelegate { } func removeMessage(withID messageID: String, cleanupFile: Bool = false) { - var removedMessage: BitchatMessage? - - if let index = context.messages.firstIndex(where: { $0.id == messageID }) { - removedMessage = context.messages.remove(at: index) - } - - if let storeRemoved = context.removeTimelineMessage(withID: messageID) { - removedMessage = removedMessage ?? storeRemoved - synchronizeAllPublicConversationStores() - } + var removedMessage = context.removePublicMessage(withID: messageID) if let removedPrivateMessage = context.removePrivateMessage(withID: messageID) { removedMessage = removedMessage ?? removedPrivateMessage @@ -335,45 +290,8 @@ final class ChatPublicConversationCoordinator: PublicMessagePipelineDelegate { context.notifyUIChanged() } - func initializeConversationStore() { - context.setConversationActiveChannel(context.activeChannel) - synchronizePublicConversationStore(for: context.activeChannel) - context.synchronizeConversationSelectionStore() - } - - func synchronizePublicConversationStore(for channel: ChannelID) { - let publicMessages = context.timelineMessages(for: channel) - context.replaceConversationMessages(publicMessages, for: channel) - if channel == context.activeChannel { - context.setConversationActiveChannel(context.activeChannel) - } - } - - func synchronizePublicConversationStore(forGeohash geohash: String) { - let channel = ChannelID.location(GeohashChannel(level: .city, geohash: geohash)) - let publicMessages = context.timelineMessages(for: channel) - context.replaceConversationMessages(publicMessages, for: .geohash(geohash.lowercased())) - } - - func synchronizeAllPublicConversationStores() { - synchronizePublicConversationStore(for: .mesh) - for geohash in context.timelineGeohashKeys() { - synchronizePublicConversationStore(forGeohash: geohash) - } - } - - func refreshVisibleMessages(from channel: ChannelID? = nil) { - let target = channel ?? context.activeChannel - context.messages = context.timelineMessages(for: target) - context.replaceConversationMessages(context.messages, for: target) - if target == context.activeChannel { - context.setConversationActiveChannel(context.activeChannel) - } - } - func clearCurrentPublicTimeline() { - context.messages.removeAll() - context.clearTimeline(for: context.activeChannel) + context.clearPublicConversation(ConversationID(channelID: context.activeChannel)) Task.detached(priority: .utility) { do { @@ -413,7 +331,7 @@ final class ChatPublicConversationCoordinator: PublicMessagePipelineDelegate { timestamp: timestamp, isRelay: false ) - context.messages.append(systemMessage) + context.appendPublicMessage(systemMessage, to: ConversationID(channelID: context.activeChannel)) } func addMeshOnlySystemMessage(_ content: String) { @@ -423,11 +341,7 @@ final class ChatPublicConversationCoordinator: PublicMessagePipelineDelegate { timestamp: Date(), isRelay: false ) - context.appendTimelineMessage(systemMessage, to: .mesh) - synchronizePublicConversationStore(for: .mesh) - refreshVisibleMessages() - context.trimMessagesIfNeeded() - context.notifyUIChanged() + context.appendPublicMessage(systemMessage, to: .mesh) } func addPublicSystemMessage(_ content: String) { @@ -437,12 +351,9 @@ final class ChatPublicConversationCoordinator: PublicMessagePipelineDelegate { timestamp: Date(), isRelay: false ) - context.appendTimelineMessage(systemMessage, to: context.activeChannel) - refreshVisibleMessages(from: context.activeChannel) + context.appendPublicMessage(systemMessage, to: ConversationID(channelID: context.activeChannel)) let contentKey = context.normalizedContentKey(systemMessage.content) context.recordContentKey(contentKey, timestamp: systemMessage.timestamp) - context.trimMessagesIfNeeded() - context.notifyUIChanged() } func addGeohashOnlySystemMessage(_ content: String) { @@ -492,7 +403,8 @@ final class ChatPublicConversationCoordinator: PublicMessagePipelineDelegate { if context.isMessageBlocked(finalMessage) { return } let isGeo = finalMessage.senderPeerID?.isGeoChat == true - let shouldRateLimit = finalMessage.sender != "system" || finalMessage.senderPeerID != nil + let isSystem = finalMessage.sender == "system" + let shouldRateLimit = !isSystem || finalMessage.senderPeerID != nil if shouldRateLimit { let senderKey = normalizedSenderKey(for: finalMessage) let contentKey = context.normalizedContentKey(finalMessage.content) @@ -501,20 +413,26 @@ final class ChatPublicConversationCoordinator: PublicMessagePipelineDelegate { } } - if finalMessage.sender != "system" && finalMessage.content.count > 16000 { return } + if !isSystem && finalMessage.content.count > 16000 { return } + // Empty content never rendered before (the old visible-array enqueue + // filtered it); with the store as the sole timeline it is dropped + // outright instead of lingering invisibly in a backing buffer. + guard !finalMessage.content.trimmed.isEmpty else { return } - if !isGeo && finalMessage.sender != "system" { - context.appendTimelineMessage(finalMessage, to: .mesh) - synchronizePublicConversationStore(for: .mesh) + // Resolve the destination conversation. System messages surface on + // the active channel (matching their old visible-only routing); geo + // messages require a current geohash, mesh messages always land in + // the mesh conversation. + let destination: ConversationID? + if isSystem { + destination = ConversationID(channelID: context.activeChannel) + } else if isGeo { + destination = context.currentGeohash.map { .geohash($0.lowercased()) } + } else { + destination = .mesh } + guard let destination else { return } - if isGeo && finalMessage.sender != "system", - let geohash = context.currentGeohash, - context.appendGeohashMessageIfAbsent(finalMessage, toGeohash: geohash) { - synchronizePublicConversationStore(forGeohash: geohash) - } - - let isSystem = finalMessage.sender == "system" let channelMatches: Bool = { switch context.activeChannel { case .mesh: return !isGeo || isSystem @@ -522,11 +440,16 @@ final class ChatPublicConversationCoordinator: PublicMessagePipelineDelegate { } }() - guard channelMatches else { return } - - if !finalMessage.content.trimmed.isEmpty, - !context.messages.contains(where: { $0.id == finalMessage.id }) { - context.enqueuePublicMessage(finalMessage) + if channelMatches { + // Visible-channel arrivals are batched: the pipeline's ~80 ms + // flush commits them to the store (which dedups by ID), keeping + // the deliberate UI flush cadence. + guard !context.publicConversationContainsMessage(withID: finalMessage.id, in: destination) else { return } + context.enqueuePublicMessage(finalMessage, to: destination) + } else { + // Background-channel arrivals have no rendering observers to + // batch for; they land in the store immediately. + context.appendPublicMessage(finalMessage, to: destination) } } @@ -584,14 +507,6 @@ final class ChatPublicConversationCoordinator: PublicMessagePipelineDelegate { #endif } - func pipelineCurrentMessages(_ pipeline: PublicMessagePipeline) -> [BitchatMessage] { - context.messages - } - - func pipeline(_ pipeline: PublicMessagePipeline, setMessages messages: [BitchatMessage]) { - context.messages = messages - } - func pipeline(_ pipeline: PublicMessagePipeline, normalizeContent content: String) -> String { context.normalizedContentKey(content) } @@ -604,8 +519,8 @@ final class ChatPublicConversationCoordinator: PublicMessagePipelineDelegate { context.recordContentKey(key, timestamp: timestamp) } - func pipelineTrimMessages(_ pipeline: PublicMessagePipeline) { - context.trimMessagesIfNeeded() + func pipeline(_ pipeline: PublicMessagePipeline, commit message: BitchatMessage, to conversationID: ConversationID) -> Bool { + context.appendPublicMessage(message, to: conversationID) } func pipelinePrewarmMessage(_ pipeline: PublicMessagePipeline, message: BitchatMessage) { diff --git a/bitchat/ViewModels/ChatViewModel.swift b/bitchat/ViewModels/ChatViewModel.swift index 948514db..8be04fc0 100644 --- a/bitchat/ViewModels/ChatViewModel.swift +++ b/bitchat/ViewModels/ChatViewModel.swift @@ -119,9 +119,24 @@ final class ChatViewModel: ObservableObject, BitchatDelegate, TransportEventDele // MARK: - Published Properties - @Published var messages: [BitchatMessage] = [] + /// Read-only derived view of the ACTIVE public channel's conversation in + /// the single-writer `ConversationStore` (migration step 3 shim; views + /// observe `Conversation` objects directly in step 5). Hot for rendering, + /// so the array is cached and invalidated from the store's `changes` + /// subject (filtered to the active conversation) and on channel switches. + /// `objectWillChange` fires on every store change via the sink in `init`. + @MainActor + var messages: [BitchatMessage] { + if let cached = visibleMessagesCache { return cached } + // Read-only lookup (never creates the conversation): this getter + // runs during SwiftUI renders, where mutating the store's + // `@Published` collections would publish mid-view-update. + let current = conversations.conversationsByID[ConversationID(channelID: activeChannel)]?.messages ?? [] + visibleMessagesCache = current + return current + } + private var visibleMessagesCache: [BitchatMessage]? @Published var currentColorScheme: ColorScheme = .light - private let maxMessages = TransportConfig.meshTimelineCap // Maximum messages before oldest are removed @Published var isConnected = false @Published var nickname: String = "" { didSet { @@ -298,9 +313,8 @@ final class ChatViewModel: ObservableObject, BitchatDelegate, TransportEventDele get { conversationStore.activeChannel } set { guard conversationStore.activeChannel != newValue else { return } - publicMessagePipeline.updateActiveChannel(newValue) conversationStore.setActiveChannel(newValue) - synchronizePublicConversationStore(for: newValue) + visibleMessagesCache = nil synchronizeConversationSelectionStore() objectWillChange.send() } @@ -352,11 +366,6 @@ final class ChatViewModel: ObservableObject, BitchatDelegate, TransportEventDele @Published var bluetoothAlertMessage = "" @Published var bluetoothState: CBManagerState = .unknown - var timelineStore = PublicTimelineStore( - meshCap: TransportConfig.meshTimelineCap, - geohashCap: TransportConfig.geoTimelineCap - ) - private func performDeliveryUpdate(_ update: @escaping @MainActor (ChatDeliveryCoordinator) -> Void) { if Thread.isMainThread { MainActor.assumeIsolated { @@ -631,6 +640,83 @@ final class ChatViewModel: ObservableObject, BitchatDelegate, TransportEventDele return removed } + // MARK: - Public Conversation Store Intents + // The sole mutation paths for public (mesh/geohash) message state, + // mirroring the private intents above. The store's per-conversation cap + // and timestamp-ordered insert replace `PublicTimelineStore`'s trim and + // the pipeline's late-insert positioning; the read-only `messages` shim + // above is derived from the same store. + + /// Appends a public message in timestamp order. Returns `false` when a + /// message with the same ID is already in that conversation (O(1) dedup + /// via the conversation's ID index). + @MainActor + @discardableResult + func appendPublicMessage(_ message: BitchatMessage, to conversationID: ConversationID) -> Bool { + conversations.append(message, to: conversationID) + } + + /// Appends a geohash message if absent. Returns `true` when stored + /// (the legacy `PublicTimelineStore.appendIfAbsent` contract). + @MainActor + @discardableResult + func appendGeohashMessageIfAbsent(_ message: BitchatMessage, toGeohash geohash: String) -> Bool { + conversations.append(message, to: .geohash(geohash.lowercased())) + } + + /// A public (mesh/geohash) channel's full timeline. + @MainActor + func publicMessages(for channel: ChannelID) -> [BitchatMessage] { + conversations.conversation(for: ConversationID(channelID: channel)).messages + } + + /// `true` when the conversation contains a message with `messageID`. + @MainActor + func publicConversationContainsMessage(withID messageID: String, in conversationID: ConversationID) -> Bool { + conversations.conversationsByID[conversationID]?.containsMessage(withID: messageID) ?? false + } + + /// Removes a message by ID from whichever public conversation contains + /// it. Returns the removed message, if any. + @MainActor + @discardableResult + func removePublicMessage(withID messageID: String) -> BitchatMessage? { + conversations.removePublicMessage(withID: messageID) + } + + /// Removes every message matching `predicate` from a geohash + /// conversation (block-user purge). + @MainActor + func removePublicMessages(fromGeohash geohash: String, where predicate: (BitchatMessage) -> Bool) { + conversations.removeMessages(from: .geohash(geohash.lowercased()), where: predicate) + } + + /// Empties a public conversation's timeline (`/clear`). + @MainActor + func clearPublicConversation(_ conversationID: ConversationID) { + conversations.clear(conversationID) + } + + /// Queues a system message for the next geohash channel visit. (Tiny + /// UI-flow queue formerly on `PublicTimelineStore`; it is notice text, + /// not conversation state, so it stays on the owner.) + @MainActor + func queueGeohashSystemMessage(_ content: String) { + pendingGeohashSystemMessages.append(content) + } + + /// Drains the queued geohash system messages (single consumer: + /// `GeohashSubscriptionManager.switchLocationChannel`). + @MainActor + func drainPendingGeohashSystemMessages() -> [String] { + defer { pendingGeohashSystemMessages.removeAll(keepingCapacity: false) } + return pendingGeohashSystemMessages + } + + // Single-writer: mutate only via `queueGeohashSystemMessage(_:)` / + // `drainPendingGeohashSystemMessages()` above. + private var pendingGeohashSystemMessages: [String] = [] + // MARK: - Initialization @MainActor @@ -717,12 +803,18 @@ final class ChatViewModel: ObservableObject, BitchatDelegate, TransportEventDele // Republish on every store change so SwiftUI observers of the // view model refresh. This replaces the UI-update role of the old - // `PrivateChatManager.@Published` dictionaries (their debounced - // Legacy synchronization sinks are gone; the bridge above feeds - // Legacy instead). + // `PrivateChatManager.@Published` dictionaries and the old + // `@Published var messages` (their debounced Legacy synchronization + // sinks are gone; the bridge above feeds Legacy instead). Changes + // touching the ACTIVE public conversation also invalidate the + // derived `messages` cache before observers re-read it. conversations.changes - .sink { [weak self] _ in - self?.objectWillChange.send() + .sink { [weak self] change in + guard let self else { return } + if self.changeAffectsActivePublicConversation(change) { + self.visibleMessagesCache = nil + } + self.objectWillChange.send() } .store(in: &cancellables) @@ -1008,13 +1100,11 @@ final class ChatViewModel: ObservableObject, BitchatDelegate, TransportEventDele func panicClearAllData() { // Messages are processed immediately - nothing to flush - // Clear all messages - messages.removeAll() - timelineStore = PublicTimelineStore( - meshCap: TransportConfig.meshTimelineCap, - geohashCap: TransportConfig.geoTimelineCap - ) - conversations.removeAllDirectConversations() + // Clear all messages (public timelines and private chats live in the + // single-writer ConversationStore; the derived `messages` view and + // the legacy mirror empty with it) + conversations.clearAll() + pendingGeohashSystemMessages.removeAll() // Delete all keychain data (including Noise and Nostr keys) _ = keychain.deleteAllKeychainData() @@ -1190,22 +1280,8 @@ final class ChatViewModel: ObservableObject, BitchatDelegate, TransportEventDele @MainActor func initializeConversationStore() { - publicConversationCoordinator.initializeConversationStore() - } - - @MainActor - func synchronizePublicConversationStore(for channel: ChannelID) { - publicConversationCoordinator.synchronizePublicConversationStore(for: channel) - } - - @MainActor - func synchronizePublicConversationStore(forGeohash geohash: String) { - publicConversationCoordinator.synchronizePublicConversationStore(forGeohash: geohash) - } - - @MainActor - func synchronizeAllPublicConversationStores() { - publicConversationCoordinator.synchronizeAllPublicConversationStores() + conversationStore.setActiveChannel(activeChannel) + synchronizeConversationSelectionStore() } /// Full Legacy-store resynchronization from the new `ConversationStore`. @@ -1226,15 +1302,34 @@ final class ChatViewModel: ObservableObject, BitchatDelegate, TransportEventDele ) } - func trimMessagesIfNeeded() { - if messages.count > maxMessages { - messages = Array(messages.suffix(maxMessages)) - } - } - + /// Invalidates the derived `messages` cache and notifies observers. + /// (Formerly pulled the channel's timeline into a stored `messages` + /// array; `messages` is now derived from the `ConversationStore`, so + /// only the invalidation remains. The `channel` parameter is kept for + /// call-site compatibility — every caller passes the active channel.) @MainActor func refreshVisibleMessages(from channel: ChannelID? = nil) { - publicConversationCoordinator.refreshVisibleMessages(from: channel) + visibleMessagesCache = nil + objectWillChange.send() + } + + /// `true` when a store change touches the active public conversation + /// (so the derived `messages` cache must be invalidated). + @MainActor + private func changeAffectsActivePublicConversation(_ change: ConversationChange) -> Bool { + let activeID = ConversationID(channelID: activeChannel) + switch change { + case .appended(let id, _), + .updated(let id, _), + .statusChanged(let id, _, _), + .messageRemoved(let id, _), + .cleared(let id), + .removed(let id), + .unreadChanged(let id, _): + return id == activeID + case .migrated(let source, let destination): + return source == activeID || destination == activeID + } } @MainActor @@ -1276,15 +1371,6 @@ final class ChatViewModel: ObservableObject, BitchatDelegate, TransportEventDele publicConversationCoordinator.clearCurrentPublicTimeline() } - // MARK: - Message Management - - private func addMessage(_ message: BitchatMessage) { - // Check for duplicates - guard !messages.contains(where: { $0.id == message.id }) else { return } - messages.append(message) - trimMessagesIfNeeded() - } - // MARK: - Peer Lookup Helpers func getPeer(byID peerID: PeerID) -> BitchatPeer? { diff --git a/bitchat/ViewModels/ChatViewModelBootstrapper.swift b/bitchat/ViewModels/ChatViewModelBootstrapper.swift index 5e1b71d4..9be99cfe 100644 --- a/bitchat/ViewModels/ChatViewModelBootstrapper.swift +++ b/bitchat/ViewModels/ChatViewModelBootstrapper.swift @@ -131,7 +131,6 @@ private extension ChatViewModelBootstrapper { viewModel.meshService.startServices() viewModel.publicMessagePipeline.delegate = viewModel.publicConversationCoordinator - viewModel.publicMessagePipeline.updateActiveChannel(viewModel.activeChannel) DispatchQueue.main.asyncAfter(deadline: .now() + 0.1) { [weak viewModel] in guard let viewModel, diff --git a/bitchat/ViewModels/GeoPresenceTracker.swift b/bitchat/ViewModels/GeoPresenceTracker.swift index 100041a9..ffe423b5 100644 --- a/bitchat/ViewModels/GeoPresenceTracker.swift +++ b/bitchat/ViewModels/GeoPresenceTracker.swift @@ -23,8 +23,10 @@ protocol GeoPresenceContext: AnyObject { func geoParticipantCount(for geohash: String) -> Int func markGeoTeleported(_ pubkeyHexLowercased: String) + /// Appends a geohash message if absent (single-writer store intent). + /// Returns `true` when stored. + @discardableResult func appendGeohashMessageIfAbsent(_ message: BitchatMessage, toGeohash geohash: String) -> Bool - func synchronizePublicConversationStore(forGeohash geohash: String) /// Posts the sampled-geohash-activity local notification. func notifyGeohashActivity(geohash: String, bodyPreview: String) @@ -32,13 +34,10 @@ protocol GeoPresenceContext: AnyObject { extension ChatViewModel: GeoPresenceContext { // `activeChannel`, `lastGeoNotificationAt`, `geoNicknames`, the Nostr - // identity/blocking members, and `synchronizePublicConversationStore` - // already have witnesses on `ChatViewModel`. The members below flatten - // nested service accesses into intent-named calls. - - func appendGeohashMessageIfAbsent(_ message: BitchatMessage, toGeohash geohash: String) -> Bool { - timelineStore.appendIfAbsent(message, toGeohash: geohash) - } + // identity/blocking members, and the + // `appendGeohashMessageIfAbsent(_:toGeohash:)` store intent already have + // witnesses on `ChatViewModel`. The members below flatten nested service + // accesses into intent-named calls. var teleportedGeoCount: Int { locationPresenceStore.teleportedGeo.count @@ -166,7 +165,6 @@ final class GeoPresenceTracker { mentions: mentions.isEmpty ? nil : mentions ) if context.appendGeohashMessageIfAbsent(message, toGeohash: gh) { - context.synchronizePublicConversationStore(forGeohash: gh) context.notifyGeohashActivity(geohash: gh, bodyPreview: preview) } } diff --git a/bitchat/ViewModels/GeohashSubscriptionManager.swift b/bitchat/ViewModels/GeohashSubscriptionManager.swift index 1553410a..4e267c59 100644 --- a/bitchat/ViewModels/GeohashSubscriptionManager.swift +++ b/bitchat/ViewModels/GeohashSubscriptionManager.swift @@ -27,8 +27,9 @@ protocol GeohashSubscriptionContext: AnyObject { // MARK: Public timeline & pipeline var messages: [BitchatMessage] { get } - func resetPublicMessagePipeline() - func updatePublicMessagePipelineChannel(_ channel: ChannelID) + /// Commits any batched-but-unflushed public messages to the store so a + /// channel switch never strands them in the pipeline buffer. + func flushPublicMessagePipeline() func refreshVisibleMessages(from channel: ChannelID?) func addPublicSystemMessage(_ content: String) func drainPendingGeohashSystemMessages() -> [String] @@ -64,16 +65,8 @@ extension ChatViewModel: GeohashSubscriptionContext { // `ChatViewModel`. The members below flatten nested service accesses into // intent-named calls. - func resetPublicMessagePipeline() { - publicMessagePipeline.reset() - } - - func updatePublicMessagePipelineChannel(_ channel: ChannelID) { - publicMessagePipeline.updateActiveChannel(channel) - } - - func drainPendingGeohashSystemMessages() -> [String] { - timelineStore.drainPendingGeohashSystemMessages() + func flushPublicMessagePipeline() { + publicMessagePipeline.flushIfNeeded() } func clearProcessedNostrEvents() { @@ -178,9 +171,8 @@ final class GeohashSubscriptionManager { @MainActor func switchLocationChannel(to channel: ChannelID) { guard let context else { return } - context.resetPublicMessagePipeline() + context.flushPublicMessagePipeline() context.activeChannel = channel - context.updatePublicMessagePipelineChannel(channel) context.clearProcessedNostrEvents() switch channel { diff --git a/bitchat/ViewModels/PublicMessagePipeline.swift b/bitchat/ViewModels/PublicMessagePipeline.swift index 516160e9..7383151b 100644 --- a/bitchat/ViewModels/PublicMessagePipeline.swift +++ b/bitchat/ViewModels/PublicMessagePipeline.swift @@ -2,7 +2,11 @@ // PublicMessagePipeline.swift // bitchat // -// Handles batching and deduplication of public chat messages before surfacing them to the UI. +// Batches visible-channel public messages before committing them to the +// ConversationStore: the deliberate ~80 ms UI flush cadence survives the +// store cutover, while ordering, dedup, and caps live in the store itself +// (its timestamp-ordered insert replaced this pipeline's late-insert +// threshold positioning; see docs/CONVERSATION-STORE-DESIGN.md). // import BitFoundation @@ -10,12 +14,13 @@ import Foundation @MainActor protocol PublicMessagePipelineDelegate: AnyObject { - func pipelineCurrentMessages(_ pipeline: PublicMessagePipeline) -> [BitchatMessage] - func pipeline(_ pipeline: PublicMessagePipeline, setMessages messages: [BitchatMessage]) func pipeline(_ pipeline: PublicMessagePipeline, normalizeContent content: String) -> String func pipeline(_ pipeline: PublicMessagePipeline, contentTimestampForKey key: String) -> Date? func pipeline(_ pipeline: PublicMessagePipeline, recordContentKey key: String, timestamp: Date) - func pipelineTrimMessages(_ pipeline: PublicMessagePipeline) + /// Commits a batched message to its conversation in the store. + /// Returns `false` when the message was already present (ID dedup). + @discardableResult + func pipeline(_ pipeline: PublicMessagePipeline, commit message: BitchatMessage, to conversationID: ConversationID) -> Bool func pipelinePrewarmMessage(_ pipeline: PublicMessagePipeline, message: BitchatMessage) func pipelineSetBatchingState(_ pipeline: PublicMessagePipeline, isBatching: Bool) } @@ -24,14 +29,13 @@ protocol PublicMessagePipelineDelegate: AnyObject { final class PublicMessagePipeline { weak var delegate: PublicMessagePipelineDelegate? - private var buffer: [BitchatMessage] = [] + private var buffer: [(message: BitchatMessage, conversationID: ConversationID)] = [] private var timer: Timer? private let baseFlushInterval: TimeInterval private var dynamicFlushInterval: TimeInterval private var recentBatchSizes: [Int] = [] private let maxRecentBatchSamples: Int private let dedupWindow: TimeInterval - private var activeChannel: ChannelID = .mesh init( baseFlushInterval: TimeInterval = TransportConfig.basePublicFlushInterval, @@ -48,25 +52,17 @@ final class PublicMessagePipeline { timer?.invalidate() } - func updateActiveChannel(_ channel: ChannelID) { - activeChannel = channel - } - - func enqueue(_ message: BitchatMessage) { - buffer.append(message) + /// Buffers a message destined for `conversationID`; the next batched + /// flush commits it to the store. Each entry carries its destination so + /// a channel switch mid-batch can never misroute buffered messages. + func enqueue(_ message: BitchatMessage, to conversationID: ConversationID) { + buffer.append((message, conversationID)) scheduleFlush() } func flushIfNeeded() { flushBuffer() } - - func reset() { - timer?.invalidate() - timer = nil - buffer.removeAll(keepingCapacity: false) - } - } private extension PublicMessagePipeline { @@ -91,57 +87,38 @@ private extension PublicMessagePipeline { delegate.pipelineSetBatchingState(self, isBatching: true) - var existingIDs = Set(delegate.pipelineCurrentMessages(self).map { $0.id }) - var pending: [(message: BitchatMessage, contentKey: String)] = [] + // Content-window dedup against recorded keys and within the batch; + // ID dedup happens in the store at commit time. + var pending: [(message: BitchatMessage, conversationID: ConversationID, contentKey: String)] = [] var batchContentLatest: [String: Date] = [:] - for message in buffer { - if existingIDs.contains(message.id) { continue } - let contentKey = delegate.pipeline(self, normalizeContent: message.content) + for item in buffer { + let contentKey = delegate.pipeline(self, normalizeContent: item.message.content) if let ts = delegate.pipeline(self, contentTimestampForKey: contentKey), - abs(ts.timeIntervalSince(message.timestamp)) < dedupWindow { + abs(ts.timeIntervalSince(item.message.timestamp)) < dedupWindow { continue } if let ts = batchContentLatest[contentKey], - abs(ts.timeIntervalSince(message.timestamp)) < dedupWindow { + abs(ts.timeIntervalSince(item.message.timestamp)) < dedupWindow { continue } - existingIDs.insert(message.id) - pending.append((message, contentKey)) - batchContentLatest[contentKey] = message.timestamp + pending.append((item.message, item.conversationID, contentKey)) + batchContentLatest[contentKey] = item.message.timestamp } buffer.removeAll(keepingCapacity: true) guard !pending.isEmpty else { delegate.pipelineSetBatchingState(self, isBatching: false) - if !buffer.isEmpty { scheduleFlush() } return } pending.sort { $0.message.timestamp < $1.message.timestamp } - var messages = delegate.pipelineCurrentMessages(self) - let threshold = lateInsertThreshold(for: activeChannel) - let lastTimestamp = messages.last?.timestamp ?? .distantPast - for item in pending { - let message = item.message - if threshold == 0 || message.timestamp < lastTimestamp.addingTimeInterval(-threshold) { - let index = insertionIndex(for: message.timestamp, in: messages) - if index >= messages.count { - messages.append(message) - } else { - messages.insert(message, at: index) - } - } else { - messages.append(message) - } - delegate.pipeline(self, recordContentKey: item.contentKey, timestamp: message.timestamp) + guard delegate.pipeline(self, commit: item.message, to: item.conversationID) else { continue } + delegate.pipeline(self, recordContentKey: item.contentKey, timestamp: item.message.timestamp) } - delegate.pipeline(self, setMessages: messages) - delegate.pipelineTrimMessages(self) - updateFlushInterval(withBatchSize: pending.count) for item in pending { @@ -165,27 +142,4 @@ private extension PublicMessagePipeline { : Double(recentBatchSizes.reduce(0, +)) / Double(recentBatchSizes.count) dynamicFlushInterval = avg > 100.0 ? 0.12 : baseFlushInterval } - - func lateInsertThreshold(for channel: ChannelID) -> TimeInterval { - switch channel { - case .mesh: - return TransportConfig.uiLateInsertThreshold - case .location: - return TransportConfig.uiLateInsertThresholdGeo - } - } - - func insertionIndex(for timestamp: Date, in messages: [BitchatMessage]) -> Int { - var low = 0 - var high = messages.count - while low < high { - let mid = (low + high) / 2 - if messages[mid].timestamp < timestamp { - low = mid + 1 - } else { - high = mid - } - } - return low - } } diff --git a/bitchat/ViewModels/PublicTimelineStore.swift b/bitchat/ViewModels/PublicTimelineStore.swift deleted file mode 100644 index ae613f2b..00000000 --- a/bitchat/ViewModels/PublicTimelineStore.swift +++ /dev/null @@ -1,148 +0,0 @@ -// -// PublicTimelineStore.swift -// bitchat -// -// Maintains mesh and geohash public timelines with simple caps and helpers. -// - -import BitFoundation -import Foundation - -struct PublicTimelineStore { - private var meshTimeline: [BitchatMessage] = [] - private var meshMessageIDs: Set = [] - private var geohashTimelines: [String: [BitchatMessage]] = [:] - private var geohashMessageIDs: [String: Set] = [:] - private var pendingGeohashSystemMessages: [String] = [] - - private let meshCap: Int - private let geohashCap: Int - - init(meshCap: Int, geohashCap: Int) { - self.meshCap = meshCap - self.geohashCap = geohashCap - } - - mutating func append(_ message: BitchatMessage, to channel: ChannelID) { - switch channel { - case .mesh: - guard !meshMessageIDs.contains(message.id) else { return } - meshTimeline.append(message) - meshMessageIDs.insert(message.id) - trimMeshTimelineIfNeeded() - case .location(let channel): - append(message, toGeohash: channel.geohash) - } - } - - mutating func append(_ message: BitchatMessage, toGeohash geohash: String) { - _ = appendGeohashMessageIfAbsent(message, geohash: geohash) - } - - /// Append message if absent, returning true when stored. - mutating func appendIfAbsent(_ message: BitchatMessage, toGeohash geohash: String) -> Bool { - appendGeohashMessageIfAbsent(message, geohash: geohash) - } - - mutating func messages(for channel: ChannelID) -> [BitchatMessage] { - switch channel { - case .mesh: - return meshTimeline - case .location(let channel): - let cleaned = geohashTimelines[channel.geohash]?.cleanedAndDeduped() ?? [] - replaceGeohashTimeline(cleaned, for: channel.geohash, keepEmpty: true) - return cleaned - } - } - - mutating func clear(channel: ChannelID) { - switch channel { - case .mesh: - meshTimeline.removeAll() - meshMessageIDs.removeAll() - case .location(let channel): - geohashTimelines[channel.geohash] = [] - geohashMessageIDs[channel.geohash] = [] - } - } - - @discardableResult - mutating func removeMessage(withID id: String) -> BitchatMessage? { - if let index = meshTimeline.firstIndex(where: { $0.id == id }) { - let removed = meshTimeline.remove(at: index) - meshMessageIDs.remove(id) - return removed - } - - for key in Array(geohashTimelines.keys) { - var timeline = geohashTimelines[key] ?? [] - if let index = timeline.firstIndex(where: { $0.id == id }) { - let removed = timeline.remove(at: index) - replaceGeohashTimeline(timeline, for: key, keepEmpty: false) - return removed - } - } - - return nil - } - - mutating func removeMessages(in geohash: String, where predicate: (BitchatMessage) -> Bool) { - var timeline = geohashTimelines[geohash] ?? [] - timeline.removeAll(where: predicate) - replaceGeohashTimeline(timeline, for: geohash, keepEmpty: false) - } - - mutating func mutateGeohash(_ geohash: String, _ transform: (inout [BitchatMessage]) -> Void) { - var timeline = geohashTimelines[geohash] ?? [] - transform(&timeline) - replaceGeohashTimeline(timeline, for: geohash, keepEmpty: false) - } - - mutating func queueGeohashSystemMessage(_ content: String) { - pendingGeohashSystemMessages.append(content) - } - - mutating func drainPendingGeohashSystemMessages() -> [String] { - defer { pendingGeohashSystemMessages.removeAll(keepingCapacity: false) } - return pendingGeohashSystemMessages - } - - func geohashKeys() -> [String] { - Array(geohashTimelines.keys) - } - - private mutating func trimMeshTimelineIfNeeded() { - guard meshTimeline.count > meshCap else { return } - meshTimeline = Array(meshTimeline.suffix(meshCap)) - meshMessageIDs = Set(meshTimeline.map(\.id)) - } - - private mutating func appendGeohashMessageIfAbsent(_ message: BitchatMessage, geohash: String) -> Bool { - var timeline = geohashTimelines[geohash] ?? [] - var messageIDs = geohashMessageIDs[geohash] ?? Set(timeline.map(\.id)) - guard messageIDs.insert(message.id).inserted else { return false } - - timeline.append(message) - trimGeohashTimelineIfNeeded(&timeline, messageIDs: &messageIDs) - geohashTimelines[geohash] = timeline - geohashMessageIDs[geohash] = messageIDs - return true - } - - private func trimGeohashTimelineIfNeeded(_ timeline: inout [BitchatMessage], messageIDs: inout Set) { - guard timeline.count > geohashCap else { return } - timeline = Array(timeline.suffix(geohashCap)) - messageIDs = Set(timeline.map(\.id)) - } - - private mutating func replaceGeohashTimeline(_ timeline: [BitchatMessage], for geohash: String, keepEmpty: Bool) { - if timeline.isEmpty && !keepEmpty { - geohashTimelines[geohash] = nil - geohashMessageIDs[geohash] = nil - return - } - - geohashTimelines[geohash] = timeline - geohashMessageIDs[geohash] = Set(timeline.map(\.id)) - } -} diff --git a/bitchatTests/ChatMediaTransferCoordinatorContextTests.swift b/bitchatTests/ChatMediaTransferCoordinatorContextTests.swift index 32cf8276..60932a5a 100644 --- a/bitchatTests/ChatMediaTransferCoordinatorContextTests.swift +++ b/bitchatTests/ChatMediaTransferCoordinatorContextTests.swift @@ -52,23 +52,17 @@ private final class MockChatMediaTransferContext: ChatMediaTransferContext { return true } - var meshTimeline: [BitchatMessage] = [] - private(set) var refreshedChannels: [ChannelID?] = [] - private(set) var trimCount = 0 + private(set) var appendedPublicMessages: [(message: BitchatMessage, conversationID: ConversationID)] = [] private(set) var removedMessages: [(messageID: String, cleanupFile: Bool)] = [] private(set) var systemMessages: [String] = [] private(set) var notifyUIChangedCount = 0 - func appendTimelineMessage(_ message: BitchatMessage, to channel: ChannelID) { - meshTimeline.append(message) + @discardableResult + func appendPublicMessage(_ message: BitchatMessage, to conversationID: ConversationID) -> Bool { + appendedPublicMessages.append((message, conversationID)) + return true } - func refreshVisibleMessages(from channel: ChannelID?) { - refreshedChannels.append(channel) - } - - func trimMessagesIfNeeded() { trimCount += 1 } - func removeMessage(withID messageID: String, cleanupFile: Bool) { removedMessages.append((messageID, cleanupFile)) } @@ -129,22 +123,21 @@ struct ChatMediaTransferCoordinatorContextTests { #expect(message.senderPeerID == context.myPeerID) #expect(message.deliveryStatus == .sending) #expect(context.recordedContentKeys == ["[voice] note.m4a"]) - #expect(context.trimCount == 1) #expect(context.notifyUIChangedCount == 1) - #expect(context.meshTimeline.isEmpty) + #expect(context.appendedPublicMessages.isEmpty) } @Test @MainActor - func enqueueMediaMessage_publicAppendsToTimelineAndRefreshes() async { + func enqueueMediaMessage_publicAppendsToActiveConversation() async { let context = MockChatMediaTransferContext() let coordinator = ChatMediaTransferCoordinator(context: context) let message = coordinator.enqueueMediaMessage(content: "[image] pic.jpg", targetPeer: nil) - #expect(context.meshTimeline.map(\.id) == [message.id]) + #expect(context.appendedPublicMessages.map(\.message.id) == [message.id]) + #expect(context.appendedPublicMessages.first?.conversationID == .mesh) #expect(!message.isPrivate) #expect(message.sender == "me") - #expect(context.refreshedChannels.count == 1) #expect(context.privateChats.isEmpty) #expect(context.notifyUIChangedCount == 1) } @@ -210,7 +203,7 @@ struct ChatMediaTransferCoordinatorContextTests { #expect(!FileManager.default.fileExists(atPath: url.path)) #expect(context.systemMessages == ["Voice notes are only available in mesh chats."]) #expect(context.privateChats.isEmpty) - #expect(context.meshTimeline.isEmpty) + #expect(context.appendedPublicMessages.isEmpty) #expect(coordinator.transferIdToMessageIDs.isEmpty) } } diff --git a/bitchatTests/ChatNostrCoordinatorContextTests.swift b/bitchatTests/ChatNostrCoordinatorContextTests.swift index d9d68c4c..7244f366 100644 --- a/bitchatTests/ChatNostrCoordinatorContextTests.swift +++ b/bitchatTests/ChatNostrCoordinatorContextTests.swift @@ -42,16 +42,13 @@ private final class MockChatNostrContext: ChatNostrContext { // Public timeline & pipeline var messages: [BitchatMessage] = [] - private(set) var pipelineResetCount = 0 - private(set) var pipelineChannelUpdates: [ChannelID] = [] + private(set) var pipelineFlushCount = 0 private(set) var refreshedChannels: [ChannelID?] = [] private(set) var publicSystemMessages: [String] = [] var pendingGeohashSystemMessages: [String] = [] private(set) var appendedGeohashMessages: [(message: BitchatMessage, geohash: String)] = [] - private(set) var synchronizedGeohashes: [String] = [] - func resetPublicMessagePipeline() { pipelineResetCount += 1 } - func updatePublicMessagePipelineChannel(_ channel: ChannelID) { pipelineChannelUpdates.append(channel) } + func flushPublicMessagePipeline() { pipelineFlushCount += 1 } func refreshVisibleMessages(from channel: ChannelID?) { refreshedChannels.append(channel) } func addPublicSystemMessage(_ content: String) { publicSystemMessages.append(content) } @@ -68,8 +65,6 @@ private final class MockChatNostrContext: ChatNostrContext { return true } - func synchronizePublicConversationStore(forGeohash geohash: String) { synchronizedGeohashes.append(geohash) } - // Inbound public messages private(set) var handledPublicMessages: [BitchatMessage] = [] private(set) var mentionCheckedMessageIDs: [String] = [] @@ -441,9 +436,8 @@ struct ChatNostrCoordinatorContextTests { coordinator.subscriptions.switchLocationChannel(to: .mesh) - #expect(context.pipelineResetCount == 1) + #expect(context.pipelineFlushCount == 1) #expect(context.activeChannel == .mesh) - #expect(context.pipelineChannelUpdates == [.mesh]) #expect(context.clearProcessedNostrEventsCount == 1) #expect(context.refreshedChannels == [.mesh]) #expect(context.refreshTimerStopCount == 1) @@ -578,7 +572,6 @@ struct GeoPresenceTrackerTests { await drainMainQueue() #expect(context.appendedGeohashMessages.isEmpty) #expect(context.lastGeoNotificationAt["9q8yy"] == recent) - #expect(context.synchronizedGeohashes.isEmpty) } @Test @MainActor @@ -605,7 +598,7 @@ struct GeoPresenceTrackerTests { #expect(context.appendGeohashMessageIfAbsent(placeholder, toGeohash: "9q8yy")) // Cooldown elapsed: the geohash is re-stamped and the append is - // attempted (and rejected as a duplicate, so no store sync either). + // attempted (and rejected as a duplicate, so no notification either). let stale = Date().addingTimeInterval(-TransportConfig.uiGeoNotifyCooldownSeconds - 1) context.lastGeoNotificationAt["9q8yy"] = stale tracker.cooldownPerGeohash("9q8yy", content: "sampled activity", event: event) @@ -614,7 +607,6 @@ struct GeoPresenceTrackerTests { let stamped = try #require(context.lastGeoNotificationAt["9q8yy"]) #expect(stamped > stale) #expect(context.appendedGeohashMessages.count == 1) - #expect(context.synchronizedGeohashes.isEmpty) } @Test @MainActor func handleFavoriteNotification_persistsFavoriteAndPostsLocalNotification() async throws { @@ -661,10 +653,9 @@ struct GeoPresenceTrackerTests { coordinator.presence.cooldownPerGeohash("u4pruyd", content: "hello geohash", event: first) await drainMainQueue() - // Sampled message recorded, store synced, and notification posted. + // Sampled message recorded in the store and notification posted. #expect(context.appendedGeohashMessages.map(\.message.id) == [first.id]) #expect(context.appendedGeohashMessages.first?.message.sender == "alice#" + String(first.pubkey.suffix(4))) - #expect(context.synchronizedGeohashes == ["u4pruyd"]) #expect(context.geohashActivityNotifications.count == 1) #expect(context.geohashActivityNotifications.first?.geohash == "u4pruyd") #expect(context.geohashActivityNotifications.first?.bodyPreview == "hello geohash") diff --git a/bitchatTests/ChatOutgoingCoordinatorContextTests.swift b/bitchatTests/ChatOutgoingCoordinatorContextTests.swift index fcb61d8f..6ed7c6da 100644 --- a/bitchatTests/ChatOutgoingCoordinatorContextTests.swift +++ b/bitchatTests/ChatOutgoingCoordinatorContextTests.swift @@ -49,21 +49,19 @@ private final class MockChatOutgoingContext: ChatOutgoingContext { } // Public timeline (local echo) - private(set) var appendedTimelineMessages: [(message: BitchatMessage, channel: ChannelID)] = [] - private(set) var refreshedChannels: [ChannelID?] = [] - private(set) var trimMessagesIfNeededCount = 0 + private(set) var appendedPublicMessages: [(message: BitchatMessage, conversationID: ConversationID)] = [] private(set) var systemMessages: [String] = [] func parseMentions(from content: String) -> [String] { content.contains("@bob") ? ["bob"] : [] } - func appendTimelineMessage(_ message: BitchatMessage, to channel: ChannelID) { - appendedTimelineMessages.append((message, channel)) + @discardableResult + func appendPublicMessage(_ message: BitchatMessage, to conversationID: ConversationID) -> Bool { + appendedPublicMessages.append((message, conversationID)) + return true } - func refreshVisibleMessages(from channel: ChannelID?) { refreshedChannels.append(channel) } - func trimMessagesIfNeeded() { trimMessagesIfNeededCount += 1 } func addSystemMessage(_ content: String) { systemMessages.append(content) } // Content dedup @@ -129,7 +127,7 @@ struct ChatOutgoingCoordinatorContextTests { await drainMainActorTasks() #expect(context.handledCommands == ["/who all"]) - #expect(context.appendedTimelineMessages.isEmpty) + #expect(context.appendedPublicMessages.isEmpty) #expect(context.sentMeshMessages.isEmpty) } @@ -153,7 +151,7 @@ struct ChatOutgoingCoordinatorContextTests { coordinator.sendMessage("dropped") await drainMainActorTasks() #expect(context.sentPrivateMessages.count == 1) - #expect(context.appendedTimelineMessages.isEmpty) + #expect(context.appendedPublicMessages.isEmpty) } @Test @MainActor @@ -165,16 +163,14 @@ struct ChatOutgoingCoordinatorContextTests { await drainMainActorTasks() // Local echo uses the trimmed content, own nickname/peer ID, mentions. - #expect(context.appendedTimelineMessages.count == 1) - let echo = context.appendedTimelineMessages[0] + #expect(context.appendedPublicMessages.count == 1) + let echo = context.appendedPublicMessages[0] #expect(echo.message.content == "hello @bob") #expect(echo.message.sender == "me") #expect(echo.message.senderPeerID == context.myPeerID) #expect(echo.message.mentions == ["bob"]) - #expect(echo.channel == .mesh) - #expect(context.refreshedChannels == [.mesh]) + #expect(echo.conversationID == .mesh) #expect(context.recordedContentKeys.map(\.key) == ["key:hello @bob"]) - #expect(context.trimMessagesIfNeededCount == 1) // The mesh send carries the original (untrimmed) content and reuses // the echo's message ID and timestamp; activity is stamped for "mesh". @@ -200,8 +196,9 @@ struct ChatOutgoingCoordinatorContextTests { // Local echo carries the geohash sender suffix (#last-4-of-pubkey) and // the signed event's ID; the send context targets the same channel. - #expect(context.appendedTimelineMessages.count == 1) - let echo = context.appendedTimelineMessages[0].message + #expect(context.appendedPublicMessages.count == 1) + let echo = context.appendedPublicMessages[0].message + #expect(context.appendedPublicMessages[0].conversationID == .geohash("u4pruydq")) #expect(echo.sender == "me#2222") #expect(context.recordedActivityKeys == ["geo:u4pruydq"]) #expect(context.sentGeohashContexts.count == 1) @@ -215,7 +212,7 @@ struct ChatOutgoingCoordinatorContextTests { coordinator.sendMessage("doomed") await drainMainActorTasks() #expect(context.systemMessages.count == 1) - #expect(context.appendedTimelineMessages.count == 1) + #expect(context.appendedPublicMessages.count == 1) #expect(context.sentGeohashContexts.count == 1) } } diff --git a/bitchatTests/ChatPublicConversationCoordinatorContextTests.swift b/bitchatTests/ChatPublicConversationCoordinatorContextTests.swift index 55b40079..9bcd2b93 100644 --- a/bitchatTests/ChatPublicConversationCoordinatorContextTests.swift +++ b/bitchatTests/ChatPublicConversationCoordinatorContextTests.swift @@ -25,15 +25,13 @@ import BitFoundation /// `ChatPublicConversationCoordinator` is testable without a `ChatViewModel`. @MainActor private final class MockChatPublicConversationContext: ChatPublicConversationContext { - // Channel & visible timeline state - var messages: [BitchatMessage] = [] + // Channel state var activeChannel: ChannelID = .mesh var currentGeohash: String? var nickname = "me" var myPeerID = PeerID(str: "0011223344556677") private(set) var isBatchingPublic = false private(set) var notifyUIChangedCount = 0 - private(set) var trimMessagesCount = 0 func setPublicBatching(_ isBatching: Bool) { isBatchingPublic = isBatching @@ -43,92 +41,59 @@ private final class MockChatPublicConversationContext: ChatPublicConversationCon notifyUIChangedCount += 1 } - func trimMessagesIfNeeded() { - trimMessagesCount += 1 - } - - // Public timeline store - var meshTimeline: [BitchatMessage] = [] - var geoTimelines: [String: [BitchatMessage]] = [:] + // Public conversation store (single-writer intents) + var conversations: [ConversationID: [BitchatMessage]] = [:] private(set) var queuedGeohashSystemMessages: [String] = [] - func timelineMessages(for channel: ChannelID) -> [BitchatMessage] { - switch channel { - case .mesh: return meshTimeline - case .location(let channel): return geoTimelines[channel.geohash] ?? [] - } + func publicMessages(in conversationID: ConversationID) -> [BitchatMessage] { + conversations[conversationID] ?? [] } - func appendTimelineMessage(_ message: BitchatMessage, to channel: ChannelID) { - switch channel { - case .mesh: meshTimeline.append(message) - case .location(let channel): geoTimelines[channel.geohash, default: []].append(message) - } - } - - func appendGeohashMessageIfAbsent(_ message: BitchatMessage, toGeohash geohash: String) -> Bool { - if geoTimelines[geohash]?.contains(where: { $0.id == message.id }) == true { + @discardableResult + func appendPublicMessage(_ message: BitchatMessage, to conversationID: ConversationID) -> Bool { + guard conversations[conversationID]?.contains(where: { $0.id == message.id }) != true else { return false } - geoTimelines[geohash, default: []].append(message) + conversations[conversationID, default: []].append(message) return true } - func removeTimelineMessage(withID id: String) -> BitchatMessage? { - if let index = meshTimeline.firstIndex(where: { $0.id == id }) { - return meshTimeline.remove(at: index) - } - for (geohash, timeline) in geoTimelines { - guard let index = timeline.firstIndex(where: { $0.id == id }) else { continue } + @discardableResult + func appendGeohashMessageIfAbsent(_ message: BitchatMessage, toGeohash geohash: String) -> Bool { + appendPublicMessage(message, to: .geohash(geohash.lowercased())) + } + + func publicConversationContainsMessage(withID messageID: String, in conversationID: ConversationID) -> Bool { + conversations[conversationID]?.contains(where: { $0.id == messageID }) == true + } + + @discardableResult + func removePublicMessage(withID messageID: String) -> BitchatMessage? { + for (conversationID, timeline) in conversations { + guard let index = timeline.firstIndex(where: { $0.id == messageID }) else { continue } var updated = timeline let removed = updated.remove(at: index) - geoTimelines[geohash] = updated + conversations[conversationID] = updated return removed } return nil } - func removeGeohashTimelineMessages(in geohash: String, where predicate: (BitchatMessage) -> Bool) { - geoTimelines[geohash]?.removeAll(where: predicate) + func removePublicMessages(fromGeohash geohash: String, where predicate: (BitchatMessage) -> Bool) { + conversations[.geohash(geohash.lowercased())]?.removeAll(where: predicate) } - func clearTimeline(for channel: ChannelID) { - switch channel { - case .mesh: meshTimeline.removeAll() - case .location(let channel): geoTimelines[channel.geohash] = [] - } - } + private(set) var clearedConversations: [ConversationID] = [] - func timelineGeohashKeys() -> [String] { - Array(geoTimelines.keys) + func clearPublicConversation(_ conversationID: ConversationID) { + clearedConversations.append(conversationID) + conversations[conversationID] = [] } func queueGeohashSystemMessage(_ content: String) { queuedGeohashSystemMessages.append(content) } - // Conversation stores - private(set) var conversationActiveChannels: [ChannelID] = [] - private(set) var replacedChannelMessages: [(channel: ChannelID, messageIDs: [String])] = [] - private(set) var replacedConversationMessages: [(conversation: ConversationID, messageIDs: [String])] = [] - private(set) var selectionStoreSyncCount = 0 - - func setConversationActiveChannel(_ channel: ChannelID) { - conversationActiveChannels.append(channel) - } - - func replaceConversationMessages(_ messages: [BitchatMessage], for channelID: ChannelID) { - replacedChannelMessages.append((channelID, messages.map(\.id))) - } - - func replaceConversationMessages(_ messages: [BitchatMessage], for conversationID: ConversationID) { - replacedConversationMessages.append((conversationID, messages.map(\.id))) - } - - func synchronizeConversationSelectionStore() { - selectionStoreSyncCount += 1 - } - // Private chats var privateChats: [PeerID: [BitchatMessage]] = [:] var unreadPrivateMessages: Set = [] @@ -222,7 +187,8 @@ private final class MockChatPublicConversationContext: ChatPublicConversationCon var blockedMessageIDs: Set = [] var rateLimitAllowed = true private(set) var rateLimitChecks: [(senderKey: String, contentKey: String)] = [] - private(set) var enqueuedMessageIDs: [String] = [] + private(set) var enqueuedMessages: [(messageID: String, conversationID: ConversationID)] = [] + var enqueuedMessageIDs: [String] { enqueuedMessages.map(\.messageID) } var stablePeerIDs: [PeerID: PeerID] = [:] func processActionMessage(_ message: BitchatMessage) -> BitchatMessage { @@ -238,8 +204,8 @@ private final class MockChatPublicConversationContext: ChatPublicConversationCon return rateLimitAllowed } - func enqueuePublicMessage(_ message: BitchatMessage) { - enqueuedMessageIDs.append(message.id) + func enqueuePublicMessage(_ message: BitchatMessage, to conversationID: ConversationID) { + enqueuedMessages.append((message.id, conversationID)) } func cachedStablePeerID(for shortPeerID: PeerID) -> PeerID? { @@ -309,24 +275,24 @@ private func makePublicMessage( struct ChatPublicConversationCoordinatorContextTests { @Test @MainActor - func handlePublicMessage_meshMessage_appendsSyncsStoreAndEnqueues() async { + func handlePublicMessage_meshMessage_enqueuesForBatchedStoreCommit() async { let context = MockChatPublicConversationContext() let coordinator = ChatPublicConversationCoordinator(context: context) let message = makePublicMessage(id: "mesh-msg-1", content: "Hello Mesh") coordinator.handlePublicMessage(message) - #expect(context.meshTimeline.map(\.id) == ["mesh-msg-1"]) - #expect(context.replacedChannelMessages.count == 1) - #expect(context.replacedChannelMessages.first?.channel == .mesh) - #expect(context.replacedChannelMessages.first?.messageIDs == ["mesh-msg-1"]) + // Visible-channel arrival: buffered for the batched pipeline flush + // (which commits to the store), not appended directly. #expect(context.rateLimitChecks.count == 1) #expect(context.rateLimitChecks.first?.senderKey == "mesh:aabbccddeeff0011") #expect(context.rateLimitChecks.first?.contentKey == "hello mesh") - #expect(context.enqueuedMessageIDs == ["mesh-msg-1"]) + #expect(context.enqueuedMessages.map(\.messageID) == ["mesh-msg-1"]) + #expect(context.enqueuedMessages.first?.conversationID == .mesh) + #expect(context.publicMessages(in: .mesh).isEmpty) - // Already visible in the timeline: stored again, but not re-enqueued. - context.messages = [message] + // Already committed to the store: not re-enqueued. + context.appendPublicMessage(message, to: .mesh) coordinator.handlePublicMessage(message) #expect(context.enqueuedMessageIDs == ["mesh-msg-1"]) } @@ -340,14 +306,14 @@ struct ChatPublicConversationCoordinatorContextTests { context.blockedMessageIDs = ["blocked-msg"] coordinator.handlePublicMessage(makePublicMessage(id: "blocked-msg")) #expect(context.rateLimitChecks.isEmpty) - #expect(context.meshTimeline.isEmpty) + #expect(context.publicMessages(in: .mesh).isEmpty) #expect(context.enqueuedMessageIDs.isEmpty) // Rate limited: consulted, then dropped before storage. context.rateLimitAllowed = false coordinator.handlePublicMessage(makePublicMessage(id: "limited-msg")) #expect(context.rateLimitChecks.count == 1) - #expect(context.meshTimeline.isEmpty) + #expect(context.publicMessages(in: .mesh).isEmpty) #expect(context.enqueuedMessageIDs.isEmpty) } @@ -364,16 +330,15 @@ struct ChatPublicConversationCoordinatorContextTests { senderPeerID: PeerID(nostr: senderHex) ) - // On mesh channel: stored in the geohash timeline but not enqueued. + // On mesh channel: a background-channel arrival lands in the geohash + // conversation immediately, with no pipeline batching. context.activeChannel = .mesh coordinator.handlePublicMessage(geoMessage) - #expect(context.geoTimelines[geohash]?.map(\.id) == ["geo-msg-1"]) - #expect(context.replacedConversationMessages.count == 1) - #expect(context.replacedConversationMessages.first?.conversation == .geohash(geohash)) - #expect(context.meshTimeline.isEmpty) + #expect(context.publicMessages(in: .geohash(geohash)).map(\.id) == ["geo-msg-1"]) + #expect(context.publicMessages(in: .mesh).isEmpty) #expect(context.enqueuedMessageIDs.isEmpty) - // On the matching location channel: enqueued for display. + // On the matching location channel: enqueued for the batched flush. context.activeChannel = .location(GeohashChannel(level: .city, geohash: geohash)) let second = makePublicMessage( id: "geo-msg-2", @@ -381,7 +346,8 @@ struct ChatPublicConversationCoordinatorContextTests { senderPeerID: PeerID(nostr: senderHex) ) coordinator.handlePublicMessage(second) - #expect(context.enqueuedMessageIDs == ["geo-msg-2"]) + #expect(context.enqueuedMessages.map(\.messageID) == ["geo-msg-2"]) + #expect(context.enqueuedMessages.first?.conversationID == .geohash(geohash)) } @Test @MainActor @@ -396,8 +362,7 @@ struct ChatPublicConversationCoordinatorContextTests { context.currentGeohash = geohash context.activeChannel = .location(GeohashChannel(level: .city, geohash: geohash)) - context.geoTimelines[geohash] = [geoMessage] - context.messages = [geoMessage] + context.conversations[.geohash(geohash)] = [geoMessage] context.nostrKeyMapping = [senderPeerID: hex, convKey: hex] context.privateChats[convKey] = [geoMessage] context.unreadPrivateMessages = [convKey] @@ -406,13 +371,14 @@ struct ChatPublicConversationCoordinatorContextTests { #expect(context.blockedNostrPubkeys.contains(hex)) #expect(context.removedGeoParticipants == [hex]) - #expect(context.geoTimelines[geohash]?.isEmpty == true) #expect(context.privateChats[convKey] == nil) #expect(context.unreadPrivateMessages.isEmpty) #expect(context.nostrKeyMapping.isEmpty) - // The blocked user's visible message is gone; a system notice was added. - #expect(!context.messages.contains(where: { $0.id == "geo-bad-1" })) - #expect(context.messages.last?.sender == "system") + // The blocked user's message is purged from the geohash conversation + // (the visible timeline is the same conversation now); a system + // notice was appended to the active conversation. + #expect(!context.publicMessages(in: .geohash(geohash)).contains(where: { $0.id == "geo-bad-1" })) + #expect(context.publicMessages(in: .geohash(geohash)).last?.sender == "system") coordinator.unblockGeohashUser(pubkeyHexLowercased: hex, displayName: "rude#abcd") #expect(!context.blockedNostrPubkeys.contains(hex)) @@ -424,36 +390,27 @@ struct ChatPublicConversationCoordinatorContextTests { let coordinator = ChatPublicConversationCoordinator(context: context) let peerID = PeerID(str: "0102030405060708") let message = makePublicMessage(id: "doomed-msg") - context.messages = [message] - context.meshTimeline = [message] + context.conversations[.mesh] = [message] context.privateChats[peerID] = [message] coordinator.removeMessage(withID: "doomed-msg", cleanupFile: true) - #expect(context.messages.isEmpty) - #expect(context.meshTimeline.isEmpty) + #expect(context.publicMessages(in: .mesh).isEmpty) #expect(context.privateChats[peerID] == nil) #expect(context.cleanedUpFileMessageIDs == ["doomed-msg"]) #expect(context.notifyUIChangedCount == 1) - // Timeline removal triggers a full conversation-store resync. - #expect(context.replacedChannelMessages.contains(where: { $0.channel == .mesh && $0.messageIDs.isEmpty })) } @Test @MainActor - func addPublicSystemMessage_appendsRefreshesAndRecordsContentKey() async { + func addPublicSystemMessage_appendsToActiveConversationAndRecordsContentKey() async { let context = MockChatPublicConversationContext() let coordinator = ChatPublicConversationCoordinator(context: context) coordinator.addPublicSystemMessage("Tor Ready") - #expect(context.meshTimeline.count == 1) - #expect(context.meshTimeline.first?.sender == "system") - // refreshVisibleMessages mirrors the timeline into the visible list. - #expect(context.messages.map(\.id) == context.meshTimeline.map(\.id)) + #expect(context.publicMessages(in: .mesh).count == 1) + #expect(context.publicMessages(in: .mesh).first?.sender == "system") #expect(context.recordedContentKeys.map(\.key) == ["tor ready"]) - #expect(context.trimMessagesCount == 1) - #expect(context.notifyUIChangedCount == 1) - #expect(context.conversationActiveChannels == [.mesh]) // On mesh, geohash-only system messages are queued for the next geo visit. coordinator.addGeohashOnlySystemMessage("geo notice") @@ -477,22 +434,20 @@ struct ChatPublicConversationCoordinatorContextTests { let coordinator = ChatPublicConversationCoordinator(context: context) let pipeline = PublicMessagePipeline() let message = makePublicMessage(id: "pipeline-msg") - context.messages = [message] context.contentTimestamps["key-1"] = Date(timeIntervalSince1970: 42) - #expect(coordinator.pipelineCurrentMessages(pipeline).map(\.id) == ["pipeline-msg"]) #expect(coordinator.pipeline(pipeline, normalizeContent: "HeLLo") == "hello") #expect(coordinator.pipeline(pipeline, contentTimestampForKey: "key-1") == Date(timeIntervalSince1970: 42)) - coordinator.pipeline(pipeline, setMessages: []) - #expect(context.messages.isEmpty) + // Commit lands in the store via the append intent; a duplicate ID + // reports `false` (the store's dedup contract). + #expect(coordinator.pipeline(pipeline, commit: message, to: .mesh)) + #expect(context.publicMessages(in: .mesh).map(\.id) == ["pipeline-msg"]) + #expect(!coordinator.pipeline(pipeline, commit: message, to: .mesh)) coordinator.pipeline(pipeline, recordContentKey: "key-2", timestamp: Date(timeIntervalSince1970: 7)) #expect(context.recordedContentKeys.map(\.key) == ["key-2"]) - coordinator.pipelineTrimMessages(pipeline) - #expect(context.trimMessagesCount == 1) - coordinator.pipelinePrewarmMessage(pipeline, message: message) #expect(context.prewarmedMessageIDs == ["pipeline-msg"]) diff --git a/bitchatTests/ChatViewModelDeliveryStatusTests.swift b/bitchatTests/ChatViewModelDeliveryStatusTests.swift index d176b249..ecebdae3 100644 --- a/bitchatTests/ChatViewModelDeliveryStatusTests.swift +++ b/bitchatTests/ChatViewModelDeliveryStatusTests.swift @@ -233,7 +233,7 @@ struct ChatViewModelDeliveryStatusTests { isPrivate: false, deliveryStatus: .sending ) - viewModel.messages.append(message) + viewModel.seedPublicMessages([message]) // Action: update to .sent viewModel.didUpdateMessageDeliveryStatus(messageID, status: .sent) @@ -253,7 +253,7 @@ struct ChatViewModelDeliveryStatusTests { let firstPeerID = PeerID(str: "0102030405060708") let secondPeerID = PeerID(str: "1112131415161718") - viewModel.messages = [ + viewModel.seedPublicMessages([ BitchatMessage( id: messageID, sender: viewModel.nickname, @@ -264,7 +264,7 @@ struct ChatViewModelDeliveryStatusTests { senderPeerID: transport.myPeerID, deliveryStatus: .sent ) - ] + ]) viewModel.seedPrivateChat([ BitchatMessage( id: messageID, diff --git a/bitchatTests/ChatViewModelRefactoringTests.swift b/bitchatTests/ChatViewModelRefactoringTests.swift index fcb3f13f..d576106d 100644 --- a/bitchatTests/ChatViewModelRefactoringTests.swift +++ b/bitchatTests/ChatViewModelRefactoringTests.swift @@ -138,7 +138,7 @@ struct ChatViewModelRefactoringTests { // Wait for async processing with proper timeout let found = await TestHelpers.waitUntil( { - viewModel.timelineStore.messages(for: .mesh).contains(where: { $0.content == "Public Hi" }) + viewModel.publicMessages(for: .mesh).contains(where: { $0.content == "Public Hi" }) }, timeout: TestConstants.defaultTimeout ) diff --git a/bitchatTests/ChatViewModelTests.swift b/bitchatTests/ChatViewModelTests.swift index 7b442ed5..ecba96d0 100644 --- a/bitchatTests/ChatViewModelTests.swift +++ b/bitchatTests/ChatViewModelTests.swift @@ -438,7 +438,7 @@ struct ChatViewModelReceivingTests { ) let found = await TestHelpers.waitUntil({ - viewModel.timelineStore.messages(for: .mesh).contains { $0.content == "Public hello from Bob" } + viewModel.publicMessages(for: .mesh).contains { $0.content == "Public hello from Bob" } }, timeout: TestConstants.defaultTimeout) #expect(found) @@ -709,10 +709,12 @@ struct ChatViewModelPublicConversationTests { let (viewModel, _) = makeTestableViewModel() viewModel.addPublicSystemMessage("system refresh test") - viewModel.messages.removeAll() viewModel.refreshVisibleMessages(from: .mesh) + // The system message lives in the mesh conversation itself, so the + // derived `messages` view still surfaces it after a refresh. #expect(viewModel.messages.last?.content == "system refresh test") + #expect(viewModel.publicMessages(for: .mesh).last?.content == "system refresh test") } @Test @MainActor @@ -726,6 +728,18 @@ struct ChatViewModelPublicConversationTests { viewModel.refreshVisibleMessages(from: .mesh) #expect(viewModel.messages.isEmpty) + #expect(viewModel.publicMessages(for: .mesh).isEmpty) + } + + @Test @MainActor + func queuedGeohashSystemMessages_drainOnce() async { + let (viewModel, _) = makeTestableViewModel() + + viewModel.queueGeohashSystemMessage("first") + viewModel.queueGeohashSystemMessage("second") + + #expect(viewModel.drainPendingGeohashSystemMessages() == ["first", "second"]) + #expect(viewModel.drainPendingGeohashSystemMessages().isEmpty) } } @@ -999,7 +1013,7 @@ struct ChatViewModelPanicTests { // Set up some state transport.connectedPeers.insert(PeerID(str: "PEER1")) - viewModel.messages = [ + viewModel.seedPublicMessages([ BitchatMessage( id: "panic-1", sender: "Tester", @@ -1007,7 +1021,7 @@ struct ChatViewModelPanicTests { timestamp: Date(), isRelay: false ) - ] + ]) viewModel.seedPrivateChat([ BitchatMessage( id: "pm-1", diff --git a/bitchatTests/ConversationStoreTests.swift b/bitchatTests/ConversationStoreTests.swift index 4a805134..8535cc57 100644 --- a/bitchatTests/ConversationStoreTests.swift +++ b/bitchatTests/ConversationStoreTests.swift @@ -468,4 +468,98 @@ struct ConversationStoreTests { store.append(makeMessage(id: "m3", timestamp: 3), to: b) #expect(bWillChangeCount > 0) } + + // MARK: - Public timelines (mesh/geohash, ex-PublicTimelineStore behavior) + + @Test("geohash conversations are separated by geohash and from mesh") + @MainActor + func geohashConversationSeparation() { + let store = ConversationStore() + store.append(makeMessage(id: "mesh-1", timestamp: 1), to: .mesh) + store.append(makeMessage(id: "geo-a-1", timestamp: 2), to: .geohash("u4pruyd")) + store.append(makeMessage(id: "geo-b-1", timestamp: 3), to: .geohash("9q8yy")) + + #expect(store.conversation(for: .mesh).messages.map(\.id) == ["mesh-1"]) + #expect(store.conversation(for: .geohash("u4pruyd")).messages.map(\.id) == ["geo-a-1"]) + #expect(store.conversation(for: .geohash("9q8yy")).messages.map(\.id) == ["geo-b-1"]) + } + + @Test("geohash append dedups by ID and reports duplicates") + @MainActor + func geohashAppendIfAbsentContract() { + let store = ConversationStore() + let message = makeMessage(id: "geo-1", timestamp: 1) + + #expect(store.append(message, to: .geohash("u4pruyd"))) + #expect(!store.append(message, to: .geohash("u4pruyd"))) + // The same ID is still fresh in a different geohash. + #expect(store.append(message, to: .geohash("9q8yy"))) + } + + @Test("removePublicMessage searches mesh and geohash conversations only") + @MainActor + func removePublicMessageSearchesPublicConversations() { + let store = ConversationStore() + let direct = makeDirectConversationID("aa") + store.append(makeMessage(id: "mesh-1", timestamp: 1), to: .mesh) + store.append(makeMessage(id: "geo-1", timestamp: 2), to: .geohash("u4pruyd")) + store.append(makeMessage(id: "dm-1", timestamp: 3, isPrivate: true), to: direct) + + #expect(store.removePublicMessage(withID: "geo-1")?.id == "geo-1") + #expect(store.conversation(for: .geohash("u4pruyd")).messages.isEmpty) + + #expect(store.removePublicMessage(withID: "mesh-1")?.id == "mesh-1") + #expect(store.conversation(for: .mesh).messages.isEmpty) + + // Direct conversations are never touched. + #expect(store.removePublicMessage(withID: "dm-1") == nil) + #expect(store.conversation(for: direct).messages.map(\.id) == ["dm-1"]) + } + + @Test("removeMessages(from:where:) purges matches and emits per removal") + @MainActor + func removeMessagesByPredicate() { + let store = ConversationStore() + let id = ConversationID.geohash("u4pruyd") + store.append(makeMessage(id: "keep-1", timestamp: 1), to: id) + store.append(makeMessage(id: "drop-1", timestamp: 2, content: "purge me"), to: id) + store.append(makeMessage(id: "drop-2", timestamp: 3, content: "purge me"), to: id) + store.append(makeMessage(id: "keep-2", timestamp: 4), to: id) + + var removedIDs: [String] = [] + var cancellables = Set() + store.changes + .sink { change in + if case .messageRemoved(_, let messageID) = change { + removedIDs.append(messageID) + } + } + .store(in: &cancellables) + + store.removeMessages(from: id, where: { $0.content == "purge me" }) + + #expect(store.conversation(for: id).messages.map(\.id) == ["keep-1", "keep-2"]) + #expect(removedIDs == ["drop-1", "drop-2"]) + // The ID index survives the purge: dedup and removal still work. + #expect(!store.append(makeMessage(id: "keep-2", timestamp: 4), to: id)) + #expect(store.removeMessage(withID: "keep-1", from: id) != nil) + #expect(store.conversation(for: id).messages.map(\.id) == ["keep-2"]) + } + + @Test("trimmed public message IDs can return after falling off the cap") + @MainActor + func trimmedMessageIDsCanReturn() { + let store = ConversationStore() + let id = ConversationID.geohash("u4pruyd") + let conversation = store.conversation(for: id) + let first = makeMessage(id: "one", timestamp: 1) + + store.append(first, to: id) + for index in 0.. [String] { [] } func appendGeohashMessageIfAbsent(_ message: BitchatMessage, toGeohash geohash: String) -> Bool { true } - func synchronizePublicConversationStore(forGeohash geohash: String) {} private(set) var handledPublicMessageCount = 0 func handlePublicMessage(_ message: BitchatMessage) { handledPublicMessageCount += 1 } diff --git a/bitchatTests/PublicMessagePipelineTests.swift b/bitchatTests/PublicMessagePipelineTests.swift index 9bf0a5cf..6d1458da 100644 --- a/bitchatTests/PublicMessagePipelineTests.swift +++ b/bitchatTests/PublicMessagePipelineTests.swift @@ -2,7 +2,10 @@ // PublicMessagePipelineTests.swift // bitchatTests // -// Tests for PublicMessagePipeline ordering and deduplication. +// Tests for PublicMessagePipeline batching, content dedup, and per-message +// conversation routing. Ordering and ID dedup live in the ConversationStore +// the flush commits into (the old late-insert threshold is gone; see +// ConversationStoreTests for ordered-insert coverage). // import Testing @@ -13,14 +16,15 @@ import BitFoundation @MainActor private final class TestPipelineDelegate: PublicMessagePipelineDelegate { private let dedupService = MessageDeduplicationService() - var messages: [BitchatMessage] = [] + /// Commits in arrival-at-commit order, per conversation. + private(set) var committed: [(message: BitchatMessage, conversationID: ConversationID)] = [] + /// Message IDs the commit rejects (simulates the store's ID dedup). + var rejectedMessageIDs: Set = [] + private(set) var recordedContentKeys: [String] = [] + private(set) var batchingStates: [Bool] = [] - func pipelineCurrentMessages(_ pipeline: PublicMessagePipeline) -> [BitchatMessage] { - messages - } - - func pipeline(_ pipeline: PublicMessagePipeline, setMessages messages: [BitchatMessage]) { - self.messages = messages + func messages(in conversationID: ConversationID) -> [BitchatMessage] { + committed.filter { $0.conversationID == conversationID }.map(\.message) } func pipeline(_ pipeline: PublicMessagePipeline, normalizeContent content: String) -> String { @@ -33,19 +37,37 @@ private final class TestPipelineDelegate: PublicMessagePipelineDelegate { func pipeline(_ pipeline: PublicMessagePipeline, recordContentKey key: String, timestamp: Date) { dedupService.recordContentKey(key, timestamp: timestamp) + recordedContentKeys.append(key) } - func pipelineTrimMessages(_ pipeline: PublicMessagePipeline) {} + func pipeline(_ pipeline: PublicMessagePipeline, commit message: BitchatMessage, to conversationID: ConversationID) -> Bool { + guard !rejectedMessageIDs.contains(message.id) else { return false } + committed.append((message, conversationID)) + return true + } func pipelinePrewarmMessage(_ pipeline: PublicMessagePipeline, message: BitchatMessage) {} - func pipelineSetBatchingState(_ pipeline: PublicMessagePipeline, isBatching: Bool) {} + func pipelineSetBatchingState(_ pipeline: PublicMessagePipeline, isBatching: Bool) { + batchingStates.append(isBatching) + } +} + +@MainActor +private func makeMessage(id: String, content: String, timestamp: Date) -> BitchatMessage { + BitchatMessage( + id: id, + sender: "A", + content: content, + timestamp: timestamp, + isRelay: false + ) } struct PublicMessagePipelineTests { @Test @MainActor - func flush_sortsByTimestamp() async { + func flush_commitsInTimestampOrder() async { let pipeline = PublicMessagePipeline() let delegate = TestPipelineDelegate() pipeline.delegate = delegate @@ -53,26 +75,13 @@ struct PublicMessagePipelineTests { let earlier = Date().addingTimeInterval(-10) let later = Date() - let messageA = BitchatMessage( - id: "a", - sender: "A", - content: "Later", - timestamp: later, - isRelay: false - ) - let messageB = BitchatMessage( - id: "b", - sender: "A", - content: "Earlier", - timestamp: earlier, - isRelay: false - ) - - pipeline.enqueue(messageA) - pipeline.enqueue(messageB) + pipeline.enqueue(makeMessage(id: "a", content: "Later", timestamp: later), to: .mesh) + pipeline.enqueue(makeMessage(id: "b", content: "Earlier", timestamp: earlier), to: .mesh) pipeline.flushIfNeeded() - #expect(delegate.messages.map { $0.id } == ["b", "a"]) + #expect(delegate.messages(in: .mesh).map { $0.id } == ["b", "a"]) + // Batching state wrapped the flush. + #expect(delegate.batchingStates == [true, false]) } @Test @MainActor @@ -82,86 +91,41 @@ struct PublicMessagePipelineTests { pipeline.delegate = delegate let now = Date() - let messageA = BitchatMessage( - id: "a", - sender: "A", - content: "Same", - timestamp: now, - isRelay: false - ) - let messageB = BitchatMessage( - id: "b", - sender: "A", - content: "Same", - timestamp: now.addingTimeInterval(0.2), - isRelay: false - ) - - pipeline.enqueue(messageA) - pipeline.enqueue(messageB) + pipeline.enqueue(makeMessage(id: "a", content: "Same", timestamp: now), to: .mesh) + pipeline.enqueue(makeMessage(id: "b", content: "Same", timestamp: now.addingTimeInterval(0.2)), to: .mesh) pipeline.flushIfNeeded() - #expect(delegate.messages.count == 1) - #expect(delegate.messages.first?.content == "Same") + #expect(delegate.messages(in: .mesh).count == 1) + #expect(delegate.messages(in: .mesh).first?.content == "Same") } @Test @MainActor - func lateInsert_meshAppendsRecentOlderMessage() async { + func flush_routesEachMessageToItsConversation() async { let pipeline = PublicMessagePipeline() let delegate = TestPipelineDelegate() pipeline.delegate = delegate - pipeline.updateActiveChannel(.mesh) let base = Date() - let newer = BitchatMessage( - id: "new", - sender: "A", - content: "New", - timestamp: base, - isRelay: false - ) - let older = BitchatMessage( - id: "old", - sender: "A", - content: "Old", - timestamp: base.addingTimeInterval(-5), - isRelay: false - ) - - delegate.messages = [newer] - pipeline.enqueue(older) + pipeline.enqueue(makeMessage(id: "mesh-1", content: "mesh hello", timestamp: base), to: .mesh) + // A channel switch mid-batch must not misroute already-buffered messages. + pipeline.enqueue(makeMessage(id: "geo-1", content: "geo hello", timestamp: base.addingTimeInterval(1)), to: .geohash("u4pruydq")) pipeline.flushIfNeeded() - #expect(delegate.messages.map { $0.id } == ["new", "old"]) + #expect(delegate.messages(in: .mesh).map { $0.id } == ["mesh-1"]) + #expect(delegate.messages(in: .geohash("u4pruydq")).map { $0.id } == ["geo-1"]) } @Test @MainActor - func lateInsert_locationInsertsByTimestamp() async { + func flush_rejectedCommitDoesNotRecordContentKey() async { let pipeline = PublicMessagePipeline() let delegate = TestPipelineDelegate() pipeline.delegate = delegate - pipeline.updateActiveChannel(.location(GeohashChannel(level: .city, geohash: "u4pruydq"))) + delegate.rejectedMessageIDs = ["dup"] - let base = Date() - let newer = BitchatMessage( - id: "new", - sender: "A", - content: "New", - timestamp: base, - isRelay: false - ) - let older = BitchatMessage( - id: "old", - sender: "A", - content: "Old", - timestamp: base.addingTimeInterval(-5), - isRelay: false - ) - - delegate.messages = [newer] - pipeline.enqueue(older) + pipeline.enqueue(makeMessage(id: "dup", content: "already stored", timestamp: Date()), to: .mesh) pipeline.flushIfNeeded() - #expect(delegate.messages.map { $0.id } == ["old", "new"]) + #expect(delegate.messages(in: .mesh).isEmpty) + #expect(delegate.recordedContentKeys.isEmpty) } } diff --git a/bitchatTests/PublicTimelineStoreTests.swift b/bitchatTests/PublicTimelineStoreTests.swift deleted file mode 100644 index 17081aca..00000000 --- a/bitchatTests/PublicTimelineStoreTests.swift +++ /dev/null @@ -1,115 +0,0 @@ -import Foundation -import BitFoundation -import Testing -@testable import bitchat - -@Suite("PublicTimelineStore Tests") -struct PublicTimelineStoreTests { - - @Test("Mesh timeline deduplicates and trims to cap") - func meshTimelineDeduplicatesAndTrims() { - var store = PublicTimelineStore(meshCap: 2, geohashCap: 2) - let first = TestHelpers.createTestMessage(content: "one") - let second = TestHelpers.createTestMessage(content: "two") - let third = TestHelpers.createTestMessage(content: "three") - - store.append(first, to: .mesh) - store.append(second, to: .mesh) - store.append(first, to: .mesh) - store.append(third, to: .mesh) - - let messages = store.messages(for: .mesh) - #expect(messages.map(\.content) == ["two", "three"]) - } - - @Test("Timeline indexes allow trimmed message IDs to return") - func timelineIndexesAllowTrimmedMessageIDsToReturn() { - var store = PublicTimelineStore(meshCap: 2, geohashCap: 2) - let first = timelineMessage(id: "one", content: "one", timestamp: 1) - let second = timelineMessage(id: "two", content: "two", timestamp: 2) - let third = timelineMessage(id: "three", content: "three", timestamp: 3) - - store.append(first, to: .mesh) - store.append(second, to: .mesh) - store.append(third, to: .mesh) - store.append(first, to: .mesh) - - #expect(store.messages(for: .mesh).map(\.content) == ["three", "one"]) - - let geohash = "u4pruydq" - let channel = ChannelID.location(GeohashChannel(level: .city, geohash: geohash)) - let geoFirst = timelineMessage(id: "geo-one", content: "geo one", timestamp: 1) - let geoSecond = timelineMessage(id: "geo-two", content: "geo two", timestamp: 2) - let geoThird = timelineMessage(id: "geo-three", content: "geo three", timestamp: 3) - - let didAppendGeoFirst = store.appendIfAbsent(geoFirst, toGeohash: geohash) - let didAppendGeoSecond = store.appendIfAbsent(geoSecond, toGeohash: geohash) - let didAppendGeoThird = store.appendIfAbsent(geoThird, toGeohash: geohash) - let didReappendGeoFirst = store.appendIfAbsent(geoFirst, toGeohash: geohash) - - #expect(didAppendGeoFirst) - #expect(didAppendGeoSecond) - #expect(didAppendGeoThird) - #expect(didReappendGeoFirst) - #expect(store.messages(for: channel).map(\.content) == ["geo one", "geo three"]) - } - - @Test("Geohash appendIfAbsent remove and clear work together") - func geohashStoreSupportsAppendRemoveAndClear() { - var store = PublicTimelineStore(meshCap: 2, geohashCap: 3) - let geohash = "u4pruydq" - let channel = ChannelID.location(GeohashChannel(level: .city, geohash: geohash)) - let first = TestHelpers.createTestMessage(content: "geo one") - let second = TestHelpers.createTestMessage(content: "geo two") - - let didAppendFirst = store.appendIfAbsent(first, toGeohash: geohash) - let didAppendDuplicate = store.appendIfAbsent(first, toGeohash: geohash) - - #expect(didAppendFirst) - #expect(!didAppendDuplicate) - store.append(second, toGeohash: geohash) - let removed = store.removeMessage(withID: first.id) - - #expect(removed?.id == first.id) - #expect(store.messages(for: channel).map(\.content) == ["geo two"]) - - store.clear(channel: channel) - #expect(store.messages(for: channel).isEmpty) - } - - @Test("Mutate geohash updates stored messages in place") - func mutateGeohashAppliesTransformation() { - var store = PublicTimelineStore(meshCap: 2, geohashCap: 3) - let geohash = "u4pruydq" - let channel = ChannelID.location(GeohashChannel(level: .city, geohash: geohash)) - let first = TestHelpers.createTestMessage(content: "geo one") - - store.append(first, toGeohash: geohash) - store.mutateGeohash(geohash) { timeline in - timeline.append(TestHelpers.createTestMessage(content: "geo two")) - } - - #expect(store.messages(for: channel).map(\.content) == ["geo one", "geo two"]) - } - - @Test("Queued geohash system messages drain once") - func pendingGeohashSystemMessagesDrainOnce() { - var store = PublicTimelineStore(meshCap: 1, geohashCap: 1) - - store.queueGeohashSystemMessage("first") - store.queueGeohashSystemMessage("second") - - #expect(store.drainPendingGeohashSystemMessages() == ["first", "second"]) - #expect(store.drainPendingGeohashSystemMessages().isEmpty) - } - - private func timelineMessage(id: String, content: String, timestamp: TimeInterval) -> BitchatMessage { - BitchatMessage( - id: id, - sender: "alice", - content: content, - timestamp: Date(timeIntervalSince1970: timestamp), - isRelay: false - ) - } -} diff --git a/bitchatTests/TestUtilities/TestHelpers.swift b/bitchatTests/TestUtilities/TestHelpers.swift index c09c60c5..21894e91 100644 --- a/bitchatTests/TestUtilities/TestHelpers.swift +++ b/bitchatTests/TestUtilities/TestHelpers.swift @@ -153,6 +153,24 @@ extension ChatViewModel { } } + /// Test-only replacement for the deleted `messages` setter: seeds a + /// public channel's conversation through the single-writer + /// `ConversationStore` intents (upsert keeps re-seeding with updated + /// copies working the way the old array assignment did). + @MainActor + func seedPublicMessages(_ messages: [BitchatMessage], for channel: ChannelID = .mesh) { + for message in messages { + conversations.upsertByID(message, in: ConversationID(channelID: channel)) + } + } + + /// Test-only replacement for `messages.removeAll()`: empties a public + /// channel's conversation. + @MainActor + func clearPublicMessages(for channel: ChannelID = .mesh) { + conversations.clear(ConversationID(channelID: channel)) + } + /// Test-only: drops every private chat and unread flag. @MainActor func clearAllPrivateChats() { diff --git a/docs/CONVERSATION-STORE-DESIGN.md b/docs/CONVERSATION-STORE-DESIGN.md index 7b01e7e9..7f63342e 100644 --- a/docs/CONVERSATION-STORE-DESIGN.md +++ b/docs/CONVERSATION-STORE-DESIGN.md @@ -1,8 +1,9 @@ # Conversation Store: Single Source of Truth -**Status:** Approved design, not yet implemented. Baselines recorded in +**Status:** Steps 1–3 implemented (additive store, private cutover, public +cutover; `PublicTimelineStore` deleted). Baselines recorded in `bitchatTests/Performance/PerformanceBaselineTests.swift` (`pipeline.privateIngest`, -`pipeline.publicIngest`). +`pipeline.publicIngest`, `store.append`). ---