Make MessageReceiptRepository getAllByType return a Flow and add a clear function

The `MessageReceiptRepository.getAllByType` function now accepts a `limit` parameter and returns a `Flow<List<MessageReceipt>>` instead of a `suspend` function returning `List<MessageReceipt>`.

A new `clear()` function has been added to `MessageReceiptRepository` to delete all receipts from the database.

Author: andremion

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

@@ -0,0 +1,34 @@
+/*
+ * Copyright (c) 2014-2025 Stream.io Inc. All rights reserved.
+ *
+ * Licensed under the Stream License;
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    https://github.com/GetStream/stream-chat-android/blob/main/LICENSE
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.getstream.chat.android.client.persistance.repository
+
+import io.getstream.chat.android.core.internal.InternalStreamChatApi
+import io.getstream.chat.android.models.MessageReceipt
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.emptyFlow
+
+@InternalStreamChatApi
+public interface MessageReceiptRepository {
+
+    public suspend fun upsert(receipts: List<MessageReceipt>) { /* no-op */ }
+
+    public fun getAllByType(type: String, limit: Int): Flow<List<MessageReceipt>> = emptyFlow()
+
+    public suspend fun deleteByMessageIds(messageIds: List<String>) { /* no-op */ }
+
+    public suspend fun clear() { /* no-op */ }
+}

stream-chat-android-offline/api/stream-chat-android-offline.api

