fix(chat): keep X-Chat-Last-Common-Read from the read marker response

app/src/main/java/com/nextcloud/talk/api/NcApi.java

@@ -501,9 +501,9 @@ Observable<GenericOverall> notificationCalls(@Header("Authorization") String aut
     // Url is: /api/{apiVersion}/chat/{token}/read
     @FormUrlEncoded
     @POST
-    Observable<GenericOverall> setChatReadMarker(@Header("Authorization") String authorization,
-                                                 @Url String url,
-                                                 @Nullable @Field("lastReadMessage") Integer lastReadMessage);
+    Observable<Response<GenericOverall>> setChatReadMarker(@Header("Authorization") String authorization,
+                                                           @Url String url,
+                                                           @Nullable @Field("lastReadMessage") Integer lastReadMessage);
 
     // Url is: /api/{apiVersion}/chat/{token}/read
     @DELETE

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

@@ -78,7 +78,11 @@ interface ChatNetworkDataSource {
 
     suspend fun deleteChatMessage(credentials: String, url: String): ChatOverallSingleMessage
     fun createRoom(credentials: String, url: String, map: Map<String, String>): Observable<RoomOverall>
-    fun setChatReadMarker(credentials: String, url: String, previousMessageId: Int): Observable<GenericOverall>
+    fun setChatReadMarker(
+        credentials: String,
+        url: String,
+        previousMessageId: Int
+    ): Observable<Response<GenericOverall>>
     suspend fun editChatMessage(credentials: String, url: String, text: String): ChatOverallSingleMessage
     suspend fun getOutOfOfficeStatusForUser(credentials: String, baseUrl: String, userId: String): UserAbsenceOverall
     suspend fun getUpcomingEvents(credentials: String, baseUrl: String, roomToken: String): UpcomingEventsOverall

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

@@ -156,7 +156,7 @@ class RetrofitChatNetwork(private val ncApi: NcApi, private val ncApiCoroutines:
         credentials: String,
         url: String,
         previousMessageId: Int
-    ): Observable<GenericOverall> = ncApi.setChatReadMarker(credentials, url, previousMessageId).map { it }
+    ): Observable<Response<GenericOverall>> = ncApi.setChatReadMarker(credentials, url, previousMessageId)
 
     override suspend fun editChatMessage(credentials: String, url: String, text: String): ChatOverallSingleMessage =
         ncApiCoroutines.editChatMessage(credentials, url, text)

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

@@ -17,6 +17,8 @@ import com.nextcloud.talk.data.database.model.ConversationEntity
 import com.nextcloud.talk.utils.Undo
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
 import kotlinx.coroutines.withContext
 import java.util.concurrent.ConcurrentHashMap
 import javax.inject.Inject
@@ -49,6 +51,24 @@ class ConversationListUpdater @Inject constructor(
      */
     private val pendingReadMarkers = ConcurrentHashMap<String, Int>()
 
+    /**
+     * The last message read by all participants as the server reported it in the answer to a read
+     * marker, with the time that answer arrived. A room list response whose request started before
+     * that moment can carry an older value, so the merge keeps this one against it.
+     */
+    private data class CommonReadWrite(val lastCommonRead: Int, val writtenAtNanos: Long)
+
+    private val commonReadWrites = ConcurrentHashMap<String, CommonReadWrite>()
+
+    /**
+     * Held by a sync (or single room refresh) from the merge through the write of its response, and by
+     * [updateLastCommonRead] for its record and write. Without it a read marker answer arriving
+     * between the merge and the write is not seen by the merge and then overwritten by the write.
+     * With it, either the answer is recorded before the merge and the merge keeps it, or its write
+     * comes after the sync's write.
+     */
+    val syncWriteLock = Mutex()
+
     /**
      * Favorite flags applied locally but not yet confirmed by a server response; guarded in the
      * merge like the pending read markers.
@@ -160,6 +180,18 @@ class ConversationListUpdater @Inject constructor(
         Log.d(TAG, "Local read state for room ${target.roomToken}: lastRead=$lastReadMessage, unread=$unreadMessages")
     }
 
+    /**
+     * Stores the last message read by all participants, which the server reports in the response to a
+     * read marker, so that the read check marks of the user's own messages are up to date without a
+     * further request.
+     */
+    suspend fun updateLastCommonRead(internalConversationId: String, lastCommonRead: Int) {
+        syncWriteLock.withLock {
+            commonReadWrites[internalConversationId] = CommonReadWrite(lastCommonRead, System.nanoTime())
+            conversationsDao.updateLastCommonRead(internalConversationId, lastCommonRead)
+        }
+    }
+
     /**
      * 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
@@ -285,10 +317,15 @@ class ConversationListUpdater @Inject constructor(
      * the local state is kept; once the server confirms it (or moved past it, e.g. read further
      * or unfavorited on another device after confirmation) the pending change is released and
      * the server state applies unchanged, so the server stays the authority.
+     *
+     * [requestedAtNanos] is the [System.nanoTime] at which the request that produced
+     * [conversationsFromServer] was started; it tells whether such a response can predate the last
+     * message read by all participants that a read marker answer reported.
      */
     fun preservePendingLocalState(
         previousConversations: Map<String, ConversationEntity>,
-        conversationsFromServer: List<ConversationEntity>
+        conversationsFromServer: List<ConversationEntity>,
+        requestedAtNanos: Long
     ): List<ConversationEntity> =
         conversationsFromServer.map { serverItem ->
             val previous = previousConversations[serverItem.internalId]
@@ -300,8 +337,25 @@ class ConversationListUpdater @Inject constructor(
             guarded = guardPendingHiddenPinnedId(guarded, previous)
             guarded = guardPendingArchived(guarded, previous)
             guarded = guardPendingTags(guarded, previous)
-            guarded
+            guardLastCommonRead(guarded, requestedAtNanos)
+        }
+
+    /**
+     * The server only moves the last message read by all participants back when someone marks a
+     * message as unread, so a response that is older than the value taken from a read marker answer
+     * must not lower it. A response whose request started after that answer arrived is current and
+     * applies unchanged, which also ends the guard for the room - nothing stays pinned.
+     */
+    private fun guardLastCommonRead(serverItem: ConversationEntity, requestedAtNanos: Long): ConversationEntity {
+        val write = commonReadWrites[serverItem.internalId] ?: return serverItem
+
+        return if (requestedAtNanos - write.writtenAtNanos < 0) {
+            serverItem.copy(lastCommonReadMessage = maxOf(serverItem.lastCommonReadMessage, write.lastCommonRead))
+        } else {
+            commonReadWrites.remove(serverItem.internalId, write)
+            serverItem
         }
+    }
 
     private fun guardPendingReadMarker(
         serverItem: ConversationEntity,

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

@@ -142,19 +142,23 @@ class OfflineFirstConversationsRepository @Inject constructor(
     override fun getRoom(user: User, roomToken: String): Job =
         scope.launch {
             try {
+                val requestedAtNanos = System.nanoTime()
                 val model = chatNetworkDataSource.getRoom(user, roomToken)
                 val existingEntity = dao.getConversationForUser(user.id!!, model.token).first()
                 model.hiddenUpcomingEvent = existingEntity?.hiddenUpcomingEvent
                 _conversationFlow.emit(model)
                 val previous = existingEntity?.let { mapOf(it.internalId to it) }.orEmpty()
-                val entityList = conversationListUpdater.preservePendingLocalState(
-                    previous,
-                    listOf(model.asEntity())
-                )
-                try {
-                    dao.upsertConversations(user.id!!, entityList)
-                } catch (e: SQLiteConstraintException) {
-                    Log.w(TAG, "Skipping conversation upsert for removed account ${user.id}", e)
+                conversationListUpdater.syncWriteLock.withLock {
+                    val entityList = conversationListUpdater.preservePendingLocalState(
+                        previous,
+                        listOf(model.asEntity()),
+                        requestedAtNanos
+                    )
+                    try {
+                        dao.upsertConversations(user.id!!, entityList)
+                    } catch (e: SQLiteConstraintException) {
+                        Log.w(TAG, "Skipping conversation upsert for removed account ${user.id}", e)
+                    }
                 }
             } catch (e: Exception) {
                 // In case network is offline, the call fails, or getRoom can't resolve a supported
@@ -254,6 +258,7 @@ class OfflineFirstConversationsRepository @Inject constructor(
         // not something the response can be read as having left out
         val knownBeforeRequest = dao.getConversationIdsForUser(accountId).toSet()
 
+        val requestedAtNanos = System.nanoTime()
         val roomList = withRetry(
             retries = NETWORK_FETCH_RETRIES,
             initialDelayMillis = NETWORK_FETCH_RETRY_INITIAL_DELAY_MS,
@@ -279,19 +284,23 @@ class OfflineFirstConversationsRepository @Inject constructor(
             keepCachedStatus(conversationsFromSync, previousConversations)
         }
 
-        dao.syncConversationsForUser(
-            accountId = accountId,
-            serverItems = conversationListUpdater.preservePendingLocalState(
-                previousConversations,
-                serverItems
-            ),
-            conversationIdsToDelete = if (roomList.wasDelta) {
-                emptyList()
-            } else {
-                determineLeftConversationIds(previousConversations, conversationsFromSync)
-                    .filter { it in knownBeforeRequest }
-            }
-        )
+        // merge and write as one step with respect to a last common read arriving from a read marker
+        conversationListUpdater.syncWriteLock.withLock {
+            dao.syncConversationsForUser(
+                accountId = accountId,
+                serverItems = conversationListUpdater.preservePendingLocalState(
+                    previousConversations,
+                    serverItems,
+                    requestedAtNanos
+                ),
+                conversationIdsToDelete = if (roomList.wasDelta) {
+                    emptyList()
+                } else {
+                    determineLeftConversationIds(previousConversations, conversationsFromSync)
+                        .filter { it in knownBeforeRequest }
+                }
+            )
+        }
 
         rememberSyncedState(accountId, roomList)
 

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

@@ -104,6 +104,20 @@ interface ConversationsDao {
     )
     suspend fun updateReadState(internalId: String, lastReadMessage: Int, unreadMessages: Int)
 
+    /**
+     * Writes the last message read by all participants, as reported by the server in the
+     * X-Chat-Last-Common-Read header of the response to a read marker.
+     * The value can be lower than the stored one: marking a chat as unread lowers it on the server.
+     */
+    @Query(
+        """
+        UPDATE Conversations
+        SET lastCommonReadMessage = :lastCommonReadMessage
+        WHERE internalId = :internalId
+        """
+    )
+    suspend fun updateLastCommonRead(internalId: String, lastCommonReadMessage: Int)
+
     /**
      * Deletes rows in the db matching the specified [conversationIds]
      */

app/src/main/java/com/nextcloud/talk/jobs/ReadMarkerSyncWorker.kt

@@ -27,6 +27,8 @@ import com.nextcloud.talk.users.UserManager
 import com.nextcloud.talk.utils.ApiUtils
 import com.nextcloud.talk.utils.bundle.BundleKeys.KEY_INTERNAL_USER_ID
 import com.nextcloud.talk.utils.bundle.BundleKeys.KEY_ROOM_TOKEN
+import kotlinx.coroutines.CancellationException
+import retrofit2.Response
 import java.util.concurrent.TimeUnit
 import javax.inject.Inject
 
@@ -57,7 +59,10 @@ class ReadMarkerSyncWorker(context: Context, workerParams: WorkerParameters) :
 
     override suspend fun doWork(): Result {
         sharedApplication!!.componentApplication.inject(this)
+        return send()
+    }
 
+    internal suspend fun send(): Result {
         val userId = inputData.getLong(KEY_INTERNAL_USER_ID, -1)
         val roomToken = inputData.getString(KEY_ROOM_TOKEN)
         val lastReadMessage = inputData.getInt(KEY_LAST_READ_MESSAGE, -1)
@@ -80,26 +85,44 @@ class ReadMarkerSyncWorker(context: Context, workerParams: WorkerParameters) :
             return fail(userId, roomToken, lastReadMessage)
         }
 
-        val sent = runCatching {
+        val response = runCatching {
             val url = ApiUtils.getUrlForChatReadMarker(
                 ApiUtils.getChatApiVersion(user.capabilities!!.spreedCapability!!, intArrayOf(ApiUtils.API_V1)),
                 user.baseUrl!!,
                 roomToken
             )
             chatNetworkDataSource.setChatReadMarker(credentials, url, lastReadMessage).blockingSingle()
+        }.mapCatching { httpResponse ->
+            check(httpResponse.isSuccessful) { "HTTP ${httpResponse.code()}" }
+            httpResponse
         }.onFailure { throwable ->
             Log.w(TAG, "Read marker $lastReadMessage for room $roomToken could not be sent: $throwable")
-        }.isSuccess
+        }.getOrNull()
 
-        return if (sent) {
+        return if (response != null) {
             Log.d(TAG, "Read marker $lastReadMessage sent for room $roomToken")
+            storeLastCommonRead(userId, roomToken, response)
             Result.success()
         } else {
             Log.w(TAG, "Sending read marker for room $roomToken failed (attempt ${runAttemptCount + 1})")
             retryOrFail(userId, roomToken, lastReadMessage)
         }
     }
 
+    /**
+     * The marker has reached the server by now, so a failure here must not send it again: the value
+     * only saves a request, and the next room list sync brings it anyway.
+     */
+    private suspend fun storeLastCommonRead(userId: Long, roomToken: String, response: Response<*>) {
+        val lastCommonRead = lastCommonReadOf(response) ?: return
+        runCatching {
+            conversationListUpdater.updateLastCommonRead("$userId@$roomToken", lastCommonRead)
+        }.onFailure { throwable ->
+            if (throwable is CancellationException) throw throwable
+            Log.w(TAG, "Last common read $lastCommonRead for room $roomToken could not be stored: $throwable")
+        }
+    }
+
     private fun retryOrFail(userId: Long, roomToken: String, lastReadMessage: Int): Result =
         if (runAttemptCount < MAX_RUN_ATTEMPTS - 1) {
             Result.retry()
@@ -120,6 +143,14 @@ class ReadMarkerSyncWorker(context: Context, workerParams: WorkerParameters) :
         private val TAG: String = ReadMarkerSyncWorker::class.java.simpleName
         private const val KEY_LAST_READ_MESSAGE = "KEY_LAST_READ_MESSAGE"
         private const val MAX_RUN_ATTEMPTS = 3
+        private const val LAST_COMMON_READ_HEADER = "X-Chat-Last-Common-Read"
+
+        /**
+         * The last message read by all participants, which the server sends along with the answer to a
+         * read marker; null when the header is missing or not a number.
+         */
+        internal fun lastCommonReadOf(response: Response<*>): Int? =
+            response.headers()[LAST_COMMON_READ_HEADER]?.toIntOrNull()
 
         fun enqueue(context: Context, userId: Long, roomToken: String, lastReadMessage: Int) {
             val data = Data.Builder()

app/src/main/java/com/nextcloud/talk/receivers/MarkAsReadReceiver.kt

@@ -28,6 +28,7 @@ import io.reactivex.android.schedulers.AndroidSchedulers
 import io.reactivex.disposables.Disposable
 import io.reactivex.schedulers.Schedulers
 import kotlinx.coroutines.runBlocking
+import retrofit2.Response
 import javax.inject.Inject
 
 @AutoInjector(NextcloudTalkApplication::class)
@@ -79,13 +80,17 @@ class MarkAsReadReceiver : BroadcastReceiver() {
         ncApi.setChatReadMarker(credentials, url, messageId)
             ?.subscribeOn(Schedulers.io())
             ?.observeOn(AndroidSchedulers.mainThread())
-            ?.subscribe(object : Observer<GenericOverall> {
+            ?.subscribe(object : Observer<Response<GenericOverall>> {
                 override fun onSubscribe(d: Disposable) {
                     // unused atm
                 }
 
-                override fun onNext(genericOverall: GenericOverall) {
-                    cancelNotification(systemNotificationId!!)
+                override fun onNext(response: Response<GenericOverall>) {
+                    if (response.isSuccessful) {
+                        cancelNotification(systemNotificationId!!)
+                    } else {
+                        Log.e(TAG, "Failed to set chat read marker: HTTP ${response.code()}")
+                    }
                 }
 
                 override fun onError(e: Throwable) {

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

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