feat(typing): send typing indicator events - epic (WP-4590)

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

@@ -238,6 +238,10 @@ interface ConversationRepository {
     suspend fun getConversationDetailsByMLSGroupId(mlsGroupId: GroupID): Either<CoreFailure, ConversationDetails>
 
     suspend fun observeUnreadArchivedConversationsCount(): Flow<Long>
+    suspend fun sendTypingIndicatorStatus(
+        conversationId: ConversationId,
+        typingStatus: Conversation.TypingIndicatorMode
+    ): Either<CoreFailure, Unit>
 }
 
 @Suppress("LongParameterList", "TooManyFunctions")
@@ -869,6 +873,13 @@ internal class ConversationDataSource internal constructor(
             .wrapStorageRequest()
             .mapToRightOr(0L)
 
+    override suspend fun sendTypingIndicatorStatus(
+        conversationId: ConversationId,
+        typingStatus: Conversation.TypingIndicatorMode
+    ): Either<CoreFailure, Unit> = wrapApiRequest {
+        conversationApi.sendTypingIndicatorNotification(conversationId.toApi(), typingStatus.toStatusDto())
+    }
+
     private suspend fun persistIncompleteConversations(
         conversationsFailed: List<NetworkQualifiedId>
     ): Either<CoreFailure, Unit> {

logic/src/commonMain/kotlin/com/wire/kalium/logic/data/conversation/TypingIndicatorRepository.kt → logic/src/commonMain/kotlin/com/wire/kalium/logic/data/conversation/TypingIndicatorIncomingRepository.kt

@@ -27,26 +27,25 @@ import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.onStart
-import kotlinx.datetime.Clock
-import kotlinx.datetime.Instant
 
-internal interface TypingIndicatorRepository {
+internal interface TypingIndicatorIncomingRepository {
     suspend fun addTypingUserInConversation(conversationId: ConversationId, userId: UserId)
     suspend fun removeTypingUserInConversation(conversationId: ConversationId, userId: UserId)
-    suspend fun observeUsersTyping(conversationId: ConversationId): Flow<Set<ExpiringUserTyping>>
+    suspend fun observeUsersTyping(conversationId: ConversationId): Flow<Set<UserId>>
+    suspend fun clearExpiredTypingIndicators()
 }
 
-internal class TypingIndicatorRepositoryImpl(
-    private val userTypingCache: ConcurrentMutableMap<ConversationId, MutableSet<ExpiringUserTyping>>,
+internal class TypingIndicatorIncomingRepositoryImpl(
+    private val userTypingCache: ConcurrentMutableMap<ConversationId, MutableSet<UserId>>,
     private val userPropertyRepository: UserPropertyRepository
-) : TypingIndicatorRepository {
+) : TypingIndicatorIncomingRepository {
 
     private val userTypingDataSourceFlow: MutableSharedFlow<Unit> =
         MutableSharedFlow(extraBufferCapacity = BUFFER_SIZE, onBufferOverflow = BufferOverflow.DROP_OLDEST)
 
     override suspend fun addTypingUserInConversation(conversationId: ConversationId, userId: UserId) {
         if (userPropertyRepository.getTypingIndicatorStatus()) {
-            userTypingCache.safeComputeAndMutateSetValue(conversationId) { ExpiringUserTyping(userId, Clock.System.now()) }
+            userTypingCache.safeComputeAndMutateSetValue(conversationId) { userId }
                 .also {
                     userTypingDataSourceFlow.tryEmit(Unit)
                 }
@@ -55,36 +54,27 @@ internal class TypingIndicatorRepositoryImpl(
 
     override suspend fun removeTypingUserInConversation(conversationId: ConversationId, userId: UserId) {
         userTypingCache.block { entry ->
-            entry[conversationId]?.apply { this.removeAll { it.userId == userId } }
+            entry[conversationId]?.apply { this.removeAll { it == userId } }
         }.also {
             userTypingDataSourceFlow.tryEmit(Unit)
         }
     }
 
-    override suspend fun observeUsersTyping(conversationId: ConversationId): Flow<Set<ExpiringUserTyping>> {
+    override suspend fun observeUsersTyping(conversationId: ConversationId): Flow<Set<UserId>> {
         return userTypingDataSourceFlow
             .map { userTypingCache[conversationId] ?: emptySet() }
             .onStart { emit(userTypingCache[conversationId] ?: emptySet()) }
     }
 
-    companion object {
-        const val BUFFER_SIZE = 32 // drop after this threshold
-    }
-}
-
-// todo expire by worker
-data class ExpiringUserTyping(
-    val userId: UserId,
-    val date: Instant
-) {
-    override fun equals(other: Any?): Boolean {
-        return other != null && when (other) {
-            is ExpiringUserTyping -> other.userId == this.userId
-            else -> false
+    override suspend fun clearExpiredTypingIndicators() {
+        userTypingCache.block { entry ->
+            entry.clear()
+        }.also {
+            userTypingDataSourceFlow.tryEmit(Unit)
         }
     }
 
-    override fun hashCode(): Int {
-        return this.userId.hashCode()
+    companion object {
+        const val BUFFER_SIZE = 32 // drop after this threshold
     }
 }

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

@@ -0,0 +1,53 @@
+/*
+ * 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.data.conversation
+
+import com.wire.kalium.logic.CoreFailure
+import com.wire.kalium.logic.data.id.ConversationId
+import com.wire.kalium.logic.data.properties.UserPropertyRepository
+import com.wire.kalium.logic.functional.Either
+
+internal interface TypingIndicatorOutgoingRepository {
+    suspend fun sendTypingIndicatorStatus(
+        conversationId: ConversationId,
+        typingStatus: Conversation.TypingIndicatorMode
+    ): Either<CoreFailure, Unit>
+
+}
+
+internal class TypingIndicatorOutgoingRepositoryImpl(
+    private val typingIndicatorSenderHandler: TypingIndicatorSenderHandler,
+    private val userPropertyRepository: UserPropertyRepository
+) : TypingIndicatorOutgoingRepository {
+
+    override suspend fun sendTypingIndicatorStatus(
+        conversationId: ConversationId,
+        typingStatus: Conversation.TypingIndicatorMode
+    ): Either<CoreFailure, Unit> {
+        if (userPropertyRepository.getTypingIndicatorStatus()) {
+            when (typingStatus) {
+                Conversation.TypingIndicatorMode.STARTED ->
+                    typingIndicatorSenderHandler.sendStartedAndEnqueueStoppingEvent(conversationId)
+
+                Conversation.TypingIndicatorMode.STOPPED -> typingIndicatorSenderHandler.sendStoppingEvent(conversationId)
+            }
+        }
+        return Either.Right(Unit)
+    }
+
+}

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

@@ -0,0 +1,106 @@
+/*
+ * 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.data.conversation
+
+import com.wire.kalium.logic.data.id.ConversationId
+import com.wire.kalium.logic.functional.fold
+import com.wire.kalium.logic.kaliumLogger
+import com.wire.kalium.util.KaliumDispatcher
+import com.wire.kalium.util.KaliumDispatcherImpl
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+import kotlin.coroutines.CoroutineContext
+import kotlin.time.DurationUnit
+import kotlin.time.toDuration
+
+/**
+ * Outgoing user typing sent events manager.
+ *
+ * - It will send started and stopped events.
+ * - For each started sent event, will 'enqueue' a stopped event after a timeout.
+ *
+ */
+internal interface TypingIndicatorSenderHandler {
+    fun sendStoppingEvent(conversationId: ConversationId)
+    fun sendStartedAndEnqueueStoppingEvent(conversationId: ConversationId)
+}
+
+internal class TypingIndicatorSenderHandlerImpl(
+    private val conversationRepository: ConversationRepository,
+    private val kaliumDispatcher: KaliumDispatcher = KaliumDispatcherImpl,
+    userSessionCoroutineScope: CoroutineScope
+) : TypingIndicatorSenderHandler, CoroutineScope by userSessionCoroutineScope {
+    override val coroutineContext: CoroutineContext
+        get() = kaliumDispatcher.default
+
+    private val outgoingStoppedQueueTypingEventsMutex = Mutex()
+    private val outgoingStoppedQueueTypingEvents = mutableMapOf<ConversationId, Unit>()
+    private val typingIndicatorTimeoutInSeconds = 10.toDuration(DurationUnit.SECONDS)
+
+    /**
+     * Sends a stopping event and removes it from the 'queue'.
+     */
+    override fun sendStoppingEvent(conversationId: ConversationId) {
+        launch {
+            outgoingStoppedQueueTypingEventsMutex.withLock {
+                if (!outgoingStoppedQueueTypingEvents.containsKey(conversationId)) {
+                    return@launch
+                }
+                outgoingStoppedQueueTypingEvents.remove(conversationId)
+                sendTypingIndicatorStatus(conversationId, Conversation.TypingIndicatorMode.STOPPED)
+            }
+        }
+    }
+
+    /**
+     * Sends a started event and enqueues a stopping event if sent successfully.
+     */
+    override fun sendStartedAndEnqueueStoppingEvent(conversationId: ConversationId) {
+        launch {
+            outgoingStoppedQueueTypingEventsMutex.withLock {
+                if (outgoingStoppedQueueTypingEvents.containsKey(conversationId)) {
+                    return@launch
+                }
+                val isSent = sendTypingIndicatorStatus(conversationId, Conversation.TypingIndicatorMode.STARTED)
+                when (isSent) {
+                    true -> {
+                        outgoingStoppedQueueTypingEvents[conversationId] = Unit
+                        delay(typingIndicatorTimeoutInSeconds)
+                        sendStoppingEvent(conversationId)
+                    }
+
+                    false -> Unit // do nothing
+                }
+            }
+        }
+    }
+
+    private suspend fun sendTypingIndicatorStatus(
+        conversationId: ConversationId,
+        typingStatus: Conversation.TypingIndicatorMode
+    ): Boolean = conversationRepository.sendTypingIndicatorStatus(conversationId, typingStatus).fold({
+        kaliumLogger.w("Skipping failed to send typing indicator status: $typingStatus")
+        false
+    }) {
+        kaliumLogger.i("Successfully sent typing started indicator status: $typingStatus")
+        true
+    }
+}

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

@@ -0,0 +1,33 @@
+/*
+ * 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.data.conversation
+
+import com.wire.kalium.network.api.base.authenticated.conversation.TypingIndicatorStatus
+import com.wire.kalium.network.api.base.authenticated.conversation.TypingIndicatorStatusDTO
+
+fun TypingIndicatorStatus.toModel(): Conversation.TypingIndicatorMode = when (this) {
+    TypingIndicatorStatus.STARTED -> Conversation.TypingIndicatorMode.STARTED
+    TypingIndicatorStatus.STOPPED -> Conversation.TypingIndicatorMode.STOPPED
+}
+
+fun Conversation.TypingIndicatorMode.toApi(): TypingIndicatorStatus = when (this) {
+    Conversation.TypingIndicatorMode.STARTED -> TypingIndicatorStatus.STARTED
+    Conversation.TypingIndicatorMode.STOPPED -> TypingIndicatorStatus.STOPPED
+}
+
+fun Conversation.TypingIndicatorMode.toStatusDto(): TypingIndicatorStatusDTO = TypingIndicatorStatusDTO(this.toApi())

logic/src/commonMain/kotlin/com/wire/kalium/logic/data/event/EventMapper.kt

@@ -27,14 +27,14 @@ import com.wire.kalium.logic.data.conversation.ConversationRoleMapper
 import com.wire.kalium.logic.data.conversation.MemberMapper
 import com.wire.kalium.logic.data.conversation.MutedConversationStatus
 import com.wire.kalium.logic.data.conversation.ReceiptModeMapper
+import com.wire.kalium.logic.data.conversation.toModel
 import com.wire.kalium.logic.data.event.Event.UserProperty.ReadReceiptModeSet
 import com.wire.kalium.logic.data.event.Event.UserProperty.TypingIndicatorModeSet
 import com.wire.kalium.logic.data.featureConfig.FeatureConfigMapper
 import com.wire.kalium.logic.data.id.SubconversationId
 import com.wire.kalium.logic.data.id.toModel
 import com.wire.kalium.logic.di.MapperProvider
 import com.wire.kalium.logic.util.Base64
-import com.wire.kalium.network.api.base.authenticated.conversation.TypingIndicatorStatus
 import com.wire.kalium.network.api.base.authenticated.featureConfigs.FeatureConfigData
 import com.wire.kalium.network.api.base.authenticated.notification.EventContentDTO
 import com.wire.kalium.network.api.base.authenticated.notification.EventResponse
@@ -114,10 +114,7 @@ class EventMapper(
             transient,
             eventContentDTO.qualifiedFrom.toModel(),
             eventContentDTO.time,
-            when (eventContentDTO.status.status) {
-                TypingIndicatorStatus.STARTED -> Conversation.TypingIndicatorMode.STARTED
-                TypingIndicatorStatus.STOPPED -> Conversation.TypingIndicatorMode.STOPPED
-            }
+            eventContentDTO.status.status.toModel()
         )
 
     private fun federationTerminated(id: String, eventContentDTO: EventContentDTO.Federation, transient: Boolean): Event =
@@ -234,6 +231,7 @@ class EventMapper(
                     )
                 }
             }
+
             else -> unknown(
                 id = id,
                 transient = transient,