fix(chat): apply every GetDelta batch, not only the first

apps/flipcash/shared/chat/src/commonMain/kotlin/com/flipcash/shared/chat/internal/delegates/EventStreamDelegate.kt

@@ -40,17 +40,18 @@ import com.getcode.opencode.model.core.ID
 import com.getcode.opencode.model.financial.Fiat
 import com.getcode.utils.TraceType
 import com.getcode.utils.trace
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.awaitCancellation
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.distinctUntilChanged
-import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.firstOrNull
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.receiveAsFlow
+import kotlinx.coroutines.flow.transformWhile
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.withTimeoutOrNull
 import javax.inject.Inject
@@ -111,6 +112,12 @@ class EventStreamDelegate @Inject constructor(
         private val GAP_FILL_DELAY = 2.seconds
         private val HEARTBEAT_INTERVAL = 30.seconds
         private val REOPEN_BACKOFF_BASE = 1.seconds
+
+        /** How many caught-up messages are written per transaction; `GetDelta` sends at most 100 per batch. */
+        internal const val DELTA_FLUSH_MESSAGES = 500
+
+        /** Bounds a catch-up stream, which runs inline in the coordinator's event handling. */
+        private val DELTA_SYNC_TIMEOUT = 60.seconds
     }
 
     sealed interface Event {
@@ -301,39 +308,82 @@ class EventStreamDelegate @Inject constructor(
      * 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.
+     *
+     * `GetDelta` streams the catch-up as batches of at most 100 messages, and only the completed
+     * stream reaches the head (`latest_sequence`). Batches are buffered and written
+     * [DELTA_FLUSH_MESSAGES] at a time, each write followed by the checkpoint it reached, so a long
+     * catch-up costs observers a few invalidations rather than one per batch. A stream that fails,
+     * drops or times out keeps what it delivered, and the next sync resumes from that checkpoint.
      */
     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 {
-            val result = messagingController.getDelta(chatId, afterSequence).first()
-            result
-                .onSuccess { delta ->
-                    if (delta.messages.isNotEmpty()) {
-                        messageDataSource.upsert(chatId, delta.messages)
+        val buffered = mutableListOf<ChatMessage>()
+        var checkpoint = afterSequence
+        var head: Long? = null
+        var applied = 0
+        var failure: Throwable? = null
+
+        // One write that only moves forward: the cursor, and the newest message only when it is
+        // newer than the stored one. See ChatMetadataDao.applyCatchUp.
+        suspend fun flush(cursor: Long) {
+            if (buffered.isNotEmpty()) messageDataSource.upsert(chatId, buffered.toList())
+            metadataDataSource.applyCatchUp(
+                chatId,
+                if (cursor > afterSequence) cursor else 0L,
+                buffered.maxByOrNull { it.messageId },
+            )
+            applied += buffered.size
+            buffered.clear()
+        }
+
+        val completed = try {
+            withTimeoutOrNull(DELTA_SYNC_TIMEOUT) {
+                messagingController.getDelta(chatId, afterSequence)
+                    .transformWhile { result ->
+                        result.onSuccess { emit(it) }.onFailure { failure = it }
+                        result.isSuccess
                     }
-                    // One write that only moves forward: the cursor, and the newest message only
-                    // when it is newer than the stored one. See ChatMetadataDao.applyCatchUp.
-                    metadataDataSource.applyCatchUp(
-                        chatId,
-                        if (delta.latestSequence > afterSequence) delta.latestSequence else 0L,
-                        delta.messages.maxByOrNull { it.messageId },
-                    )
-                    sequenceTracker.resetTo(chatId, delta.latestSequence)
-                    trace(tag = TAG, message = "Delta sync complete: ${delta.messages.size} messages, sequence ${delta.latestSequence}", type = TraceType.Process)
-                }
-                .onFailure { error ->
-                    if (error is GetDeltaError.ResetRequired) {
-                        trace(tag = TAG, message = "Delta sync reset required for $chatId, falling back to full load", type = TraceType.Process)
-                        sequenceTracker.clear(chatId)
-                        _events.send(Event.LoadMessages(chatId))
-                    } else {
-                        trace(tag = TAG, message = "Delta sync failed for $chatId: ${error.message}", type = TraceType.Error)
+                    .collect { delta ->
+                        buffered += delta.messages
+                        checkpoint = maxOf(checkpoint, delta.checkpointSequence)
+                        head = delta.latestSequence
+                        if (buffered.size >= DELTA_FLUSH_MESSAGES) flush(checkpoint)
                     }
-                }
+                true
+            } ?: false
+        } catch (e: CancellationException) {
+            throw e
         } catch (e: Exception) {
             trace(tag = TAG, message = "Delta sync exception for $chatId: ${e.message}", type = TraceType.Error)
+            false
+        }
+
+        val latest = head
+        if (completed && failure == null && latest != null) {
+            flush(maxOf(latest, checkpoint))
+            sequenceTracker.resetTo(chatId, latest)
+            trace(tag = TAG, message = "Delta sync complete: $applied messages, sequence $latest", type = TraceType.Process)
+            return
+        }
+
+        // Every batch received before the stream stopped was whole; keep them, and resume from
+        // their checkpoint rather than the head they never reached.
+        flush(checkpoint)
+        val error = failure
+        when {
+            error is GetDeltaError.ResetRequired -> {
+                trace(tag = TAG, message = "Delta sync reset required for $chatId, falling back to full load", type = TraceType.Process)
+                sequenceTracker.clear(chatId)
+                _events.send(Event.LoadMessages(chatId))
+            }
+            error != null ->
+                trace(tag = TAG, message = "Delta sync failed for $chatId: ${error.message}", type = TraceType.Error)
+            completed ->
+                trace(tag = TAG, message = "Delta sync for $chatId ended without a response", type = TraceType.Error)
+            else ->
+                trace(tag = TAG, message = "Delta sync for $chatId stopped at sequence $checkpoint after $applied messages", type = TraceType.Error)
         }
     }
 

apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/DeltaSyncBatchesTest.kt

@@ -0,0 +1,144 @@
+package com.flipcash.shared.chat
+
+import com.flipcash.app.persistence.sources.ChatMessageDataSource
+import com.flipcash.app.persistence.sources.ChatMetadataDataSource
+import com.flipcash.services.controllers.ChatMessagingController
+import com.flipcash.services.controllers.EventStreamingController
+import com.flipcash.services.models.GetDeltaError
+import com.flipcash.services.models.chat.ChatId
+import com.flipcash.services.models.chat.ChatMessage
+import com.flipcash.services.models.chat.MessageContent
+import com.flipcash.services.repository.DeltaUpdate
+import com.flipcash.services.user.UserManager
+import com.flipcash.shared.chat.internal.ChatStateHolder
+import com.flipcash.shared.chat.internal.delegates.EventStreamDelegate
+import io.mockk.coEvery
+import io.mockk.every
+import io.mockk.mockk
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.emptyFlow
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.test.runTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+import kotlin.time.Instant
+
+/**
+ * `GetDelta` is a server stream: the catch-up arrives as a run of `MessageBatch`es of up to 100
+ * messages each, and the cursor may only move to `latest_sequence` once the stream completes.
+ */
+class DeltaSyncBatchesTest {
+
+    private val chat = ChatId("aabbccdd")
+
+    private val written = mutableListOf<List<ChatMessage>>()
+    private val cursors = mutableListOf<Long>()
+    private var stored = 0L
+
+    private val messageDataSource = mockk<ChatMessageDataSource>(relaxed = true).also {
+        coEvery { it.upsert(chat, any()) } answers { written += secondArg<List<ChatMessage>>() }
+    }
+    private val metadataDataSource = mockk<ChatMetadataDataSource>(relaxed = true).also {
+        coEvery { it.getLatestEventSequence(chat) } answers { stored }
+        coEvery { it.applyCatchUp(chat, any(), any()) } answers {
+            val sequence = secondArg<Long>()
+            if (sequence > 0) {
+                cursors += sequence
+                stored = maxOf(stored, sequence)
+            }
+        }
+    }
+    private val messagingController = mockk<ChatMessagingController>(relaxed = true)
+
+    private val delegate = EventStreamDelegate(
+        eventStreamingController = mockk<EventStreamingController>(relaxed = true).also {
+            every { it.streamFailures } returns emptyFlow()
+        },
+        messagingController = messagingController,
+        metadataDataSource = metadataDataSource,
+        messageDataSource = messageDataSource,
+        memberDataSource = mockk(relaxed = true),
+        tokenCoordinator = mockk(relaxed = true),
+        userManager = mockk<UserManager>(relaxed = true),
+        stateHolder = ChatStateHolder(),
+        analytics = mockk(relaxed = true),
+        exchange = mockk(relaxed = true),
+    )
+
+    private fun message(sequence: Long) = ChatMessage(
+        messageId = sequence,
+        senderId = listOf(2),
+        content = listOf(MessageContent.Text("m$sequence")),
+        timestamp = Instant.fromEpochMilliseconds(sequence * 1000),
+        unreadSeq = sequence,
+        eventSequence = sequence,
+    )
+
+    /** [count] batches of [size] messages, sequences 1 through count * size. */
+    private fun batches(count: Int, size: Int = 100): List<DeltaUpdate> {
+        val head = (count * size).toLong()
+        return (0 until count).map { b ->
+            val first = b * size + 1L
+            val last = first + size - 1
+            DeltaUpdate(
+                messages = (first..last).map(::message),
+                latestSequence = head,
+                checkpointSequence = last,
+            )
+        }
+    }
+
+    private fun stream(responses: List<DeltaUpdate>, thenFail: Throwable? = null): Flow<Result<DeltaUpdate>> = flow {
+        for (response in responses) emit(Result.success(response))
+        if (thenFail != null) throw thenFail
+    }
+
+    @Test
+    fun `every batch in the stream is written and the cursor ends at the head`() = runTest {
+        every { messagingController.getDelta(chat, 0L, any()) } returns stream(batches(3))
+
+        delegate.performDeltaSync(chat)
+
+        assertEquals((1L..300L).toList(), written.flatten().map { it.messageId }.sorted())
+        assertEquals(300L, stored)
+    }
+
+    @Test
+    fun `a stream that drops keeps what it delivered and resumes from the last checkpoint`() = runTest {
+        every { messagingController.getDelta(chat, 0L, any()) } returns
+            stream(batches(3).take(2), thenFail = RuntimeException("stream reset"))
+
+        delegate.performDeltaSync(chat)
+
+        assertEquals((1L..200L).toList(), written.flatten().map { it.messageId }.sorted())
+        assertEquals(200L, stored, "the cursor must not move past what was applied")
+    }
+
+    @Test
+    fun `a reset after the first batch keeps that batch and does not jump to the head`() = runTest {
+        val responses = batches(3)
+        every { messagingController.getDelta(chat, 0L, any()) } returns flow {
+            emit(Result.success(responses[0]))
+            emit(Result.failure(GetDeltaError.ResetRequired()))
+            emit(Result.success(responses[1]))
+        }
+
+        delegate.performDeltaSync(chat)
+
+        assertEquals((1L..100L).toList(), written.flatten().map { it.messageId }.sorted())
+        assertEquals(100L, stored)
+    }
+
+    @Test
+    fun `a long catch-up is written in fewer transactions than it has batches`() = runTest {
+        every { messagingController.getDelta(chat, 0L, any()) } returns stream(batches(12))
+
+        delegate.performDeltaSync(chat)
+
+        assertEquals(1200, written.sumOf { it.size })
+        assertTrue(written.size < 12, "wrote ${written.size} times for 12 batches")
+        assertEquals(cursors.sorted(), cursors, "the cursor only moves forward")
+        assertEquals(1200L, stored)
+    }
+}