diff --git a/Flipcash/Core/Controllers/ConversationController.swift b/Flipcash/Core/Controllers/ConversationController.swift index b8539e6b7..682e36d94 100644 --- a/Flipcash/Core/Controllers/ConversationController.swift +++ b/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 diff --git a/FlipcashCore/Sources/FlipcashCore/Models/Conversation/ConversationStore.swift b/FlipcashCore/Sources/FlipcashCore/Models/Conversation/ConversationStore.swift index 64b24cc10..c52d2918e 100644 --- a/FlipcashCore/Sources/FlipcashCore/Models/Conversation/ConversationStore.swift +++ b/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): diff --git a/FlipcashCore/Sources/FlipcashCore/Models/Conversation/ConversationStreamEvent.swift b/FlipcashCore/Sources/FlipcashCore/Models/Conversation/ConversationStreamEvent.swift index fbbc0b901..5dc749284 100644 --- a/FlipcashCore/Sources/FlipcashCore/Models/Conversation/ConversationStreamEvent.swift +++ b/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): diff --git a/FlipcashCore/Tests/FlipcashCoreTests/ConversationStoreTests.swift b/FlipcashCore/Tests/FlipcashCoreTests/ConversationStoreTests.swift index 717d8a0d3..e46ab4a9c 100644 --- a/FlipcashCore/Tests/FlipcashCoreTests/ConversationStoreTests.swift +++ b/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() diff --git a/FlipcashCore/Tests/FlipcashCoreTests/ConversationStreamEventDecodeTests.swift b/FlipcashCore/Tests/FlipcashCoreTests/ConversationStreamEventDecodeTests.swift index 482983a09..a7c1b523c 100644 --- a/FlipcashCore/Tests/FlipcashCoreTests/ConversationStreamEventDecodeTests.swift +++ b/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 { diff --git a/FlipcashTests/Chat/ConversationReceiptWiringTests.swift b/FlipcashTests/Chat/ConversationReceiptWiringTests.swift index 1262bca38..b7f335edc 100644 --- a/FlipcashTests/Chat/ConversationReceiptWiringTests.swift +++ b/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)) diff --git a/FlipcashTests/ConversationControllerTests.swift b/FlipcashTests/ConversationControllerTests.swift index 5cf44da36..f24366b3b 100644 --- a/FlipcashTests/ConversationControllerTests.swift +++ b/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 } diff --git a/FlipcashTests/TestSupport/Conversation+TestSupport.swift b/FlipcashTests/TestSupport/Conversation+TestSupport.swift index 6c4215773..21c700f02 100644 --- a/FlipcashTests/TestSupport/Conversation+TestSupport.swift +++ b/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) } + ) + ]) + } +}