fix(chat): do not send a read marker below the user's own message

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

@@ -79,6 +79,14 @@ interface ChatMessageRepository : LifecycleAwareManager {
      */
     suspend fun updateLocalReadState(lastReadMessage: Int)
 
+    /**
+     * Like [updateLocalReadState], but for a marker that cannot be sent yet (an own message is still on its
+     * way): only the unread count of the conversation entry is lowered, the stored read marker stays the
+     * one the server knows and no pending marker is registered. So the server stays the authority for the
+     * read state, and the marker is still sent as soon as the own message is confirmed or failed.
+     */
+    suspend fun updateLocalUnreadCountOnly(lastReadMessage: Int)
+
     /**
      * Writes the local read state for a message the user marked as unread: the read marker moves back
      * to [lastReadMessage] and the conversation keeps at least one unread message until the server

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

@@ -430,6 +430,10 @@ class OfflineFirstChatRepository @Inject constructor(
         conversationListUpdater.updateLocalReadState(syncTarget, lastReadMessage)
     }
 
+    override suspend fun updateLocalUnreadCountOnly(lastReadMessage: Int) {
+        conversationListUpdater.updateLocalUnreadCountOnly(syncTarget, lastReadMessage)
+    }
+
     override suspend fun updateLocalUnreadState(lastReadMessage: Int) {
         conversationListUpdater.updateLocalUnreadState(syncTarget, lastReadMessage)
     }

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

@@ -44,6 +44,7 @@ import com.nextcloud.talk.conversationlist.viewmodels.ConversationsListViewModel
 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.database.model.SendStatus
 import com.nextcloud.talk.data.user.model.User
 import com.nextcloud.talk.jobs.ReadMarkerSyncWorker
 import com.nextcloud.talk.jobs.SendMessageWorker
@@ -362,6 +363,12 @@ class ChatViewModel @AssistedInject constructor(
 
     private var localLastReadMessage: Int = 0
 
+    // Newest message list from the database, updated before the throttled rebuild of the UI items. It is the
+    // source of truth for the user's own sends (temporary copy with its send status, then the accepted
+    // message) when the read marker is decided, see updateRemoteLastReadMessageIfNeeded.
+    @Volatile
+    private var latestMessages: List<ChatMessage> = emptyList()
+
     /**
      * Set once the user marked a message as unread, so the automatic read marker handling keeps its
      * hands off the conversation for the rest of this visit.
@@ -1308,7 +1315,9 @@ class ChatViewModel @AssistedInject constructor(
         // getLastCommonReadFlow provides the "last read by all" value for read-receipt checkmarks.
         // These are two different concepts that must not be conflated.
         combine(
-            messagesFlow,
+            // the raw message list is tapped before the throttle and the UI mapping below: the read marker
+            // decisions must not depend on how far the rendered items lag behind the database
+            messagesFlow.onEach(::onMessagesChanged),
             getLastCommonReadFlow.onStart { emit(0) },
             parentMessagesFlow,
             combine(conversationFlow, _spreedCapabilities) { conv, caps -> conv to caps },
@@ -2095,6 +2104,19 @@ class ChatViewModel @AssistedInject constructor(
         }
     }
 
+    // Runs on the background dispatcher of the rebuild pipeline (flowOn in observeMessages): it only
+    // publishes the immutable list through a volatile field. localLastReadMessage is written on the main
+    // thread only, so the own message is combined with it when the marker is decided (readMarkerToSend).
+    //
+    // Only the list of the newest messages counts. A window around a quoted or searched message (search and
+    // old chat block mode) can lie far from them and miss the own message just sent, which must still be
+    // known when the chat is left from there.
+    private fun onMessagesChanged(messages: List<ChatMessage>) {
+        if (chatMode.value == ChatMode.DEFAULT_MODE || contextAnchorMessageId.value == null) {
+            latestMessages = messages
+        }
+    }
+
     fun advanceLocalLastReadMessageIfNeeded(messageId: Int) {
         Log.d(TAG, "advanceLocalLastReadMessageIfNeeded, messageId: $messageId")
         Log.d(TAG, "advanceLocalLastReadMessageIfNeeded, localLastReadMessage: $localLastReadMessage")
@@ -2122,10 +2144,25 @@ class ChatViewModel @AssistedInject constructor(
         val conversationLastReadMessage = _uiState.value.conversation?.lastReadMessage ?: return
         Log.d(TAG, "updateRemoteLastReadMessageIfNeeded, conversation.lastReadMessage: $conversationLastReadMessage")
 
-        if (localLastReadMessage > conversationLastReadMessage) {
-            Log.d(TAG, "updateRemoteLastReadMessageIfNeeded, setChatReadMessage...")
+        val messages = latestMessages
+        val newestOwnMessageId = newestOwnMessageId(messages)
+        val marker = localReadMarker(
+            localLastReadMessage = localLastReadMessage,
+            conversationLastReadMessage = conversationLastReadMessage,
+            newestOwnMessageId = newestOwnMessageId
+        )
+        if (marker != null) {
+            // The local read state is always written; only the server request waits for an own message
+            // that is still on its way (offline queue included), see readMarkerToSend.
+            val sendToServer = readMarkerToSend(
+                localLastReadMessage = localLastReadMessage,
+                conversationLastReadMessage = conversationLastReadMessage,
+                newestOwnMessageId = newestOwnMessageId,
+                hasUnconfirmedOwnMessage = hasUnconfirmedOwnMessage(messages)
+            ) != null
+            Log.d(TAG, "updateRemoteLastReadMessageIfNeeded, setChatReadMessage, sendToServer: $sendToServer")
 
-            setChatReadMessage(localLastReadMessage)
+            setChatReadMessage(marker, sendToServer)
         }
     }
 
@@ -2168,11 +2205,20 @@ class ChatViewModel @AssistedInject constructor(
         )
     }
 
-    fun setChatReadMessage(lastReadMessage: Int) {
+    fun setChatReadMessage(lastReadMessage: Int, sendToServer: Boolean = true) {
         // marking as read is the explicit counterpart of marking as unread and hands the read marker
         // back to the automatic handling
         keepMarkedAsUnread = false
         _uiState.update { it.copy(markedAsUnreadByUser = false) }
+        if (!sendToServer) {
+            // The marker waits for an own message that is on its way: show the lower unread count, but keep
+            // the stored marker and register nothing pending, so the marker is decided again (and sent) by
+            // the next call once the message is confirmed or failed.
+            viewModelScope.launch {
+                chatRepository.updateLocalUnreadCountOnly(lastReadMessage)
+            }
+            return
+        }
         chatRepository.markPendingReadMarker(lastReadMessage)
         viewModelScope.launch {
             chatRepository.updateLocalReadState(lastReadMessage)
@@ -2804,6 +2850,58 @@ class ChatViewModel @AssistedInject constructor(
         fun holdsShare(messages: List<ChatMessageDto>?, referenceId: String): Boolean? =
             messages?.any { it.referenceId == referenceId }
 
+        /**
+         * Id of the newest message sent by the user that the server has already accepted, or null; the
+         * server marks everything up to it as read. Temporary messages (negative ids) are not accepted yet
+         * and are ignored.
+         */
+        internal fun newestOwnMessageId(messages: List<ChatMessage>): Int? =
+            messages.filter { !it.incoming && it.jsonMessageId > 0 }
+                .maxOfOrNull { it.jsonMessageId }
+
+        /**
+         * True while a text message of the user is on its way to the server: its temporary copy is stored
+         * and not failed, so the server may have accepted it (and moved the user's read marker to it)
+         * already, while its server copy is not in [messages] yet. This client must not send an older
+         * marker over it. A file placeholder is not counted before its upload is finished: the server has
+         * no message for it until then.
+         */
+        internal fun hasUnconfirmedOwnMessage(messages: List<ChatMessage>): Boolean =
+            messages.any { message ->
+                val isFileUploading = message.sendStatus != SendStatus.SENT_PENDING_ACK &&
+                    message.messageParameters.orEmpty().containsKey("file")
+                !message.incoming &&
+                    message.isTemporary &&
+                    message.sendStatus != SendStatus.FAILED &&
+                    !isFileUploading
+            }
+
+        /**
+         * The read marker to send to the server, or null when nothing has to be sent. The server sets the
+         * marker to a message at the moment the user sends it, so the client never sends a marker
+         * below the newest own message: that would move the marker back and leave the message with a
+         * single check mark for the other participants.
+         */
+        internal fun readMarkerToSend(
+            localLastReadMessage: Int,
+            conversationLastReadMessage: Int,
+            newestOwnMessageId: Int?,
+            hasUnconfirmedOwnMessage: Boolean
+        ): Int? =
+            localReadMarker(localLastReadMessage, conversationLastReadMessage, newestOwnMessageId)
+                .takeUnless { hasUnconfirmedOwnMessage }
+
+        /**
+         * The read marker to write into the local conversation state, or null when it has it already.
+         * Unlike [readMarkerToSend] it does not wait for an own message that is on its way: the local state
+         * is not sent anywhere, so the conversation list loses its unread count right away, also offline.
+         */
+        internal fun localReadMarker(
+            localLastReadMessage: Int,
+            conversationLastReadMessage: Int,
+            newestOwnMessageId: Int?
+        ): Int? = maxOf(localLastReadMessage, newestOwnMessageId ?: 0).takeIf { it > conversationLastReadMessage }
+
         /**
          * Ids of the messages a delete of [messageId] applies to: every already-uploaded message of the
          * media group the message is shown in, or just the message itself.

app/src/main/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdater.kt

@@ -160,6 +160,25 @@ class ConversationListUpdater @Inject constructor(
         Log.d(TAG, "Local read state for room ${target.roomToken}: lastRead=$lastReadMessage, unread=$unreadMessages")
     }
 
+    /**
+     * Lowers only the unread count of the conversation entry to what lies above [lastReadMessage]. The stored
+     * read marker stays and no pending marker is registered: the marker has not been sent, so the server
+     * state remains the authority and a room list sync may restore it. Used while an own message is on
+     * its way and the marker has to wait for it.
+     */
+    suspend fun updateLocalUnreadCountOnly(target: ChatMessageSyncer.SyncTarget, lastReadMessage: Int) {
+        val conversation =
+            conversationsDao.getConversationForUser(target.accountId, target.roomToken).first() ?: return
+        val unreadMessages = chatDao.countMessagesNewerThan(
+            internalConversationId = target.internalConversationId,
+            messageId = lastReadMessage.toLong(),
+            excludedActorId = target.user.userId!!
+        )
+        // only the count: the marker read above must not be written back over one stored while counting
+        conversationsDao.updateUnreadMessages(conversation.internalId, unreadMessages)
+        Log.d(TAG, "Local unread count for room ${target.roomToken}: unread=$unreadMessages (marker not sent yet)")
+    }
+
     /**
      * Writes the read state for a message the user marked as unread: the read marker moves back to
      * [lastReadMessage] and the conversation is left with at least one unread message, so the entry

app/src/main/java/com/nextcloud/talk/data/database/dao/ConversationsDao.kt

@@ -104,6 +104,15 @@ interface ConversationsDao {
     )
     suspend fun updateReadState(internalId: String, lastReadMessage: Int, unreadMessages: Int)
 
+    @Query(
+        """
+        UPDATE Conversations
+        SET unreadMessages = :unreadMessages
+        WHERE internalId = :internalId
+        """
+    )
+    suspend fun updateUnreadMessages(internalId: String, unreadMessages: Int)
+
     /**
      * Deletes rows in the db matching the specified [conversationIds]
      */

app/src/main/java/com/nextcloud/talk/utils/preview/ComposePreviewUtilsDaos.kt

@@ -320,6 +320,10 @@ class DummyConversationDaoImpl : ConversationsDao {
         /* */
     }
 
+    override suspend fun updateUnreadMessages(internalId: String, unreadMessages: Int) {
+        /* */
+    }
+
     override fun insertConversation(conversation: ConversationEntity) {
         /* */
     }