diff --git a/app/src/main/java/com/nextcloud/talk/api/NcApi.java b/app/src/main/java/com/nextcloud/talk/api/NcApi.java index d87b8cb8f5..ce6cbdd817 100644 --- a/app/src/main/java/com/nextcloud/talk/api/NcApi.java +++ b/app/src/main/java/com/nextcloud/talk/api/NcApi.java @@ -501,9 +501,9 @@ Observable notificationCalls(@Header("Authorization") String aut // Url is: /api/{apiVersion}/chat/{token}/read @FormUrlEncoded @POST - Observable setChatReadMarker(@Header("Authorization") String authorization, - @Url String url, - @Nullable @Field("lastReadMessage") Integer lastReadMessage); + Observable> setChatReadMarker(@Header("Authorization") String authorization, + @Url String url, + @Nullable @Field("lastReadMessage") Integer lastReadMessage); // Url is: /api/{apiVersion}/chat/{token}/read @DELETE diff --git a/app/src/main/java/com/nextcloud/talk/chat/data/network/ChatNetworkDataSource.kt b/app/src/main/java/com/nextcloud/talk/chat/data/network/ChatNetworkDataSource.kt index 553f1e1aea..ab7bbc64bb 100644 --- a/app/src/main/java/com/nextcloud/talk/chat/data/network/ChatNetworkDataSource.kt +++ b/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): Observable - fun setChatReadMarker(credentials: String, url: String, previousMessageId: Int): Observable + fun setChatReadMarker( + credentials: String, + url: String, + previousMessageId: Int + ): Observable> 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 diff --git a/app/src/main/java/com/nextcloud/talk/chat/data/network/RetrofitChatNetwork.kt b/app/src/main/java/com/nextcloud/talk/chat/data/network/RetrofitChatNetwork.kt index 606642563c..0e3b3269b8 100644 --- a/app/src/main/java/com/nextcloud/talk/chat/data/network/RetrofitChatNetwork.kt +++ b/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 = ncApi.setChatReadMarker(credentials, url, previousMessageId).map { it } + ): Observable> = ncApi.setChatReadMarker(credentials, url, previousMessageId) override suspend fun editChatMessage(credentials: String, url: String, text: String): ChatOverallSingleMessage = ncApiCoroutines.editChatMessage(credentials, url, text) diff --git a/app/src/main/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdater.kt b/app/src/main/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdater.kt index 8866b601d4..845367494a 100644 --- a/app/src/main/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdater.kt +++ b/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() + /** + * 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() + + /** + * 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, - conversationsFromServer: List + conversationsFromServer: List, + requestedAtNanos: Long ): List = 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, diff --git a/app/src/main/java/com/nextcloud/talk/conversationlist/data/network/OfflineFirstConversationsRepository.kt b/app/src/main/java/com/nextcloud/talk/conversationlist/data/network/OfflineFirstConversationsRepository.kt index a9cd035e0e..5967cb94b9 100644 --- a/app/src/main/java/com/nextcloud/talk/conversationlist/data/network/OfflineFirstConversationsRepository.kt +++ b/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) diff --git a/app/src/main/java/com/nextcloud/talk/data/database/dao/ConversationsDao.kt b/app/src/main/java/com/nextcloud/talk/data/database/dao/ConversationsDao.kt index bbf19adb22..d79c100ff5 100644 --- a/app/src/main/java/com/nextcloud/talk/data/database/dao/ConversationsDao.kt +++ b/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] */ diff --git a/app/src/main/java/com/nextcloud/talk/jobs/ReadMarkerSyncWorker.kt b/app/src/main/java/com/nextcloud/talk/jobs/ReadMarkerSyncWorker.kt index ca631fd361..fd6540b2fa 100644 --- a/app/src/main/java/com/nextcloud/talk/jobs/ReadMarkerSyncWorker.kt +++ b/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,19 +85,23 @@ 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})") @@ -100,6 +109,20 @@ class ReadMarkerSyncWorker(context: Context, workerParams: WorkerParameters) : } } + /** + * 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() diff --git a/app/src/main/java/com/nextcloud/talk/receivers/MarkAsReadReceiver.kt b/app/src/main/java/com/nextcloud/talk/receivers/MarkAsReadReceiver.kt index f83da396e6..34db15aaf5 100644 --- a/app/src/main/java/com/nextcloud/talk/receivers/MarkAsReadReceiver.kt +++ b/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 { + ?.subscribe(object : Observer> { override fun onSubscribe(d: Disposable) { // unused atm } - override fun onNext(genericOverall: GenericOverall) { - cancelNotification(systemNotificationId!!) + override fun onNext(response: Response) { + if (response.isSuccessful) { + cancelNotification(systemNotificationId!!) + } else { + Log.e(TAG, "Failed to set chat read marker: HTTP ${response.code()}") + } } override fun onError(e: Throwable) { diff --git a/app/src/main/java/com/nextcloud/talk/utils/preview/ComposePreviewUtilsDaos.kt b/app/src/main/java/com/nextcloud/talk/utils/preview/ComposePreviewUtilsDaos.kt index 2ce7abc503..cf3d025b14 100644 --- a/app/src/main/java/com/nextcloud/talk/utils/preview/ComposePreviewUtilsDaos.kt +++ b/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) { /* */ } diff --git a/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListFreshnessIntegrationTest.kt b/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListFreshnessIntegrationTest.kt index d5f86bc33f..40a90c4f9e 100644 --- a/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListFreshnessIntegrationTest.kt +++ b/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListFreshnessIntegrationTest.kt @@ -16,6 +16,7 @@ import com.nextcloud.talk.arbitrarystorage.ArbitraryStorageManager import com.nextcloud.talk.chat.data.model.ChatMessage import com.nextcloud.talk.chat.data.network.ChatMessageSyncer import com.nextcloud.talk.chat.data.network.ChatNetworkDataSource +import com.nextcloud.talk.data.database.dao.ConversationsDao import com.nextcloud.talk.data.database.mappers.asEntity import com.nextcloud.talk.data.database.model.ConversationEntity import com.nextcloud.talk.data.network.NetworkMonitor @@ -35,12 +36,15 @@ import com.nextcloud.talk.utils.ApiUtils import io.reactivex.Observable import io.reactivex.android.plugins.RxAndroidPlugins import io.reactivex.schedulers.Schedulers +import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.Job import kotlinx.coroutines.delay import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.first import kotlinx.coroutines.launch import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeoutOrNull import org.junit.After import org.junit.Assert.assertEquals import org.junit.Assert.assertFalse @@ -213,6 +217,85 @@ class ConversationListFreshnessIntegrationTest { } } + @Test + fun `a last common read arriving between the merge and the write of a room list sync is not overwritten`() { + val user = user(withKeepNotificationsCapability = false) + val dao = DaoAnsweringBeforeTheWrite() + whenever(conversationsNetwork.getRooms(any(), any(), any(), anyOrNull())).thenReturn( + roomList(listOf(staleServerRoom(lastReadMessage = 12, unreadMessages = 0).copy(lastCommonReadMessage = 10))) + ) + + runBlocking { + seedConversation(lastActivity = 12, lastReadMessage = 12, unreadMessages = 0, lastCommonRead = 10) + + repository(dao).getRooms(user).join() + dao.answer.join() + + assertEquals(20, conversationEntity().lastCommonReadMessage) + } + } + + @Test + fun `a last common read arriving between the merge and the write of a room refresh is not overwritten`() { + val user = user(withKeepNotificationsCapability = false) + val dao = DaoAnsweringBeforeTheWrite() + wheneverBlocking { chatNetwork.getRoom(any(), any()) }.thenReturn( + ConversationModel.mapToConversationModel( + staleServerRoom(lastReadMessage = 12, unreadMessages = 0).copy(lastCommonReadMessage = 10), + user + ) + ) + + runBlocking { + seedConversation(lastActivity = 12, lastReadMessage = 12, unreadMessages = 0, lastCommonRead = 10) + + repository(dao).getRoom(user, ROOM_TOKEN).join() + dao.answer.join() + + assertEquals(20, conversationEntity().lastCommonReadMessage) + } + } + + @Test + fun `a single-room refresh requested before the read marker answer does not lower the last common read`() { + val user = user(withKeepNotificationsCapability = false) + wheneverBlocking { chatNetwork.getRoom(any(), any()) }.thenAnswer { + // the answer to the read marker arrives while the room request is on its way + runBlocking { conversationListUpdater.updateLastCommonRead(INTERNAL_CONVERSATION_ID, lastCommonRead = 20) } + ConversationModel.mapToConversationModel(staleServerRoom(lastReadMessage = 12, unreadMessages = 0), user) + } + + runBlocking { + seedConversation(lastActivity = 12, lastReadMessage = 12, unreadMessages = 0, lastCommonRead = 10) + + repository().getRoom(user, ROOM_TOKEN).join() + + assertEquals(20, conversationEntity().lastCommonReadMessage) + } + } + + @Test + fun `a lower last common read from the read marker answer is written and a later refresh keeps it`() { + val user = user(withKeepNotificationsCapability = false) + wheneverBlocking { chatNetwork.getRoom(any(), any()) }.thenReturn( + ConversationModel.mapToConversationModel( + staleServerRoom(lastReadMessage = 12, unreadMessages = 0).copy(lastCommonReadMessage = 15), + user + ) + ) + + runBlocking { + seedConversation(lastActivity = 12, lastReadMessage = 12, unreadMessages = 0, lastCommonRead = 20) + + // the user marked the chat as unread, so the server reports less than the stored value + conversationListUpdater.updateLastCommonRead(INTERNAL_CONVERSATION_ID, lastCommonRead = 15) + assertEquals(15, conversationEntity().lastCommonReadMessage) + + repository().getRoom(user, ROOM_TOKEN).join() + assertEquals(15, conversationEntity().lastCommonReadMessage) + } + } + @Test fun `background catch-up never regresses a newer conversation entry`() { runBlocking { @@ -413,9 +496,9 @@ class ConversationListFreshnessIntegrationTest { } } - private fun repository(): OfflineFirstConversationsRepository = + private fun repository(dao: ConversationsDao = db.conversationsDao()): OfflineFirstConversationsRepository = OfflineFirstConversationsRepository( - db.conversationsDao(), + dao, conversationsNetwork, chatNetwork, networkMonitor, @@ -426,6 +509,37 @@ class ConversationListFreshnessIntegrationTest { mock() ) + /** + * The conversations DAO with the answer to a read marker (last common read 20) arriving right + * before the response of a sync or room refresh is written, that is after its merge. The answer + * gets a moment to complete: it does so only when nothing keeps it from writing in that window. + */ + private inner class DaoAnsweringBeforeTheWrite(private val real: ConversationsDao = db.conversationsDao()) : + ConversationsDao by real { + lateinit var answer: Job + + override suspend fun syncConversationsForUser( + accountId: Long, + serverItems: List, + conversationIdsToDelete: List + ) { + answerArrives() + real.syncConversationsForUser(accountId, serverItems, conversationIdsToDelete) + } + + override suspend fun upsertConversations(accountId: Long, serverItems: List) { + answerArrives() + real.upsertConversations(accountId, serverItems) + } + + private suspend fun answerArrives() { + answer = CoroutineScope(Dispatchers.Default).launch { + conversationListUpdater.updateLastCommonRead(INTERNAL_CONVERSATION_ID, lastCommonRead = 20) + } + withTimeoutOrNull(ANSWER_WAIT_MILLIS) { answer.join() } + } + } + private fun staleServerRoom( lastReadMessage: Int, unreadMessages: Int, @@ -441,11 +555,17 @@ class ConversationListFreshnessIntegrationTest { hasArchived = hasArchived ) - private suspend fun seedConversation(lastActivity: Long, lastReadMessage: Int, unreadMessages: Int) { + private suspend fun seedConversation( + lastActivity: Long, + lastReadMessage: Int, + unreadMessages: Int, + lastCommonRead: Int = 0 + ) { val entity = ConversationDto( token = ROOM_TOKEN, lastActivity = lastActivity, lastReadMessage = lastReadMessage, + lastCommonReadMessage = lastCommonRead, unreadMessages = unreadMessages, lastMessage = message(lastReadMessage.toLong()) ).asEntity(ACCOUNT_ID) @@ -521,6 +641,7 @@ class ConversationListFreshnessIntegrationTest { private const val INTERNAL_CONVERSATION_ID = "$ACCOUNT_ID@$ROOM_TOKEN" private const val TIMEOUT_MILLIS = 10_000L private const val POLL_INTERVAL_MILLIS = 50L + private const val ANSWER_WAIT_MILLIS = 300L } } diff --git a/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdaterCommonReadTest.kt b/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdaterCommonReadTest.kt new file mode 100644 index 0000000000..4e5d0c2b16 --- /dev/null +++ b/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdaterCommonReadTest.kt @@ -0,0 +1,94 @@ +/* + * Nextcloud Talk - Android Client + * + * SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors + * SPDX-License-Identifier: GPL-3.0-or-later + */ + +package com.nextcloud.talk.conversationlist.data.network + +import com.nextcloud.talk.data.database.dao.ChatBlocksDao +import com.nextcloud.talk.data.database.dao.ChatMessagesDao +import com.nextcloud.talk.data.database.dao.ConversationsDao +import com.nextcloud.talk.data.database.mappers.asEntity +import com.nextcloud.talk.data.database.model.ConversationEntity +import com.nextcloud.talk.models.json.conversations.ConversationDto +import kotlinx.coroutines.test.runTest +import org.junit.Assert.assertEquals +import org.junit.Test +import org.mockito.kotlin.mock +import org.mockito.kotlin.verifyBlocking + +/** + * The last message read by all participants comes with the answer to a read marker. A room list + * response whose request started before that answer must not lower it again; a later one is + * current and applies as the server sent it. + */ +class ConversationListUpdaterCommonReadTest { + + private val conversationsDao: ConversationsDao = mock() + private val updater = ConversationListUpdater(mock(), mock(), conversationsDao) + + private val stored: ConversationEntity = ConversationDto(token = ROOM_TOKEN).asEntity(ACCOUNT_ID) + + @Test + fun `last common read is written to the conversation`() = + runTest { + updater.updateLastCommonRead(stored.internalId, NEWER) + + verifyBlocking(conversationsDao) { updateLastCommonRead(stored.internalId, NEWER) } + } + + @Test + fun `a room list response requested before the read marker answer does not lower it`() = + runTest { + val requestedAt = System.nanoTime() + updater.updateLastCommonRead(stored.internalId, NEWER) + + assertEquals(NEWER, preserve(serverLastCommonRead = OLDER, requestedAt).lastCommonReadMessage) + // the guard stays for every response of that age, it is not used up by the first one + assertEquals(NEWER, preserve(serverLastCommonRead = OLDER, requestedAt).lastCommonReadMessage) + } + + @Test + fun `a room list response requested before the read marker answer still brings a higher value`() = + runTest { + val requestedAt = System.nanoTime() + updater.updateLastCommonRead(stored.internalId, NEWER) + + assertEquals(NEWEST, preserve(serverLastCommonRead = NEWEST, requestedAt).lastCommonReadMessage) + } + + @Test + fun `a room list response requested after the read marker answer applies unchanged and ends the guard`() = + runTest { + val requestedBefore = System.nanoTime() + updater.updateLastCommonRead(stored.internalId, NEWER) + val requestedAfter = System.nanoTime() + + // someone marked a message as unread, so the server really reports less now + assertEquals(OLDER, preserve(serverLastCommonRead = OLDER, requestedAfter).lastCommonReadMessage) + // the guard is gone: not even an older request is held back any more + assertEquals(OLDER, preserve(serverLastCommonRead = OLDER, requestedBefore).lastCommonReadMessage) + } + + @Test + fun `a room without a stored common read takes the server value`() { + assertEquals(OLDER, preserve(serverLastCommonRead = OLDER, System.nanoTime()).lastCommonReadMessage) + } + + private fun preserve(serverLastCommonRead: Int, requestedAtNanos: Long): ConversationEntity = + updater.preservePendingLocalState( + previousConversations = mapOf(stored.internalId to stored), + conversationsFromServer = listOf(stored.copy(lastCommonReadMessage = serverLastCommonRead)), + requestedAtNanos = requestedAtNanos + ).single() + + companion object { + private const val ACCOUNT_ID = 7L + private const val ROOM_TOKEN = "token" + private const val OLDER = 2620 + private const val NEWER = 2624 + private const val NEWEST = 2630 + } +} diff --git a/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdaterPinTest.kt b/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdaterPinTest.kt index 18bdc19918..c9518a3dd0 100644 --- a/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdaterPinTest.kt +++ b/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdaterPinTest.kt @@ -112,13 +112,15 @@ class ConversationListUpdaterPinTest { private fun preserveHidden(serverHiddenId: Long?): ConversationEntity = updater.preservePendingLocalState( previousConversations = mapOf(stored.internalId to stored), - conversationsFromServer = listOf(stored.copy(hiddenPinnedId = serverHiddenId)) + conversationsFromServer = listOf(stored.copy(hiddenPinnedId = serverHiddenId)), + requestedAtNanos = System.nanoTime() ).single() private fun preserve(serverPinnedId: Long?): ConversationEntity = updater.preservePendingLocalState( previousConversations = mapOf(stored.internalId to stored), - conversationsFromServer = listOf(stored.copy(lastPinnedId = serverPinnedId)) + conversationsFromServer = listOf(stored.copy(lastPinnedId = serverPinnedId)), + requestedAtNanos = System.nanoTime() ).single() private fun givenCachedConversation(pinnedId: Long?) { diff --git a/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdaterUnreadTest.kt b/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdaterUnreadTest.kt index 8a016090fb..2a798e3d90 100644 --- a/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdaterUnreadTest.kt +++ b/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdaterUnreadTest.kt @@ -123,7 +123,8 @@ class ConversationListUpdaterUnreadTest { previousConversations = mapOf(stored.internalId to stored), conversationsFromServer = listOf( stored.copy(lastReadMessage = serverLastRead, unreadMessages = serverUnread) - ) + ), + requestedAtNanos = System.nanoTime() ).single() private fun givenCachedMessagesNewerThanMarker(count: Int) { diff --git a/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/OfflineFirstConversationsRepositoryTest.kt b/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/OfflineFirstConversationsRepositoryTest.kt index 5439300811..2b2a7b6dfd 100644 --- a/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/OfflineFirstConversationsRepositoryTest.kt +++ b/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/OfflineFirstConversationsRepositoryTest.kt @@ -38,6 +38,7 @@ import kotlinx.coroutines.flow.first import kotlinx.coroutines.flow.flowOf import kotlinx.coroutines.launch import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.test.runTest import okhttp3.ResponseBody.Companion.toResponseBody import org.junit.After @@ -111,7 +112,8 @@ class OfflineFirstConversationsRepositoryTest { stubStoredConversations() whenever(arbitraryStorageManager.getStorageSetting(any(), any(), any())) .thenReturn(Maybe.empty()) - whenever(conversationListUpdater.preservePendingLocalState(any(), any())) + whenever(conversationListUpdater.syncWriteLock).thenReturn(Mutex()) + whenever(conversationListUpdater.preservePendingLocalState(any(), any(), any())) .thenAnswer { invocation -> invocation.getArgument>(1) } repository = OfflineFirstConversationsRepository( @@ -374,7 +376,8 @@ class OfflineFirstConversationsRepositoryTest { val previousCaptor = argumentCaptor>() val serverCaptor = argumentCaptor>() - verify(conversationListUpdater).preservePendingLocalState(previousCaptor.capture(), serverCaptor.capture()) + verify(conversationListUpdater) + .preservePendingLocalState(previousCaptor.capture(), serverCaptor.capture(), any()) assertEquals(mapOf(previous.internalId to previous), previousCaptor.firstValue) assertEquals(6L, serverCaptor.firstValue.first().lastActivity) } diff --git a/app/src/test/java/com/nextcloud/talk/jobs/ReadMarkerSyncWorkerTest.kt b/app/src/test/java/com/nextcloud/talk/jobs/ReadMarkerSyncWorkerTest.kt new file mode 100644 index 0000000000..da500aebbd --- /dev/null +++ b/app/src/test/java/com/nextcloud/talk/jobs/ReadMarkerSyncWorkerTest.kt @@ -0,0 +1,164 @@ +/* + * Nextcloud Talk - Android Client + * + * SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors + * SPDX-License-Identifier: GPL-3.0-or-later + */ + +package com.nextcloud.talk.jobs + +import android.app.Application +import android.content.Context +import androidx.test.core.app.ApplicationProvider +import androidx.work.Data +import androidx.work.ListenableWorker +import androidx.work.testing.TestListenableWorkerBuilder +import com.nextcloud.talk.chat.data.network.ChatNetworkDataSource +import com.nextcloud.talk.conversationlist.data.network.ConversationListUpdater +import com.nextcloud.talk.data.user.model.User +import com.nextcloud.talk.models.json.capabilities.CapabilitiesDto +import com.nextcloud.talk.models.json.capabilities.SpreedCapabilityDto +import com.nextcloud.talk.models.json.generic.GenericOverall +import com.nextcloud.talk.users.UserManager +import com.nextcloud.talk.utils.SpreedFeatures +import com.nextcloud.talk.utils.bundle.BundleKeys.KEY_INTERNAL_USER_ID +import com.nextcloud.talk.utils.bundle.BundleKeys.KEY_ROOM_TOKEN +import io.reactivex.Observable +import kotlinx.coroutines.runBlocking +import okhttp3.Headers +import okhttp3.MediaType.Companion.toMediaType +import okhttp3.ResponseBody.Companion.toResponseBody +import org.junit.Assert.assertEquals +import org.junit.Assert.assertNull +import org.junit.Test +import org.junit.runner.RunWith +import org.mockito.kotlin.any +import org.mockito.kotlin.mock +import org.mockito.kotlin.never +import org.mockito.kotlin.verifyBlocking +import org.mockito.kotlin.whenever +import org.mockito.kotlin.wheneverBlocking +import org.robolectric.RobolectricTestRunner +import org.robolectric.annotation.Config +import retrofit2.Response + +/** + * The answer to a read marker carries the last message read by all participants. It has to reach the + * database, otherwise the check marks of the own messages stay behind until the next sync. + */ +@RunWith(RobolectricTestRunner::class) +@Config(application = Application::class, sdk = [33]) +class ReadMarkerSyncWorkerTest { + + private val userManager: UserManager = mock() + private val chatNetworkDataSource: ChatNetworkDataSource = mock() + private val conversationListUpdater: ConversationListUpdater = mock() + + @Test + fun `last common read is taken from the header of the read marker response`() { + val response = responseWithHeader("2624") + + assertEquals(2624, ReadMarkerSyncWorker.lastCommonReadOf(response)) + } + + @Test + fun `missing or broken header gives no last common read`() { + assertNull(ReadMarkerSyncWorker.lastCommonReadOf(responseWithHeader(null))) + assertNull(ReadMarkerSyncWorker.lastCommonReadOf(responseWithHeader("abc"))) + } + + @Test + fun `a delivered read marker stores the last common read of its response`() { + givenMarkerAnswer(responseWithHeader("2624")) + + val result = runBlocking { worker().send() } + + assertEquals(ListenableWorker.Result.success(), result) + verifyBlocking(conversationListUpdater) { updateLastCommonRead("$USER_ID@$ROOM_TOKEN", 2624) } + } + + @Test + fun `a response without the header stores nothing but still counts as delivered`() { + givenMarkerAnswer(responseWithHeader(null)) + + val result = runBlocking { worker().send() } + + assertEquals(ListenableWorker.Result.success(), result) + verifyBlocking(conversationListUpdater, never()) { updateLastCommonRead(any(), any()) } + } + + @Test + fun `a failed answer stores nothing and asks for another attempt`() { + givenMarkerAnswer(responseWithHeader("2624", code = 500)) + + val result = runBlocking { worker().send() } + + assertEquals(ListenableWorker.Result.retry(), result) + verifyBlocking(conversationListUpdater, never()) { updateLastCommonRead(any(), any()) } + } + + @Test + fun `a failing database write does not send the delivered marker again`() { + givenMarkerAnswer(responseWithHeader("2624")) + wheneverBlocking { conversationListUpdater.updateLastCommonRead(any(), any()) } + .thenThrow(IllegalStateException("database closed")) + + val result = runBlocking { worker().send() } + + assertEquals(ListenableWorker.Result.success(), result) + } + + private fun givenMarkerAnswer(response: Response) { + wheneverBlocking { userManager.getUserWithId(USER_ID) }.thenReturn(user()) + whenever(chatNetworkDataSource.setChatReadMarker(any(), any(), any())).thenReturn(Observable.just(response)) + } + + private fun worker(): ReadMarkerSyncWorker = + TestListenableWorkerBuilder(applicationContext()) + .setInputData( + Data.Builder() + .putLong(KEY_INTERNAL_USER_ID, USER_ID) + .putString(KEY_ROOM_TOKEN, ROOM_TOKEN) + .putInt("KEY_LAST_READ_MESSAGE", MARKER) + .build() + ) + .build() + .also { + it.userManager = userManager + it.chatNetworkDataSource = chatNetworkDataSource + it.conversationListUpdater = conversationListUpdater + } + + private fun applicationContext(): Context = ApplicationProvider.getApplicationContext() + + private fun user(): User = + User( + id = USER_ID, + userId = "me", + username = "me", + baseUrl = "https://cloud.example", + token = "app-password", + capabilities = CapabilitiesDto().apply { + spreedCapability = SpreedCapabilityDto().apply { features = listOf(SpreedFeatures.CHAT_V2.value) } + } + ) + + private fun responseWithHeader(value: String?, code: Int = 200): Response { + val headers = Headers.Builder().apply { value?.let { add("X-Chat-Last-Common-Read", it) } }.build() + val raw = okhttp3.Response.Builder() + .request(okhttp3.Request.Builder().url("https://cloud.example/").build()) + .protocol(okhttp3.Protocol.HTTP_1_1) + .code(code) + .message("message") + .headers(headers) + .body("".toResponseBody("application/json".toMediaType())) + .build() + return if (code in 200..299) Response.success(null, raw) else Response.error(raw.body!!, raw) + } + + companion object { + private const val USER_ID = 7L + private const val ROOM_TOKEN = "token" + private const val MARKER = 2624 + } +}