fix(chat): back chats up by their applied event cursor, not by message presence

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

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 {

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))
                     }
                 }
 

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())

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<FeedSyncDelegate.Event> {
+        val chat = metadata(CHAT_HEX, serverHead)
+
+        val chatController = mockk<ChatController>(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<ChatMetadataDataSource>(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<ChatMessageDataSource>(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<FeedSyncDelegate.Event>()
+        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"
+    }
+}