From 6f360fc1d533dbd92f6aadb5f6f8ba5c1ab77781 Mon Sep 17 00:00:00 2001 From: Brandon McAnsh Date: Fri, 6 Nov 2026 17:40:41 -0500 Subject: [PATCH 1/2] perf(chat): patch the open transcript window instead of re-reading it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `messageRevision` was one counter per controller and part of the confirmed-window cache key. A write to any chat, a reaction, and every catch-up batch re-decoded the open chat's whole window on the main actor, at about 20 µs per row. Writes now report which chat and which message rows they touched. `noteMessageWrite` still bumps `messageRevision` for the views that observe it, and then: - leaves the cached window alone for a write to another chat - patches the written rows into the window for a write to the open chat, by binary search, and drops rows an arrival pushes out of the newest-N window - drops the window when a write's rows aren't known (reset resync, initial load, the notification-extension reload) A catch-up reports its writes once, when the run ends, not once per batch. `ConversationReactions.didPersist` now passes the chat and message ids. --- ...versationController+MessageMutations.swift | 2 +- .../Controllers/ConversationController.swift | 172 +++++++++++++--- .../Controllers/ConversationReactions.swift | 9 +- .../Database+Conversations.swift | 30 +++ .../Chat/ConfirmedWindowPatchTests.swift | 192 ++++++++++++++++++ 5 files changed, 373 insertions(+), 32 deletions(-) create mode 100644 FlipcashTests/Chat/ConfirmedWindowPatchTests.swift diff --git a/Flipcash/Core/Controllers/ConversationController+MessageMutations.swift b/Flipcash/Core/Controllers/ConversationController+MessageMutations.swift index 942de2b93..4420733bd 100644 --- a/Flipcash/Core/Controllers/ConversationController+MessageMutations.swift +++ b/Flipcash/Core/Controllers/ConversationController+MessageMutations.swift @@ -154,7 +154,7 @@ extension ConversationController { in conversationID: ConversationID, operation: String ) { - _ = persist(operation: operation) { + _ = persist(operation: operation, scope: .messages([outcome.message.id], in: conversationID)) { try database.upsertConversationMessages([outcome.message], conversationID: conversationID) } store.dropMutation(for: messageID, in: conversationID) diff --git a/Flipcash/Core/Controllers/ConversationController.swift b/Flipcash/Core/Controllers/ConversationController.swift index 29485452b..1a35e67b4 100644 --- a/Flipcash/Core/Controllers/ConversationController.swift +++ b/Flipcash/Core/Controllers/ConversationController.swift @@ -634,7 +634,9 @@ final class ConversationController { selfUserID: selfUserID ) self.pins = ConversationPins(messaging: messaging, owner: owner, selfUserID: selfUserID) - reactions.didPersist = { [weak self] in self?.bumpMessageRevision() } + reactions.didPersist = { [weak self] conversationID, messageIDs in + self?.noteMessageWrite(.messages(messageIDs, in: conversationID)) + } pins.didChange = { [weak self] in self?.bumpMessageRevision() } } @@ -815,6 +817,13 @@ final class ConversationController { guard !catchUpInFlight.contains(conversationID) else { return } catchUpInFlight.insert(conversationID) defer { catchUpInFlight.remove(conversationID) } + // Every batch is reported once, when the run ends, so a multi-batch catch-up re-reads the + // open transcript once instead of once per batch. `catchUpInFlight` keeps this slot to one run. + defer { + if let written = catchUpWrites.removeValue(forKey: conversationID), !written.isEmpty { + noteMessageWrite(.messages(written, in: conversationID)) + } + } let after = store.appliedCursor(for: conversationID) // One baseline for the whole run. A per-batch watermark would let the second @@ -828,9 +837,12 @@ final class ConversationController { // Batch + checkpoint cursor persist atomically; advance the in-memory checkpoint only // after the write succeeds, so a rolled-back batch isn't skipped on the next catch-up. let cursor = checkpoint ?? self.store.appliedCursor(for: conversationID) - let outcome = self.persistChanges(operation: "delta-batch") { + let outcome = self.persistChanges(operation: "delta-batch", scope: nil) { try self.database.persistMessages(reconciled, cursor: cursor, conversationID: conversationID) } + if outcome == .changed { + self.catchUpWrites[conversationID, default: []] += reconciled.map(\.id) + } if outcome.succeeded { self.commitReconciled(pairs, in: conversationID) self.receipts.countReceived(reconciled, countedThrough: countedThrough, delivery: .catchUp) @@ -923,8 +935,9 @@ final class ConversationController { /// Re-reads every known conversation's transcript and feed preview straight from disk, with no /// network involved. Covers the notification extension's writes: it persists pushed messages into /// the shared store, which the running app's in-memory `store` never observes, since - /// `Database.transaction` only posts `.databaseDidChange` for writes made by this process. Bumping - /// `messageRevision` invalidates the cached transcript window so the next read re-queries the DB; + /// `Database.transaction` only posts `.databaseDidChange` for writes made by this process. Which + /// chats it wrote isn't known here, so this reports a write to every chat, which bumps + /// `messageRevision` and drops the cached transcript window so the next read re-queries the DB; /// `refreshFeedPreview` does the same for each chat's last-message preview and unread state, which /// don't come from the network-only catch-up paths. The extension also advances a chat's catch-up /// cursor when its delta completed (see `NotificationService.persist`), so the stored cursors are @@ -1001,7 +1014,7 @@ final class ConversationController { let floor = messages.map(\.eventSequence).filter({ $0 > 0 }).min() ?? 0 // Page + floor cursor land atomically (a failed write must not seat the cursor), and a // non-overlapping page drops the stale older epoch first. - let ok = persist(operation: "reset-resync") { + let ok = persist(operation: "reset-resync", scope: .conversation(conversationID)) { try dropStaleEpochIfNeeded(before: messages, in: conversationID) try database.persistMessages(reconciled, cursor: floor, conversationID: conversationID) } @@ -1693,7 +1706,7 @@ final class ConversationController { // in-memory cursor optimistically, so if this write rolls back, re-seat the cursor to the // persisted value and catch up — otherwise the un-persisted messages are skipped by the next // GetDelta and lost (the store no longer holds a confirmed copy). - let outcome = persistChanges(operation: "apply-chat-events") { + let outcome = persistChanges(operation: "apply-chat-events", scope: .messages(reconciled.map(\.id), in: conversationID)) { try database.persistMessages(reconciled, cursor: store.appliedCursor(for: conversationID), conversationID: conversationID) } if outcome.succeeded { @@ -1823,7 +1836,9 @@ final class ConversationController { } private func persistConversation(_ conversation: Conversation) { - persistChanges(operation: "upsert-conversation") { try database.upsertConversation(withStoreMembers(conversation)) } + // The one message row this writes is the chat's last message. + let scope: MessageWriteScope = conversation.lastMessage.map { .messages([$0.id], in: conversation.id) } ?? .conversation(conversation.id) + persistChanges(operation: "upsert-conversation", scope: scope) { try database.upsertConversation(withStoreMembers(conversation)) } } /// A server copy of a group carrying the store's member list instead of its own. A group's server @@ -1840,13 +1855,14 @@ final class ConversationController { return conversation } + /// - Parameter scope: the message rows the write touches, for a message-write operation. @discardableResult - func persist(operation: String, _ write: () throws -> Void) -> Bool { + func persist(operation: String, scope: MessageWriteScope = .allConversations, _ write: () throws -> Void) -> Bool { guard attempt(operation: operation, write) else { return false } - // A successful confirmed-message write invalidates the DB-backed transcript window; bump the - // revision the coordinator observes so it re-reads. + // A successful confirmed-message write invalidates the DB-backed transcript window; report it + // so the window re-reads and the views observing `messageRevision` re-evaluate. if Self.messageWriteOperations.contains(operation) { - messageRevision &+= 1 + noteMessageWrite(scope) } return true } @@ -1869,12 +1885,17 @@ final class ConversationController { /// Runs a write that returns whether it changed a message row, bumping `messageRevision` only /// when it did. A catch-up that re-fetches what the notification extension already stored then /// redraws nothing, instead of invalidating every chat row's unread count once per batch. + /// + /// - Parameter scope: the message rows the write touches. Pass nil when the caller reports the + /// write itself, once for several writes. @discardableResult - func persistChanges(operation: String, _ write: () throws -> Bool) -> PersistOutcome { + func persistChanges(operation: String, scope: MessageWriteScope? = .allConversations, _ write: () throws -> Bool) -> PersistOutcome { var changed = false guard attempt(operation: operation, { changed = try write() }) else { return .failed } guard changed else { return .unchanged } - messageRevision &+= 1 + if let scope { + noteMessageWrite(scope) + } return .changed } @@ -2160,11 +2181,15 @@ final class ConversationController { windowedMessages(for: conversationID, limit: Self.recentWindow) } - /// Bumped after every successful confirmed-message DB write, so the coordinator — which reads the - /// transcript from the DB, not the store — knows to re-read its window. Pending-overlay changes - /// re-fire through the store directly. + /// Bumped after every successful confirmed-message DB write, so the views that read the transcript + /// from the DB, not the store, re-evaluate. Pending-overlay changes re-fire through the store + /// directly. The cached transcript window is not keyed on this: ``noteMessageWrite(_:)`` drops or + /// patches it only for the conversation a write touched. private(set) var messageRevision = 0 + /// Batch writes of each running catch-up, reported once when the run ends. + @ObservationIgnored private var catchUpWrites: [ConversationID: [MessageID]] = [:] + /// Own sends the server has confirmed this session, per chat. Observed by the jump badge, which /// reads a rise as the viewer having caught up with the transcript. private(set) var confirmedOwnSendCounts: [ConversationID: Int] = [:] @@ -2172,7 +2197,99 @@ final class ConversationController { /// Forces the transcript to re-read its window. `persist(operation:)` does this for database /// writes; an overlay change writes nothing, so it has to say so explicitly. func bumpMessageRevision() { + noteMessageWrite(.allConversations) + } + + /// The message rows a confirmed write touched. + enum MessageWriteScope { + /// Unknown, or across the feed: every chat's window re-reads. + case allConversations + /// One chat's rows may have been removed or rewritten wholesale: its window re-reads. + case conversation(ConversationID) + /// Exactly these rows of one chat were inserted or rewritten. + case messages([MessageID], in: ConversationID) + } + + /// Reports a confirmed write: bumps ``messageRevision`` and brings the cached transcript window up + /// to date. A write to another chat leaves the window alone, and a write of a few rows to the + /// window's own chat patches those rows in, so neither re-decodes the whole window. + func noteMessageWrite(_ scope: MessageWriteScope) { messageRevision &+= 1 + guard let cached = confirmedWindowCache else { return } + switch scope { + case .allConversations: + confirmedWindowCache = nil + case .conversation(let conversationID): + if cached.key.conversationID == conversationID { + confirmedWindowCache = nil + } + case .messages(let messageIDs, let conversationID): + guard cached.key.conversationID == conversationID else { return } + confirmedWindowCache = patchedWindowCache(cached, writing: messageIDs) + } + } + + /// `cached` with the rows of `messageIDs` re-read, or nil when re-reading the window is cheaper + /// or a read fails. + private func patchedWindowCache( + _ cached: (key: ConfirmedWindowKey, messages: [ConversationMessage]), + writing messageIDs: [MessageID] + ) -> (key: ConfirmedWindowKey, messages: [ConversationMessage])? { + let key = cached.key + do { + // An anchored window starts at its anchor. The newest-N window starts at the oldest of the + // newest N stored rows, which an arrival moves up. + let lowerBound = try key.startID ?? database.newestWindowFloor(conversationID: key.conversationID, limit: key.limit) + let inWindow = messageIDs.filter { id in lowerBound.map { id.value >= $0 } ?? true } + // Re-reading most of the window row by row costs more than one window read. + guard inWindow.count * 4 <= max(cached.messages.count, 4) else { return nil } + let rows = try database.messages(conversationID: key.conversationID, ids: inWindow) + return (key, Self.patchedWindow(cached.messages, written: inWindow, rows: rows, from: lowerBound)) + } catch { + return nil + } + } + + /// A confirmed window after a write: `window` with every `written` row replaced by its stored copy + /// in `rows` (dropped when absent, inserted in id order when new), and every row below + /// `lowerBound` dropped. Equal to a fresh window read when `window` was current before the write + /// and `written` lists every row the write touched. + static func patchedWindow( + _ window: [ConversationMessage], + written: [MessageID], + rows: [ConversationMessage], + from lowerBound: UInt64? + ) -> [ConversationMessage] { + // `window` is in ascending id order, so each written row is found by binary search and the + // patch costs O(written · log window) plus the array shifts, not a pass over every row. + var patched = window + if let lowerBound { + patched.removeFirst(insertionIndex(of: lowerBound, in: patched)) + } + let stored = Dictionary(rows.map { ($0.id.value, $0) }, uniquingKeysWith: { _, last in last }) + for id in Set(written.map(\.value)) { + if let lowerBound, id < lowerBound { continue } + let index = insertionIndex(of: id, in: patched) + let present = index < patched.count && patched[index].id.value == id + switch (present, stored[id]) { + case (true, let row?): patched[index] = row + case (true, nil): patched.remove(at: index) + case (false, let row?): patched.insert(row, at: index) + case (false, nil): break + } + } + return patched + } + + /// The first index in `messages` (ascending by id) whose id is not below `id`. + private static func insertionIndex(of id: UInt64, in messages: [ConversationMessage]) -> Int { + var low = 0 + var high = messages.count + while low < high { + let mid = (low + high) / 2 + if messages[mid].id.value < id { low = mid + 1 } else { high = mid } + } + return low } /// Set when a mutation needs to be reported to the person who made it. The screen presents it @@ -2227,16 +2344,16 @@ final class ConversationController { return reactions.displayed(displayed, in: conversationID) } - /// The confirmed rows behind ``windowedMessages(for:startingAt:limit:)``, cached against - /// ``messageRevision``. + /// The confirmed rows behind ``windowedMessages(for:startingAt:limit:)``, cached until + /// ``noteMessageWrite(_:)`` reports a write to this chat that it can't patch in. /// /// The window read runs synchronously on the main actor on every observation tick, and an /// id-anchored window grows without bound as the reader pages back, so re-decoding it per tick /// is what a long group transcript feels as scroll jank. Nothing but a confirmed write can - /// change these rows, and every such write bumps the revision — the pending overlay is applied - /// on top by the caller and stays live. + /// change these rows, and every such write is reported — the pending overlay is applied on top by + /// the caller and stays live. private func confirmedWindow(for conversationID: ConversationID, startingAt startID: UInt64?, limit: Int) -> [ConversationMessage] { - let key = ConfirmedWindowKey(conversationID: conversationID, startID: startID, limit: limit, revision: messageRevision) + let key = ConfirmedWindowKey(conversationID: conversationID, startID: startID, limit: limit) if let cached = confirmedWindowCache, cached.key == key { return cached.messages } @@ -2254,7 +2371,6 @@ final class ConversationController { let conversationID: ConversationID let startID: UInt64? let limit: Int - let revision: Int } /// One entry, not a table: a screen reads one chat's window, and the next open replaces it. @@ -2388,7 +2504,7 @@ final class ConversationController { // drops the stale older epoch first. let (reconciled, pairs) = reconciledForPersist(messages, in: conversationID) let head = messages.map(\.eventSequence).max() ?? 0 - let ok = persist(operation: "load-messages") { + let ok = persist(operation: "load-messages", scope: .conversation(conversationID)) { try dropStaleEpochIfNeeded(before: messages, in: conversationID) try database.persistMessages(reconciled, cursor: head, conversationID: conversationID) } @@ -2462,7 +2578,9 @@ final class ConversationController { var transaction = Transaction() transaction.disablesAnimations = true withTransaction(transaction) { - _ = persist(operation: "load-older") { try database.upsertConversationMessages(older, conversationID: conversationID) } + _ = persist(operation: "load-older", scope: .messages(older.map(\.id), in: conversationID)) { + try database.upsertConversationMessages(older, conversationID: conversationID) + } } } } catch { @@ -2583,7 +2701,7 @@ final class ConversationController { // reconciles it later) instead of vanishing it. var confirmed = message confirmed.clientMessageID = clientMessageID - let ok = persist(operation: "send-message") { try database.upsertConversationMessages([confirmed], conversationID: conversationID) } + let ok = persist(operation: "send-message", scope: .messages([confirmed.id], in: conversationID)) { try database.upsertConversationMessages([confirmed], conversationID: conversationID) } if ok { store.dropPending(clientMessageID: clientMessageID, confirmedAt: message.id, in: conversationID) } else { @@ -2654,7 +2772,7 @@ final class ConversationController { clientMessageID: clientMessageID ) confirmed.clientMessageID = clientMessageID - if persist(operation: "send-payment-request", { try database.upsertConversationMessages([confirmed], conversationID: conversationID) }) { + if persist(operation: "send-payment-request", scope: .messages([confirmed.id], in: conversationID), { try database.upsertConversationMessages([confirmed], conversationID: conversationID) }) { store.dropPending(clientMessageID: clientMessageID, confirmedAt: confirmed.id, in: conversationID) } else { scheduleGapCatchUp(conversationID) @@ -3027,7 +3145,7 @@ final class ConversationController { chip.progress.finish() var confirmed = message confirmed.clientMessageID = clientMessageID - let ok = persist(operation: "send-media-message") { try database.upsertConversationMessages([confirmed], conversationID: conversationID) } + let ok = persist(operation: "send-media-message", scope: .messages([confirmed.id], in: conversationID)) { try database.upsertConversationMessages([confirmed], conversationID: conversationID) } if ok { keepSentImage(chip.image, clientMessageID: clientMessageID) store.dropPending(clientMessageID: clientMessageID, confirmedAt: message.id, in: conversationID) diff --git a/Flipcash/Core/Controllers/ConversationReactions.swift b/Flipcash/Core/Controllers/ConversationReactions.swift index 47b61b08f..36269a946 100644 --- a/Flipcash/Core/Controllers/ConversationReactions.swift +++ b/Flipcash/Core/Controllers/ConversationReactions.swift @@ -35,8 +35,9 @@ final class ConversationReactions { /// Counts the user's reaction adds for the strip's recents. Wired by the session before use. @ObservationIgnored var recents: RecentReactionsStore? - /// Called after every write to a stored message's reactions, so the transcript re-reads its rows. - @ObservationIgnored var didPersist: () -> Void = {} + /// Called after every write to stored messages' reactions, with the chat and the messages written, + /// so the transcript re-reads those rows. + @ObservationIgnored var didPersist: (ConversationID, [MessageID]) -> Void = { _, _ in } @ObservationIgnored private let messaging: any ConversationMessaging @ObservationIgnored private let database: Database @@ -140,7 +141,7 @@ final class ConversationReactions { } do { if try database.mergeReactions(summaries, conversationID: conversationID) > 0 { - didPersist() + didPersist(conversationID, Array(summaries.keys)) } } catch { logger.error("Failed to persist message reactions", metadata: [ @@ -248,7 +249,7 @@ final class ConversationReactions { private func write(_ transforms: [MessageID: (inout ReactionState) -> Void], in conversationID: ConversationID, operation: String) { do { try database.updateReactions(transforms, conversationID: conversationID) - didPersist() + didPersist(conversationID, Array(transforms.keys)) } catch { logger.error("Failed to persist message reactions", metadata: [ "operation": "\(operation)", diff --git a/FlipcashCore/Sources/FlipcashStore/Database+Conversations.swift b/FlipcashCore/Sources/FlipcashStore/Database+Conversations.swift index 08bb78f18..97e4aa8bf 100644 --- a/FlipcashCore/Sources/FlipcashStore/Database+Conversations.swift +++ b/FlipcashCore/Sources/FlipcashStore/Database+Conversations.swift @@ -200,6 +200,36 @@ nonisolated extension Database { return Array(try rows.map { conversationMessage(from: $0) }.compactMap { $0 }.reversed()) } + /// The oldest id among the newest `limit` stored rows — the lower bound of + /// ``messagesWindow(conversationID:before:limit:)`` with no `before` — or nil when the conversation + /// stores fewer than `limit` rows. One index seek, no decode. + public func newestWindowFloor(conversationID: ConversationID, limit: Int) throws -> UInt64? { + guard limit > 0 else { return nil } + let m = ConversationMessageTable() + let query = m.table.select(m.id) + .filter(m.conversationId == conversationID.data) + .order(m.id.desc) + .limit(1, offset: limit - 1) + return try reader.pluck(query)?[m.id] + } + + /// The stored messages among `ids`, oldest-first. An id that isn't stored, or whose row doesn't + /// decode, is absent — as it would be from a window read. + public func messages(conversationID: ConversationID, ids: [MessageID]) throws -> [ConversationMessage] { + let m = ConversationMessageTable() + let unique = Array(Set(ids.map(\.value))) + var messages: [ConversationMessage] = [] + // Chunked to stay under SQLite's bound-variable limit. + for start in stride(from: 0, to: unique.count, by: 500) { + let chunk = Array(unique[start.. ConversationController { + ConversationController( + fetching: server, membership: server, viewerSettings: server, messaging: server, streaming: server, + contactNaming: MockDMContactNaming(), + database: database, + owner: .generate()!, selfUserID: UUID(), + typingHeartbeatInterval: .seconds(3), incomingTypingExpiry: .seconds(10) + ) + } + + private func message(_ id: UInt64, _ text: String? = nil, sequence: UInt64 = 1) -> ConversationMessage { + ConversationMessage(id: MessageID(value: id), senderID: nil, content: .text(text ?? "m\(id)"), + date: Date(timeIntervalSince1970: TimeInterval(id)), unreadSeq: id, eventSequence: sequence) + } + + private func seeded(_ count: UInt64 = 200) throws -> (Database, URL) { + let (database, url) = try Database.makeTemp() + try database.upsertConversationMessages((1...count).map { message($0) }, conversationID: chat) + try database.upsertConversationMessages((1...count).map { message($0) }, conversationID: otherChat) + return (database, url) + } + + // MARK: - patchedWindow - + + @Test("patching an anchored window equals a fresh read after appends, an in-place edit and a write below the anchor") + func anchoredPatchEqualsFreshRead() throws { + let (database, url) = try seeded() + defer { Database.removeTemp(at: url) } + let window = try database.messages(conversationID: chat, from: 141) + + let written = [message(150, "edited", sequence: 2), message(201), message(202), message(10, "old edit", sequence: 2)] + try database.upsertConversationMessages(written, conversationID: chat) + let ids = written.map(\.id) + let patched = ConversationController.patchedWindow( + window, written: ids, rows: try database.messages(conversationID: chat, ids: ids), from: 141 + ) + + #expect(patched == (try database.messages(conversationID: chat, from: 141))) + #expect(patched.first?.id.value == 141) + #expect(patched.last?.id.value == 202) + } + + @Test("patching the newest-N window equals a fresh read: arrivals push the oldest rows out") + func newestWindowPatchEqualsFreshRead() throws { + let (database, url) = try seeded() + defer { Database.removeTemp(at: url) } + let window = try database.messagesWindow(conversationID: chat, limit: 60) + + let written = [message(201), message(202), message(203)] + try database.upsertConversationMessages(written, conversationID: chat) + let ids = written.map(\.id) + let floor = try database.newestWindowFloor(conversationID: chat, limit: 60) + let patched = ConversationController.patchedWindow( + window, written: ids, rows: try database.messages(conversationID: chat, ids: ids), from: floor + ) + + #expect(floor == 144) + #expect(patched == (try database.messagesWindow(conversationID: chat, limit: 60))) + } + + @Test("a written row with no stored copy is dropped from the window, not kept stale") + func writtenRowWithoutStoredCopyIsDropped() { + let window = [message(1), message(2), message(3)] + let patched = ConversationController.patchedWindow(window, written: [MessageID(value: 2)], rows: [], from: nil) + #expect(patched.map(\.id.value) == [1, 3]) + } + + // MARK: - Controller - + + @Test("a write reported for another chat keeps the open chat's cached window") + func writeInAnotherChatKeepsWindow() throws { + let (database, url) = try seeded() + defer { Database.removeTemp(at: url) } + let controller = makeController(database) + let before = controller.windowedMessages(for: chat, limit: 60) + + try database.upsertConversationMessages([message(201)], conversationID: chat) // unreported + controller.persistChanges(operation: "apply-chat-events", scope: .messages([MessageID(value: 201)], in: otherChat)) { + try database.persistMessages([message(201)], cursor: 0, conversationID: otherChat) + } + + #expect(controller.windowedMessages(for: chat, limit: 60) == before) + } + + @Test("an arrival in the open chat patches in only the reported row") + func arrivalInOpenChatPatchesReportedRow() throws { + let (database, url) = try seeded() + defer { Database.removeTemp(at: url) } + let controller = makeController(database) + _ = controller.windowedMessages(for: chat, startingAt: 101, limit: 60) + + try database.upsertConversationMessages([message(150, "unreported edit", sequence: 2)], conversationID: chat) + controller.persistChanges(operation: "apply-chat-events", scope: .messages([MessageID(value: 201)], in: chat)) { + try database.persistMessages([message(201)], cursor: 0, conversationID: chat) + } + + let window = controller.windowedMessages(for: chat, startingAt: 101, limit: 60) + #expect(window.map(\.id.value) == Array(101...201)) + #expect(window.first { $0.id.value == 150 }?.content == .text("m150")) // not re-read + } + + @Test("an arrival in the open chat's newest-N window matches a fresh read") + func arrivalInNewestWindowMatchesFreshRead() throws { + let (database, url) = try seeded() + defer { Database.removeTemp(at: url) } + let controller = makeController(database) + _ = controller.windowedMessages(for: chat, limit: 60) + + controller.persistChanges(operation: "apply-chat-events", scope: .messages([MessageID(value: 201)], in: chat)) { + try database.persistMessages([message(201)], cursor: 0, conversationID: chat) + } + + #expect(controller.windowedMessages(for: chat, limit: 60) == (try database.messagesWindow(conversationID: chat, limit: 60))) + } + + @Test("a persisted reaction patches the reacted row into the cached window") + func reactionPatchesRow() throws { + let (database, url) = try seeded() + defer { Database.removeTemp(at: url) } + let controller = makeController(database) + _ = controller.windowedMessages(for: chat, limit: 60) + let target = MessageID(value: 190) + + try database.updateReactions(messageID: target, conversationID: chat) { + $0.applyUpdate(emoji: "🔥", actorIsSelf: false, added: true, count: 1, version: 1, reactedAt: nil) + } + controller.reactions.didPersist(chat, [target]) + + let window = controller.windowedMessages(for: chat, limit: 60) + #expect(window == (try database.messagesWindow(conversationID: chat, limit: 60))) + #expect(window.first { $0.id == target }?.reactionState != nil) + } + + @Test("a write that can drop rows in the open chat re-reads its window") + func conversationScopeReReads() throws { + let (database, url) = try seeded() + defer { Database.removeTemp(at: url) } + let controller = makeController(database) + _ = controller.windowedMessages(for: chat, limit: 60) + + try database.upsertConversationMessages([message(201)], conversationID: chat) // unreported + controller.noteMessageWrite(.conversation(chat)) + + #expect(controller.windowedMessages(for: chat, limit: 60).last?.id.value == 201) + } + + @Test("a multi-batch catch-up reports its writes once, when it ends") + func catchUpReportsOnce() async throws { + let (database, url) = try seeded() + defer { Database.removeTemp(at: url) } + let server = MockConversations() + server.deltaBatches = [ + MockConversations.DeltaBatch(messages: [message(201, sequence: 201)], checkpoint: 201), + MockConversations.DeltaBatch(messages: [message(202, sequence: 202)], checkpoint: 202), + MockConversations.DeltaBatch(messages: [message(203, sequence: 203)], checkpoint: 203), + ] + server.deltaHead = 203 + let controller = makeController(database, server: server) + _ = controller.windowedMessages(for: chat, limit: 60) + let revision = controller.messageRevision + + await controller.catchUp(conversationID: chat) + + #expect(controller.messageRevision == revision + 1) + #expect(controller.windowedMessages(for: chat, limit: 60) == (try database.messagesWindow(conversationID: chat, limit: 60))) + #expect(controller.windowedMessages(for: chat, limit: 60).last?.id.value == 203) + } +} From 23d2ac68943ba47b799835afb59a6870b63b7665 Mon Sep 17 00:00:00 2001 From: Brandon McAnsh Date: Fri, 6 Nov 2026 17:58:02 -0500 Subject: [PATCH 2/2] fix(chat): re-read the newest-N window when a written row is gone A fresh read pulls an older row in to fill the limit, which the patch can't see. --- Flipcash/Core/Controllers/ConversationController.swift | 3 +++ 1 file changed, 3 insertions(+) diff --git a/Flipcash/Core/Controllers/ConversationController.swift b/Flipcash/Core/Controllers/ConversationController.swift index 1a35e67b4..8cee9ee78 100644 --- a/Flipcash/Core/Controllers/ConversationController.swift +++ b/Flipcash/Core/Controllers/ConversationController.swift @@ -2244,6 +2244,9 @@ final class ConversationController { // Re-reading most of the window row by row costs more than one window read. guard inWindow.count * 4 <= max(cached.messages.count, 4) else { return nil } let rows = try database.messages(conversationID: key.conversationID, ids: inWindow) + // A row gone from the newest-N window lets a fresh read pull an older row in to fill the + // limit, which a patch can't see, so re-read instead. + guard key.startID != nil || rows.count == inWindow.count else { return nil } return (key, Self.patchedWindow(cached.messages, written: inWindow, rows: rows, from: lowerBound)) } catch { return nil