Stop polling message receipts while idle

stream-chat-android-client/src/main/java/io/getstream/chat/android/client/persistence/db/dao/MessageReceiptDao.kt

@@ -21,6 +21,7 @@ import androidx.room.Insert
 import androidx.room.OnConflictStrategy
 import androidx.room.Query
 import io.getstream.chat.android.client.persistence.db.entity.MessageReceiptEntity
+import kotlinx.coroutines.flow.Flow
 
 @Dao
 internal interface MessageReceiptDao {
@@ -31,6 +32,9 @@ internal interface MessageReceiptDao {
     @Query("SELECT * FROM message_receipt ORDER BY createdAt ASC LIMIT :limit")
     suspend fun selectAll(limit: Int): List<MessageReceiptEntity>
 
+    @Query("SELECT COUNT(*) FROM message_receipt")
+    fun observeCount(): Flow<Int>
+
     @Query("DELETE FROM message_receipt WHERE messageId IN (:messageIds)")
     suspend fun deleteByMessageIds(messageIds: List<String>)
 

stream-chat-android-client/src/main/java/io/getstream/chat/android/client/persistence/repository/MessageReceiptRepository.kt

@@ -20,6 +20,7 @@ import io.getstream.chat.android.client.persistence.db.ChatClientDatabase
 import io.getstream.chat.android.client.persistence.db.dao.MessageReceiptDao
 import io.getstream.chat.android.client.persistence.db.entity.MessageReceiptEntity
 import io.getstream.chat.android.client.receipts.MessageReceipt
+import kotlinx.coroutines.flow.Flow
 
 internal interface MessageReceiptRepository {
 
@@ -34,6 +35,8 @@ internal interface MessageReceiptRepository {
 
     suspend fun selectMessageReceipts(limit: Int): List<MessageReceipt>
 
+    fun observeMessageReceiptCount(): Flow<Int>
+
     suspend fun deleteMessageReceiptsByMessageIds(messageIds: List<String>)
 
     suspend fun clearMessageReceipts()
@@ -50,6 +53,8 @@ internal class MessageReceiptRepositoryImpl(
     override suspend fun selectMessageReceipts(limit: Int): List<MessageReceipt> =
         dao.selectAll(limit).map(MessageReceiptEntity::toModel)
 
+    override fun observeMessageReceiptCount(): Flow<Int> = dao.observeCount()
+
     override suspend fun deleteMessageReceiptsByMessageIds(messageIds: List<String>) {
         dao.deleteByMessageIds(messageIds)
     }

stream-chat-android-client/src/main/java/io/getstream/chat/android/client/receipts/MessageReceiptReporter.kt

@@ -22,13 +22,17 @@ import io.getstream.chat.android.models.Message
 import io.getstream.log.taggedLogger
 import io.getstream.result.onSuccessSuspend
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
 import kotlinx.coroutines.delay
-import kotlinx.coroutines.isActive
+import kotlinx.coroutines.flow.conflate
 import kotlinx.coroutines.launch
 
 /**
- * Reports message delivery receipts to the server in batches of [MAX_BATCH_SIZE]
- * every [REPORT_INTERVAL_IN_MS] milliseconds.
+ * Reports message delivery receipts to the server.
+ *
+ * Idle sessions do not poll the repository. The receipt-count flow wakes reporting after any
+ * write, and each [start] drains receipts that were saved before the job began.
+ * See [start] for batching, pacing, and retry behavior.
  */
 internal class MessageReceiptReporter(
     private val scope: CoroutineScope,
@@ -38,44 +42,79 @@ internal class MessageReceiptReporter(
 
     private val logger by taggedLogger("Chat:MessageReceiptReporter")
 
+    private var reportingJob: Job? = null
+
+    /**
+     * Starts reporting queued delivery receipts for the current user session.
+     *
+     * Collects [MessageReceiptRepository.observeMessageReceiptCount] and drains while the count is
+     * positive. Receipts persisted before this call are included in the initial emission. While the
+     * queue is non-empty, batches of at most [MAX_BATCH_SIZE] are reported at least
+     * [REPORT_INTERVAL_IN_MS] apart, including the selection that finds the queue empty. A zero
+     * count suspends until a later write. Failed deliveries stay queued and are retried on that
+     * cadence without another write. A second call while the reporting job is still active does
+     * nothing.
+     */
     fun start() {
+        if (reportingJob?.isActive == true) {
+            logger.d { "Reporter is already active" }
+            return
+        }
         logger.d { "Starting reporter…" }
-        scope.launch {
+        reportingJob = scope.launch {
             try {
-                while (isActive) {
-                    val messages = messageReceiptRepository
-                        .selectMessageReceipts(limit = MAX_BATCH_SIZE)
-                        .map { receipt ->
-                            Message(
-                                id = receipt.messageId,
-                                cid = receipt.cid,
-                            )
+                messageReceiptRepository.observeMessageReceiptCount()
+                    .conflate()
+                    .collect { count ->
+                        if (count > 0) {
+                            drainQueuedReceipts()
                         }
-
-                    if (messages.isNotEmpty()) {
-                        logger.d { "Reporting delivery receipts for ${messages.size} messages…" }
-                        api.markDelivered(messages)
-                            .execute()
-                            .onSuccessSuspend {
-                                logger.d { "Successfully reported delivery receipts for ${messages.size} messages" }
-                                val deliveredMessageIds = messages.map(Message::id)
-                                messageReceiptRepository.deleteMessageReceiptsByMessageIds(deliveredMessageIds)
-                            }
-                            .onError { error ->
-                                logger.e {
-                                    "Failed to report delivery receipts for ${messages.size} messages: " +
-                                        error.message
-                                }
-                            }
                     }
-
-                    delay(REPORT_INTERVAL_IN_MS)
-                }
             } finally {
                 logger.d { "Reporter is no longer active" }
             }
         }
     }
+
+    /**
+     * Selects and reports until the repository returns an empty batch.
+     *
+     * Failed deliveries stay queued and are retried after [REPORT_INTERVAL_IN_MS].
+     * Cancellation from [delay] or [ChatApi.markDelivered] propagates to the reporting job.
+     */
+    private suspend fun drainQueuedReceipts() {
+        while (true) {
+            val messages = messageReceiptRepository
+                .selectMessageReceipts(limit = MAX_BATCH_SIZE)
+                .map { receipt ->
+                    Message(
+                        id = receipt.messageId,
+                        cid = receipt.cid,
+                    )
+                }
+
+            if (messages.isEmpty()) {
+                return
+            }
+
+            logger.d { "Reporting delivery receipts for ${messages.size} messages…" }
+            api.markDelivered(messages)
+                .await()
+                .onSuccessSuspend {
+                    logger.d { "Successfully reported delivery receipts for ${messages.size} messages" }
+                    val deliveredMessageIds = messages.map(Message::id)
+                    messageReceiptRepository.deleteMessageReceiptsByMessageIds(deliveredMessageIds)
+                }
+                .onError { error ->
+                    logger.e {
+                        "Failed to report delivery receipts for ${messages.size} messages: " +
+                            error.message
+                    }
+                }
+
+            delay(REPORT_INTERVAL_IN_MS)
+        }
+    }
 }
 
 private const val REPORT_INTERVAL_IN_MS = 1000L

stream-chat-android-client/src/test/java/io/getstream/chat/android/client/receipts/MessageReceiptManagerTest.kt

@@ -381,13 +381,25 @@ internal class MessageReceiptManagerTest {
         fixture.verifyUpsertMessageReceiptsCalled(never())
     }
 
+    @Test
+    fun `should propagate failure when persisting receipts fails`() = runTest {
+        val fixture = Fixture().givenUpsertFails()
+        val sut = fixture.get()
+
+        val failure = runCatching { sut.markMessageAsDelivered(DeliverableMessage) }
+
+        assertTrue(failure.exceptionOrNull() is IllegalStateException)
+        fixture.verifyUpsertMessageReceiptsCalled()
+    }
+
     private class Fixture {
+        private val messageReceiptRepository: MessageReceiptRepository = mock()
+
         private val mockRepositoryFacade = mock<RepositoryFacade> {
             onBlocking { selectUser("me") } doReturn CurrentUser
             onBlocking { selectChannel(DeliverableChannel.cid) } doReturn DeliverableChannel
             onBlocking { selectMessage(DeliverableMessage.id) } doReturn DeliverableMessage
         }
-        private val mockMessageReceiptRepository = mock<MessageReceiptRepository>()
         private val mockChatApi = mock<ChatApi> {
             on {
                 queryChannel(
@@ -403,6 +415,12 @@ internal class MessageReceiptManagerTest {
             wheneverBlocking { mockRepositoryFacade.selectUser("me") } doReturn user
         }
 
+        fun givenUpsertFails() = apply {
+            wheneverBlocking { messageReceiptRepository.upsertMessageReceipts(any()) }.thenAnswer {
+                throw IllegalStateException("upsert failed")
+            }
+        }
+
         fun givenChannelNotFoundFromRepository() = apply {
             wheneverBlocking { mockRepositoryFacade.selectChannel(cid = any()) } doReturn null
         }
@@ -429,15 +447,15 @@ internal class MessageReceiptManagerTest {
             mode: VerificationMode = times(1),
             receipts: List<MessageReceipt>? = null,
         ) {
-            verifyBlocking(mockMessageReceiptRepository, mode) {
+            verifyBlocking(messageReceiptRepository, mode) {
                 upsertMessageReceipts(receipts ?: any())
             }
         }
 
         fun get() = MessageReceiptManager(
             now = { Now },
             getRepositoryFacade = { mockRepositoryFacade },
-            messageReceiptRepository = mockMessageReceiptRepository,
+            messageReceiptRepository = messageReceiptRepository,
             api = mockChatApi,
         )
     }