diff --git a/bitchat/ViewModels/ChatViewModel.swift b/bitchat/ViewModels/ChatViewModel.swift index fce48174..702ed36e 100644 --- a/bitchat/ViewModels/ChatViewModel.swift +++ b/bitchat/ViewModels/ChatViewModel.swift @@ -365,7 +365,7 @@ final class ChatViewModel: ObservableObject, BitchatDelegate { private let keychain: KeychainManagerProtocol private let nicknameKey = "bitchat.nickname" // Location channel state (macOS supports manual geohash selection) - @Published private var activeChannel: ChannelID = .mesh + @Published private(set) var activeChannel: ChannelID = .mesh private var geoSubscriptionID: String? = nil private var geoDmSubscriptionID: String? = nil private var currentGeohash: String? = nil @@ -454,14 +454,8 @@ final class ChatViewModel: ObservableObject, BitchatDelegate { private var lastMutualToastAt: [String: Date] = [:] // key: fingerprint // MARK: - Public message batching (UI perf) - // Buffer incoming public messages and flush in small batches to reduce UI invalidations - private var publicBuffer: [BitchatMessage] = [] - private var publicBufferTimer: Timer? = nil - private let basePublicFlushInterval: TimeInterval = TransportConfig.basePublicFlushInterval - private var dynamicPublicFlushInterval: TimeInterval = TransportConfig.basePublicFlushInterval - private var recentBatchSizes: [Int] = [] + private let publicMessagePipeline: PublicMessagePipeline @Published private(set) var isBatchingPublic: Bool = false - private let lateInsertThreshold: TimeInterval = TransportConfig.uiLateInsertThreshold // Track sent read receipts to avoid duplicates (persisted across launches) // Note: Persistence happens automatically in didSet, no lifecycle observers needed @@ -507,6 +501,7 @@ final class ChatViewModel: ObservableObject, BitchatDelegate { self.idBridge = idBridge self.identityManager = identityManager self.meshService = BLEService(keychain: keychain, idBridge: idBridge, identityManager: identityManager) + self.publicMessagePipeline = PublicMessagePipeline() // Load persisted read receipts if let data = UserDefaults.standard.data(forKey: "sentReadReceipts"), @@ -559,6 +554,9 @@ final class ChatViewModel: ObservableObject, BitchatDelegate { // Start mesh service immediately meshService.startServices() + publicMessagePipeline.delegate = self + publicMessagePipeline.updateActiveChannel(activeChannel) + // Check initial Bluetooth state after a brief delay to allow centralManager initialization DispatchQueue.main.asyncAfter(deadline: .now() + 0.1) { [weak self] in guard let self = self else { return } @@ -1501,10 +1499,10 @@ final class ChatViewModel: ObservableObject, BitchatDelegate { @MainActor private func switchLocationChannel(to channel: ChannelID) { - // Flush pending public buffer to avoid cross-channel bleed - publicBufferTimer?.invalidate(); publicBufferTimer = nil - publicBuffer.removeAll(keepingCapacity: false) + // Reset pending public batches to avoid cross-channel bleed + publicMessagePipeline.reset() activeChannel = channel + publicMessagePipeline.updateActiveChannel(channel) // Reset deduplication set and optionally hydrate timeline for mesh processedNostrEvents.removeAll() processedNostrEventOrder.removeAll() @@ -6043,113 +6041,10 @@ final class ChatViewModel: ObservableObject, BitchatDelegate { // Append via batching buffer (skip empty content) with simple dedup by ID if !finalMessage.content.trimmingCharacters(in: .whitespacesAndNewlines).isEmpty { if !messages.contains(where: { $0.id == finalMessage.id }) { - enqueuePublic(finalMessage) + publicMessagePipeline.enqueue(finalMessage) } } } - - // MARK: - Public message batching helpers - @MainActor - private func enqueuePublic(_ message: BitchatMessage) { - publicBuffer.append(message) - schedulePublicFlush() - } - - @MainActor - private func schedulePublicFlush() { - if publicBufferTimer != nil { return } - publicBufferTimer = Timer.scheduledTimer(timeInterval: dynamicPublicFlushInterval, - target: self, - selector: #selector(onPublicBufferTimerFired(_:)), - userInfo: nil, - repeats: false) - } - - @MainActor - private func flushPublicBuffer() { - publicBufferTimer?.invalidate() - publicBufferTimer = nil - guard !publicBuffer.isEmpty else { return } - - // Dedup against existing by id and near-duplicate messages by content (within ~1s), across senders - var seenIDs = Set(messages.map { $0.id }) - var added: [BitchatMessage] = [] - var batchContentLatest: [String: Date] = [:] - for m in publicBuffer { - if seenIDs.contains(m.id) { continue } - let ckey = normalizedContentKey(m.content) - if let ts = contentLRUMap[ckey], abs(ts.timeIntervalSince(m.timestamp)) < 1.0 { continue } - if let ts = batchContentLatest[ckey], abs(ts.timeIntervalSince(m.timestamp)) < 1.0 { continue } - seenIDs.insert(m.id) - added.append(m) - batchContentLatest[ckey] = m.timestamp - } - publicBuffer.removeAll(keepingCapacity: true) - guard !added.isEmpty else { return } - - // Indicate batching for conditional UI animations - isBatchingPublic = true - // Rough chronological order: sort the batch by timestamp before inserting - added.sort { $0.timestamp < $1.timestamp } - // Channel-aware insertion policy: geohash uses strict ordering; mesh allows small out-of-order appends - let threshold: TimeInterval = { - switch activeChannel { - case .location: return TransportConfig.uiLateInsertThresholdGeo - case .mesh: return TransportConfig.uiLateInsertThreshold - } - }() - let lastTs = messages.last?.timestamp ?? .distantPast - for m in added { - if m.timestamp < lastTs.addingTimeInterval(-threshold) { - let idx = insertionIndexByTimestamp(m.timestamp) - if idx >= messages.count { messages.append(m) } else { messages.insert(m, at: idx) } - } else if threshold == 0 { - // Strict ordering for geohash: always insert by timestamp - let idx = insertionIndexByTimestamp(m.timestamp) - if idx >= messages.count { messages.append(m) } else { messages.insert(m, at: idx) } - } else { - messages.append(m) - } - // Record content key for LRU - let ckey = normalizedContentKey(m.content) - recordContentKey(ckey, timestamp: m.timestamp) - } - trimMessagesIfNeeded() - // Update batch size stats and adjust interval - recentBatchSizes.append(added.count) - if recentBatchSizes.count > 10 { recentBatchSizes.removeFirst(recentBatchSizes.count - 10) } - let avg = recentBatchSizes.isEmpty ? 0.0 : Double(recentBatchSizes.reduce(0, +)) / Double(recentBatchSizes.count) - dynamicPublicFlushInterval = avg > 100.0 ? 0.12 : basePublicFlushInterval - // Prewarm formatting cache for current UI color scheme only - for m in added { - _ = self.formatMessageAsText(m, colorScheme: currentColorScheme) - } - // Reset batching flag (already on main actor) - isBatchingPublic = false - // If new items arrived during this flush, coalesce by flushing once more next tick - if !publicBuffer.isEmpty { schedulePublicFlush() } - } - - // Timer selector to avoid @Sendable closure capture issues under Swift 6 - @MainActor @objc - private func onPublicBufferTimerFired(_ timer: Timer) { - flushPublicBuffer() - } - - @MainActor - private func insertionIndexByTimestamp(_ ts: Date) -> Int { - var low = 0 - var high = messages.count - while low < high { - let mid = (low + high) / 2 - if messages[mid].timestamp < ts { - low = mid + 1 - } else { - high = mid - } - } - return low - } /// Check for mentions and send notifications @@ -6216,3 +6111,37 @@ private func checkForMentions(_ message: BitchatMessage) { } } // End of ChatViewModel class + +extension ChatViewModel: PublicMessagePipelineDelegate { + func pipelineCurrentMessages(_ pipeline: PublicMessagePipeline) -> [BitchatMessage] { + messages + } + + func pipeline(_ pipeline: PublicMessagePipeline, setMessages messages: [BitchatMessage]) { + self.messages = messages + } + + func pipeline(_ pipeline: PublicMessagePipeline, normalizeContent content: String) -> String { + normalizedContentKey(content) + } + + func pipeline(_ pipeline: PublicMessagePipeline, contentTimestampForKey key: String) -> Date? { + contentLRUMap[key] + } + + func pipeline(_ pipeline: PublicMessagePipeline, recordContentKey key: String, timestamp: Date) { + recordContentKey(key, timestamp: timestamp) + } + + func pipelineTrimMessages(_ pipeline: PublicMessagePipeline) { + trimMessagesIfNeeded() + } + + func pipelinePrewarmMessage(_ pipeline: PublicMessagePipeline, message: BitchatMessage) { + _ = formatMessageAsText(message, colorScheme: currentColorScheme) + } + + func pipelineSetBatchingState(_ pipeline: PublicMessagePipeline, isBatching: Bool) { + isBatchingPublic = isBatching + } +} diff --git a/bitchat/ViewModels/GeoChannelCoordinator.swift b/bitchat/ViewModels/GeoChannelCoordinator.swift index 8df6331a..60879334 100644 --- a/bitchat/ViewModels/GeoChannelCoordinator.swift +++ b/bitchat/ViewModels/GeoChannelCoordinator.swift @@ -31,9 +31,9 @@ final class GeoChannelCoordinator { beginSampling: @escaping ([String]) -> Void, endSampling: @escaping () -> Void ) { - self.locationManager = locationManager ?? LocationChannelManager.shared + self.locationManager = locationManager ?? Self.defaultLocationManager() self.bookmarksStore = bookmarksStore ?? GeohashBookmarksStore.shared - self.torManager = torManager ?? TorManager.shared + self.torManager = torManager ?? Self.defaultTorManager() self.onChannelSwitch = onChannelSwitch self.beginSampling = beginSampling self.endSampling = endSampling @@ -107,4 +107,12 @@ final class GeoChannelCoordinator { func refreshSampling() { updateSampling() } + private static func defaultLocationManager() -> LocationChannelManager { + LocationChannelManager.shared + } + + @MainActor + private static func defaultTorManager() -> TorManager { + TorManager.shared + } } diff --git a/bitchat/ViewModels/PublicMessagePipeline.swift b/bitchat/ViewModels/PublicMessagePipeline.swift new file mode 100644 index 00000000..a1b794e8 --- /dev/null +++ b/bitchat/ViewModels/PublicMessagePipeline.swift @@ -0,0 +1,190 @@ +// +// PublicMessagePipeline.swift +// bitchat +// +// Handles batching and deduplication of public chat messages before surfacing them to the UI. +// + +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) + func pipelinePrewarmMessage(_ pipeline: PublicMessagePipeline, message: BitchatMessage) + func pipelineSetBatchingState(_ pipeline: PublicMessagePipeline, isBatching: Bool) +} + +@MainActor +final class PublicMessagePipeline { + weak var delegate: PublicMessagePipelineDelegate? + + private var buffer: [BitchatMessage] = [] + 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, + maxRecentBatchSamples: Int = 10, + dedupWindow: TimeInterval = 1.0 + ) { + self.baseFlushInterval = baseFlushInterval + self.dynamicFlushInterval = baseFlushInterval + self.maxRecentBatchSamples = maxRecentBatchSamples + self.dedupWindow = dedupWindow + } + + deinit { + timer?.invalidate() + } + + func updateActiveChannel(_ channel: ChannelID) { + activeChannel = channel + } + + func enqueue(_ message: BitchatMessage) { + buffer.append(message) + scheduleFlush() + } + + func flushIfNeeded() { + flushBuffer() + } + + func reset() { + timer?.invalidate() + timer = nil + buffer.removeAll(keepingCapacity: false) + } + +} + +private extension PublicMessagePipeline { + func scheduleFlush() { + guard timer == nil else { return } + timer = Timer.scheduledTimer(withTimeInterval: dynamicFlushInterval, repeats: false) { [weak self] _ in + guard let self else { return } + Task { @MainActor in + self.flushBuffer() + } + } + } + + func flushBuffer() { + timer?.invalidate() + timer = nil + guard !buffer.isEmpty else { return } + guard let delegate = delegate else { + buffer.removeAll(keepingCapacity: false) + return + } + + delegate.pipelineSetBatchingState(self, isBatching: true) + + var existingIDs = Set(delegate.pipelineCurrentMessages(self).map { $0.id }) + var pending: [(message: BitchatMessage, contentKey: String)] = [] + var batchContentLatest: [String: Date] = [:] + + for message in buffer { + if existingIDs.contains(message.id) { continue } + let contentKey = delegate.pipeline(self, normalizeContent: message.content) + if let ts = delegate.pipeline(self, contentTimestampForKey: contentKey), + abs(ts.timeIntervalSince(message.timestamp)) < dedupWindow { + continue + } + if let ts = batchContentLatest[contentKey], + abs(ts.timeIntervalSince(message.timestamp)) < dedupWindow { + continue + } + existingIDs.insert(message.id) + pending.append((message, contentKey)) + batchContentLatest[contentKey] = 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) + } + + delegate.pipeline(self, setMessages: messages) + delegate.pipelineTrimMessages(self) + + updateFlushInterval(withBatchSize: pending.count) + + for item in pending { + delegate.pipelinePrewarmMessage(self, message: item.message) + } + + delegate.pipelineSetBatchingState(self, isBatching: false) + + if !buffer.isEmpty { + scheduleFlush() + } + } + + func updateFlushInterval(withBatchSize size: Int) { + recentBatchSizes.append(size) + if recentBatchSizes.count > maxRecentBatchSamples { + recentBatchSizes.removeFirst(recentBatchSizes.count - maxRecentBatchSamples) + } + let avg = recentBatchSizes.isEmpty + ? 0.0 + : 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 + } +}