Extract public message pipeline from ChatViewModel (#870)

* Extract public timeline helpers and rate limiter

* Extract public message pipeline

* Fix Swift concurrency warnings

* Incorporate public chat review feedback

* Tighten nostr key removal loop

* Address review feedback

---------

Co-authored-by: jack <jackjackbits@users.noreply.github.com>

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 let publicMessagePipeline: PublicMessagePipeline
-    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] = []
     @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
+        // Reset pending public batches to avoid cross-channel bleed
-        publicBufferTimer?.invalidate(); publicBufferTimer = nil
+        publicMessagePipeline.reset()
-        publicBuffer.removeAll(keepingCapacity: false)
         activeChannel = channel
+        publicMessagePipeline.updateActiveChannel(channel)
         // Reset deduplication set and optionally hydrate timeline for mesh
         processedNostrEvents.removeAll()
         processedNostrEventOrder.removeAll()
@@ -6043,114 +6041,11 @@ 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
     
 private func checkForMentions(_ message: BitchatMessage) {
@@ -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
+    }
+}

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
+    }
 }

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
+    }
+}