refactor(chat): drop the deprecated new_messages overlay

Flipcash/Core/Controllers/ConversationController.swift

@@ -451,7 +451,7 @@ final class ConversationController {
     private func hydrateIfUnknown(_ event: ConversationStreamEvent) {
         let conversationID: ConversationID
         switch event {
-        case .newMessages(let id, _), .chatEvents(let id, _), .lastActivityChanged(let id, _), .readPointersChanged(let id, _):
+        case .chatEvents(let id, _), .lastActivityChanged(let id, _), .readPointersChanged(let id, _):
             conversationID = id
         case .metadataRefresh:
             return
@@ -614,22 +614,9 @@ final class ConversationController {
     /// post-`apply` state so monotonic rules (read pointers) hold.
     private func persist(event: ConversationStreamEvent) {
         switch event {
-        case .newMessages(let conversationID, let messages):
+        case .chatEvents(let conversationID, let events):
             // Read before the write: the newest stored id is the analytics watermark,
             // and after the upsert it would already include this batch.
-            let countedThrough = (try? database.newestMessageID(conversationID: conversationID)) ?? nil
-            let (reconciled, pairs) = reconciledForPersist(messages, in: conversationID)
-            let ok = persist(operation: "upsert-messages") { try database.upsertConversationMessages(reconciled, conversationID: conversationID) }
-            if ok {
-                commitReconciled(pairs, in: conversationID)
-                receipts.countReceived(reconciled, countedThrough: countedThrough, delivery: .live)
-            } else {
-                // The delivered batch is in neither the DB nor the store — refetch it from the event log.
-                scheduleGapCatchUp(conversationID)
-            }
-            refreshFeedPreview(for: conversationID)
-            persistConversation(conversationID)
-        case .chatEvents(let conversationID, let events):
             let countedThrough = (try? database.newestMessageID(conversationID: conversationID)) ?? nil
             let (reconciled, pairs) = reconciledForPersist(events.flatMap { $0.mutations.map(\.message) }, in: conversationID)
             // Messages + the advanced cursor persist atomically. `store.apply` already advanced the

FlipcashCore/Sources/FlipcashCore/Models/Conversation/ConversationStore.swift

@@ -288,11 +288,6 @@ public struct ConversationStore: Sendable {
     @discardableResult
     public mutating func apply(_ event: ConversationStreamEvent) -> GapSignal {
         switch event {
-        case .newMessages(let conversationID, let messages):
-            if let latest = messages.max(by: { $0.id < $1.id }) {
-                advanceLastActivity(to: latest.date, in: conversationID)
-            }
-            return .none
         case .chatEvents(let conversationID, let events):
             return applyChatEvents(events, into: conversationID)
         case .metadataRefresh(let conversation):

FlipcashCore/Sources/FlipcashCore/Models/Conversation/ConversationStreamEvent.swift

@@ -13,12 +13,9 @@ import FlipcashAPI
 /// apply them without touching proto types.
 public enum ConversationStreamEvent: Sendable {
 
-    /// New messages arrived in a conversation.
-    case newMessages(conversationID: ConversationID, messages: [ConversationMessage])
-
     /// Durable, sequenced event-log mutations for a conversation (message sent/edited/deleted). The
     /// store applies them last-writer-wins by `event_sequence` and gap-detects via `sequence`/`count`,
-    /// catching up with `GetDelta` on a gap. Supersedes the deprecated `newMessages` overlay.
+    /// catching up with `GetDelta` on a gap.
     case chatEvents(conversationID: ConversationID, events: [DecodedChatEvent])
 
     /// A conversation's full metadata was refreshed (members/last message/last activity).
@@ -93,21 +90,12 @@ extension ConversationStreamEvent {
         let conversationID = ConversationID(update.chat)
         var events: [ConversationStreamEvent] = []
 
-        // The sequenced event log (message sent/edited/deleted). Additive with `new_messages`: both may
-        // carry the same send during the server's migration window, and last-writer-wins by
-        // `event_sequence` in the store lands it exactly once.
+        // The sequenced event log (message sent/edited/deleted).
         let chatEvents = update.events.events.map(DecodedChatEvent.init)
         if !chatEvents.isEmpty {
             events.append(.chatEvents(conversationID: conversationID, events: chatEvents))
         }
 
-        // The deprecated real-time overlay. Decoded regardless of `events` so a message that arrives
-        // only here (an events-empty or malformed batch) is never dropped; the store dedups by version.
-        let messages = update.newMessages.messages.compactMap(ConversationMessage.init)
-        if !messages.isEmpty {
-            events.append(.newMessages(conversationID: conversationID, messages: messages))
-        }
-
         for metadataUpdate in update.metadataUpdates {
             switch metadataUpdate.kind {
             case .fullRefresh(let refresh):

FlipcashCore/Tests/FlipcashCoreTests/ConversationStoreTests.swift

@@ -380,14 +380,6 @@ struct ConversationStoreTests {
         #expect(store.appliedCursor(for: conversationID(1)) == 3)
     }
 
-    @Test("newMessages bumps the conversation's last activity")
-    func newMessagesBumpsActivity() {
-        var store = ConversationStore()
-        store.setFeed([conversation(1, lastActivity: 100), conversation(2, lastActivity: 200)])
-        store.apply(.newMessages(conversationID: conversationID(1), messages: [message(5, "yo", at: 500)]))
-        #expect(store.conversations.first?.id == conversationID(1))
-    }
-
     @Test("readPointersChanged advances a member's READ watermark monotonically")
     func readPointers() {
         let me = UUID()

FlipcashCore/Tests/FlipcashCoreTests/ConversationStreamEventDecodeTests.swift

@@ -22,24 +22,6 @@ struct ConversationStreamEventDecodeTests {
         }
     }
 
-    @Test("New messages decode to a newMessages event")
-    func newMessages() {
-        let event = Flipcash_Event_V1_Event.with {
-            $0.chatUpdate = .with {
-                $0.chat = .with { $0.value = conversationBytes }
-                $0.newMessages = .with { $0.messages = [textMessage(1, "hi"), textMessage(2, "yo")] }
-            }
-        }
-
-        let decoded = ConversationStreamEvent.decode(event)
-        #expect(decoded.count == 1)
-        guard case .newMessages(let conversationID, let messages) = decoded.first else {
-            Issue.record("expected .newMessages"); return
-        }
-        #expect(conversationID == ConversationID(data: conversationBytes))
-        #expect(messages.map(\.content) == [.text("hi"), .text("yo")])
-    }
-
     @Test("FullRefresh metadata decodes to a metadataRefresh event")
     func metadataRefresh() {
         let event = Flipcash_Event_V1_Event.with {
@@ -82,12 +64,12 @@ struct ConversationStreamEventDecodeTests {
         #expect(date == Date(timeIntervalSince1970: 900))
     }
 
-    @Test("New messages and a metadata update decode to both events in order")
+    @Test("An event batch and a metadata update decode to both events in order")
     func combined() {
         let event = Flipcash_Event_V1_Event.with {
             $0.chatUpdate = .with {
                 $0.chat = .with { $0.value = conversationBytes }
-                $0.newMessages = .with { $0.messages = [textMessage(5, "ping")] }
+                $0.events = .with { $0.events = [.with { $0.sequence = 5; $0.count = 1; $0.mutations = [sentMutation(5, "ping")] }] }
                 $0.metadataUpdates = [.with {
                     $0.lastActivityChanged = .with { $0.newLastActivity = .init(date: Date(timeIntervalSince1970: 1)) }
                 }]
@@ -96,7 +78,7 @@ struct ConversationStreamEventDecodeTests {
 
         let decoded = ConversationStreamEvent.decode(event)
         #expect(decoded.count == 2)
-        if case .newMessages = decoded.first {} else { Issue.record("first should be .newMessages") }
+        if case .chatEvents = decoded.first {} else { Issue.record("first should be .chatEvents") }
         if case .lastActivityChanged = decoded.last {} else { Issue.record("last should be .lastActivityChanged") }
     }
 
@@ -185,13 +167,13 @@ struct ConversationStreamEventDecodeTests {
         let event = Flipcash_Event_V1_Event.with {
             $0.chatUpdate = .with {
                 $0.chat = .with { $0.value = conversationBytes }
-                $0.newMessages = .with { $0.messages = [textMessage(1, "hi")] }
+                $0.events = .with { $0.events = [.with { $0.sequence = 1; $0.count = 1; $0.mutations = [sentMutation(1, "hi")] }] }
                 $0.isTypingNotifications = .with { $0.isTypingNotifications = [typing(u1, .startedTyping)] }
             }
         }
         let decoded = ConversationStreamEvent.decode(event)
         #expect(decoded.count == 2)
-        #expect(decoded.contains { if case .newMessages = $0 { true } else { false } })
+        #expect(decoded.contains { if case .chatEvents = $0 { true } else { false } })
         #expect(decoded.contains { if case .typingChanged = $0 { true } else { false } })
     }
 
@@ -229,33 +211,6 @@ struct ConversationStreamEventDecodeTests {
         #expect(tombstone.id.value == 3)
     }
 
-    @Test("both events and new_messages present decode to both (additive migration gate)")
-    func chatEventsAndNewMessagesAdditive() {
-        let event = Flipcash_Event_V1_Event.with {
-            $0.chatUpdate = .with {
-                $0.chat = .with { $0.value = conversationBytes }
-                $0.newMessages = .with { $0.messages = [textMessage(9, "dup")] }
-                $0.events = .with { $0.events = [.with { $0.sequence = 9; $0.count = 1; $0.mutations = [sentMutation(9, "dup")] }] }
-            }
-        }
-        let decoded = ConversationStreamEvent.decode(event)
-        #expect(decoded.contains { if case .chatEvents = $0 { true } else { false } })
-        #expect(decoded.contains { if case .newMessages = $0 { true } else { false } })
-    }
-
-    @Test("an absent events batch does not suppress new_messages (deprecated path still works)")
-    func emptyEventsKeepsNewMessages() {
-        let event = Flipcash_Event_V1_Event.with {
-            $0.chatUpdate = .with {
-                $0.chat = .with { $0.value = conversationBytes }
-                $0.newMessages = .with { $0.messages = [textMessage(1, "keep")] }
-            }
-        }
-        let decoded = ConversationStreamEvent.decode(event)
-        #expect(decoded.contains { if case .newMessages = $0 { true } else { false } })
-        #expect(!decoded.contains { if case .chatEvents = $0 { true } else { false } })
-    }
-
     @Test("an event whose only mutation is unrepresentable still carries sequence/count so the cursor advances")
     func unrepresentableMutationStillAdvances() {
         let event = Flipcash_Event_V1_Event.with {

FlipcashTests/Chat/ConversationReceiptWiringTests.swift

@@ -76,11 +76,11 @@ struct ConversationReceiptWiringTests {
         controller.start()
         try await waitUntil { mock.streamOpened }
 
-        mock.emit(.newMessages(conversationID: ConversationID.test(1), messages: [inboundTip(id: 5, from: them)]))
+        mock.emit(.sent([inboundTip(id: 5, from: them)], in: ConversationID.test(1)))
         try await waitUntil { spy.count(of: .tips) == 1 }
 
         // The same message delivered again (a reconnect replay) must not re-credit.
-        mock.emit(.newMessages(conversationID: ConversationID.test(1), messages: [inboundTip(id: 5, from: them)]))
+        mock.emit(.sent([inboundTip(id: 5, from: them)], in: ConversationID.test(1)))
         try await Task.sleep(for: .milliseconds(100))
 
         #expect(spy.count(of: .tips) == 1)
@@ -126,10 +126,10 @@ struct ConversationReceiptWiringTests {
         controller.start()
         try await waitUntil { mock.streamOpened && !controller.conversations.isEmpty }
 
-        mock.emit(.newMessages(conversationID: ConversationID.test(1), messages: [
+        mock.emit(.sent([
             inboundTip(id: 2, from: them),
             inboundTip(id: 3, from: them),
-        ]))
+        ], in: ConversationID.test(1)))
         try await waitUntil { ((try? database.newestMessageID(conversationID: ConversationID.test(1))) ?? nil) == MessageID(value: 3) }
 
         await controller.markRead(conversationID: ConversationID.test(1))
@@ -153,7 +153,7 @@ struct ConversationReceiptWiringTests {
         controller.start()
         try await waitUntil { mock.streamOpened && !controller.conversations.isEmpty }
 
-        mock.emit(.newMessages(conversationID: ConversationID.test(1), messages: [inboundTip(id: 2, from: them)]))
+        mock.emit(.sent([inboundTip(id: 2, from: them)], in: ConversationID.test(1)))
         try await waitUntil { ((try? database.newestMessageID(conversationID: ConversationID.test(1))) ?? nil) == MessageID(value: 2) }
 
         await controller.markRead(conversationID: ConversationID.test(1))

FlipcashTests/ConversationControllerTests.swift

@@ -506,7 +506,7 @@ struct ConversationControllerTests {
         try await waitUntil { mock.streamOpened }
 
         let message = ConversationMessage(id: MessageID(value: 9), senderID: nil, content: .text("live"), date: Date(timeIntervalSince1970: 0), unreadSeq: 0)
-        mock.emit(.newMessages(conversationID: ConversationID.test(1), messages: [message]))
+        mock.emit(.sent([message], in: ConversationID.test(1)))
 
         // The stream is consumed on a Task; poll briefly for it to apply.
         try await waitUntil { !controller.messages(for: ConversationID.test(1)).isEmpty }
@@ -535,7 +535,7 @@ struct ConversationControllerTests {
         )
         mock.feed = [existing, newConversation]
         let message = ConversationMessage(id: MessageID(value: 1), senderID: nil, content: .text("first"), date: Date(timeIntervalSince1970: 200), unreadSeq: 0)
-        mock.emit(.newMessages(conversationID: ConversationID.test(2), messages: [message]))
+        mock.emit(.sent([message], in: ConversationID.test(2)))
 
         // The stream is consumed on a Task; poll briefly for the hydration.
         try await waitUntil { controller.conversations.count >= 2 }
@@ -957,7 +957,7 @@ struct ConversationControllerTests {
         let backlog = (1...150).map {
             ConversationMessage(id: MessageID(value: UInt64($0)), senderID: nil, content: .text("m\($0)"), date: Date(timeIntervalSince1970: TimeInterval($0)), unreadSeq: UInt64($0))
         }
-        mock.emit(.newMessages(conversationID: ConversationID.test(1), messages: backlog))
+        mock.emit(.sent(backlog, in: ConversationID.test(1)))
 
         // The whole backlog lands in the DB (nothing dropped), but the accessor reads a bounded window.
         try await waitUntil { ((try? database.messageCount(conversationID: ConversationID.test(1))) ?? 0) == 150 }

FlipcashTests/TestSupport/Conversation+TestSupport.swift

@@ -12,3 +12,18 @@ extension ConversationID {
         ConversationID(data: Data(repeating: byte, count: 32))
     }
 }
+
+extension ConversationStreamEvent {
+    /// A live `.chatEvents` update carrying `messages` as one contiguous run of `.sent` mutations,
+    /// as the server delivers a send. The run is sequenced from zero, so it lands on a conversation
+    /// whose frontier the test has not seeded.
+    static func sent(_ messages: [ConversationMessage], in conversationID: ConversationID) -> ConversationStreamEvent {
+        .chatEvents(conversationID: conversationID, events: [
+            DecodedChatEvent(
+                sequence: UInt64(messages.count),
+                count: UInt64(messages.count),
+                mutations: messages.map { .sent($0) }
+            )
+        ])
+    }
+}