Add conversation-store design doc and end-to-end ingest baselines

docs/CONVERSATION-STORE-DESIGN.md records the approved design: a single-writer ConversationStore of per-conversation ObservableObjects (per-conversation publishing, incremental ID index, folded caps, typed change subject) replacing today's four-store/three-bridge topology, with a five-step migration plan and explicit deletions/non-goals. New pipeline benchmarks measure the CURRENT architecture end-to-end so every migration step is judged against real before-numbers: pipeline.privateIngest ~9.7k msg/s, pipeline.publicIngest ~6.8k msg/s (200-message passes, stable within 1.5%). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-24 22:45:19 +00:00 · 2026-06-11 10:51:24 +02:00
parent 75dd83d9cc
commit 45650854e7
2 changed files with 307 additions and 0 deletions
@@ -328,6 +328,139 @@ final class PerformanceBaselineTests: XCTestCase {
        reportThroughput("formatting.formatMessage", samples: samples, operations: batchSize, unit: "messages")
    }

+    // MARK: - 6a. End-to-end private ingest pipeline (current architecture)
+
+    /// Baseline for the full private-message ingest cycle through a real
+    /// `ChatViewModel`: `handlePrivateMessage` → `privateChats` passthrough →
+    /// `PrivateChatManager`'s `@Published` dict → bootstrapper Combine sink →
+    /// `Task.yield`-debounced `ConversationStore.synchronizePrivateChats`
+    /// (full-dict replace) → `PrivateInboxModel.refreshMessages` (full
+    /// rebuild). Measures wall time from first ingest until the store AND the
+    /// feature model both reflect every message. The peer is not mesh-active
+    /// and no chat is selected, so notification/read-receipt side paths stay
+    /// cold (notifications are no-ops under test anyway).
+    func testPipelinePrivateIngest() {
+        let messageCount = 200
+        // 64-hex (stable Noise key) peer ID: skips the short-ID consolidation
+        // and ephemeral-mirror paths so every pass does identical work.
+        let peerID = PeerID(str: String(repeating: "ab", count: 32))
+        let base = Date(timeIntervalSince1970: 1_700_000_000)
+        let messages: [BitchatMessage] = (0..<messageCount).map { i in
+            BitchatMessage(
+                id: "perf-dm-\(i)",
+                sender: "perfsender",
+                content: "private pipeline message \(i)",
+                timestamp: base.addingTimeInterval(Double(i)),
+                isRelay: false,
+                isPrivate: true,
+                recipientNickname: "me",
+                senderPeerID: peerID
+            )
+        }
+        // Fresh fixture per pass so dedup scans and store syncs start from the
+        // same empty state every iteration. Kept alive so weakly captured
+        // coordinator Task hops stay valid.
+        var keepAlive: [PerfPipelineFixture] = []
+        var samples: [TimeInterval] = []
+
+        measure {
+            let fixture = PerfPipelineFixture()
+            keepAlive.append(fixture)
+            let start = Date()
+            for message in messages {
+                fixture.viewModel.handlePrivateMessage(message)
+            }
+            // Drain the main queue until the debounced ConversationStore sync
+            // ran and the PrivateInboxModel mirror caught up.
+            let consistent = spinMainRunLoop(timeout: 10) {
+                fixture.conversationStore.directMessagesByPeerID()[peerID]?.count == messageCount
+                    && fixture.privateInbox.messagesByPeerID[peerID]?.count == messageCount
+            }
+            samples.append(Date().timeIntervalSince(start))
+            XCTAssertTrue(consistent, "ConversationStore/PrivateInboxModel never converged")
+            XCTAssertEqual(fixture.viewModel.privateChats[peerID]?.count, messageCount)
+        }
+
+        reportThroughput("pipeline.privateIngest", samples: samples, operations: messageCount, unit: "messages")
+    }
+
+    // MARK: - 6b. End-to-end public ingest pipeline (current architecture)
+
+    /// Baseline for the full public-message ingest cycle through a real
+    /// `ChatViewModel`: `didReceivePublicMessage` (transport delegate entry,
+    /// main-actor Task hop per message) → `handlePublicMessage` (rate limit,
+    /// `PublicTimelineStore` append, per-message full-array
+    /// `ConversationStore` sync) → `PublicMessagePipeline` timer-batched
+    /// flush into `ChatViewModel.messages` → `PublicChatModel` mirror.
+    /// Measures until `messages` and the feature model reflect every message,
+    /// so the pipeline's flush latency is part of the cycle. Senders are
+    /// spread 4-per-peer to stay under the 5-token sender rate bucket.
+    func testPipelinePublicIngest() {
+        let messageCount = 200
+        let senderCount = 50
+        let base = Date(timeIntervalSince1970: 1_700_000_000)
+        struct InboundPublic {
+            let peerID: PeerID
+            let nickname: String
+            let content: String
+            let timestamp: Date
+            let messageID: String
+        }
+        let items: [InboundPublic] = (0..<messageCount).map { i in
+            let sender = i % senderCount
+            return InboundPublic(
+                peerID: PeerID(str: String(format: "%016x", 0xC0DE_0000 + sender)),
+                nickname: "perfpeer\(sender)",
+                content: "public pipeline message \(i) from sender \(sender)",
+                timestamp: base.addingTimeInterval(Double(i)),
+                messageID: "perf-pub-\(i)"
+            )
+        }
+        let expectedIDs = Set(items.map(\.messageID))
+        var keepAlive: [PerfPipelineFixture] = []
+        var samples: [TimeInterval] = []
+
+        measure {
+            let fixture = PerfPipelineFixture()
+            keepAlive.append(fixture)
+            let start = Date()
+            for item in items {
+                fixture.viewModel.didReceivePublicMessage(
+                    from: item.peerID,
+                    nickname: item.nickname,
+                    content: item.content,
+                    timestamp: item.timestamp,
+                    messageID: item.messageID
+                )
+            }
+            // Drain the per-message main-actor hops and whichever surfacing
+            // path wins (the pipeline's batched timer flush or the startup
+            // channel apply's `refreshVisibleMessages`, which pulls the whole
+            // timeline into `messages`), plus the PublicChatModel mirror.
+            let consistent = spinMainRunLoop(timeout: 10) {
+                fixture.viewModel.messages.count >= messageCount
+                    && fixture.publicChat.messages.count >= messageCount
+            }
+            samples.append(Date().timeIntervalSince(start))
+            XCTAssertTrue(consistent, "messages/PublicChatModel never converged")
+            let ingested = fixture.viewModel.messages.filter { expectedIDs.contains($0.id) }
+            XCTAssertEqual(ingested.count, messageCount)
+        }
+
+        reportThroughput("pipeline.publicIngest", samples: samples, operations: messageCount, unit: "messages")
+    }
+
+    /// Spins the main run loop in small slices (draining main-queue tasks and
+    /// timers) until `condition` holds or `timeout` elapses.
+    private func spinMainRunLoop(timeout: TimeInterval, until condition: () -> Bool) -> Bool {
+        let deadline = Date().addingTimeInterval(timeout)
+        while !condition() {
+            guard Date() < deadline else { return false }
+            RunLoop.current.run(until: Date().addingTimeInterval(0.005))
+        }
+        return true
+    }
+
    // MARK: - Fixtures

    /// Builds deterministic signed kind-20000 geohash events. Content cycles
