fix(chat): report stream liveness honestly and stop losing read pointers

apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/RealChatCoordinator.kt

@@ -145,6 +145,8 @@ class RealChatCoordinator @Inject constructor(
                         messagingDelegate.loadMessages(event.chatId)
                     is FeedSyncDelegate.Event.DeltaSyncNeeded ->
                         eventStreamDelegate.performDeltaSync(event.chatId)
+                    is FeedSyncDelegate.Event.ReadPointerUnreported ->
+                        messagingDelegate.reportReadPointer(event.chatId, event.messageId)
                     // 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

@@ -29,17 +29,21 @@ import com.getcode.utils.TraceType
 import com.getcode.utils.trace
 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.launch
+import kotlinx.coroutines.withTimeoutOrNull
 import javax.inject.Inject
 import javax.inject.Singleton
 import kotlin.time.Clock
+import kotlin.time.Duration
 import kotlin.time.Duration.Companion.seconds
 
 /**
@@ -86,6 +90,8 @@ class EventStreamDelegate @Inject constructor(
     companion object {
         private const val TAG = "EventStreamDelegate"
         private val GAP_FILL_DELAY = 2.seconds
+        private val HEARTBEAT_INTERVAL = 30.seconds
+        private val REOPEN_BACKOFF_BASE = 1.seconds
     }
 
     sealed interface Event {
@@ -164,22 +170,57 @@ class EventStreamDelegate @Inject constructor(
         eventStreamingController.close()
     }
 
+    /**
+     * Supervises the stream: whenever it is found down, syncs the feed and reopens it.
+     *
+     * Each pass waits for whichever comes first, a [HEARTBEAT_INTERVAL] tick or
+     * [EventStreamingController.streamFailures] reporting a stream that ended with nothing
+     * retrying it. The failure signal is what keeps such a stream from costing a whole interval:
+     * a status outside the reconnect loop's retryable set, or a ping timeout, is never retried by
+     * the loop, and the tick was the only thing that noticed.
+     *
+     * Reopening backs off from the second consecutive attempt on, so a stream that fails as soon
+     * as it opens cannot spin the loop. The backoff tops out at [HEARTBEAT_INTERVAL], which is
+     * where every failure used to wait.
+     */
     internal fun startHeartbeat(onReconnect: () -> Unit) {
         val scope = scope ?: return
         stopHeartbeat()
         heartbeatJob = scope.launch {
+            var consecutiveReopens = 0
+
             while (true) {
-                delay(30.seconds)
-                if (!eventStreamingController.isStreamActive) {
-                    trace(tag = TAG, message = "Heartbeat: event stream dead, syncing feed and reconnecting", type = TraceType.Process)
-                    onReconnect()
-                    eventStreamingController.close()
-                    open()
+                // A failures flow that completes means no signal is coming, not that one just
+                // arrived — waiting out the tick keeps this a supervisor rather than a spin.
+                val failed = withTimeoutOrNull(HEARTBEAT_INTERVAL) {
+                    eventStreamingController.streamFailures.firstOrNull() ?: awaitCancellation()
+                } != null
+
+                // Only the failure-driven wake can arrive back to back; the tick already spaces
+                // itself out. Liveness is read after the wait, so a stream another trigger
+                // reopened in the meantime is left alone.
+                if (failed) delay(reopenBackoff(consecutiveReopens))
+
+                if (eventStreamingController.isStreamActive) {
+                    consecutiveReopens = 0
+                    continue
                 }
+                consecutiveReopens++
+
+                trace(tag = TAG, message = "Event stream down, syncing feed and reconnecting", type = TraceType.Process)
+                onReconnect()
+                eventStreamingController.close()
+                open()
             }
         }
     }
 
+    private fun reopenBackoff(consecutiveReopens: Int): Duration = when {
+        consecutiveReopens <= 0 -> Duration.ZERO
+        else -> (REOPEN_BACKOFF_BASE * (1 shl (consecutiveReopens - 1).coerceAtMost(5)))
+            .coerceAtMost(HEARTBEAT_INTERVAL)
+    }
+
     internal fun stopHeartbeat() {
         heartbeatJob?.cancel()
         heartbeatJob = null
@@ -301,8 +342,10 @@ class EventStreamDelegate @Inject constructor(
             when (metaUpdate) {
                 is MetadataUpdate.FullRefresh -> {
                     metadataDataSource.upsert(metaUpdate.metadata)
-                    memberDataSource.deleteForChat(metaUpdate.metadata.chatId)
-                    memberDataSource.upsert(metaUpdate.metadata.chatId, metaUpdate.metadata.members)
+                    memberDataSource.replaceMembers(
+                        metaUpdate.metadata.chatId,
+                        metaUpdate.metadata.members,
+                    )
                     metaUpdate.metadata.lastMessage?.let { msg ->
                         messageDataSource.upsert(metaUpdate.metadata.chatId, listOf(msg))
                     }

apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/FeedSyncDelegate.kt

@@ -17,6 +17,7 @@ import com.flipcash.shared.chat.FeedOperations
 import com.flipcash.shared.chat.FeedSyncState
 import com.flipcash.shared.chat.internal.ChatStateHolder
 import com.flipcash.services.user.UserManager
+import com.getcode.opencode.model.core.ID
 import com.getcode.utils.TraceType
 import com.getcode.utils.trace
 import kotlinx.coroutines.CoroutineScope
@@ -71,6 +72,12 @@ class FeedSyncDelegate @Inject constructor(
          */
         data class DeltaSyncNeeded(val chatId: ChatId) : Event
 
+        /**
+         * The client's own READ pointer for [chatId] is ahead of the copy the feed just returned,
+         * so the server never took the advance. The consumer re-reports [messageId].
+         */
+        data class ReadPointerUnreported(val chatId: ChatId, val messageId: Long) : Event
+
         /**
          * Emitted last by every successful sync, after any catch-up above it.
          *
@@ -184,6 +191,32 @@ class FeedSyncDelegate @Inject constructor(
         feedObserverJob = null
     }
 
+    /**
+     * Emits [Event.ReadPointerUnreported] when the stored READ pointer for [chat] is ahead of the
+     * one the feed just returned.
+     *
+     * A read is written locally the moment the message is on screen and reported to the server
+     * afterwards, and nothing retries a report that fails. The local pointer survives — the member
+     * row keeps whichever value is further ahead — so the two copies can disagree indefinitely,
+     * leaving every other device and the pushes this one receives treating the chat as unread. A
+     * feed payload is the server's own copy, so the sync is where they can be compared.
+     */
+    private suspend fun reportUnreportedRead(chat: ChatMetadata, selfId: ID) {
+        val server = chat.members
+            .firstOrNull { it.userId == selfId }
+            ?.pointers
+            ?.firstOrNull { it.type == PointerType.READ }
+            ?.value
+            ?: 0L
+
+        // Read after the merge above, so this is max(local, server): ahead of `server` only when
+        // a local advance never reached it.
+        val local = memberDataSource.getSelfReadPointer(chat.chatId, selfId)
+        if (local > server) {
+            _events.send(Event.ReadPointerUnreported(chat.chatId, local))
+        }
+    }
+
     private suspend fun buildFeedFromDb(
         metadataEntities: List<ChatMetadataEntity>,
         membersByChat: Map<String, List<ChatMember>>,
@@ -229,7 +262,11 @@ class FeedSyncDelegate @Inject constructor(
                 stateHolder.update { it.copy(feedSyncState = FeedSyncState.Synced) }
                 trace(tag = TAG, message = "Feed synced: ${chats.size} chats", type = TraceType.Process)
 
+                val selfId = userManager.accountId
+
                 for (chat in chats) {
+                    if (selfId != null) reportUnreportedRead(chat, selfId)
+
                     // 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

apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/MessagingDelegate.kt

@@ -28,6 +28,8 @@ import com.flipcash.shared.chat.MessagingOperations
 import com.flipcash.shared.chat.internal.ChatStateHolder
 import com.flipcash.services.user.UserManager
 import com.getcode.opencode.model.core.ID
+import com.getcode.utils.TraceType
+import com.getcode.utils.trace
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.distinctUntilChanged
@@ -235,6 +237,22 @@ class MessagingDelegate @Inject constructor(
         return messagingController.advancePointer(chatId, PointerType.READ, messageId)
     }
 
+    /**
+     * Re-sends a READ pointer the server never took, without touching the local copy or the
+     * analytics that go with a genuine read — both already happened when the message was seen.
+     */
+    internal suspend fun reportReadPointer(chatId: ChatId, messageId: Long) {
+        messagingController.advancePointer(chatId, PointerType.READ, messageId)
+            .onFailure {
+                trace(
+                    tag = TAG,
+                    message = "Re-reporting read pointer $messageId failed",
+                    type = TraceType.Error,
+                    error = it,
+                )
+            }
+    }
+
     override suspend fun markAsRead(chatId: ChatId): Result<Unit> {
         val messageId = stateHolder.current.feed
             .firstOrNull { it.chatId == chatId }
@@ -292,4 +310,8 @@ class MessagingDelegate @Inject constructor(
     }
 
     // endregion
+
+    private companion object {
+        const val TAG = "MessagingDelegate"
+    }
 }

apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/EventStreamHeartbeatTest.kt

@@ -0,0 +1,106 @@
+package com.flipcash.shared.chat
+
+import com.flipcash.services.controllers.EventStreamingController
+import com.flipcash.shared.chat.internal.ChatStateHolder
+import com.flipcash.shared.chat.internal.delegates.EventStreamDelegate
+import io.mockk.every
+import io.mockk.mockk
+import io.mockk.verify
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.flow.emptyFlow
+import kotlinx.coroutines.flow.receiveAsFlow
+import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.advanceTimeBy
+import kotlinx.coroutines.test.runCurrent
+import kotlinx.coroutines.test.runTest
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.robolectric.RobolectricTestRunner
+import kotlin.time.Duration.Companion.milliseconds
+import kotlin.time.Duration.Companion.seconds
+
+/**
+ * The heartbeat is the only thing that reopens a stream the reconnect loop has stopped retrying —
+ * a status outside its retryable set, a ping timeout, or its attempts running out. Polling for
+ * that on a fixed interval costs up to the whole interval in missed messages.
+ */
+@OptIn(ExperimentalCoroutinesApi::class)
+@RunWith(RobolectricTestRunner::class)
+class EventStreamHeartbeatTest {
+
+    private class Harness {
+        val failures = Channel<Unit>(capacity = Channel.CONFLATED)
+
+        val controller = mockk<EventStreamingController>(relaxed = true).also { controller ->
+            every { controller.streamFailures } returns failures.receiveAsFlow()
+            every { controller.chatUpdates } returns emptyFlow()
+            every { controller.blobUpdates } returns emptyFlow()
+            // Nothing is connected for the whole test: every wake finds the stream down.
+            every { controller.isStreamActive } returns false
+            every { controller.isConnected } returns false
+        }
+
+        val delegate = EventStreamDelegate(
+            eventStreamingController = controller,
+            messagingController = mockk(relaxed = true),
+            metadataDataSource = mockk(relaxed = true),
+            messageDataSource = mockk(relaxed = true),
+            memberDataSource = mockk(relaxed = true),
+            tokenCoordinator = mockk(relaxed = true),
+            userManager = mockk(relaxed = true),
+            stateHolder = ChatStateHolder(),
+            analytics = mockk(relaxed = true),
+            exchange = mockk(relaxed = true),
+        )
+
+        fun start(scope: TestScope) {
+            delegate.initialize(scope.backgroundScope)
+            delegate.startHeartbeat { }
+            scope.runCurrent()
+        }
+    }
+
+    @Test
+    fun `a stream that gives up is reopened without waiting out the tick`() = runTest {
+        val harness = Harness()
+        harness.start(this)
+
+        harness.failures.trySend(Unit)
+        advanceTimeBy(100.milliseconds)
+        runCurrent()
+
+        verify(exactly = 1) { harness.controller.open(any()) }
+    }
+
+    @Test
+    fun `a stream failing as soon as it opens is not reopened in a hot loop`() = runTest {
+        val harness = Harness()
+        harness.start(this)
+
+        harness.failures.trySend(Unit)
+        advanceTimeBy(100.milliseconds)
+        runCurrent()
+
+        // The reopened stream fails immediately too.
+        harness.failures.trySend(Unit)
+        advanceTimeBy(500.milliseconds)
+        runCurrent()
+        verify(exactly = 1) { harness.controller.open(any()) }
+
+        advanceTimeBy(1.seconds)
+        runCurrent()
+        verify(exactly = 2) { harness.controller.open(any()) }
+    }
+
+    @Test
+    fun `a dead stream is still reopened at the tick`() = runTest {
+        val harness = Harness()
+        harness.start(this)
+
+        advanceTimeBy(31.seconds)
+        runCurrent()
+
+        verify(exactly = 1) { harness.controller.open(any()) }
+    }
+}