Send text messages by worker

app/src/main/java/com/nextcloud/talk/chat/ChatActivity.kt

@@ -1030,11 +1030,7 @@ class ChatActivity :
                             showEdit = sendingFailed || !isOnline,
                             showDelete = sendingFailed || !isOnline,
                             onResend = {
-                                chatViewModel.resendMessage(
-                                    conversationUser!!.getCredentials(),
-                                    ApiUtils.getUrlForChat(chatApiVersion, conversationUser!!.baseUrl!!, roomToken),
-                                    msg
-                                )
+                                chatViewModel.resendMessage(msg)
                             },
                             onEdit = { messageInputViewModel.edit(msg) },
                             onDelete = { chatViewModel.deleteTempMessage(msg) },

app/src/main/java/com/nextcloud/talk/chat/MessageInputFragment.kt

@@ -1049,12 +1049,8 @@ class MessageInputFragment : Fragment() {
         chatActivity.chatViewModel.onMessageSent()
 
         messageInputViewModel.sendChatMessage(
-            credentials = chatActivity.conversationUser!!.getCredentials(),
-            url = ApiUtils.getUrlForChat(
-                chatActivity.chatApiVersion,
-                chatActivity.conversationUser!!.baseUrl!!,
-                chatActivity.roomToken
-            ),
+            userId = chatActivity.conversationUser!!.id!!,
+            roomToken = chatActivity.roomToken,
             message = message,
             displayName = chatActivity.conversationUser!!.displayName ?: "",
             replyTo = chatActivity.getReplyToMessageId(),

app/src/main/java/com/nextcloud/talk/chat/data/ChatMessageRepository.kt

@@ -138,16 +138,11 @@ interface ChatMessageRepository : LifecycleAwareManager {
         threadTitle: String?
     ): Flow<Result<ChatMessage?>>
 
-    @Suppress("LongParameterList")
-    suspend fun resendChatMessage(
-        credentials: String,
-        url: String,
-        message: String,
-        displayName: String,
-        replyTo: Int,
-        sendWithoutNotification: Boolean,
-        referenceId: String
-    ): Flow<Result<ChatMessage?>>
+    /**
+     * Resets a previously failed temporary message back to PENDING so it can be handed to
+     * SendMessageWorker for another send attempt. Does not itself send anything.
+     */
+    suspend fun markMessageForResend(referenceId: String): Flow<Result<ChatMessage?>>
 
     suspend fun addTemporaryMessage(
         message: CharSequence,

app/src/main/java/com/nextcloud/talk/chat/data/network/OfflineFirstChatRepository.kt

@@ -32,6 +32,7 @@ import com.nextcloud.talk.models.json.generic.GenericOverall
 import com.nextcloud.talk.models.json.participants.Participant
 import com.nextcloud.talk.utils.bundle.BundleKeys
 import com.nextcloud.talk.utils.message.SendMessageUtils
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
@@ -656,16 +657,7 @@ class OfflineFirstChatRepository @Inject constructor(
             }
     }
 
-    @Suppress("LongParameterList")
-    override suspend fun resendChatMessage(
-        credentials: String,
-        url: String,
-        message: String,
-        displayName: String,
-        replyTo: Int,
-        sendWithoutNotification: Boolean,
-        referenceId: String
-    ): Flow<Result<ChatMessage?>> {
+    override suspend fun markMessageForResend(referenceId: String): Flow<Result<ChatMessage?>> {
         val messageToResend = chatDao.getTempMessageForConversation(
             internalConversationId,
             referenceId,
@@ -678,16 +670,9 @@ class OfflineFirstChatRepository @Inject constructor(
             val messageToResendModel = messageToResend.toDomainModel()
             _updateMessageFlow.emit(messageToResendModel)
 
-            sendChatMessage(
-                credentials = credentials,
-                url = url,
-                message = message,
-                displayName = displayName,
-                replyTo = replyTo,
-                sendWithoutNotification = sendWithoutNotification,
-                referenceId = referenceId,
-                threadTitle = null
-            )
+            flow {
+                emit(Result.success(messageToResendModel))
+            }
         } else {
             flow {
                 emit(Result.failure(IllegalStateException("No temporary message found to resend")))
@@ -713,6 +698,12 @@ class OfflineFirstChatRepository @Inject constructor(
                     referenceId
                 )
                 chatDao.upsertChatMessage(tempChatMessageEntity)
+                emit(Result.success(tempChatMessageEntity.toDomainModel()))
+            } catch (e: CancellationException) {
+                // a collector (e.g. first()/take(1)) is done with the flow, not a real failure -
+                // rethrow instead of turning it into a Result.failure emission, which would violate
+                // flow exception transparency since the collector already stopped listening
+                throw e
             } catch (e: Exception) {
                 Log.e(TAG, "Something went wrong when adding temporary message", e)
                 emit(Result.failure(e))

app/src/main/java/com/nextcloud/talk/chat/viewmodels/ChatViewModel.kt

@@ -41,8 +41,8 @@ import com.nextcloud.talk.dagger.modules.ApplicationScope
 import com.nextcloud.talk.data.database.mappers.toDomainModel
 import com.nextcloud.talk.data.database.model.ChatMessageEntity
 import com.nextcloud.talk.data.user.model.User
-import com.nextcloud.talk.extensions.toIntOrZero
 import com.nextcloud.talk.jobs.ReadMarkerSyncWorker
+import com.nextcloud.talk.jobs.SendMessageWorker
 import com.nextcloud.talk.jobs.ShareOperationWorker
 import androidx.lifecycle.asFlow
 import androidx.work.WorkManager
@@ -2559,19 +2559,17 @@ class ChatViewModel @AssistedInject constructor(
         }
     }
 
-    fun resendMessage(credentials: String, urlForChat: String, message: ChatMessage) {
+    fun resendMessage(message: ChatMessage) {
+        val referenceId = message.referenceId.orEmpty()
         viewModelScope.launch {
-            chatRepository.resendChatMessage(
-                credentials,
-                urlForChat,
-                message.message.orEmpty(),
-                message.actorDisplayName.orEmpty(),
-                message.parentMessageId?.toIntOrZero() ?: 0,
-                false,
-                message.referenceId.orEmpty()
-            ).collect { result ->
+            chatRepository.markMessageForResend(referenceId).collect { result ->
                 if (result.isSuccess) {
-                    Log.d(TAG, "resend successful")
+                    Log.d(TAG, "message marked pending for resend")
+                    SendMessageWorker.enqueue(
+                        internalConversationId = "${currentUser.id}@$chatRoomToken",
+                        referenceId = referenceId,
+                        threadTitle = null
+                    )
                 } else {
                     Log.e(TAG, "resend failed")
                 }

app/src/main/java/com/nextcloud/talk/chat/viewmodels/MessageInputViewModel.kt

@@ -23,6 +23,7 @@ import com.nextcloud.talk.chat.data.io.AudioRecorderManager
 import com.nextcloud.talk.chat.data.io.MediaPlayerManager
 import com.nextcloud.talk.chat.data.model.ChatMessage
 import com.nextcloud.talk.chat.data.network.ChatNetworkDataSource
+import com.nextcloud.talk.jobs.SendMessageWorker
 import com.nextcloud.talk.models.MessageDraft
 import com.nextcloud.talk.models.json.chat.ChatOverallSingleMessage
 import com.nextcloud.talk.models.json.chat.ChatUtils
@@ -33,6 +34,8 @@ import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.update
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
 import javax.inject.Inject
 
 @Suppress("Detekt.TooManyFunctions")
@@ -77,6 +80,13 @@ class MessageInputViewModel :
     lateinit var currentLifeCycleFlag: LifeCycleFlag
     val disposableSet = mutableSetOf<Disposable>()
 
+    // Serializes addTemporaryMessage()+SendMessageWorker.enqueue() across sendChatMessage() calls.
+    // Without it, several messages sent in quick succession each await their own independent local
+    // DB write before enqueueing, and those writes aren't guaranteed to finish in the order they
+    // were started - so the worker chain (and therefore the order messages actually reach the
+    // server) could end up scrambled relative to the order the user sent them.
+    private val sendMessageMutex = Mutex()
+
     fun setData(chatMessageRepository: ChatMessageRepository) {
         chatRepository = chatMessageRepository
     }
@@ -164,8 +174,8 @@ class MessageInputViewModel :
 
     @Suppress("LongParameterList")
     fun sendChatMessage(
-        credentials: String,
-        url: String,
+        userId: Long,
+        roomToken: String,
         message: String,
         displayName: String,
         replyTo: Int,
@@ -176,40 +186,28 @@ class MessageInputViewModel :
         Log.d(TAG, "Random SHA-256 Hash: $referenceId")
 
         viewModelScope.launch {
-            chatRepository.addTemporaryMessage(
-                message,
-                displayName,
-                replyTo,
-                sendWithoutNotification,
-                referenceId
-            ).collect { result ->
-                if (result.isSuccess) {
-                    Log.d(TAG, "temp message ref id: " + (result.getOrNull()?.referenceId ?: "none"))
-
-                    _sendChatMessageViewState.value = SendChatMessageSuccessState(message)
-                } else {
-                    _sendChatMessageViewState.value = SendChatMessageErrorState(message)
-                }
-            }
-        }
-
-        viewModelScope.launch {
-            chatRepository.sendChatMessage(
-                credentials,
-                url,
-                message,
-                displayName,
-                replyTo,
-                sendWithoutNotification,
-                referenceId,
-                threadTitle
-            ).collect { result ->
-                if (result.isSuccess) {
-                    Log.d(TAG, "received ref id: " + (result.getOrNull()?.referenceId ?: "none"))
-
-                    _sendChatMessageViewState.value = SendChatMessageSuccessState(message)
-                } else {
-                    _sendChatMessageViewState.value = SendChatMessageErrorState(message)
+            // Holding the lock across the DB write and the enqueue call, rather than just around
+            // the enqueue call, is what actually guarantees ordering: it forces this whole
+            // insert-then-enqueue step for one message to finish before the next queued send is
+            // even allowed to start its own DB write.
+            sendMessageMutex.withLock {
+                chatRepository.addTemporaryMessage(
+                    message,
+                    displayName,
+                    replyTo,
+                    sendWithoutNotification,
+                    referenceId
+                ).collect { result ->
+                    if (result.isSuccess) {
+                        _sendChatMessageViewState.value = SendChatMessageSuccessState(message)
+                        SendMessageWorker.enqueue(
+                            internalConversationId = "$userId@$roomToken",
+                            referenceId = referenceId,
+                            threadTitle = threadTitle
+                        )
+                    } else {
+                        _sendChatMessageViewState.value = SendChatMessageErrorState(message)
+                    }
                 }
             }
         }