feat: unread archived conversations count [WPB-4437]

logic/src/commonMain/kotlin/com/wire/kalium/logic/data/conversation/ConversationRepository.kt

@@ -46,6 +46,7 @@ import com.wire.kalium.logic.functional.isLeft
 import com.wire.kalium.logic.functional.isRight
 import com.wire.kalium.logic.functional.map
 import com.wire.kalium.logic.functional.mapRight
+import com.wire.kalium.logic.functional.mapToRightOr
 import com.wire.kalium.logic.functional.onFailure
 import com.wire.kalium.logic.functional.onSuccess
 import com.wire.kalium.logic.kaliumLogger
@@ -234,6 +235,8 @@ interface ConversationRepository {
     ): Either<CoreFailure, Unit>
 
     suspend fun getConversationDetailsByMLSGroupId(mlsGroupId: GroupID): Either<CoreFailure, ConversationDetails>
+
+    suspend fun observeUnreadArchivedConversationsCount(): Flow<Long>
 }
 
 @Suppress("LongParameterList", "TooManyFunctions")
@@ -860,6 +863,11 @@ internal class ConversationDataSource internal constructor(
         wrapStorageRequest { conversationDAO.getConversationByGroupID(mlsGroupId.value) }
             .map { conversationMapper.fromDaoModelToDetails(it, null, mapOf()) }
 
+    override suspend fun observeUnreadArchivedConversationsCount(): Flow<Long> =
+        conversationDAO.observeUnreadArchivedConversationsCount()
+            .wrapStorageRequest()
+            .mapToRightOr(0L)
+
     private suspend fun persistIncompleteConversations(
         conversationsFailed: List<NetworkQualifiedId>
     ): Either<CoreFailure, Unit> {

logic/src/commonMain/kotlin/com/wire/kalium/logic/feature/conversation/ConversationScope.kt

@@ -261,6 +261,9 @@ class ConversationScope internal constructor(
             selfUserId
         )
 
+    val observeArchivedUnreadConversationsCount: ObserveArchivedUnreadConversationsCountUseCase
+        get() = ObserveArchivedUnreadConversationsCountUseCaseImpl(conversationRepository)
+
     internal val typingIndicatorRepository = TypingIndicatorRepositoryImpl(ConcurrentMutableMap())
 
     val observeUsersTyping: ObserveUsersTypingUseCase

logic/src/commonMain/kotlin/com/wire/kalium/logic/feature/conversation/ObserveArchivedUnreadConversationsCountUseCase.kt

@@ -0,0 +1,36 @@
+/*
+ * Wire
+ * Copyright (C) 2023 Wire Swiss GmbH
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see http://www.gnu.org/licenses/.
+ */
+package com.wire.kalium.logic.feature.conversation
+
+import com.wire.kalium.logic.data.conversation.ConversationRepository
+import kotlinx.coroutines.flow.Flow
+
+/**
+ * UseCase for observing the count of archived conversations that have unread events (e.g., messages, pings, missed calls).
+ * The result is presented as a continuous flow, updating whenever the count changes.
+ */
+interface ObserveArchivedUnreadConversationsCountUseCase {
+    suspend operator fun invoke(): Flow<Long>
+}
+
+class ObserveArchivedUnreadConversationsCountUseCaseImpl(
+    private val conversationRepository: ConversationRepository
+) : ObserveArchivedUnreadConversationsCountUseCase {
+
+    override suspend fun invoke(): Flow<Long> = conversationRepository.observeUnreadArchivedConversationsCount()
+}

logic/src/commonTest/kotlin/com/wire/kalium/logic/data/conversation/ConversationRepositoryTest.kt

@@ -97,6 +97,7 @@ import io.mockative.once
 import io.mockative.thenDoNothing
 import io.mockative.verify
 import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.flowOf
 import kotlinx.coroutines.test.runTest
 import kotlinx.datetime.Clock
@@ -1083,6 +1084,17 @@ class ConversationRepositoryTest {
             .wasInvoked(exactly = once)
     }
 
+    @Test
+    fun givenUnreadArchivedConversationsCount_WhenObserving_ThenShouldReturnSuccess() = runTest {
+        val unreadCount = 10L
+        val (arrange, conversationRepository) = Arrangement()
+            .withUnreadArchivedConversationsCount(unreadCount)
+            .arrange()
+
+        val result = conversationRepository.observeUnreadArchivedConversationsCount().first()
+        assertEquals(unreadCount, result)
+    }
+
     private class Arrangement :
         MemberDAOArrangement by MemberDAOArrangementImpl() {
         @Mock
@@ -1222,6 +1234,13 @@ class ConversationRepositoryTest {
                 .thenReturn(flowOf(unreadEvents))
         }
 
+        fun withUnreadArchivedConversationsCount(unreadCount: Long) = apply {
+            given(conversationDAO)
+                .suspendFunction(conversationDAO::observeUnreadArchivedConversationsCount)
+                .whenInvoked()
+                .thenReturn(flowOf(unreadCount))
+        }
+
         fun withUnreadMessageCounter(unreadCounter: Map<ConversationIDEntity, Int>) = apply {
             given(messageDAO)
                 .suspendFunction(messageDAO::observeUnreadMessageCounter)

logic/src/commonTest/kotlin/com/wire/kalium/logic/feature/conversation/ObserveArchivedUnreadConversationsCountUseCaseTest.kt

@@ -0,0 +1,68 @@
+/*
+ * Wire
+ * Copyright (C) 2023 Wire Swiss GmbH
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see http://www.gnu.org/licenses/.
+ */
+
+package com.wire.kalium.logic.feature.conversation
+
+import app.cash.turbine.test
+import com.wire.kalium.logic.data.conversation.ConversationRepository
+import io.mockative.Mock
+import io.mockative.given
+import io.mockative.mock
+import io.mockative.once
+import io.mockative.verify
+import kotlinx.coroutines.flow.flowOf
+import kotlinx.coroutines.test.runTest
+import kotlin.test.BeforeTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+
+class ObserveArchivedUnreadConversationsCountUseCaseTest {
+
+    @Mock
+    private val conversationRepository: ConversationRepository = mock(ConversationRepository::class)
+
+    private lateinit var observeArchivedUnreadConversationsCount: ObserveArchivedUnreadConversationsCountUseCase
+
+    @BeforeTest
+    fun setup() {
+        observeArchivedUnreadConversationsCount = ObserveArchivedUnreadConversationsCountUseCaseImpl(conversationRepository)
+    }
+
+    @Test
+    fun givenArchivedUnreadConversationsCount_whenObserving_thenCorrectCountShouldBeReturned() = runTest {
+        // Given
+        val unreadCount = 10L
+
+        given(conversationRepository)
+            .suspendFunction(conversationRepository::observeUnreadArchivedConversationsCount)
+            .whenInvoked()
+            .then { flowOf(unreadCount) }
+
+        // When
+        observeArchivedUnreadConversationsCount().test {
+            // Then
+            val result = awaitItem()
+            verify(conversationRepository)
+                .suspendFunction(conversationRepository::observeUnreadArchivedConversationsCount)
+                .wasInvoked(exactly = once)
+
+            assertEquals(unreadCount, result)
+            awaitComplete()
+        }
+    }
+}

persistence/src/commonMain/db_user/com/wire/kalium/persistence/UnreadEvents.sq

@@ -38,3 +38,8 @@ SELECT * FROM UnreadEvent LIMIT ? OFFSET ?;
 
 getConversationUnreadEventsCount:
 SELECT COUNT(*) FROM UnreadEvent WHERE  conversation_id = ?;
+
+getUnreadArchivedConversationsCount:
+SELECT COUNT(DISTINCT ue.conversation_id) FROM UnreadEvent ue
+INNER JOIN Conversation c ON ue.conversation_id = c.qualified_id
+WHERE c.archived = 1;

persistence/src/commonMain/kotlin/com/wire/kalium/persistence/dao/conversation/ConversationDAO.kt

@@ -94,4 +94,5 @@ interface ConversationDAO {
     suspend fun clearContent(conversationId: QualifiedIDEntity)
     suspend fun updateVerificationStatus(verificationStatus: ConversationEntity.VerificationStatus, conversationId: QualifiedIDEntity)
     suspend fun getConversationByGroupID(groupID: String): ConversationViewEntity
+    suspend fun observeUnreadArchivedConversationsCount(): Flow<Long>
 }

persistence/src/commonMain/kotlin/com/wire/kalium/persistence/dao/conversation/ConversationDAOImpl.kt

@@ -25,6 +25,7 @@ import com.wire.kalium.persistence.UnreadEventsQueries
 import com.wire.kalium.persistence.dao.QualifiedIDEntity
 import com.wire.kalium.persistence.dao.UserIDEntity
 import com.wire.kalium.persistence.util.mapToList
+import com.wire.kalium.persistence.util.mapToOne
 import com.wire.kalium.persistence.util.mapToOneOrNull
 import com.wire.kalium.util.DateTimeUtil
 import com.wire.kalium.util.DateTimeUtil.toIsoDateTimeString
@@ -342,4 +343,7 @@ internal class ConversationDAOImpl internal constructor(
     ) = withContext(coroutineContext) {
         conversationQueries.updateVerificationStatus(verificationStatus, conversationId)
     }
+
+    override suspend fun observeUnreadArchivedConversationsCount(): Flow<Long> =
+        unreadEventsQueries.getUnreadArchivedConversationsCount().asFlow().mapToOne()
 }

persistence/src/commonTest/kotlin/com/wire/kalium/persistence/dao/ConversationDAOTest.kt

@@ -977,6 +977,59 @@ class ConversationDAOTest : BaseDatabaseTest() {
         assertEquals(conversation2.id, result[0].id)
     }
 
+    @Test
+    fun givenArchivedConversations_whenObservingUnreadConversationCount_thenReturnedCorrectCount() = runTest {
+        // given
+        val conversation1 = conversationEntity1.copy(
+            id = ConversationIDEntity("convNullName", "domain"),
+            name = null,
+            type = ConversationEntity.Type.GROUP,
+            hasIncompleteMetadata = false,
+            lastModifiedDate = "2021-03-30T15:36:00.000Z".toInstant(),
+            lastReadDate = "2021-03-30T15:36:00.000Z".toInstant(),
+            archived = true
+        )
+
+        val conversation2 = conversationEntity2.copy(
+            id = ConversationIDEntity("convWithName", "domain"),
+            name = "name",
+            type = ConversationEntity.Type.GROUP,
+            hasIncompleteMetadata = false,
+            lastModifiedDate = "2021-03-30T15:36:00.000Z".toInstant(),
+            lastReadDate = "2021-03-30T15:36:00.000Z".toInstant(),
+            archived = false
+        )
+
+        val instant = Clock.System.now()
+
+        conversationDAO.insertConversation(conversation1)
+        conversationDAO.insertConversation(conversation2)
+        insertTeamUserAndMember(team, user1, conversation1.id)
+        insertTeamUserAndMember(team, user1, conversation2.id)
+
+        repeat(5) {
+            newRegularMessageEntity(
+                id = Random.nextBytes(10).decodeToString(),
+                conversationId = conversation1.id,
+                senderUserId = user1.id,
+                date = instant
+            ).also { messageDAO.insertOrIgnoreMessage(it) }
+
+            newRegularMessageEntity(
+                id = Random.nextBytes(10).decodeToString(),
+                conversationId = conversation2.id,
+                senderUserId = user1.id,
+                date = instant
+            ).also { messageDAO.insertOrIgnoreMessage(it) }
+        }
+
+        // when
+        val result = conversationDAO.observeUnreadArchivedConversationsCount().first()
+
+        // then
+        assertTrue(result == 1L)
+    }
+
     private suspend fun insertTeamUserAndMember(team: TeamEntity, user: UserEntity, conversationId: QualifiedIDEntity) {
         teamDAO.insertTeam(team)
         userDAO.insertUser(user)