@@ -517,6 +650,42 @@ private final class PerfDeliveryContext: ChatDeliveryContext {
    }
 }

+// MARK: - End-to-end pipeline fixture
+
+/// A real `ChatViewModel` over `MockTransport` plus the AppRuntime-style
+/// feature models (`PrivateInboxModel` / `PublicChatModel`) bound to the same
+/// `ConversationStore`, so end-to-end ingest benchmarks cover the full
+/// current store-synchronization chain. Mirrors the construction used by
+/// `ChatViewModelExtensionsTests`.
+@MainActor
+private final class PerfPipelineFixture {
+    let viewModel: ChatViewModel
+    let transport: MockTransport
+    let conversationStore: ConversationStore
+    let privateInbox: PrivateInboxModel
+    let publicChat: PublicChatModel
+
+    init() {
+        let keychain = MockKeychain()
+        let idBridge = NostrIdentityBridge(keychain: MockKeychainHelper())
+        let identityManager = MockIdentityManager(keychain)
+        let transport = MockTransport()
+        let conversationStore = ConversationStore()
+
+        self.transport = transport
+        self.conversationStore = conversationStore
+        self.viewModel = ChatViewModel(
+            keychain: keychain,
+            idBridge: idBridge,
+            identityManager: identityManager,
+            transport: transport,
+            conversationStore: conversationStore
+        )
+        self.privateInbox = PrivateInboxModel(conversationStore: conversationStore)
+        self.publicChat = PublicChatModel(conversationStore: conversationStore)
+    }
+}
+
 // MARK: - Mock MessageFormattingContext