@@ -190,9 +190,10 @@ public final class io/getstream/chat/android/offline/repository/domain/reaction/
 
 public final class io/getstream/chat/android/offline/repository/domain/receipts/MessageReceiptDao_Impl : io/getstream/chat/android/offline/repository/domain/receipts/MessageReceiptDao {
 	public fun <init> (Landroidx/room/RoomDatabase;)V
+	public fun deleteAll (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public fun deleteByMessageIds (Ljava/util/List;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public static fun getRequiredConverters ()Ljava/util/List;
-	public fun selectAllByType (Ljava/lang/String;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public fun selectAllByType (Ljava/lang/String;I)Lkotlinx/coroutines/flow/Flow;
 	public fun upsert (Ljava/util/List;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 }
 

stream-chat-android-offline/src/main/java/io/getstream/chat/android/offline/repository/domain/receipts/MessageReceiptDao.kt

@@ -4,16 +4,25 @@ import androidx.room.Dao
 import androidx.room.Insert
 import androidx.room.OnConflictStrategy
 import androidx.room.Query
+import kotlinx.coroutines.flow.Flow
 
 @Dao
 internal interface MessageReceiptDao {
 
     @Insert(onConflict = OnConflictStrategy.REPLACE)
     suspend fun upsert(receipts: List<MessageReceiptEntity>)
 
-    @Query("SELECT * FROM stream_chat_message_receipt WHERE type = :type ORDER BY createdAt ASC")
-    suspend fun selectAllByType(type: String): List<MessageReceiptEntity>
+    @Query(
+        "SELECT * FROM stream_chat_message_receipt " +
+            "WHERE type = :type " +
+            "ORDER BY createdAt ASC " +
+            "LIMIT :limit"
+    )
+    fun selectAllByType(type: String, limit: Int): Flow<List<MessageReceiptEntity>>
 
     @Query("DELETE FROM stream_chat_message_receipt WHERE messageId IN (:messageIds)")
     suspend fun deleteByMessageIds(messageIds: List<String>)
+
+    @Query("DELETE FROM stream_chat_message_receipt")
+    suspend fun deleteAll()
 }

stream-chat-android-offline/src/main/java/io/getstream/chat/android/offline/repository/domain/receipts/MessageReceiptRepositoryImpl.kt

@@ -2,6 +2,8 @@ package io.getstream.chat.android.offline.repository.domain.receipts
 
 import io.getstream.chat.android.client.persistance.repository.MessageReceiptRepository
 import io.getstream.chat.android.models.MessageReceipt
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.map
 
 internal class MessageReceiptRepositoryImpl(
     private val dao: MessageReceiptDao,
@@ -11,10 +13,17 @@ internal class MessageReceiptRepositoryImpl(
         dao.upsert(receipts.map(MessageReceipt::toEntity))
     }
 
-    override suspend fun getAllByType(type: String): List<MessageReceipt> =
-        dao.selectAllByType(type).map(MessageReceiptEntity::toModel)
+    override fun getAllByType(type: String, limit: Int): Flow<List<MessageReceipt>> =
+        dao.selectAllByType(type, limit)
+            .map { receipts ->
+                receipts.map(MessageReceiptEntity::toModel)
+            }
 
     override suspend fun deleteByMessageIds(messageIds: List<String>) {
         dao.deleteByMessageIds(messageIds)
     }
+
+    override suspend fun clear() {
+        dao.deleteAll()
+    }
 }

stream-chat-android-offline/src/test/java/io/getstream/chat/android/offline/repository/domain/receipts/MessageReceiptRepositoryImplTest.kt

@@ -1,9 +1,12 @@
 package io.getstream.chat.android.offline.repository.domain.receipts
 
+import app.cash.turbine.test
 import io.getstream.chat.android.models.MessageReceipt
 import io.getstream.chat.android.offline.randomMessageReceiptEntity
+import io.getstream.chat.android.randomInt
 import io.getstream.chat.android.randomMessageReceipt
 import io.getstream.chat.android.randomString
+import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.test.runTest
 import org.junit.Test
 import org.junit.jupiter.api.Assertions.assertEquals
@@ -31,31 +34,35 @@ internal class MessageReceiptRepositoryImplTest {
                 cid = receipt.cid,
             )
         )
-        fixture.verifyUpsert(expectedReceipts)
+        fixture.verifyUpsertCalled(expectedReceipts)
     }
 
     @Test
     fun `get receipts by type`() = runTest {
         val type = randomString()
+        val limit = randomInt()
         val receipt = randomMessageReceiptEntity()
         val fixture = Fixture()
             .givenReceiptsByType(
                 type = type,
+                limit = limit,
                 receipts = listOf(receipt),
             )
         val sut = fixture.get()
 
-        val actual = sut.getAllByType(type)
+        sut.getAllByType(type, limit).test {
+            val actual = awaitItem()
 
-        val expected = listOf(
-            MessageReceipt(
-                messageId = receipt.messageId,
-                type = receipt.type,
-                createdAt = receipt.createdAt,
-                cid = receipt.cid,
+            val expected = listOf(
+                MessageReceipt(
+                    messageId = receipt.messageId,
+                    type = receipt.type,
+                    createdAt = receipt.createdAt,
+                    cid = receipt.cid,
+                )
             )
-        )
-        assertEquals(expected, actual)
+            assertEquals(expected, actual)
+        }
     }
 
     @Test
@@ -66,27 +73,45 @@ internal class MessageReceiptRepositoryImplTest {
 
         sut.deleteByMessageIds(messageIds)
 
-        fixture.verifyDeleteByMessageIds(messageIds)
+        fixture.verifyDeleteByMessageIdsCalled(messageIds)
+    }
+
+    @Test
+    fun `clear receipts`() = runTest {
+        val fixture = Fixture()
+        val sut = fixture.get()
+
+        sut.clear()
+
+        fixture.verifyDeleteAllCalled()
     }
 
     private class Fixture {
+
+        private val receiptsStateFlow = MutableStateFlow<List<MessageReceiptEntity>>(emptyList())
+
         private val mockDao = mock<MessageReceiptDao> {
             onBlocking { upsert(any()) } doReturn Unit
             onBlocking { deleteByMessageIds(any()) } doReturn Unit
         }
 
-        fun givenReceiptsByType(type: String, receipts: List<MessageReceiptEntity>) = apply {
-            wheneverBlocking { mockDao.selectAllByType(type) } doReturn receipts
+        fun givenReceiptsByType(type: String, limit: Int, receipts: List<MessageReceiptEntity>) = apply {
+            wheneverBlocking { mockDao.selectAllByType(type, limit) } doReturn
+                receiptsStateFlow.apply { value = receipts }
         }
 
-        fun verifyUpsert(receipts: List<MessageReceiptEntity>) {
+        fun verifyUpsertCalled(receipts: List<MessageReceiptEntity>) {
             verifyBlocking(mockDao) { upsert(receipts) }
         }
 
-        fun verifyDeleteByMessageIds(messageIds: List<String>) {
+        fun verifyDeleteByMessageIdsCalled(messageIds: List<String>) {
             verifyBlocking(mockDao) { deleteByMessageIds(messageIds) }
         }
 
+        fun verifyDeleteAllCalled() {
+            verifyBlocking(mockDao) { deleteAll() }
+        }
+
         fun get() = MessageReceiptRepositoryImpl(dao = mockDao)
     }
 }