fix(chat): recover a message left sending by an app kill

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

@@ -129,6 +129,11 @@ class RealChatCoordinator @Inject constructor(
         feedDelegate.initialize(scope)
         eventStreamDelegate.initialize(scope)
         groupFeedDelegate.initialize(scope)
+        // Before the first refresh, and awaited rather than launched: a refresh that carries a
+        // self-authored message deletes this chat's `SENDING` rows, so a send a previous process
+        // was killed mid-flight has to be marked failed first or it is destroyed instead of
+        // becoming retryable. See [MessagingDelegate.recoverInterruptedSends].
+        messagingDelegate.recoverInterruptedSends()
         feedDelegate.observeFeedFromDb()
         syncFeeds()
         eventStreamDelegate.open()

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

@@ -500,6 +500,27 @@ class MessagingDelegate @Inject constructor(
         }
     }
 
+    /**
+     * Fails the sends a previous process left in flight, so the transcript offers to retry them
+     * instead of spinning on a send that nothing is still attempting.
+     *
+     * A send is marked failed from the `onFailure` of the coroutine that issued it ([sendMessage],
+     * [retryMessage]). Kill the app mid-send and that coroutine dies with the process, leaving a
+     * `SENDING` row on disk that nothing will ever move — until the first refresh deletes it and the
+     * user's text with it.
+     *
+     * Must complete before the first refresh of this session. The refresh path
+     * ([com.flipcash.app.persistence.sources.ChatMessageDataSource.upsert] on a self-authored
+     * message) deletes `SENDING` rows and spares `FAILED` ones, so sweeping first makes them
+     * durable and sweeping second does nothing at all. That is why this is `suspend` and awaited by
+     * [com.flipcash.shared.chat.internal.RealChatCoordinator.onUserLoggedIn] rather than launched
+     * from an `initialize(scope)` hook like the other delegates' — the refresh is itself a `launch`,
+     * and two coroutines would leave the order to chance.
+     */
+    internal suspend fun recoverInterruptedSends() {
+        messageDataSource.failInterruptedSends()
+    }
+
     internal suspend fun clear() {
         pendingMutations.value = emptyMap()
         metadataDataSource.clear()

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

@@ -0,0 +1,194 @@
+package com.flipcash.shared.chat
+
+import com.flipcash.app.core.dispatchers.TestDispatchers
+import com.flipcash.app.persistence.sources.ChatMemberDataSource
+import com.flipcash.app.persistence.sources.ChatMessageDataSource
+import com.flipcash.app.persistence.sources.ChatMetadataDataSource
+import com.flipcash.app.persistence.sources.ContactDataSource
+import com.flipcash.services.controllers.ChatController
+import com.flipcash.services.controllers.ChatMessagingController
+import com.flipcash.services.controllers.EventStreamingController
+import com.flipcash.services.models.chat.ChatUpdate
+import com.flipcash.services.user.UserManager
+import com.flipcash.shared.chat.internal.ChatIdGenerator
+import com.flipcash.shared.chat.internal.ChatStateHolder
+import com.flipcash.shared.chat.internal.RealChatCoordinator
+import com.flipcash.shared.chat.internal.delegates.DmChatResolverDelegate
+import com.flipcash.shared.chat.internal.delegates.EventStreamDelegate
+import com.flipcash.shared.chat.internal.delegates.FeedSyncDelegate
+import com.flipcash.shared.chat.internal.delegates.MessagingDelegate
+import com.getcode.utils.network.NetworkConnectivityListener
+import io.mockk.coEvery
+import io.mockk.coVerify
+import io.mockk.every
+import io.mockk.mockk
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.receiveAsFlow
+import kotlinx.coroutines.test.TestCoroutineScheduler
+import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.advanceTimeBy
+import kotlinx.coroutines.test.runCurrent
+import kotlinx.coroutines.test.runTest
+import org.junit.Before
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.robolectric.RobolectricTestRunner
+import kotlin.test.assertTrue
+import kotlin.time.Duration.Companion.milliseconds
+
+/**
+ * A send the app was killed in the middle of has to be marked failed before the session's first
+ * refresh, or the refresh deletes it.
+ *
+ * Nothing in-process survives the kill: a send is marked failed from the `onFailure` of the
+ * coroutine that issued it, so the row is left `SENDING` on disk. The first refresh that carries a
+ * self-authored message runs `deleteAllPending`, which is scoped to `SENDING` — so the sweep either
+ * runs first and makes the row durable, or runs second and finds nothing. There is no third
+ * outcome, which is why the order is worth a test rather than a comment.
+ */
+@OptIn(ExperimentalCoroutinesApi::class)
+@RunWith(RobolectricTestRunner::class)
+class InterruptedSendRecoveryTest {
+
+    private val chatUpdatesChannel = Channel<ChatUpdate>(capacity = Channel.UNLIMITED)
+
+    private lateinit var coordinator: RealChatCoordinator
+    private lateinit var messageDataSource: ChatMessageDataSource
+    private lateinit var testDispatchers: TestDispatchers
+
+    /** Set when the sweep returns; read by the feed fetch to report what it raced. */
+    private var sweepFinished = false
+    private var feedFetchSawFinishedSweep: Boolean? = null
+
+    @Before
+    fun setUp() {
+        val userManager = mockk<UserManager>(relaxed = true)
+        every { userManager.accountId } returns listOf<Byte>(1, 2, 3)
+
+        val eventStreamingController = mockk<EventStreamingController>(relaxed = true)
+        every { eventStreamingController.chatUpdates } returns chatUpdatesChannel.receiveAsFlow()
+        every { eventStreamingController.isConnected } returns true
+        every { eventStreamingController.isStreamActive } returns true
+
+        val chatController = mockk<ChatController>(relaxed = true)
+        coEvery { chatController.getDmChatFeed(any(), any()) } coAnswers {
+            feedFetchSawFinishedSweep = sweepFinished
+            Result.failure(RuntimeException("feed contents are not what this test is about"))
+        }
+
+        testDispatchers = TestDispatchers(TestCoroutineScheduler())
+
+        val stateHolder = ChatStateHolder()
+        val memberDataSource = mockk<ChatMemberDataSource>(relaxed = true)
+        val messagingController = mockk<ChatMessagingController>(relaxed = true)
+        val metadataDataSource = mockk<ChatMetadataDataSource>(relaxed = true)
+        messageDataSource = mockk(relaxed = true)
+        // The sweep is a Room write, so it is modelled with a suspension point. Without one the
+        // test cannot tell the fix from the bug: the refresh is a `launch`, so on a single-threaded
+        // test dispatcher it could not start before a sweep that never suspends had already
+        // returned, and a sweep placed after it would still appear to have gone first.
+        coEvery { messageDataSource.failInterruptedSends() } coAnswers {
+            delay(1)
+            sweepFinished = true
+        }
+
+        val feedDelegate = FeedSyncDelegate(
+            chatController = chatController,
+            metadataDataSource = metadataDataSource,
+            messageDataSource = messageDataSource,
+            memberDataSource = memberDataSource,
+            stateHolder = stateHolder,
+            userManager = userManager,
+        )
+
+        val eventStreamDelegate = EventStreamDelegate(
+            eventStreamingController = eventStreamingController,
+            messagingController = messagingController,
+            metadataDataSource = metadataDataSource,
+            messageDataSource = messageDataSource,
+            memberDataSource = memberDataSource,
+            tokenCoordinator = mockk(relaxed = true),
+            userManager = userManager,
+            stateHolder = stateHolder,
+            analytics = mockk(relaxed = true),
+            exchange = mockk(relaxed = true),
+        )
+
+        val messagingDelegate = MessagingDelegate(
+            chatController = chatController,
+            messagingController = messagingController,
+            metadataDataSource = metadataDataSource,
+            messageDataSource = messageDataSource,
+            memberDataSource = memberDataSource,
+            notificationManager = mockk(relaxed = true),
+            userManager = userManager,
+            stateHolder = stateHolder,
+            analytics = mockk(relaxed = true),
+            senderResolver = mockk(relaxed = true),
+        )
+
+        val dmChatResolverDelegate = DmChatResolverDelegate(
+            chatIdGenerator = ChatIdGenerator(),
+            userManager = userManager,
+            contactDataSource = mockk<ContactDataSource>(relaxed = true),
+            memberDataSource = memberDataSource,
+        )
+
+        coordinator = RealChatCoordinator(
+            feedDelegate = feedDelegate,
+            eventStreamDelegate = eventStreamDelegate,
+            dmChatResolverDelegate = dmChatResolverDelegate,
+            messagingDelegate = messagingDelegate,
+            stateHolder = stateHolder,
+            draftStore = mockk<ChatDraftStore>(relaxed = true),
+            userManager = userManager,
+            networkObserver = mockk<NetworkConnectivityListener>(relaxed = true),
+            dispatchers = testDispatchers,
+            groupFeedDelegate = mockk(relaxed = true),
+        )
+    }
+
+    /**
+     * Signs in, lets the login work settle, and tears the coordinator down however [block] ends.
+     *
+     * The `finally` is load-bearing, for the reason the other coordinator tests give: logging in
+     * starts a heartbeat that is a `while (true)` of delays on the coordinator's own scope, which
+     * is not [TestScope.backgroundScope] and so outlives the body. [runTest] drains the scheduler
+     * once the body returns, so a failed assertion that skipped the teardown would leave that loop
+     * spinning virtual time and bury the failure in a run that never ends.
+     *
+     * The advance is bounded for the same reason — `advanceUntilIdle` would never come back.
+     */
+    private suspend fun TestScope.loggedIn(block: () -> Unit) {
+        try {
+            coordinator.onUserLoggedIn(mockk(relaxed = true))
+            advanceTimeBy(10.milliseconds)
+            runCurrent()
+            block()
+        } finally {
+            coordinator.teardown()
+        }
+    }
+
+    @Test
+    fun `login sweeps interrupted sends before it fetches the feed`() =
+        runTest(testDispatchers.dispatcher) {
+            loggedIn {
+                assertTrue(
+                    feedFetchSawFinishedSweep == true,
+                    "The feed fetch ran while the sweep was still in flight, so a refresh " +
+                        "carrying a self-authored message could delete a SENDING row before it " +
+                        "had been marked failed.",
+                )
+            }
+        }
+
+    @Test
+    fun `the sweep runs once per login`() = runTest(testDispatchers.dispatcher) {
+        loggedIn {
+            coVerify(exactly = 1) { messageDataSource.failInterruptedSends() }
+        }
+    }
+}

apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMessageDao.kt

@@ -123,6 +123,26 @@ interface ChatMessageDao {
     @Query("DELETE FROM chat_messages WHERE chat_id_hex = :chatIdHex AND status = 'SENDING'")
     suspend fun deleteAllPending(chatIdHex: String)
 
+    /**
+     * Fails every row still marked `SENDING`, in every chat.
+     *
+     * A send is only ever marked failed from the `onFailure` of the coroutine that issued it, so a
+     * process killed mid-send leaves its row `SENDING` on disk with nothing left alive to move it.
+     * The transcript shows that row spinning — retry is offered for `FAILED`, not `SENDING` — until
+     * the next refresh reaches [upsertAndClearPending] and deletes it, taking the user's text with
+     * it and offering no way to get it back.
+     *
+     * Run once per process before the first refresh. [deleteAllPending] is scoped to `SENDING` and
+     * spares `FAILED`, so a row swept here survives every later refresh and picks up the retry
+     * affordance that already exists. `pending_client_id_hex` is untouched, and it is what the retry
+     * re-sends under — the same `client_message_id` the server dedupes on, so the re-send replaces
+     * the original rather than duplicating it.
+     *
+     * Deliberately not scoped to a chat: the caller runs at login, before it knows which chats exist.
+     */
+    @Query("UPDATE chat_messages SET status = 'FAILED' WHERE status = 'SENDING'")
+    suspend fun failInterruptedSends()
+
     /**
      * The optimistic row is written before the server has stamped the message, so every
      * server-assigned field is written here — `event_sequence` included. Leaving it at the pending

apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMessageDaoTest.kt

@@ -183,6 +183,47 @@ class ChatMessageDaoTest {
         assertEquals(2, dao.countNewerThan(CHAT_HEX, 3 * 1_000))
     }
 
+    /**
+     * Killing the app mid-send leaves the row `SENDING` with nothing alive to move it: failure is
+     * only ever written from the `onFailure` of the coroutine that issued the send. The sweep is
+     * what a fresh process has instead of that coroutine.
+     */
+    @Test
+    fun `the sweep fails a send a dead process left in flight`() = runTest {
+        dao.upsert(pending("half-sent"))
+
+        dao.failInterruptedSends()
+
+        val stored = dao.getByClientId(CHAT_HEX, CLIENT_HEX)!!
+        assertEquals(MessageStatus.FAILED, stored.status)
+        // The retry re-sends under this id, so losing it would lose the affordance with it.
+        assertEquals(CLIENT_HEX, stored.pendingClientIdHex)
+    }
+
+    /**
+     * The point of failing the row rather than leaving it pending. `deleteAllPending` — which every
+     * refresh carrying a self-authored message runs — is scoped to `SENDING`, so a swept row is
+     * durable where a pending one is deleted along with the text the user typed.
+     */
+    @Test
+    fun `a swept row survives the refresh that would have deleted it`() = runTest {
+        dao.upsert(pending("half-sent"))
+        dao.failInterruptedSends()
+
+        dao.upsertAndClearPending(CHAT_HEX, listOf(text(1, "from the server")))
+
+        assertEquals(MessageStatus.FAILED, dao.getByClientId(CHAT_HEX, CLIENT_HEX)?.status)
+    }
+
+    @Test
+    fun `the sweep leaves a sent message alone`() = runTest {
+        dao.upsert(text(1, "already sent"))
+
+        dao.failInterruptedSends()
+
+        assertEquals(MessageStatus.SENT, dao.getMessage(CHAT_HEX, 1)?.status)
+    }
+
     private companion object {
         const val CHAT_HEX = "aabb"
         const val OTHER_HEX = "ccdd"

apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/ChatMessageDataSource.kt

@@ -2,6 +2,7 @@ package com.flipcash.app.persistence.sources
 
 import androidx.paging.PagingSource
 import com.flipcash.app.persistence.FlipcashDatabase
+import com.flipcash.app.persistence.dao.ChatMessageDao
 import com.flipcash.app.persistence.entities.ChatMessageEntity
 import com.flipcash.app.persistence.sources.mapper.chat.ChatEntityMapper
 import com.flipcash.services.models.chat.ChatId
@@ -216,6 +217,13 @@ class ChatMessageDataSource @Inject constructor(
         )
     }
 
+    /**
+     * Fails the sends a previous process left in flight. See [ChatMessageDao.failInterruptedSends].
+     */
+    suspend fun failInterruptedSends() {
+        db?.chatMessageDao()?.failInterruptedSends()
+    }
+
     suspend fun retryPending(chatId: ChatId, pendingClientIdHex: String): ClientMessageId {
         val clientMessageId = mapper.clientMessageIdFromHex(pendingClientIdHex)
         db?.chatMessageDao()?.updatePendingStatus(