@MainActor
@@ -0,0 +1,138 @@
+# Conversation Store: Single Source of Truth
+
+**Status:** Approved design, not yet implemented. Baselines recorded in
+`bitchatTests/Performance/PerformanceBaselineTests.swift` (`pipeline.privateIngest`,
+`pipeline.publicIngest`).
+
+---
+
+## 1. Problem
+
+Message state is replicated across four stores, kept eventually consistent by three
+async bridges. One inbound private message today:
+
+1. `ChatPrivateConversationCoordinator.handlePrivateMessage` writes through
+   `ChatViewModel.privateChats` — a passthrough computed property
+   (`ChatViewModel.swift:167-173`) into `PrivateChatManager`'s
+   `@Published var privateChats` (`PrivateChatManager.swift:16`).
+2. **Bridge 1:** the bootstrapper subscribes `privateChatManager.$privateChats` and
+   `$unreadMessages` with `.receive(on: DispatchQueue.main)` sinks
+   (`ChatViewModelBootstrapper.swift:92-108`).
+3. **Bridge 2:** each sink calls `schedulePrivateConversationStoreSynchronization`,
+   a `Task.yield`-debounced task (`ChatViewModel.swift:1084-1092`) that eventually runs
+   `synchronizePrivateConversationStore` (`ChatViewModel.swift:1095-1101`).
+4. That calls `ConversationStore.synchronizePrivateChats` — a **full-dict replace**: every
+   conversation is re-normalized (dedup + `O(n log n)` sort) and diff-compared on every
+   sync (`AppArchitecture.swift:304-346`, `normalized` at `AppArchitecture.swift:359-372`).
+5. **Bridge 3:** `PrivateInboxModel` subscribes `conversationStore.$messagesByConversation`
+   (again `.receive(on: DispatchQueue.main)`) and rebuilds its entire
+   `messagesByPeerID` dictionary via `refreshMessages`
+   (`PrivateConversationModels.swift:43-48`, `54-68`).
+6. SwiftUI finally observes the feature model.
+
+Costs and hazards:
+
+- **O(total messages) × 3 layers per single message.** One append re-sorts, re-compares,
+  and re-publishes every conversation through store and feature-model layers. The ingest
+  path itself is also quadratic: `isDuplicateMessage` linearly scans *all* private chats
+  per inbound message (`ChatPrivateConversationCoordinator.swift:622-630`) and
+  `sanitizeChat` re-sorts the whole chat per append (`PrivateChatManager.swift:213-234`).
+- **Delivery status mutates two copies.** `ChatDeliveryCoordinator` patches both
+  `context.messages` and a value-copied `context.privateChats`
+  (`ChatDeliveryCoordinator.swift:105-139`), navigating with a positional
+  `messageLocationIndex` (`ChatDeliveryCoordinator.swift:40`) that any non-append
+  mutation invalidates, forcing a full rebuild over every message location
+  (`ChatDeliveryCoordinator.swift:298-320`).
+- **Transient disagreement.** Between the `@Published` write and the debounced sync,
+  `privateChatManager.privateChats` and `ConversationStore.messagesByConversation`
+  disagree; anything reading the store mid-flight sees stale data.
+
+The public path has the same shape: `@Published var messages`
+(`ChatViewModel.swift:122`) is the render copy, `PublicTimelineStore`
+(`ChatViewModel.swift:342-345`) is the backing copy, and `handlePublicMessage` appends to
+the timeline, then full-replaces the conversation store **per message**
+(`ChatPublicConversationCoordinator.swift:504-545` calling
+`synchronizePublicConversationStore` at `:358-364`, which funnels into
+`ConversationStore.replaceMessages`'s whole-array compare at `AppArchitecture.swift:249-253`),
+while a timer-batched `PublicMessagePipeline` mutates `messages` ~80 ms later
+(`PublicMessagePipeline.swift`, `TransportConfig.basePublicFlushInterval`).
+`PublicChatModel` then mirrors the store again (`PublicChatModel.swift`).
+
+## 2. Design
+
+`ConversationStore` (already `@MainActor` and owned by `AppRuntime`,
+`AppRuntime.swift:46`) becomes the **sole writer and sole holder** of message state.
+
+- **`Conversation` is a reference-type `ObservableObject`**, one instance per
+  `ConversationID` (`.mesh` / `.geohash` / `.direct`), with `@Published private(set)`
+  `messages` and unread state. Each conversation maintains its message-ID index
+  **incrementally** (insert on append, never rebuilt from scratch) and owns its cap
+  policy: `TransportConfig.meshTimelineCap` / `geoTimelineCap` / `privateChatCap`
+  (`TransportConfig.swift:17-19`) fold into the store; `PublicTimelineStore`'s trim logic
+  and `PrivateChatManager`'s cap disappear.
+- **Publishing granularity is per conversation.** Views observe ONE `Conversation`
+  object. An append to chat A never invalidates observers of chat B — unlike today,
+  where any write republishes the entire `messagesByConversation` dictionary
+  (`AppArchitecture.swift:205`) and every bound feature model rebuilds.
+- **Store-level `changes: PassthroughSubject<ConversationChange, Never>`** for non-UI
+  consumers (delivery tracking, notifications, gossip/sync) that need "a message was
+  appended / status changed in conversation X" without subscribing to message arrays.
+- **Mutations go through an intent API only**, mirroring the codebase's existing
+  single-writer intent ops (`ChatViewModel.swift:421-424`, the `private(set)` +
+  dedicated-mutator pattern):
+  - `append(_:to:)` — incremental, dedup via the ID index
+  - `upsertByID(_:in:)` — replace-or-append (media progress, edits)
+  - `setDeliveryStatus(_:for:in:)` — keyed by message ID, no positional index
+  - `markRead(_:)` / `markUnread(_:)`
+  - `migrateConversation(from:to:)` — the ephemeral↔stable peer-ID handoff that today
+    is hand-rolled dictionary surgery in three places
+  - `clear(_:)`
+  Backing collections are `private(set)`; coordinators receive the intent surface, not
+  the dictionaries.
+- **Reads are synchronous.** Because writers and readers share the main actor and there
+  is one copy, "await the sync" disappears: after `append` returns, every observer of
+  that `Conversation` sees the message.
+
+## 3. Deleted at end state
+
+- `PublicTimelineStore` (`bitchat/ViewModels/PublicTimelineStore.swift`) — folded into
+  `Conversation` cap/dedup policy.
+- `PrivateChatManager`'s message dict and trim/sanitize logic — the manager shrinks to
+  read-receipt policy (`markAsRead`, `syncReadReceiptsForSentMessages`).
+- `ChatDeliveryCoordinator.messageLocationIndex` and its growth/rebuild machinery
+  (`ChatDeliveryCoordinator.swift:40-45`, `221-320`) — replaced by
+  `setDeliveryStatus(for:in:)` against the per-conversation ID index.
+- Both bootstrapper sync bridges (`ChatViewModelBootstrapper.swift:92-108`).
+- `schedulePrivateConversationStoreSynchronization` /
+  `synchronizePrivateConversationStore` and the public equivalents
+  (`ChatViewModel.swift:1084-1101`, `ChatPublicConversationCoordinator.swift:351-386`).
+- Feature-model mirror collections: `PrivateInboxModel.messagesByPeerID`,
+  `PublicChatModel.messages` (they observe `Conversation` objects directly).
+- `ChatViewModel.messages` / `ChatViewModel.privateChats` as stored/owning properties.
+
+## 4. Migration plan
+
+Each step lands green against the full suite plus the `PerformanceBaselineTests`
+numbers (no pipeline throughput regression at any step).
+
+1. **Additive store.** Introduce `Conversation` objects and the intent API inside
+   `ConversationStore` alongside the existing replace-based API. Nothing reads them yet.
+2. **Private cutover with compat shims.** Inbound/outbound private paths write through
+   the intent API. `ChatViewModel.privateChats` becomes a derived **read-only** view of
+   the store; `PrivateChatManager`'s dict and the private sync bridges are bypassed but
+   the property surface stays so coordinators/tests compile unchanged.
+3. **Public cutover.** `handlePublicMessage` and the `PublicMessagePipeline` flush write
+   to the store; `PublicTimelineStore` folds in; `ChatViewModel.messages` becomes a
+   derived view of the active conversation.
+4. **Delivery via store.** `ChatDeliveryCoordinator` switches to
+   `setDeliveryStatus(for:in:)`; `messageLocationIndex` is deleted.
+5. **View cutover.** Views and feature models observe `Conversation` objects directly;
+   delete all shims, mirrors, and the replace-based store API.
+
+## 5. Non-goals
+
+- **No message persistence.** bitchat is ephemeral by design; the store stays in-memory.
+- **`sentReadReceipts` UserDefaults persistence stays put** (`ChatViewModel.swift:394-406`);
+  it is receipt-protocol state, not conversation state.
+- **`MessageRouter`'s outbox remains the SSOT for unsent messages**; the store records
+  delivery status but never owns retry/resend queues.