From c0bfbd87a4a837e3ab2eb47e99a87f8af5597db2 Mon Sep 17 00:00:00 2001 From: Krainov Gleb Date: Fri, 9 Oct 2026 19:22:42 +0000 Subject: [PATCH 1/2] fix(chat): keep X-Chat-Last-Common-Read from the read marker response The answer to POST chat//read carries the last message read by all participants in the X-Chat-Last-Common-Read header. The read marker worker dropped it, so the read check marks of the own messages stayed behind until the next chat sync or room list refresh. Return the whole response from setChatReadMarker, write the header value to the conversation entry and let the open chat pick it up from there. A non-2xx answer is now treated as a failed attempt explicitly, as the response is no longer thrown as an error. Assisted-by: Claude-Code:claude-sonnet-5-5 Signed-off-by: Krainov Gleb --- .../java/com/nextcloud/talk/api/NcApi.java | 6 +- .../data/network/ChatNetworkDataSource.kt | 6 +- .../chat/data/network/RetrofitChatNetwork.kt | 2 +- .../data/network/ConversationListUpdater.kt | 9 +++ .../data/database/dao/ConversationsDao.kt | 13 ++++ .../talk/jobs/ReadMarkerSyncWorker.kt | 21 +++++- .../talk/receivers/MarkAsReadReceiver.kt | 11 ++- .../utils/preview/ComposePreviewUtilsDaos.kt | 4 ++ .../talk/jobs/ReadMarkerSyncWorkerTest.kt | 67 +++++++++++++++++++ 9 files changed, 128 insertions(+), 11 deletions(-) create mode 100644 app/src/test/java/com/nextcloud/talk/jobs/ReadMarkerSyncWorkerTest.kt 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 d87b8cb8f54..ce6cbdd8178 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 553f1e1aeac..ab7bbc64bb4 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 606642563c8..0e3b3269b8e 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 8866b601d44..2516499c4e6 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 @@ -160,6 +160,15 @@ 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) { + 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 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 bbf19adb224..ae64257ccd9 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,19 @@ 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. + */ + @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 ca631fd3614..fb6d4f32624 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,7 @@ 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 retrofit2.Response import java.util.concurrent.TimeUnit import javax.inject.Inject @@ -80,19 +81,25 @@ 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") + lastCommonReadOf(response)?.let { + conversationListUpdater.updateLastCommonRead("$userId@$roomToken", it) + } Result.success() } else { Log.w(TAG, "Sending read marker for room $roomToken failed (attempt ${runAttemptCount + 1})") @@ -120,6 +127,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 f83da396e6e..34db15aaf5f 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 2ce7abc5034..cf3d025b148 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/jobs/ReadMarkerSyncWorkerTest.kt b/app/src/test/java/com/nextcloud/talk/jobs/ReadMarkerSyncWorkerTest.kt new file mode 100644 index 00000000000..1ebe7f68cb0 --- /dev/null +++ b/app/src/test/java/com/nextcloud/talk/jobs/ReadMarkerSyncWorkerTest.kt @@ -0,0 +1,67 @@ +/* + * 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 com.nextcloud.talk.conversationlist.data.network.ConversationListUpdater +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 kotlinx.coroutines.test.runTest +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.mockito.kotlin.mock +import org.mockito.kotlin.verifyBlocking +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. + */ +class ReadMarkerSyncWorkerTest { + + @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 `last common read is written to the conversation`() = + runTest { + val conversationsDao: ConversationsDao = mock() + val updater = ConversationListUpdater(mock(), mock(), conversationsDao) + + updater.updateLastCommonRead("7@token", 2624) + + verifyBlocking(conversationsDao) { updateLastCommonRead("7@token", 2624) } + } + + private fun responseWithHeader(value: String?): 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(200) + .message("OK") + .headers(headers) + .body("".toResponseBody("application/json".toMediaType())) + .build() + return Response.success(null, raw) + } +} From 9a3375ac32b862056cb39dcae9136357ca95842f Mon Sep 17 00:00:00 2001 From: Krainov Gleb Date: Sun, 11 Oct 2026 14:02:43 +0000 Subject: [PATCH 2/2] fix(chat): keep the last common read against older room list responses A room list response requested before the read marker answer reached the client can still carry an older X-Chat-Last-Common-Read value, and applying it after the answer lowered the read check marks again until the next sync. The read marker pending guard cannot catch this, it only covers lastReadMessage. Remember when the answer arrived and give preservePendingLocalState the start time of the request that produced the response: a response requested before that moment cannot lower the value, one requested after it applies unchanged and ends the guard. A sync holds a lock from its merge through its write and the answer takes the same lock, so an answer cannot slip in between the two and be overwritten. The value written from an answer may be lower than the stored one, as marking a chat as unread lowers it on the server. A failure while storing the value no longer leaves the worker, the marker has been delivered and must not be sent again. The worker test now runs the worker. Assisted-by: Claude-Code:claude-opus-5-5 Signed-off-by: Krainov Gleb --- .../data/network/ConversationListUpdater.kt | 51 ++++++- .../OfflineFirstConversationsRepository.kt | 51 ++++--- .../data/database/dao/ConversationsDao.kt | 1 + .../talk/jobs/ReadMarkerSyncWorker.kt | 22 ++- ...onversationListFreshnessIntegrationTest.kt | 127 +++++++++++++++++- .../ConversationListUpdaterCommonReadTest.kt | 94 +++++++++++++ .../network/ConversationListUpdaterPinTest.kt | 6 +- .../ConversationListUpdaterUnreadTest.kt | 3 +- ...OfflineFirstConversationsRepositoryTest.kt | 7 +- .../talk/jobs/ReadMarkerSyncWorkerTest.kt | 127 +++++++++++++++--- 10 files changed, 439 insertions(+), 50 deletions(-) create mode 100644 app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdaterCommonReadTest.kt 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 2516499c4e6..845367494af 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. @@ -166,7 +186,10 @@ class ConversationListUpdater @Inject constructor( * further request. */ suspend fun updateLastCommonRead(internalConversationId: String, lastCommonRead: Int) { - conversationsDao.updateLastCommonRead(internalConversationId, lastCommonRead) + syncWriteLock.withLock { + commonReadWrites[internalConversationId] = CommonReadWrite(lastCommonRead, System.nanoTime()) + conversationsDao.updateLastCommonRead(internalConversationId, lastCommonRead) + } } /** @@ -294,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] @@ -309,9 +337,26 @@ 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, previous: 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 a9cd035e0ef..5967cb94b93 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 ae64257ccd9..d79c100ff59 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 @@ -107,6 +107,7 @@ interface ConversationsDao { /** * 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( """ 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 fb6d4f32624..fd6540b2fa1 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,7 @@ 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 @@ -58,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) @@ -97,9 +101,7 @@ class ReadMarkerSyncWorker(context: Context, workerParams: WorkerParameters) : return if (response != null) { Log.d(TAG, "Read marker $lastReadMessage sent for room $roomToken") - lastCommonReadOf(response)?.let { - conversationListUpdater.updateLastCommonRead("$userId@$roomToken", it) - } + storeLastCommonRead(userId, roomToken, response) Result.success() } else { Log.w(TAG, "Sending read marker for room $roomToken failed (attempt ${runAttemptCount + 1})") @@ -107,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() 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 d5f86bc33fd..40a90c4f9ea 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 00000000000..4e5d0c2b165 --- /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 18bdc199183..c9518a3dd0e 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 8a016090fba..2a798e3d90d 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 54393008112..2b2a7b6dfdc 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 index 1ebe7f68cb0..da500aebbd7 100644 --- a/app/src/test/java/com/nextcloud/talk/jobs/ReadMarkerSyncWorkerTest.kt +++ b/app/src/test/java/com/nextcloud/talk/jobs/ReadMarkerSyncWorkerTest.kt @@ -7,27 +7,53 @@ 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.database.dao.ChatBlocksDao -import com.nextcloud.talk.data.database.dao.ChatMessagesDao -import com.nextcloud.talk.data.database.dao.ConversationsDao -import kotlinx.coroutines.test.runTest +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") @@ -42,26 +68,97 @@ class ReadMarkerSyncWorkerTest { } @Test - fun `last common read is written to the conversation`() = - runTest { - val conversationsDao: ConversationsDao = mock() - val updater = ConversationListUpdater(mock(), mock(), conversationsDao) + fun `a delivered read marker stores the last common read of its response`() { + givenMarkerAnswer(responseWithHeader("2624")) + + val result = runBlocking { worker().send() } - updater.updateLastCommonRead("7@token", 2624) + assertEquals(ListenableWorker.Result.success(), result) + verifyBlocking(conversationListUpdater) { updateLastCommonRead("$USER_ID@$ROOM_TOKEN", 2624) } + } - verifyBlocking(conversationsDao) { updateLastCommonRead("7@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()) } + } - private fun responseWithHeader(value: String?): Response { + @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(200) - .message("OK") + .code(code) + .message("message") .headers(headers) .body("".toResponseBody("application/json".toMediaType())) .build() - return Response.success(null, raw) + 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 } }