From 3d0d09640b745c098d099d71ab2e4e175ec36af4 Mon Sep 17 00:00:00 2001 From: Brandon McAnsh Date: Fri, 21 Aug 2026 22:57:57 -0400 Subject: [PATCH] fix(chat): back chats up by their applied event cursor, not by message presence MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A feed sync never backfilled a transcript. It asked `hasMessages` to decide whether a chat had ever been fetched, but the same sync writes each chat's last-message preview as a message row moments earlier — so the answer was always yes, and the newest page was never pulled. Chats opened to a single preview message until the user scrolled. Underneath that sat a worse problem: `latest_event_sequence` was written by two parties meaning two different things. The event stream advanced it to the sequence the client had actually applied; the feed sync overwrote it with the server's reported head via `@Insert(REPLACE)`. Once a sync landed, the next `GetDelta(after:)` resumed from head and silently skipped every event in between. The same whole-row replace also zeroed `analytics_counted_through`, re-counting received messages already counted. The column now has one meaning — the cursor the client has applied: - `ChatMetadataDao.upsert` inserts, or updates only the server-owned columns in place. The two client-owned watermarks are never touched by a sync. - `ChatEntityMapper` stops round-tripping the server head. A row starts at 0, meaning "this transcript has never been fetched"; a chat rebuilt from the database reports 0 for the head — unknown, not "no events". - `MessagingDelegate.loadMessages` seats the cursor at the newest page's frontier, and only ever forward. - The sync's decision reads the cursor: unseated means load the newest page (never a delta, which would re-pull the whole history from sequence 0), behind the server head means delta-sync the missed window. The pre-sync snapshot #1331 added comes out with it. That workaround read the cursor and `hasMessages` *before* the sync wrote, and carried the old cursor on `DeltaSyncNeeded`, because the write clobbered the row it was about to read. Nothing clobbers it now: the sync reads the cursor after its own write, and `performDeltaSync` reads it for itself again. This matches the iOS fix in code-ios-app#628. --- .../chat/internal/RealChatCoordinator.kt | 2 +- .../internal/delegates/EventStreamDelegate.kt | 14 +- .../internal/delegates/FeedSyncDelegate.kt | 60 +++----- .../internal/delegates/MessagingDelegate.kt | 9 ++ .../shared/chat/FeedSyncBackfillTest.kt | 145 ++++++++++++++++++ .../shared/chat/FeedSyncCatchUpTest.kt | 67 ++++---- .../shared/chat/MessagingLoadCursorTest.kt | 86 +++++++++++ .../app/persistence/dao/ChatMetadataDao.kt | 53 ++++++- .../persistence/dao/ChatMetadataDaoTest.kt | 123 +++++++++++++++ .../sources/mapper/chat/ChatEntityMapper.kt | 13 +- .../mapper/chat/ChatEntityMapperTest.kt | 63 ++++++++ 11 files changed, 539 insertions(+), 96 deletions(-) create mode 100644 apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/FeedSyncBackfillTest.kt create mode 100644 apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/MessagingLoadCursorTest.kt create mode 100644 apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMetadataDaoTest.kt create mode 100644 apps/flipcash/shared/persistence/sources/src/test/kotlin/com/flipcash/app/persistence/sources/mapper/chat/ChatEntityMapperTest.kt diff --git a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/RealChatCoordinator.kt b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/RealChatCoordinator.kt index 3ba5b7d26..535b70adc 100644 --- a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/RealChatCoordinator.kt +++ b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/RealChatCoordinator.kt @@ -144,7 +144,7 @@ class RealChatCoordinator @Inject constructor( is FeedSyncDelegate.Event.LoadMessages -> messagingDelegate.loadMessages(event.chatId) is FeedSyncDelegate.Event.DeltaSyncNeeded -> - eventStreamDelegate.performDeltaSync(event.chatId, event.afterSequence) + eventStreamDelegate.performDeltaSync(event.chatId) // Arrives after every catch-up item above it, because this is one sequential // collector over a FIFO channel. Anything reading chat history as evidence — // the wallet's "send a tip" milestone — waits for this rather than for the diff --git a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/EventStreamDelegate.kt b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/EventStreamDelegate.kt index e740da77d..13109f481 100644 --- a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/EventStreamDelegate.kt +++ b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/EventStreamDelegate.kt @@ -186,16 +186,14 @@ class EventStreamDelegate @Inject constructor( } /** - * Backfills a chat from [afterSequence] onward. + * Backfills a chat from its applied cursor onward. * - * Callers that have already written the server's sequence to `chat_metadata` before asking for - * a delta must pass the sequence the cache held *beforehand* — reading it here would return - * the value they just wrote and ask the server for everything after its own latest event, - * which is always nothing. The live gap-fill path has no such write in front of it and omits - * the argument. + * The cursor is read from `chat_metadata`, which holds what the client has actually applied: + * nothing writes the server's head there — a feed sync refreshes only the server-owned columns + * (`ChatMetadataDao.upsert`), and the mapper drops the head on the way in. */ - internal suspend fun performDeltaSync(chatId: ChatId, afterSequence: Long? = null) { - val afterSequence = afterSequence ?: metadataDataSource.getLatestEventSequence(chatId) + internal suspend fun performDeltaSync(chatId: ChatId) { + val afterSequence = metadataDataSource.getLatestEventSequence(chatId) trace(tag = TAG, message = "Delta sync for $chatId from sequence $afterSequence", type = TraceType.Process) try { diff --git a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/FeedSyncDelegate.kt b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/FeedSyncDelegate.kt index b4c1e6e6e..8008b7403 100644 --- a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/FeedSyncDelegate.kt +++ b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/FeedSyncDelegate.kt @@ -64,10 +64,11 @@ class FeedSyncDelegate @Inject constructor( sealed interface Event { data class LoadMessages(val chatId: ChatId) : Event /** - * @param afterSequence the sequence the local cache held *before* this sync overwrote it; - * the delta must be requested from there, not from the row we just wrote. + * The chat's applied cursor is behind the feed's head. The consumer reads that cursor from + * `chat_metadata` itself: a feed sync never writes it (see `ChatMetadataDao.upsert`), so + * the row still holds what the client has actually applied. */ - data class DeltaSyncNeeded(val chatId: ChatId, val afterSequence: Long) : Event + data class DeltaSyncNeeded(val chatId: ChatId) : Event /** * Emitted last by every successful sync, after any catch-up above it. @@ -216,35 +217,10 @@ class FeedSyncDelegate @Inject constructor( Result.success(contact.chats + (tip?.chats ?: emptyList())) } - /** What the local cache held for a chat before a sync wrote to it. */ - private data class CachedChatState( - val hasMessages: Boolean, - val latestEventSequence: Long, - ) - - private suspend fun performFeedSync() { + internal suspend fun performFeedSync() { stateHolder.update { it.copy(feedSyncState = FeedSyncState.Syncing) } fetchCombinedFeed() .onSuccess { chats -> - // Snapshot the cache before writing to it. The writes below stamp the server's - // latestEventSequence onto every metadata row and give every chat at least one - // message, so the catch-up checks at the end of this sync — which are reads of - // exactly those two things — would otherwise always find the chat current and - // never fire. - // - // Both branches being inert had the same consequence: after a re-login, where - // logout has cleared the chat cache (ChatCoordinator.reset), a chat is left - // holding only the one message the feed carried. Anything derived from chat - // history then reads as if the rest never happened — the wallet's "send a tip" - // milestone looks for an outgoing TIPPED message and re-shows the new-user - // tutorial to an account that has already tipped. - val cached = chats.associate { chat -> - chat.chatId to CachedChatState( - hasMessages = messageDataSource.hasMessages(chat.chatId), - latestEventSequence = metadataDataSource.getLatestEventSequence(chat.chatId), - ) - } - metadataDataSource.upsert(chats) for (chat in chats) { @@ -258,19 +234,19 @@ class FeedSyncDelegate @Inject constructor( trace(tag = TAG, message = "Feed synced: ${chats.size} chats", type = TraceType.Process) for (chat in chats) { - val before = cached[chat.chatId] ?: continue - if (chat.latestEventSequence > 0) { - if (before.latestEventSequence > 0 && - before.latestEventSequence < chat.latestEventSequence - ) { - _events.send( - Event.DeltaSyncNeeded(chat.chatId, before.latestEventSequence) - ) - continue - } - } - if (!before.hasMessages) { - _events.send(Event.LoadMessages(chat.chatId)) + // The applied cursor, not the presence of messages, is what says whether a + // transcript was ever pulled: the loop above persists each chat's last-message + // preview, so "has messages" is true for nearly every chat in a feed the client + // has otherwise never fetched. Only a message load or an applied delta seats a + // cursor. + val cursor = metadataDataSource.getLatestEventSequence(chat.chatId) + when { + // Never fetched: take the newest page. Resuming a delta from 0 would instead + // re-pull the entire history as a "gap". + cursor <= 0L -> _events.send(Event.LoadMessages(chat.chatId)) + // Fetched, but the server has moved on: stream the missed window. + chat.latestEventSequence > cursor -> + _events.send(Event.DeltaSyncNeeded(chat.chatId)) } } diff --git a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/MessagingDelegate.kt b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/MessagingDelegate.kt index fcf3b2619..4babb09c8 100644 --- a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/MessagingDelegate.kt +++ b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/MessagingDelegate.kt @@ -155,6 +155,15 @@ class MessagingDelegate @Inject constructor( .onSuccess { messages -> messageDataSource.upsert(chatId, messages) + // Seat the event-log cursor at the newest page's frontier. It is what marks this + // transcript as fetched, and it lets a following catch-up resume from head and + // append genuinely newer messages instead of re-pulling the whole history from + // sequence 0. Only ever advanced — a page older than the cursor must not rewind it. + val head = messages.maxOfOrNull { it.eventSequence } ?: 0L + if (head > metadataDataSource.getLatestEventSequence(chatId)) { + metadataDataSource.updateLatestEventSequence(chatId, head) + } + val latest = messages.maxByOrNull { it.messageId } ?: return@onSuccess metadataDataSource.updateLastMessageId(chatId, latest.messageId) metadataDataSource.updateLastActivity(chatId, latest.timestamp.toEpochMilliseconds()) diff --git a/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/FeedSyncBackfillTest.kt b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/FeedSyncBackfillTest.kt new file mode 100644 index 000000000..bd172b977 --- /dev/null +++ b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/FeedSyncBackfillTest.kt @@ -0,0 +1,145 @@ +package com.flipcash.shared.chat + +import com.flipcash.app.persistence.sources.ChatMessageDataSource +import com.flipcash.app.persistence.sources.ChatMetadataDataSource +import com.flipcash.services.controllers.ChatController +import com.flipcash.services.models.chat.ChatFeedPage +import com.flipcash.services.models.chat.ChatId +import com.flipcash.services.models.chat.ChatMetadata +import com.flipcash.services.models.chat.ChatType +import com.flipcash.shared.chat.internal.delegates.FeedSyncDelegate +import io.mockk.coEvery +import io.mockk.mockk +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.launch +import kotlinx.coroutines.test.TestScope +import kotlinx.coroutines.test.UnconfinedTestDispatcher +import kotlinx.coroutines.test.runCurrent +import kotlinx.coroutines.test.runTest +import kotlin.time.Instant +import org.junit.Assert.assertEquals +import org.junit.Test + +/** + * Covers which chats a feed sync decides to backfill. The decision is keyed on the locally + * applied event cursor — never on whether the chat holds any message rows, because the sync + * itself persists every chat's last-message preview as a row. + */ +@OptIn(ExperimentalCoroutinesApi::class) +class FeedSyncBackfillTest { + + private fun metadata(hex: String, latestEventSequence: Long) = ChatMetadata( + chatId = ChatId(hex), + type = ChatType.CONTACT_DM, + members = emptyList(), + lastMessage = null, + lastActivity = Instant.fromEpochSeconds(1000), + latestEventSequence = latestEventSequence, + ) + + /** + * Runs one feed sync over a single chat whose server head is [serverHead] and whose locally + * applied cursor is [localCursor], and returns the events it emitted. Every successful sync + * ends with [FeedSyncDelegate.Event.CatchUpComplete], so that marker closes each expectation + * below; what varies is the backfill decision in front of it. + */ + private suspend fun TestScope.backfillEvents( + serverHead: Long, + localCursor: Long, + ): List { + val chat = metadata(CHAT_HEX, serverHead) + + val chatController = mockk(relaxed = true) + coEvery { chatController.getDmChatFeed(ChatType.CONTACT_DM, any()) } returns + Result.success(ChatFeedPage(listOf(chat), null, false)) + coEvery { chatController.getDmChatFeed(ChatType.TIP_DM, any()) } returns + Result.success(ChatFeedPage(emptyList(), null, false)) + + val metadataDataSource = mockk(relaxed = true) + coEvery { metadataDataSource.getLatestEventSequence(chat.chatId) } returns localCursor + + // Nothing here stubs `hasMessages`: the decision must not consult it. The sync persists + // every chat's last-message preview, so it would answer true even for a chat whose + // transcript was never pulled. + val messageDataSource = mockk(relaxed = true) + + val delegate = FeedSyncDelegate( + chatController = chatController, + metadataDataSource = metadataDataSource, + messageDataSource = messageDataSource, + memberDataSource = mockk(relaxed = true), + stateHolder = mockk(relaxed = true), + userManager = mockk(relaxed = true), + ) + + val received = mutableListOf() + val collector = launch(UnconfinedTestDispatcher(testScheduler)) { + delegate.events.collect { received += it } + } + delegate.performFeedSync() + runCurrent() + collector.cancel() + return received + } + + @Test + fun `chat that was never fetched is backfilled with its newest page`() = runTest { + // The sync has just written this chat's last-message preview, so it holds a message row — + // only the unseated cursor reveals that its transcript was never pulled. + val events = backfillEvents(serverHead = 12, localCursor = 0) + + assertEquals( + listOf( + FeedSyncDelegate.Event.LoadMessages(ChatId(CHAT_HEX)), + FeedSyncDelegate.Event.CatchUpComplete, + ), + events, + ) + } + + @Test + fun `chat that was never fetched is never delta synced from zero`() = runTest { + // A delta after sequence 0 re-pulls the entire history instead of the newest page. + val events = backfillEvents(serverHead = 0, localCursor = 0) + + assertEquals( + listOf( + FeedSyncDelegate.Event.LoadMessages(ChatId(CHAT_HEX)), + FeedSyncDelegate.Event.CatchUpComplete, + ), + events, + ) + } + + @Test + fun `chat behind the server head catches up with a delta sync`() = runTest { + val events = backfillEvents(serverHead = 12, localCursor = 5) + + assertEquals( + listOf( + FeedSyncDelegate.Event.DeltaSyncNeeded(ChatId(CHAT_HEX)), + FeedSyncDelegate.Event.CatchUpComplete, + ), + events, + ) + } + + @Test + fun `chat already at the server head needs no backfill`() = runTest { + val events = backfillEvents(serverHead = 12, localCursor = 12) + + assertEquals(listOf(FeedSyncDelegate.Event.CatchUpComplete), events) + } + + @Test + fun `chat ahead of a stale feed head needs no backfill`() = runTest { + // The stream can apply events the feed page predates; that is not a gap. + val events = backfillEvents(serverHead = 12, localCursor = 20) + + assertEquals(listOf(FeedSyncDelegate.Event.CatchUpComplete), events) + } + + private companion object { + const val CHAT_HEX = "aabbccdd" + } +} diff --git a/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/FeedSyncCatchUpTest.kt b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/FeedSyncCatchUpTest.kt index 741b1b6ce..194c7340c 100644 --- a/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/FeedSyncCatchUpTest.kt +++ b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/FeedSyncCatchUpTest.kt @@ -14,6 +14,7 @@ import com.flipcash.services.user.UserManager import com.flipcash.shared.chat.internal.ChatStateHolder import com.flipcash.shared.chat.internal.delegates.FeedSyncDelegate import io.mockk.coEvery +import io.mockk.coVerify import io.mockk.mockk import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.flow.launchIn @@ -26,9 +27,10 @@ import kotlin.test.assertEquals import kotlin.time.Instant /** - * A feed sync writes each chat's `lastMessage` and the server's `latestEventSequence` before it - * decides whether that chat needs catching up. These tests hold the decision to what the cache held - * *before* the sync wrote to it. + * A feed sync writes each chat's `lastMessage` before it decides whether that chat needs catching + * up, so the presence of message rows says nothing about whether a transcript was ever pulled. + * These tests hold the decision to the applied event cursor instead, and hold the terminal marker + * to its position after it. * * The user-visible failure: after a re-login the chat cache is empty (logout clears it via * `ChatCoordinator.reset`), the sync repopulates one message per chat, and anything derived from @@ -38,7 +40,6 @@ import kotlin.time.Instant @OptIn(ExperimentalCoroutinesApi::class) class FeedSyncCatchUpTest { - private val selfId = listOf(1, 2, 3) private val otherId = listOf(4, 5, 6) private val chatId = ChatId("aabbccdd") @@ -71,22 +72,17 @@ class FeedSyncCatchUpTest { * Stands in for the per-user Room database: what the sync writes is what a later read sees. * Relaxed mocks would answer `false`/`0` regardless of the writes, which is exactly the coupling * under test. + * + * `upsert` deliberately leaves [seededCursors] alone. That is the DAO's contract — the feed + * payload carries the server's head, and `ChatMetadataDao.upsert` refreshes only the columns + * the server owns, so a chat's applied cursor survives a sync (see `ChatMetadataDaoTest`). */ - private class Harness(chats: List, seededChatsWithMessages: Set = emptySet()) { - val chatsWithMessages = seededChatsWithMessages.toMutableSet() - val storedSequences = mutableMapOf() + private class Harness(chats: List, seededCursors: Map = emptyMap()) { + val storedSequences = seededCursors.toMutableMap() - val messageDataSource = mockk(relaxed = true).also { source -> - coEvery { source.upsert(any(), any()) } answers { - chatsWithMessages += firstArg() - } - coEvery { source.hasMessages(any()) } answers { firstArg() in chatsWithMessages } - } + val messageDataSource = mockk(relaxed = true) val metadataDataSource = mockk(relaxed = true).also { source -> - coEvery { source.upsert(any>()) } answers { - firstArg>().forEach { storedSequences[it.chatId] = it.latestEventSequence } - } coEvery { source.getLatestEventSequence(any()) } answers { storedSequences[firstArg()] ?: 0L } @@ -136,8 +132,8 @@ class FeedSyncCatchUpTest { @Test fun `warm cache does not refetch history`() = runTest { val harness = Harness( - chats = listOf(metadata(lastMessage = message(20, otherId))), - seededChatsWithMessages = setOf(chatId), + chats = listOf(metadata(lastMessage = message(20, otherId), latestEventSequence = 20)), + seededCursors = mapOf(chatId to 20L), ) val events = harness.sync(this) @@ -153,47 +149,40 @@ class FeedSyncCatchUpTest { fun `a local sequence behind the server's triggers a delta sync`() = runTest { val harness = Harness( chats = listOf(metadata(lastMessage = message(20, otherId), latestEventSequence = 99)), - seededChatsWithMessages = setOf(chatId), - ).apply { storedSequences[chatId] = 42 } + seededCursors = mapOf(chatId to 42L), + ) val events = harness.sync(this) assertEquals( listOf( - FeedSyncDelegate.Event.DeltaSyncNeeded(chatId, afterSequence = 42), + FeedSyncDelegate.Event.DeltaSyncNeeded(chatId), FeedSyncDelegate.Event.CatchUpComplete, ), events, - "the gap must be measured against the sequence the cache held before the sync overwrote it", + "the gap must be measured against the sequence the cache holds, not against message rows", ) } /** - * The event carrying its own `afterSequence` is the whole point: the delta consumer reads - * `chat_metadata` when not given one, and by then this sync has already stamped the server's - * sequence onto that row — so a re-read would ask the server for everything after its own - * latest event and get nothing back. + * The delta consumer reads the cursor out of `chat_metadata` for itself, which is only safe + * because a feed sync never seats one: a chat is caught up by a message load or by an applied + * delta, and by nothing the feed reports about the server's head. */ @Test - fun `the delta request starts from the pre-sync sequence, not the row the sync wrote`() = runTest { + fun `a feed sync never advances the applied cursor`() = runTest { val harness = Harness( chats = listOf(metadata(lastMessage = message(20, otherId), latestEventSequence = 99)), - seededChatsWithMessages = setOf(chatId), - ).apply { storedSequences[chatId] = 42 } + seededCursors = mapOf(chatId to 42L), + ) - val event = harness.sync(this) - .filterIsInstance() - .single() + harness.sync(this) - assertEquals( - 99L, - harness.metadataDataSource.getLatestEventSequence(chatId), - "precondition: the sync has overwritten the stored sequence with the server's", - ) + coVerify(exactly = 0) { harness.metadataDataSource.updateLatestEventSequence(any(), any()) } assertEquals( 42L, - event.afterSequence, - "the delta must be requested from 42, or the backfill silently fetches nothing", + harness.metadataDataSource.getLatestEventSequence(chatId), + "the delta must still start from 42, or the backfill silently fetches nothing", ) } diff --git a/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/MessagingLoadCursorTest.kt b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/MessagingLoadCursorTest.kt new file mode 100644 index 000000000..a5473ab98 --- /dev/null +++ b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/MessagingLoadCursorTest.kt @@ -0,0 +1,86 @@ +package com.flipcash.shared.chat + +import com.flipcash.app.persistence.sources.ChatMetadataDataSource +import com.flipcash.services.controllers.ChatMessagingController +import com.flipcash.services.models.chat.ChatId +import com.flipcash.services.models.chat.ChatMessage +import com.flipcash.services.models.chat.MessageContent +import com.flipcash.shared.chat.internal.delegates.MessagingDelegate +import io.mockk.coEvery +import io.mockk.coVerify +import io.mockk.mockk +import kotlinx.coroutines.test.runTest +import org.junit.Test +import kotlin.time.Instant + +/** + * Covers the event cursor a newest-page load leaves behind. The cursor is what marks a + * transcript as fetched — a load that does not seat it makes every later feed sync re-fetch + * the same page — and it must only ever move forward. + */ +class MessagingLoadCursorTest { + + private val chatId = ChatId("aabbccdd") + + private fun message(id: Long, eventSequence: Long) = ChatMessage( + messageId = id, + senderId = listOf(4, 5, 6), + content = listOf(MessageContent.Text("msg-$id")), + timestamp = Instant.fromEpochSeconds(1000 + id), + unreadSeq = 0, + eventSequence = eventSequence, + ) + + private fun delegateWith( + messagingController: ChatMessagingController, + metadataDataSource: ChatMetadataDataSource, + ) = MessagingDelegate( + chatController = mockk(relaxed = true), + messagingController = messagingController, + metadataDataSource = metadataDataSource, + messageDataSource = mockk(relaxed = true), + memberDataSource = mockk(relaxed = true), + notificationManager = mockk(relaxed = true), + userManager = mockk(relaxed = true), + stateHolder = mockk(relaxed = true), + analytics = mockk(relaxed = true), + ) + + @Test + fun `newest page seats the cursor at its highest event sequence`() = runTest { + val messagingController = mockk(relaxed = true) + coEvery { messagingController.getMessages(chatId, any()) } returns + Result.success(listOf(message(id = 1, eventSequence = 7), message(id = 2, eventSequence = 9))) + val metadataDataSource = mockk(relaxed = true) + coEvery { metadataDataSource.getLatestEventSequence(chatId) } returns 0 + + delegateWith(messagingController, metadataDataSource).loadMessages(chatId) + + coVerify(exactly = 1) { metadataDataSource.updateLatestEventSequence(chatId, 9) } + } + + @Test + fun `page older than the cursor does not rewind it`() = runTest { + val messagingController = mockk(relaxed = true) + coEvery { messagingController.getMessages(chatId, any()) } returns + Result.success(listOf(message(id = 1, eventSequence = 4))) + val metadataDataSource = mockk(relaxed = true) + coEvery { metadataDataSource.getLatestEventSequence(chatId) } returns 9 + + delegateWith(messagingController, metadataDataSource).loadMessages(chatId) + + coVerify(exactly = 0) { metadataDataSource.updateLatestEventSequence(chatId, any()) } + } + + @Test + fun `empty page leaves the cursor unseated`() = runTest { + val messagingController = mockk(relaxed = true) + coEvery { messagingController.getMessages(chatId, any()) } returns Result.success(emptyList()) + val metadataDataSource = mockk(relaxed = true) + coEvery { metadataDataSource.getLatestEventSequence(chatId) } returns 0 + + delegateWith(messagingController, metadataDataSource).loadMessages(chatId) + + coVerify(exactly = 0) { metadataDataSource.updateLatestEventSequence(chatId, any()) } + } +} diff --git a/apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMetadataDao.kt b/apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMetadataDao.kt index b8ab6a3dd..0ca8e5b43 100644 --- a/apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMetadataDao.kt +++ b/apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMetadataDao.kt @@ -4,6 +4,7 @@ import androidx.room.Dao import androidx.room.Insert import androidx.room.OnConflictStrategy import androidx.room.Query +import androidx.room.Transaction import com.flipcash.app.persistence.entities.ChatMetadataEntity import kotlinx.coroutines.flow.Flow @@ -16,11 +17,55 @@ interface ChatMetadataDao { @Query("SELECT * FROM chat_metadata WHERE chat_id_hex = :chatIdHex") suspend fun getById(chatIdHex: String): ChatMetadataEntity? - @Insert(onConflict = OnConflictStrategy.REPLACE) - suspend fun upsert(entity: ChatMetadataEntity) + @Insert(onConflict = OnConflictStrategy.IGNORE) + suspend fun insertIfAbsent(entity: ChatMetadataEntity): Long - @Insert(onConflict = OnConflictStrategy.REPLACE) - suspend fun upsert(entities: List) + /** + * Overwrites only the columns the server owns. `latest_event_sequence` and + * `analytics_counted_through` are client-owned watermarks that no server payload + * carries, so they are deliberately absent here. + */ + @Query( + "UPDATE chat_metadata SET chat_type = :chatType, " + + "last_activity_epoch_ms = :lastActivityEpochMs, " + + "last_message_id = :lastMessageId, " + + "is_hidden = :isHidden " + + "WHERE chat_id_hex = :chatIdHex" + ) + suspend fun updateServerOwnedFields( + chatIdHex: String, + chatType: String, + lastActivityEpochMs: Long, + lastMessageId: Long?, + isHidden: Boolean, + ) + + /** + * Inserts a new chat, or refreshes an existing one's server-owned columns in place. + * + * Deliberately not a whole-row REPLACE: `latest_event_sequence` is the cursor the + * client has actually applied from the event log, and `analytics_counted_through` + * is the replay guard for received-message analytics. Neither is carried by a feed + * payload, so replacing the row would reset both — the cursor to whatever the server + * reported as its head (skipping every unapplied event on the next `GetDelta`) and + * the analytics watermark to zero (re-counting messages already counted). + */ + @Transaction + suspend fun upsert(entity: ChatMetadataEntity) { + if (insertIfAbsent(entity) != -1L) return + updateServerOwnedFields( + chatIdHex = entity.chatIdHex, + chatType = entity.chatType, + lastActivityEpochMs = entity.lastActivityEpochMs, + lastMessageId = entity.lastMessageId, + isHidden = entity.isHidden, + ) + } + + @Transaction + suspend fun upsert(entities: List) { + for (entity in entities) upsert(entity) + } @Query("SELECT last_activity_epoch_ms FROM chat_metadata WHERE chat_id_hex = :chatIdHex") suspend fun getLastActivity(chatIdHex: String): Long? diff --git a/apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMetadataDaoTest.kt b/apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMetadataDaoTest.kt new file mode 100644 index 000000000..411d1e1e4 --- /dev/null +++ b/apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMetadataDaoTest.kt @@ -0,0 +1,123 @@ +package com.flipcash.app.persistence.dao + +import android.content.Context +import androidx.room.Room +import androidx.test.core.app.ApplicationProvider +import com.flipcash.app.persistence.FlipcashDatabase +import com.flipcash.app.persistence.entities.ChatMetadataEntity +import kotlinx.coroutines.test.runTest +import org.junit.After +import org.junit.Before +import org.junit.Test +import org.junit.runner.RunWith +import org.robolectric.RobolectricTestRunner +import kotlin.test.assertEquals + +/** + * Covers what a feed sync is allowed to overwrite. `latest_event_sequence` (the applied + * catch-up cursor) and `analytics_counted_through` (the received-message replay guard) are + * client-owned watermarks that no server payload carries, so an upsert of server truth must + * leave them alone — a whole-row replace would rewind both. + */ +@RunWith(RobolectricTestRunner::class) +class ChatMetadataDaoTest { + + private lateinit var db: FlipcashDatabase + private lateinit var dao: ChatMetadataDao + + @Before + fun setUp() { + val context = ApplicationProvider.getApplicationContext() + db = Room.inMemoryDatabaseBuilder(context, FlipcashDatabase::class.java) + .allowMainThreadQueries() + .build() + dao = db.chatMetadataDao() + } + + @After + fun tearDown() { + db.close() + } + + private fun entity( + chatIdHex: String = CHAT_HEX, + chatType: String = "CONTACT_DM", + lastActivityEpochMs: Long = 1_000, + lastMessageId: Long? = null, + latestEventSequence: Long = 0, + isHidden: Boolean = false, + analyticsCountedThrough: Long = 0, + ) = ChatMetadataEntity( + chatIdHex = chatIdHex, + chatType = chatType, + lastActivityEpochMs = lastActivityEpochMs, + lastMessageId = lastMessageId, + latestEventSequence = latestEventSequence, + isHidden = isHidden, + analyticsCountedThrough = analyticsCountedThrough, + ) + + @Test + fun `upsert preserves the client-owned watermarks`() = runTest { + dao.upsert(entity()) + dao.updateLatestEventSequence(CHAT_HEX, 9) + dao.advanceAnalyticsCountedThrough(CHAT_HEX, 42) + + // A later feed sync carries neither watermark — the mapper never sets them. + dao.upsert(entity(lastActivityEpochMs = 2_000, lastMessageId = 77)) + + assertEquals(9L, dao.getLatestEventSequence(CHAT_HEX)) + assertEquals(42L, dao.getAnalyticsCountedThrough(CHAT_HEX)) + } + + @Test + fun `upsert refreshes the server-owned columns`() = runTest { + dao.upsert(entity(chatType = "CONTACT_DM", lastActivityEpochMs = 1_000, lastMessageId = 1)) + + dao.upsert( + entity( + chatType = "TIP_DM", + lastActivityEpochMs = 2_000, + lastMessageId = 77, + isHidden = true, + ) + ) + + val stored = requireNotNull(dao.getById(CHAT_HEX)) + assertEquals("TIP_DM", stored.chatType) + assertEquals(2_000L, stored.lastActivityEpochMs) + assertEquals(77L, stored.lastMessageId) + assertEquals(true, stored.isHidden) + } + + @Test + fun `upsert inserts a chat the database has not seen`() = runTest { + dao.upsert(entity(lastMessageId = 5)) + + val stored = requireNotNull(dao.getById(CHAT_HEX)) + assertEquals(5L, stored.lastMessageId) + assertEquals(0L, stored.latestEventSequence) + } + + @Test + fun `list upsert preserves each row's cursor`() = runTest { + dao.upsert(listOf(entity(chatIdHex = CHAT_HEX), entity(chatIdHex = OTHER_HEX))) + dao.updateLatestEventSequence(CHAT_HEX, 9) + dao.updateLatestEventSequence(OTHER_HEX, 4) + + dao.upsert( + listOf( + entity(chatIdHex = CHAT_HEX, lastActivityEpochMs = 2_000), + entity(chatIdHex = OTHER_HEX, lastActivityEpochMs = 3_000), + ) + ) + + assertEquals(9L, dao.getLatestEventSequence(CHAT_HEX)) + assertEquals(4L, dao.getLatestEventSequence(OTHER_HEX)) + } + + private companion object { + const val CHAT_HEX = "aabbccdd" + const val OTHER_HEX = "eeff0011" + } +} diff --git a/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/mapper/chat/ChatEntityMapper.kt b/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/mapper/chat/ChatEntityMapper.kt index 42a153227..c9e6e6f24 100644 --- a/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/mapper/chat/ChatEntityMapper.kt +++ b/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/mapper/chat/ChatEntityMapper.kt @@ -51,13 +51,20 @@ class ChatEntityMapper @Inject constructor() { // region ChatMetadata + /** + * Maps server truth onto the row. [ChatMetadata.latestEventSequence] is the server's + * head, not something the client has applied, so it is deliberately not carried into + * `latestEventSequence` — that column is the applied catch-up cursor, advanced only by + * a message load or a delta sync. A fresh insert therefore starts at 0, meaning "this + * transcript has never been fetched"; `ChatMetadataDao.upsert` leaves the column alone + * on an existing row. + */ fun toEntity(metadata: ChatMetadata): ChatMetadataEntity { return ChatMetadataEntity( chatIdHex = metadata.chatId.bytes.toList().hexEncodedString(), chatType = metadata.type.name, lastActivityEpochMs = metadata.lastActivity.toEpochMilliseconds(), lastMessageId = metadata.lastMessage?.messageId, - latestEventSequence = metadata.latestEventSequence, isHidden = metadata.isHidden, ) } @@ -73,7 +80,9 @@ class ChatEntityMapper @Inject constructor() { members = members, lastMessage = lastMessage, lastActivity = Instant.fromEpochMilliseconds(entity.lastActivityEpochMs), - latestEventSequence = entity.latestEventSequence, + // The server's head is valid only at fetch time and is never stored, so a chat + // rebuilt from the database reports 0 — "unknown", not "no events". + latestEventSequence = 0, isHidden = entity.isHidden, ) } diff --git a/apps/flipcash/shared/persistence/sources/src/test/kotlin/com/flipcash/app/persistence/sources/mapper/chat/ChatEntityMapperTest.kt b/apps/flipcash/shared/persistence/sources/src/test/kotlin/com/flipcash/app/persistence/sources/mapper/chat/ChatEntityMapperTest.kt new file mode 100644 index 000000000..9a31dc7d7 --- /dev/null +++ b/apps/flipcash/shared/persistence/sources/src/test/kotlin/com/flipcash/app/persistence/sources/mapper/chat/ChatEntityMapperTest.kt @@ -0,0 +1,63 @@ +package com.flipcash.app.persistence.sources.mapper.chat + +import com.flipcash.app.persistence.entities.ChatMetadataEntity +import com.flipcash.services.models.chat.ChatId +import com.flipcash.services.models.chat.ChatMetadata +import com.flipcash.services.models.chat.ChatType +import org.junit.Assert.assertEquals +import org.junit.Test +import kotlin.time.Instant + +/** + * `ChatMetadata.latestEventSequence` is the server's head at fetch time; the entity column of + * the same name is the cursor the client has actually applied. The mapper must not conflate + * them — carrying the head into the row would mark an unfetched transcript as caught up. + */ +class ChatEntityMapperTest { + + private val mapper = ChatEntityMapper() + + private fun metadata(latestEventSequence: Long) = ChatMetadata( + chatId = ChatId(CHAT_HEX), + type = ChatType.CONTACT_DM, + members = emptyList(), + lastMessage = null, + lastActivity = Instant.fromEpochSeconds(1_000), + latestEventSequence = latestEventSequence, + ) + + @Test + fun `server head is not written as the applied cursor`() { + val entity = mapper.toEntity(metadata(latestEventSequence = 12)) + + assertEquals(0L, entity.latestEventSequence) + } + + @Test + fun `mapped entity carries the server-owned fields`() { + val entity = mapper.toEntity(metadata(latestEventSequence = 12)) + + assertEquals(CHAT_HEX, entity.chatIdHex) + assertEquals(ChatType.CONTACT_DM.name, entity.chatType) + assertEquals(1_000_000L, entity.lastActivityEpochMs) + } + + @Test + fun `chat rebuilt from the database reports an unknown server head`() { + val entity = ChatMetadataEntity( + chatIdHex = CHAT_HEX, + chatType = ChatType.CONTACT_DM.name, + lastActivityEpochMs = 1_000_000, + lastMessageId = null, + latestEventSequence = 9, + ) + + val metadata = mapper.toMetadata(entity, members = emptyList(), lastMessage = null) + + assertEquals(0L, metadata.latestEventSequence) + } + + private companion object { + const val CHAT_HEX = "aabbccdd" + } +}