From e19b5314ed461de271603b38120fba92aea7e2c0 Mon Sep 17 00:00:00 2001 From: Andy Scherzinger Date: Fri, 14 Aug 2026 18:11:28 +0200 Subject: [PATCH 01/10] refactor(conversations): derive the room list from the database The room list was fed by manual emissions: getRooms emitted a database snapshot and a post-sync snapshot, and nothing else. Any change to the conversations table in between stayed invisible until the next full fetch cycle. roomListFlow now observes the conversations table for the selected account, making the database the single source of truth: every write reaches the UI reactively, whoever made it. getRooms shrinks to selecting the account and triggering the background sync, which stays in place unchanged as the authority and self-healing safeguard (deletions, statuses, drift correction). The sync applies deletions and upserts in one transaction, so observers see a single consistent update per sync instead of intermediate states. Unlike the MutableSharedFlow it replaces, the database-backed flow can throw, and getRoomsStateFlow collects it eagerly in the viewModelScope - an uncaught exception there would crash the app, so it is caught and surfaced as GetRoomsErrorState. updateConversationLocallyAndEmit lost its manual emission and with it any difference to updateConversation, so it is removed and all call sites point at updateConversation. Assisted-by: Claude Code:claude-fable-5 Signed-off-by: Andy Scherzinger --- .../data/OfflineConversationsRepository.kt | 17 +++-- .../OfflineFirstConversationsRepository.kt | 74 ++++++++++--------- .../viewmodels/ConversationsListViewModel.kt | 20 +++-- .../viewmodels/ConversationTagsViewModel.kt | 4 +- .../data/database/dao/ConversationsDao.kt | 17 +++++ 5 files changed, 79 insertions(+), 53 deletions(-) diff --git a/app/src/main/java/com/nextcloud/talk/conversationlist/data/OfflineConversationsRepository.kt b/app/src/main/java/com/nextcloud/talk/conversationlist/data/OfflineConversationsRepository.kt index e19bc2b749..8468cbe90e 100644 --- a/app/src/main/java/com/nextcloud/talk/conversationlist/data/OfflineConversationsRepository.kt +++ b/app/src/main/java/com/nextcloud/talk/conversationlist/data/OfflineConversationsRepository.kt @@ -16,7 +16,9 @@ import kotlinx.coroutines.flow.Flow interface OfflineConversationsRepository { /** - * Stream of a list of rooms, for use in the conversation list. + * Live stream of the observed account's conversations, for use in the conversation list. + * Backed by the local database: it re-emits whenever conversation rows change (room list + * sync, background catch-up, optimistic updates), with unchanged lists deduplicated. */ val roomListFlow: Flow> @@ -27,10 +29,9 @@ interface OfflineConversationsRepository { val conversationFlow: Flow /** - * Loads rooms from local storage. If the rooms are not found, then it - * synchronizes the database with the server, before retrying exactly once. Only - * emits to [roomListFlow] if the rooms list is not empty. - * + * Selects the account observed by [roomListFlow] and synchronizes its conversations with + * the server (when online). The synced changes surface through [roomListFlow], which + * observes the database. */ @Deprecated("use observeConversation") fun getRooms(user: User): Job @@ -42,10 +43,12 @@ interface OfflineConversationsRepository { @Deprecated("use observeConversation") fun getRoom(user: User, roomToken: String): Job + /** + * Updates a single conversation in the local database. [roomListFlow] observes the database + * and re-emits the updated list on its own. + */ suspend fun updateConversation(conversationModel: ConversationModel) - suspend fun updateConversationLocallyAndEmit(user: User, conversation: ConversationModel) - @Deprecated("use observeConversation") suspend fun getLocallyStoredConversation(user: User, roomToken: String): ConversationModel? 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 32661535ee..7a7dfd4050 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 @@ -31,10 +31,15 @@ import io.reactivex.disposables.Disposable import io.reactivex.schedulers.Schedulers import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.Job import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.distinctUntilChanged +import kotlinx.coroutines.flow.filterNotNull import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.flatMapLatest import kotlinx.coroutines.flow.map import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.launch @@ -52,9 +57,25 @@ class OfflineFirstConversationsRepository @Inject constructor( private val chatMessageSyncer: ChatMessageSyncer, private val context: Context ) : OfflineConversationsRepository { - override val roomListFlow: Flow> - get() = _roomListFlow - private val _roomListFlow: MutableSharedFlow> = MutableSharedFlow() + private val observedAccountId = MutableStateFlow(null) + + /** + * The conversation list as a live view of the local database — the single source of truth. + * Every write to the conversations table (room list sync, background message catch-up, + * optimistic read state, drafts) reaches collectors reactively; [getRooms] only selects the + * account to observe and triggers the background sync, which stays in place as the authority + * and self-healing safeguard. + */ + @OptIn(ExperimentalCoroutinesApi::class) + override val roomListFlow: Flow> = + observedAccountId + .filterNotNull() + .distinctUntilChanged() + .flatMapLatest { accountId -> + dao.getConversationsForUser(accountId) + .map { entities -> entities.map(ConversationEntity::toDomainModel) } + } + .distinctUntilChanged() override val conversationFlow: Flow get() = _conversationFlow @@ -82,15 +103,10 @@ class OfflineFirstConversationsRepository @Inject constructor( override fun getRooms(user: User): Job = scope.launch { - val initialConversationModels = getListOfConversations(user.id!!) - _roomListFlow.emit(initialConversationModels) + observedAccountId.value = user.id!! if (networkMonitor.isOnline.value) { - val conversationEntitiesFromSync = getRoomsFromServer(user) - if (!conversationEntitiesFromSync.isNullOrEmpty()) { - val conversationModelsFromSync = getListOfConversations(user.id!!) - _roomListFlow.emit(conversationModelsFromSync) - } + getRoomsFromServer(user) } } @@ -138,12 +154,6 @@ class OfflineFirstConversationsRepository @Inject constructor( dao.updateConversation(entity) } - override suspend fun updateConversationLocallyAndEmit(user: User, conversation: ConversationModel) { - dao.updateConversation(conversation.asEntity()) - val updatedList = getListOfConversations(user.id!!) - _roomListFlow.emit(updatedList) - } - override suspend fun getLocallyStoredConversation(user: User, roomToken: String): ConversationModel? { val id = user.id!! return getConversation(id, roomToken) @@ -173,11 +183,11 @@ class OfflineFirstConversationsRepository @Inject constructor( val previousConversations = dao.getConversationsForUser(user.id!!).first() .associateBy { it.internalId } - deleteLeftConversations( - user, - conversationsFromSync + dao.syncConversationsForUser( + accountId = user.id!!, + serverItems = conversationsFromSync, + conversationIdsToDelete = determineLeftConversationIds(previousConversations, conversationsFromSync) ) - dao.upsertConversations(user.id!!, conversationsFromSync) val roomsWithNewMessages = getRoomsWithNewMessages(conversationsFromSync, previousConversations) scope.launch { catchUpRoomsWithNewMessages(user, roomsWithNewMessages) } @@ -285,36 +295,28 @@ class OfflineFirstConversationsRepository @Inject constructor( connectivityManager.restrictBackgroundStatus == ConnectivityManager.RESTRICT_BACKGROUND_STATUS_ENABLED } - private suspend fun deleteLeftConversations(user: User, conversationsFromSync: List) { - val oldConversationsFromDb = dao.getConversationsForUser(user.id!!).first() - - if (conversationsFromSync.isEmpty() && oldConversationsFromDb.isNotEmpty()) { + private fun determineLeftConversationIds( + previousConversations: Map, + conversationsFromSync: List + ): List { + if (conversationsFromSync.isEmpty() && previousConversations.isNotEmpty()) { // A sync that suddenly contains no conversations at all is most likely a broken or // partial server response. Deleting the local conversations in that case would also // wipe their cached chat messages and chat blocks via foreign key cascade, destroying // the offline cache. Skip and let a later successful sync reconcile. Log.w( TAG, - "Sync returned no conversations while ${oldConversationsFromDb.size} exist locally, " + + "Sync returned no conversations while ${previousConversations.size} exist locally, " + "skipping deletion of left conversations" ) - return + return emptyList() } val conversationsFromSyncIds = conversationsFromSync.map { it.internalId }.toSet() - val conversationIdsToDelete = oldConversationsFromDb - .map { it.internalId } - .filterNot { it in conversationsFromSyncIds } - - dao.deleteConversations(conversationIdsToDelete) + return previousConversations.keys.filterNot { it in conversationsFromSyncIds } } - private suspend fun getListOfConversations(accountId: Long): List = - dao.getConversationsForUser(accountId).map { - it.map(ConversationEntity::toDomainModel) - }.first() - private suspend fun getConversation(accountId: Long, token: String): ConversationModel? { val entity = dao.getConversationForUser(accountId, token).first() return entity?.toDomainModel() diff --git a/app/src/main/java/com/nextcloud/talk/conversationlist/viewmodels/ConversationsListViewModel.kt b/app/src/main/java/com/nextcloud/talk/conversationlist/viewmodels/ConversationsListViewModel.kt index f926ce667b..4b2e6ab334 100644 --- a/app/src/main/java/com/nextcloud/talk/conversationlist/viewmodels/ConversationsListViewModel.kt +++ b/app/src/main/java/com/nextcloud/talk/conversationlist/viewmodels/ConversationsListViewModel.kt @@ -149,6 +149,10 @@ class ConversationsListViewModel @Inject constructor( val getRoomsStateFlow = repository .roomListFlow + .catch { throwable -> + Log.e(TAG, "Error observing the conversation list", throwable) + _getRoomsViewState.value = GetRoomsErrorState(throwable) + } .stateIn(viewModelScope, SharingStarted.Eagerly, listOf()) /** @@ -711,7 +715,7 @@ class ConversationsListViewModel @Inject constructor( val url = ApiUtils.getUrlForChatReadMarker(apiVersion, currentUser.baseUrl, conversation.token) viewModelScope.launch { withContext(Dispatchers.IO) { - repository.updateConversationLocallyAndEmit(currentUser, optimistic) + repository.updateConversation(optimistic) } try { withContext(Dispatchers.IO) { @@ -720,7 +724,7 @@ class ConversationsListViewModel @Inject constructor( _readUnreadState.value = ConversationReadUnreadUiState.Success } catch (e: Exception) { withContext(Dispatchers.IO) { - repository.updateConversationLocallyAndEmit(currentUser, original) + repository.updateConversation(original) } _readUnreadState.value = ConversationReadUnreadUiState.Error } @@ -738,7 +742,7 @@ class ConversationsListViewModel @Inject constructor( val url = ApiUtils.getUrlForChatReadMarker(apiVersion, currentUser.baseUrl, conversation.token) viewModelScope.launch { withContext(Dispatchers.IO) { - repository.updateConversationLocallyAndEmit(currentUser, optimistic) + repository.updateConversation(optimistic) } try { withContext(Dispatchers.IO) { @@ -747,7 +751,7 @@ class ConversationsListViewModel @Inject constructor( _readUnreadState.value = ConversationReadUnreadUiState.Success } catch (e: Exception) { withContext(Dispatchers.IO) { - repository.updateConversationLocallyAndEmit(currentUser, original) + repository.updateConversation(original) } _readUnreadState.value = ConversationReadUnreadUiState.Error } @@ -766,7 +770,7 @@ class ConversationsListViewModel @Inject constructor( val url = ApiUtils.getUrlForRoomFavorite(apiVersion, currentUser.baseUrl, conversation.token) viewModelScope.launch { withContext(Dispatchers.IO) { - repository.updateConversationLocallyAndEmit(currentUser, optimistic) + repository.updateConversation(optimistic) } try { withContext(Dispatchers.IO) { @@ -775,7 +779,7 @@ class ConversationsListViewModel @Inject constructor( _favoriteState.value = FavoriteUiState.Success } catch (e: Exception) { withContext(Dispatchers.IO) { - repository.updateConversationLocallyAndEmit(currentUser, original) + repository.updateConversation(original) } _favoriteState.value = FavoriteUiState.Error } @@ -790,7 +794,7 @@ class ConversationsListViewModel @Inject constructor( val url = ApiUtils.getUrlForRoomFavorite(apiVersion, currentUser.baseUrl, conversation.token) viewModelScope.launch { withContext(Dispatchers.IO) { - repository.updateConversationLocallyAndEmit(currentUser, optimistic) + repository.updateConversation(optimistic) } try { withContext(Dispatchers.IO) { @@ -799,7 +803,7 @@ class ConversationsListViewModel @Inject constructor( _favoriteState.value = FavoriteUiState.Success } catch (e: Exception) { withContext(Dispatchers.IO) { - repository.updateConversationLocallyAndEmit(currentUser, original) + repository.updateConversation(original) } _favoriteState.value = FavoriteUiState.Error } diff --git a/app/src/main/java/com/nextcloud/talk/conversationtags/viewmodels/ConversationTagsViewModel.kt b/app/src/main/java/com/nextcloud/talk/conversationtags/viewmodels/ConversationTagsViewModel.kt index 4d01050ee9..e3a70ec636 100644 --- a/app/src/main/java/com/nextcloud/talk/conversationtags/viewmodels/ConversationTagsViewModel.kt +++ b/app/src/main/java/com/nextcloud/talk/conversationtags/viewmodels/ConversationTagsViewModel.kt @@ -165,7 +165,7 @@ class ConversationTagsViewModel @Inject constructor( replaceConversationForTagAssignment(conversation.token, optimistic) viewModelScope.launch { withContext(Dispatchers.IO) { - repository.updateConversationLocallyAndEmit(currentUser, optimistic) + repository.updateConversation(optimistic) } try { withContext(Dispatchers.IO) { @@ -179,7 +179,7 @@ class ConversationTagsViewModel @Inject constructor( } catch (e: Exception) { replaceConversationForTagAssignment(conversation.token, original) withContext(Dispatchers.IO) { - repository.updateConversationLocallyAndEmit(currentUser, original) + repository.updateConversation(original) } Log.e(TAG, "Failed to assign conversation tags", e) } 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 0e1604eed2..23bc660511 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 @@ -25,6 +25,23 @@ interface ConversationsDao { @Query("SELECT * FROM Conversations where accountId = :accountId AND token = :token") fun getConversationForUser(accountId: Long, token: String): Flow + /** + * Applies a full room list sync atomically: left conversations are deleted and the server + * items are upserted in one transaction, so observers of the conversations table see a single + * consistent update per sync instead of intermediate states. + */ + @Transaction + suspend fun syncConversationsForUser( + accountId: Long, + serverItems: List, + conversationIdsToDelete: List + ) { + if (conversationIdsToDelete.isNotEmpty()) { + deleteConversations(conversationIdsToDelete) + } + upsertConversations(accountId, serverItems) + } + @Transaction suspend fun upsertConversations(accountId: Long, serverItems: List) { serverItems.forEach { serverItem -> From 5bd4fba889cb1adcde8c970f1e20e0a0d75950c3 Mon Sep 17 00:00:00 2001 From: Andy Scherzinger Date: Fri, 14 Aug 2026 18:14:46 +0200 Subject: [PATCH 02/10] feat(conversations): update conversation entries from background catch-ups Background message catch-ups (room list prefetch, push notifications) cached the messages but never touched the conversation entry, so the conversation list kept showing the stale last message and unread count until the next full room list sync answered - even though the fresh data was already on the device. After a room-level catch-up that persisted messages, the room's conversation entry is now updated with the newest persisted message (skipping system messages that never become a conversation's preview), its activity timestamp and a locally derived unread count. The count mirrors the server's calculation (spreed's ChatManager.getUnreadCount counts the comment and object_shared verbs only), is only derived when the latest chat block reaches back to the last read message, and excludes the user's own messages since the server advances the author's read marker on every post - which the locally cached marker may lag behind. Concurrency with the room list sync is handled without locking: the write is a single guarded UPDATE that only applies while the derived state is newer than the stored one, so the sync - which remains the authority and self-healing safeguard - can never be overwritten with older data and no read-modify-write window exists. The derivation runs inside the existing per-room catch-up mutex. Assisted-by: Claude Code:claude-fable-5 Signed-off-by: Andy Scherzinger --- .../chat/data/network/ChatMessageSyncer.kt | 114 +++++++++++++++++- .../talk/dagger/modules/RepositoryModule.kt | 2 + .../talk/data/database/dao/ChatMessagesDao.kt | 27 ++++- .../data/database/dao/ConversationsDao.kt | 26 ++++ .../talk/utils/preview/ComposePreviewUtils.kt | 1 + .../utils/preview/ComposePreviewUtilsDaos.kt | 13 ++ .../data/network/ChatMessageSyncerTest.kt | 5 +- .../network/OfflineFirstChatRepositoryTest.kt | 4 +- .../RoomListMessagePrefetchIntegrationTest.kt | 8 +- 9 files changed, 188 insertions(+), 12 deletions(-) diff --git a/app/src/main/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncer.kt b/app/src/main/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncer.kt index 641d10d581..9562994b31 100644 --- a/app/src/main/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncer.kt +++ b/app/src/main/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncer.kt @@ -10,10 +10,12 @@ package com.nextcloud.talk.chat.data.network import android.database.sqlite.SQLiteConstraintException import android.os.SystemClock import android.util.Log +import com.bluelinelabs.logansquare.LoganSquare import com.nextcloud.talk.chat.data.model.ChatMessage import com.nextcloud.talk.chat.domain.ChatPullResult 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.ChatBlockEntity import com.nextcloud.talk.data.database.model.ChatMessageEntity @@ -43,10 +45,11 @@ import javax.inject.Inject * per-room coalescing bookkeeping of [catchUpRoom], which collapses bursts of catch-up requests * (e.g. one push notification per incoming message) into few actual fetches. */ -@Suppress("TooManyFunctions") +@Suppress("TooManyFunctions", "LargeClass") class ChatMessageSyncer @Inject constructor( private val chatDao: ChatMessagesDao, private val chatBlocksDao: ChatBlocksDao, + private val conversationsDao: ConversationsDao, private val network: ChatNetworkDataSource, private val networkMonitor: NetworkMonitor ) { @@ -91,13 +94,18 @@ class ChatMessageSyncer @Inject constructor( * [syncFailed] is true when the sync ended in a transient error (offline, failed request), so * callers like a background worker can retry later. It stays false for skips that retrying * would not change, e.g. a missing server capability. + * + * [newestPersistedMessage] is the newest persisted message that qualifies as a conversation's + * last message (system messages that never change a conversation's preview are skipped). It + * feeds the conversation list update after a background catch-up. */ data class SyncOutcome( val persistedNewMessages: Boolean, val newestPersistedMessageId: Long?, val oldestPersistedMessageId: Long? = null, val persistedMessageCount: Int = 0, - val syncFailed: Boolean = false + val syncFailed: Boolean = false, + val newestPersistedMessage: ChatMessageJson? = null ) /** @@ -357,9 +365,71 @@ class ChatMessageSyncer @Inject constructor( Log.d(TAG, "Background catch-up for room ${target.roomToken}: no new messages") } + updateConversationFromCatchUp(target, outcome) + return outcome } + /** + * Reflects a background catch-up in the conversation list: the room's cached conversation + * entry is updated with the newest persisted message, its activity timestamp and a locally + * derived unread count, so the list is up to date the moment it is opened — before the room + * list sync (which stays in place as the authority and self-healing safeguard) has answered. + * + * Concurrency: runs inside the per-room catch-up mutex and writes via a single guarded UPDATE + * that only applies when the derived state is newer than the stored one, so a concurrently + * finishing room list sync can neither be overwritten with older data nor interleave with a + * read-modify-write. Thread catch-ups don't describe the room itself and are skipped. + */ + private suspend fun updateConversationFromCatchUp(target: SyncTarget, outcome: SyncOutcome) { + val newestMessage = outcome.newestPersistedMessage + if (target.threadId != null || !outcome.persistedNewMessages || newestMessage == null) { + return + } + val conversation = + conversationsDao.getConversationForUser(target.accountId, target.roomToken).first() ?: return + + val unreadMessages = deriveUnreadMessagesCount(target, conversation.lastReadMessage) + val updatedRows = conversationsDao.updateConversationFromCatchUp( + internalId = conversation.internalId, + lastMessageJson = LoganSquare.serialize(newestMessage), + lastActivity = newestMessage.timestamp, + unreadMessages = unreadMessages + ) + Log.d( + TAG, + "Conversation list update for room ${target.roomToken} from catch-up " + + "(lastActivity=${newestMessage.timestamp}, unread=$unreadMessages): " + + if (updatedRows > 0) "applied" else "skipped, stored state is newer" + ) + } + + /** + * Derives the room's unread count from the cached messages, or [UNREAD_COUNT_UNKNOWN] when it + * cannot be derived — the count is only trustworthy when the latest chat block reaches back to + * the last read message. Own messages don't count: sending from another device advances the + * server-side read marker, which the locally cached [lastReadMessage] may lag behind. + */ + private suspend fun deriveUnreadMessagesCount(target: SyncTarget, lastReadMessage: Int): Int { + val latestBlock = if (lastReadMessage > 0) { + chatBlocksDao.getLatestChatBlock(target.internalConversationId, null).first() + } else { + null + } + val blockReachesUnreadBoundary = latestBlock != null && + (latestBlock.oldestMessageId <= lastReadMessage || !latestBlock.hasHistory) + + return if (blockReachesUnreadBoundary) { + chatDao.countMessagesNewerThan( + internalConversationId = target.internalConversationId, + messageId = lastReadMessage.toLong(), + excludedActorId = target.user.userId!! + ) + } else { + UNREAD_COUNT_UNKNOWN + } + } + /** * First fetch for a conversation whose cached messages do not reach the unread boundary yet * (typically a conversation without any chat block). @@ -467,7 +537,8 @@ class ChatMessageSyncer @Inject constructor( newestPersistedMessageId = backlogOutcome.newestPersistedMessageId ?: nextAnchor, oldestPersistedMessageId = anchorOutcome.oldestPersistedMessageId, persistedMessageCount = anchorOutcome.persistedMessageCount + backlogOutcome.persistedMessageCount, - syncFailed = backlogOutcome.syncFailed + syncFailed = backlogOutcome.syncFailed, + newestPersistedMessage = backlogOutcome.newestPersistedMessage ?: anchorOutcome.newestPersistedMessage ) } @@ -497,6 +568,7 @@ class ChatMessageSyncer @Inject constructor( var totalCount = 0 var oldestPersisted: Long? = null var newestPersisted: Long? = null + var newestPersistedMessage: ChatMessageJson? = null repeat(MAX_BACKLOG_ROUNDS) { val fieldMap = buildFieldMap( @@ -515,6 +587,7 @@ class ChatMessageSyncer @Inject constructor( totalCount += roundOutcome.persistedMessageCount oldestPersisted = oldestPersisted ?: roundOutcome.oldestPersistedMessageId newestPersisted = roundOutcome.newestPersistedMessageId ?: newestPersisted + newestPersistedMessage = roundOutcome.newestPersistedMessage ?: newestPersistedMessage } val caughtUp = !roundOutcome.persistedNewMessages || roundOutcome.persistedMessageCount < limit @@ -525,7 +598,8 @@ class ChatMessageSyncer @Inject constructor( newestPersistedMessageId = newestPersisted, oldestPersistedMessageId = oldestPersisted, persistedMessageCount = totalCount, - syncFailed = roundOutcome.syncFailed + syncFailed = roundOutcome.syncFailed, + newestPersistedMessage = newestPersistedMessage ) } anchor = nextAnchor @@ -556,7 +630,8 @@ class ChatMessageSyncer @Inject constructor( newestPersistedMessageId = fallbackOutcome.newestPersistedMessageId, oldestPersistedMessageId = fallbackOutcome.oldestPersistedMessageId, persistedMessageCount = fallbackOutcome.persistedMessageCount, - syncFailed = fallbackOutcome.syncFailed + syncFailed = fallbackOutcome.syncFailed, + newestPersistedMessage = fallbackOutcome.newestPersistedMessage ?: newestPersistedMessage ) } @@ -681,11 +756,19 @@ class ChatMessageSyncer @Inject constructor( events ) persistedMessages.maxOfOrNull { it.id }?.let { recordHttpSyncedMessageId(target, it) } + val newestPersistedMessage = if (persistedMessages.isNotEmpty()) { + result.messages + .filter { it.systemMessageType !in LAST_MESSAGE_HIDDEN_SYSTEM_TYPES } + .maxByOrNull { it.id } + } else { + null + } SyncOutcome( persistedNewMessages = persistedMessages.isNotEmpty(), newestPersistedMessageId = persistedMessages.maxOfOrNull { it.id }, oldestPersistedMessageId = persistedMessages.minOfOrNull { it.id }, - persistedMessageCount = persistedMessages.size + persistedMessageCount = persistedMessages.size, + newestPersistedMessage = newestPersistedMessage ) } else { Log.d(TAG, "No new messages to update") @@ -967,6 +1050,25 @@ class ChatMessageSyncer @Inject constructor( SyncOutcome(persistedNewMessages = false, newestPersistedMessageId = null, syncFailed = true) private const val DEFAULT_MESSAGES_LIMIT = 100 + + /** + * Marks that the unread count could not be derived locally; the guarded conversation + * update keeps the stored count in that case. + */ + private const val UNREAD_COUNT_UNKNOWN = -1 + + /** + * System messages that never become a conversation's last message on the server, so a + * locally derived conversation preview must skip them as well. + */ + private val LAST_MESSAGE_HIDDEN_SYSTEM_TYPES = setOf( + ChatMessage.SystemMessageType.REACTION, + ChatMessage.SystemMessageType.REACTION_REVOKED, + ChatMessage.SystemMessageType.REACTION_DELETED, + ChatMessage.SystemMessageType.MESSAGE_DELETED, + ChatMessage.SystemMessageType.MESSAGE_EDITED, + ChatMessage.SystemMessageType.POLL_VOTED + ) private const val MILLIS_PER_SECOND = 1000L private const val ROOM_REFRESH_MAX_AGE_MILLIS = 3 * 60 * 60 * 1000L // 3 hours private const val CATCH_UP_COOLDOWN_MILLIS = 5_000L diff --git a/app/src/main/java/com/nextcloud/talk/dagger/modules/RepositoryModule.kt b/app/src/main/java/com/nextcloud/talk/dagger/modules/RepositoryModule.kt index 800315dfea..e2c4e106ce 100644 --- a/app/src/main/java/com/nextcloud/talk/dagger/modules/RepositoryModule.kt +++ b/app/src/main/java/com/nextcloud/talk/dagger/modules/RepositoryModule.kt @@ -145,12 +145,14 @@ class RepositoryModule { fun provideChatMessageSyncer( chatMessagesDao: ChatMessagesDao, chatBlocksDao: ChatBlocksDao, + conversationsDao: ConversationsDao, dataSource: ChatNetworkDataSource, networkMonitor: NetworkMonitor ): ChatMessageSyncer = ChatMessageSyncer( chatMessagesDao, chatBlocksDao, + conversationsDao, dataSource, networkMonitor ) diff --git a/app/src/main/java/com/nextcloud/talk/data/database/dao/ChatMessagesDao.kt b/app/src/main/java/com/nextcloud/talk/data/database/dao/ChatMessagesDao.kt index 66a743da2b..95c1fe48e8 100644 --- a/app/src/main/java/com/nextcloud/talk/data/database/dao/ChatMessagesDao.kt +++ b/app/src/main/java/com/nextcloud/talk/data/database/dao/ChatMessagesDao.kt @@ -262,11 +262,32 @@ interface ChatMessagesDao { threadId: Long? ): Flow> + /** + * Counts the cached messages above [messageId] the way the server counts unread messages: + * spreed's ChatManager.getUnreadCount only counts the verbs 'comment' and 'object_shared', + * which the chat API surfaces as every messageType except 'system', 'comment_deleted', + * 'command' and 'reaction'. The user's own messages ([excludedActorId]) are additionally + * excluded: the server advances the author's read marker on every post, so own messages are + * never above the marker server-side — the exclusion compensates a lagging cached marker. + */ @Query( """ - SELECT COUNT(*) - FROM ChatMessages - WHERE internalConversationId = :internalConversationId + SELECT COUNT(*) + FROM ChatMessages + WHERE internalConversationId = :internalConversationId + AND isTemporary = 0 + AND id > :messageId + AND messageType NOT IN ('system', 'comment_deleted', 'command', 'reaction') + AND NOT (actorType = 'users' AND actorId = :excludedActorId) + """ + ) + suspend fun countMessagesNewerThan(internalConversationId: String, messageId: Long, excludedActorId: String): Int + + @Query( + """ + SELECT COUNT(*) + FROM ChatMessages + WHERE internalConversationId = :internalConversationId AND isTemporary = 0 AND (:threadId IS NULL OR threadId = :threadId) AND id BETWEEN :newestMessageId AND :oldestMessageId 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 23bc660511..386866c0d3 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 @@ -57,6 +57,32 @@ interface ConversationsDao { } } + /** + * Reflects a background message catch-up in the conversation entry. A single guarded UPDATE: + * it only applies while the derived state is newer than the stored one ([lastActivity] guard), + * so a concurrently running room list sync — which stays authoritative — can never be + * overwritten with older data and no read-modify-write window exists. An [unreadMessages] + * value below zero keeps the stored count (used when the count cannot be derived locally). + * + * @return the number of updated rows: 1 when applied, 0 when the stored state was newer. + */ + @Query( + """ + UPDATE Conversations + SET lastMessage = :lastMessageJson, + lastActivity = :lastActivity, + unreadMessages = CASE WHEN :unreadMessages >= 0 THEN :unreadMessages ELSE unreadMessages END + WHERE internalId = :internalId + AND lastActivity < :lastActivity + """ + ) + suspend fun updateConversationFromCatchUp( + internalId: String, + lastMessageJson: String, + lastActivity: Long, + unreadMessages: Int + ): Int + /** * Deletes rows in the db matching the specified [conversationIds] */ diff --git a/app/src/main/java/com/nextcloud/talk/utils/preview/ComposePreviewUtils.kt b/app/src/main/java/com/nextcloud/talk/utils/preview/ComposePreviewUtils.kt index 0da2a13a4c..31d963bbad 100644 --- a/app/src/main/java/com/nextcloud/talk/utils/preview/ComposePreviewUtils.kt +++ b/app/src/main/java/com/nextcloud/talk/utils/preview/ComposePreviewUtils.kt @@ -153,6 +153,7 @@ class ComposePreviewUtils private constructor(context: Context) { get() = ChatMessageSyncer( chatMessagesDao, chatBlocksDao, + conversationsDao, chatNetworkDataSource, networkMonitor ) 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 65dc4524f5..81337ae51b 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 @@ -117,6 +117,12 @@ class DummyChatMessagesDaoImpl : ChatMessagesDao { threadId: Long? ): Flow> = flowOf() + override suspend fun countMessagesNewerThan( + internalConversationId: String, + messageId: Long, + excludedActorId: String + ): Int = 0 + override fun getCountBetweenMessageIds( internalConversationId: String, oldestMessageId: Long, @@ -264,6 +270,13 @@ class DummyConversationDaoImpl : ConversationsDao { /* */ } + override suspend fun updateConversationFromCatchUp( + internalId: String, + lastMessageJson: String, + lastActivity: Long, + unreadMessages: Int + ): Int = 0 + override fun insertConversation(conversation: ConversationEntity) { /* */ } diff --git a/app/src/test/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncerTest.kt b/app/src/test/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncerTest.kt index 4ab3f36f5f..a346da40e3 100644 --- a/app/src/test/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncerTest.kt +++ b/app/src/test/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncerTest.kt @@ -11,6 +11,7 @@ import android.database.sqlite.SQLiteConstraintException import com.nextcloud.talk.chat.data.model.ChatMessage 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.model.ChatBlockEntity import com.nextcloud.talk.data.network.NetworkMonitor import com.nextcloud.talk.data.user.model.User @@ -52,6 +53,7 @@ class ChatMessageSyncerTest { private val chatDao: ChatMessagesDao = mock() private val chatBlocksDao: ChatBlocksDao = mock() + private val conversationsDao: ConversationsDao = mock() private val network: ChatNetworkDataSource = mock() private val networkMonitor: NetworkMonitor = mock() @@ -59,8 +61,9 @@ class ChatMessageSyncerTest { @Before fun setUp() { - syncer = ChatMessageSyncer(chatDao, chatBlocksDao, network, networkMonitor) + syncer = ChatMessageSyncer(chatDao, chatBlocksDao, conversationsDao, network, networkMonitor) whenever(networkMonitor.isOnline).thenReturn(MutableStateFlow(true)) + whenever(conversationsDao.getConversationForUser(any(), any())).thenReturn(flowOf(null)) } @Test diff --git a/app/src/test/java/com/nextcloud/talk/chat/data/network/OfflineFirstChatRepositoryTest.kt b/app/src/test/java/com/nextcloud/talk/chat/data/network/OfflineFirstChatRepositoryTest.kt index 968bf59d0b..1bc74d7f11 100644 --- a/app/src/test/java/com/nextcloud/talk/chat/data/network/OfflineFirstChatRepositoryTest.kt +++ b/app/src/test/java/com/nextcloud/talk/chat/data/network/OfflineFirstChatRepositoryTest.kt @@ -11,6 +11,7 @@ import android.os.Bundle import com.nextcloud.talk.chat.data.model.ChatMessage 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.model.ChatBlockEntity import com.nextcloud.talk.data.network.NetworkMonitor import com.nextcloud.talk.data.user.model.User @@ -50,6 +51,7 @@ class OfflineFirstChatRepositoryTest { private val logger: Logger = mock() private val chatDao: ChatMessagesDao = mock() private val chatBlocksDao: ChatBlocksDao = mock() + private val conversationsDao: ConversationsDao = mock() private val network: ChatNetworkDataSource = mock() private val networkMonitor: NetworkMonitor = mock() @@ -68,7 +70,7 @@ class OfflineFirstChatRepositoryTest { chatBlocksDao, network, networkMonitor, - ChatMessageSyncer(chatDao, chatBlocksDao, network, networkMonitor) + ChatMessageSyncer(chatDao, chatBlocksDao, conversationsDao, network, networkMonitor) ) repository.initData(user(), CREDENTIALS, CHAT_URL, ROOM_TOKEN, null) } diff --git a/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/RoomListMessagePrefetchIntegrationTest.kt b/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/RoomListMessagePrefetchIntegrationTest.kt index ee64de6ea6..9d49f2d4c6 100644 --- a/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/RoomListMessagePrefetchIntegrationTest.kt +++ b/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/RoomListMessagePrefetchIntegrationTest.kt @@ -75,7 +75,13 @@ class RoomListMessagePrefetchIntegrationTest { whenever(networkMonitor.isOnline).thenReturn(MutableStateFlow(true)) - syncer = ChatMessageSyncer(db.chatMessagesDao(), db.chatBlocksDao(), chatNetwork, networkMonitor) + syncer = ChatMessageSyncer( + db.chatMessagesDao(), + db.chatBlocksDao(), + db.conversationsDao(), + chatNetwork, + networkMonitor + ) repository = OfflineFirstConversationsRepository( db.conversationsDao(), conversationsNetwork, From fb1cc5f9c6ed09daacea7b9acdd2a237f1c647a5 Mon Sep 17 00:00:00 2001 From: Andy Scherzinger Date: Fri, 14 Aug 2026 18:16:39 +0200 Subject: [PATCH 03/10] feat(chat): write the read state locally and sync the marker durably Marking a chat as read (or unread) only sent the marker to the server in a fire-and-forget call: the local conversation entry was never updated, so the conversation list showed a stale unread badge until the next room list sync answered - and a failed call was only logged, silently losing the read state. The read state is now written into the local conversation entry immediately (with the unread count recounted from the cached messages the way the server counts), so the list reflects it the moment the user leaves the chat. Sending the marker moves to a ReadMarkerSyncWorker: network-constrained, exponential backoff, at most three attempts, and unique per room with REPLACE so the newest marker always wins and a retry can never ship a stale marker backwards. The server remains the authority, with one narrowly scoped exception: a server response computed before a concurrently sent marker reached the server would revert the entry to unread until the next sync. Markers are therefore tracked as pending until a sync confirms their delivery, and while the server's read state is provably behind a pending marker it is kept out of the merge - for the room list sync and the single-room refresh alike. Once the server has caught up (or moved past the marker, e.g. read further on another device) the server state applies unchanged, so marking as unread from another device keeps working. When sending ultimately fails the worker releases the pending marker and the next sync restores the server state, so the inconsistency window stays bounded by the retry backoff or one sync cycle. Assisted-by: Claude Code:claude-fable-5 Signed-off-by: Andy Scherzinger --- .../com/nextcloud/talk/chat/ChatActivity.kt | 35 +---- .../talk/chat/data/ChatMessageRepository.kt | 7 + .../chat/data/network/ChatMessageSyncer.kt | 46 ++++++ .../network/OfflineFirstChatRepository.kt | 4 + .../talk/chat/viewmodels/ChatViewModel.kt | 55 ++++--- .../OfflineFirstConversationsRepository.kt | 42 +++++- .../data/database/dao/ConversationsDao.kt | 15 ++ .../talk/jobs/ReadMarkerSyncWorker.kt | 138 ++++++++++++++++++ .../utils/preview/ComposePreviewUtilsDaos.kt | 4 + 9 files changed, 282 insertions(+), 64 deletions(-) create mode 100644 app/src/main/java/com/nextcloud/talk/jobs/ReadMarkerSyncWorker.kt diff --git a/app/src/main/java/com/nextcloud/talk/chat/ChatActivity.kt b/app/src/main/java/com/nextcloud/talk/chat/ChatActivity.kt index b700ff83b0..c1a9921d28 100644 --- a/app/src/main/java/com/nextcloud/talk/chat/ChatActivity.kt +++ b/app/src/main/java/com/nextcloud/talk/chat/ChatActivity.kt @@ -2790,20 +2790,7 @@ class ChatActivity : } private fun updateRemoteLastReadMessageIfNeeded() { - if (this::spreedCapabilities.isInitialized) { - spreedCapabilities?.let { - val url = ApiUtils.getUrlForChatReadMarker( - ApiUtils.getChatApiVersion(it, intArrayOf(ApiUtils.API_V1)), - conversationUser.baseUrl!!, - roomToken - ) - - chatViewModel.updateRemoteLastReadMessageIfNeeded( - credentials = credentials!!, - url = url - ) - } - } + chatViewModel.updateRemoteLastReadMessageIfNeeded() } private fun isActivityNotChangingConfigurations(): Boolean = !isChangingConfigurations @@ -3487,15 +3474,7 @@ class ChatActivity : } private fun markAsRead(messageId: Int) { - chatViewModel.setChatReadMessage( - credentials!!, - ApiUtils.getUrlForChatReadMarker( - ApiUtils.getChatApiVersion(spreedCapabilities, intArrayOf(ApiUtils.API_V1)), - conversationUser?.baseUrl!!, - roomToken - ), - messageId - ) + chatViewModel.setChatReadMessage(messageId) } fun markAsUnread(chatMessage: ChatMessage) { @@ -3510,15 +3489,7 @@ class ChatActivity : } else { 0 } - chatViewModel.setChatReadMessage( - credentials!!, - ApiUtils.getUrlForChatReadMarker( - ApiUtils.getChatApiVersion(spreedCapabilities, intArrayOf(ApiUtils.API_V1)), - conversationUser.baseUrl!!, - roomToken - ), - lastReadMessage - ) + chatViewModel.setChatReadMessage(lastReadMessage) } fun copyMessage(message: ChatMessage?) { diff --git a/app/src/main/java/com/nextcloud/talk/chat/data/ChatMessageRepository.kt b/app/src/main/java/com/nextcloud/talk/chat/data/ChatMessageRepository.kt index 5cbb1fb120..1751f73dd6 100644 --- a/app/src/main/java/com/nextcloud/talk/chat/data/ChatMessageRepository.kt +++ b/app/src/main/java/com/nextcloud/talk/chat/data/ChatMessageRepository.kt @@ -72,6 +72,13 @@ interface ChatMessageRepository : LifecycleAwareManager { */ suspend fun fetchNewMessages(): Boolean + /** + * Optimistically writes the user's read state into the local conversation entry, so the + * conversation list reflects it immediately. Sending the read marker to the server is the + * caller's concern; the next room list sync re-asserts the server state either way. + */ + suspend fun updateLocalReadState(lastReadMessage: Int) + /** * Loads messages from local storage. If the messages are not found, then it * synchronizes the database with the server, before retrying exactly once. Only diff --git a/app/src/main/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncer.kt b/app/src/main/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncer.kt index 9562994b31..240ff5e500 100644 --- a/app/src/main/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncer.kt +++ b/app/src/main/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncer.kt @@ -404,6 +404,52 @@ class ChatMessageSyncer @Inject constructor( ) } + /** + * Read markers that were written locally but whose delivery the server has not confirmed yet. + * + * A room list sync request can be answered before a marker sent at the same time reaches the + * server, so the sync response carries a provably stale read state — applying it would revert + * the conversation entry to unread until the next sync. While a marker is pending, such stale + * responses are kept out of the merge; the entry is only released once a sync confirms the + * marker (server read state caught up) or sending it ultimately failed, so the server's + * authority is restored either way and marking as unread from another device stays possible. + */ + private val pendingReadMarkers = ConcurrentHashMap() + + /** + * The locally written but not yet server-confirmed read marker of the conversation, or null. + */ + fun pendingReadMarker(internalConversationId: String): Int? = pendingReadMarkers[internalConversationId] + + /** + * Releases a pending read marker: called when a room list sync confirmed it or when sending it + * ultimately failed. Only removes [lastReadMessage] itself, so a newer marker written in the + * meantime stays pending. + */ + fun clearPendingReadMarker(internalConversationId: String, lastReadMessage: Int) { + pendingReadMarkers.remove(internalConversationId, lastReadMessage) + } + + /** + * Optimistically writes the user's read state into the conversation entry, so the + * conversation list reflects it immediately — before (and independent of) the read marker + * reaching the server. The unread count is recounted from the cached messages above the new + * marker. The server stays the authority: the next room list sync re-asserts its state, with + * [pendingReadMarkers] bridging the window until the marker's delivery is confirmed. + */ + suspend fun updateLocalReadState(target: SyncTarget, lastReadMessage: Int) { + val conversation = + conversationsDao.getConversationForUser(target.accountId, target.roomToken).first() ?: return + val unreadMessages = chatDao.countMessagesNewerThan( + internalConversationId = target.internalConversationId, + messageId = lastReadMessage.toLong(), + excludedActorId = target.user.userId!! + ) + pendingReadMarkers[target.internalConversationId] = lastReadMessage + conversationsDao.updateReadState(conversation.internalId, lastReadMessage, unreadMessages) + Log.d(TAG, "Local read state for room ${target.roomToken}: lastRead=$lastReadMessage, unread=$unreadMessages") + } + /** * Derives the room's unread count from the cached messages, or [UNREAD_COUNT_UNKNOWN] when it * cannot be derived — the count is only trustworthy when the latest chat block reaches back to diff --git a/app/src/main/java/com/nextcloud/talk/chat/data/network/OfflineFirstChatRepository.kt b/app/src/main/java/com/nextcloud/talk/chat/data/network/OfflineFirstChatRepository.kt index 9f54fddbd1..57d06b98e6 100644 --- a/app/src/main/java/com/nextcloud/talk/chat/data/network/OfflineFirstChatRepository.kt +++ b/app/src/main/java/com/nextcloud/talk/chat/data/network/OfflineFirstChatRepository.kt @@ -363,6 +363,10 @@ class OfflineFirstChatRepository @Inject constructor( return outcome.persistedNewMessages } + override suspend fun updateLocalReadState(lastReadMessage: Int) { + syncer.updateLocalReadState(syncTarget, lastReadMessage) + } + override suspend fun loadMoreMessages( anchorMessageId: Long, direction: ChatMessageRepository.LoadMoreDirection, diff --git a/app/src/main/java/com/nextcloud/talk/chat/viewmodels/ChatViewModel.kt b/app/src/main/java/com/nextcloud/talk/chat/viewmodels/ChatViewModel.kt index 3bbcfc3716..4130e6caed 100644 --- a/app/src/main/java/com/nextcloud/talk/chat/viewmodels/ChatViewModel.kt +++ b/app/src/main/java/com/nextcloud/talk/chat/viewmodels/ChatViewModel.kt @@ -40,6 +40,7 @@ import com.nextcloud.talk.data.database.mappers.toDomainModel import com.nextcloud.talk.data.database.model.ChatMessageEntity import com.nextcloud.talk.data.user.model.User import com.nextcloud.talk.extensions.toIntOrZero +import com.nextcloud.talk.jobs.ReadMarkerSyncWorker import com.nextcloud.talk.jobs.ShareOperationWorker import com.nextcloud.talk.jobs.UploadAndShareFilesWorker import com.nextcloud.talk.logger.Logger @@ -1799,45 +1800,39 @@ class ChatViewModel @AssistedInject constructor( /** * Please use with caution to not spam the server */ - fun updateRemoteLastReadMessageIfNeeded(credentials: String, url: String) { + fun updateRemoteLastReadMessageIfNeeded() { Log.d(TAG, "updateRemoteLastReadMessageIfNeeded, localLastReadMessage: $localLastReadMessage") - Log.d( - TAG, - "updateRemoteLastReadMessageIfNeeded, _uiState.value.conversation!!.lastReadMessage: " + - _uiState.value.conversation!!.lastReadMessage - ) + val conversationLastReadMessage = _uiState.value.conversation?.lastReadMessage ?: return + Log.d(TAG, "updateRemoteLastReadMessageIfNeeded, conversation.lastReadMessage: $conversationLastReadMessage") - if (localLastReadMessage > _uiState.value.conversation!!.lastReadMessage) { + if (localLastReadMessage > conversationLastReadMessage) { Log.d(TAG, "updateRemoteLastReadMessageIfNeeded, setChatReadMessage...") - setChatReadMessage(credentials, url, localLastReadMessage) + setChatReadMessage(localLastReadMessage) } } /** - * Please use with caution to not spam the server + * Marks the chat as read up to [lastReadMessage]: the local conversation entry is updated + * immediately (optimistic, so the conversation list reflects it right away) while sending the + * marker to the server is delegated to [ReadMarkerSyncWorker], which retries transient + * failures with backoff. The server stays the authority — every room list sync re-asserts its + * read state, so a marker that ultimately could not be sent falls back to the server state + * instead of leaving the client diverged. */ - fun setChatReadMessage(credentials: String, url: String, lastReadMessage: Int) { - chatNetworkDataSource.setChatReadMarker(credentials, url, lastReadMessage) - .subscribeOn(Schedulers.io()) - .observeOn(AndroidSchedulers.mainThread()) - .subscribe(object : Observer { - override fun onSubscribe(d: Disposable) { - disposableSet.add(d) - } - - override fun onError(e: Throwable) { - Log.e(TAG, e.message, e) - } - - override fun onComplete() { - // unused atm - } - - override fun onNext(t: GenericOverall) { - // unused atm - } - }) + fun setChatReadMessage(lastReadMessage: Int) { + if (!this::currentUser.isInitialized) { + return + } + viewModelScope.launch { + chatRepository.updateLocalReadState(lastReadMessage) + } + ReadMarkerSyncWorker.enqueue( + context = NextcloudTalkApplication.sharedApplication!!.applicationContext, + userId = currentUser.id!!, + roomToken = chatRoomToken, + lastReadMessage = lastReadMessage + ) } fun shareToNotes(credentials: String, url: String, message: String, displayName: String) { 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 7a7dfd4050..d153639888 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,7 +142,9 @@ class OfflineFirstConversationsRepository @Inject constructor( val existingEntity = dao.getConversationForUser(user.id!!, model.token).first() model.hiddenUpcomingEvent = existingEntity?.hiddenUpcomingEvent _conversationFlow.emit(model) - val entityList = listOf(model.asEntity()) + val previous = existingEntity?.let { mapOf(it.internalId to it) }.orEmpty() + val entityList = + preserveReadStateOfPendingMarkers(previous, listOf(model.asEntity())) dao.upsertConversations(user.id!!, entityList) } } @@ -185,7 +187,7 @@ class OfflineFirstConversationsRepository @Inject constructor( dao.syncConversationsForUser( accountId = user.id!!, - serverItems = conversationsFromSync, + serverItems = preserveReadStateOfPendingMarkers(previousConversations, conversationsFromSync), conversationIdsToDelete = determineLeftConversationIds(previousConversations, conversationsFromSync) ) @@ -295,6 +297,42 @@ class OfflineFirstConversationsRepository @Inject constructor( connectivityManager.restrictBackgroundStatus == ConnectivityManager.RESTRICT_BACKGROUND_STATUS_ENABLED } + /** + * Keeps provably stale read states out of the merge of server responses (the room list sync + * and the single-room refresh of [getRoom]): a response computed before a concurrently sent + * read marker reached the server still reports the room as unread, and applying it would + * revert the conversation entry until the next sync. While a marker for the room is pending + * and the server's read state is still behind it, the local read state is kept; once the + * server has caught up (or moved past it, e.g. read further on another device) the marker is + * released and the server state applies unchanged — including a lower one, so marking as + * unread from another device keeps working. + */ + private fun preserveReadStateOfPendingMarkers( + previousConversations: Map, + conversationsFromSync: List + ): List = + conversationsFromSync.map { serverItem -> + val pendingMarker = chatMessageSyncer.pendingReadMarker(serverItem.internalId) + ?: return@map serverItem + val previous = previousConversations[serverItem.internalId] + ?: return@map serverItem + + if (serverItem.lastReadMessage < pendingMarker) { + Log.d( + TAG, + "Keeping local read state for room ${serverItem.token}: server lastRead=" + + "${serverItem.lastReadMessage} is behind the pending read marker $pendingMarker" + ) + serverItem.copy( + lastReadMessage = previous.lastReadMessage, + unreadMessages = previous.unreadMessages + ) + } else { + chatMessageSyncer.clearPendingReadMarker(serverItem.internalId, pendingMarker) + serverItem + } + } + private fun determineLeftConversationIds( previousConversations: Map, conversationsFromSync: List 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 386866c0d3..1008111556 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 @@ -83,6 +83,21 @@ interface ConversationsDao { unreadMessages: Int ): Int + /** + * Optimistically writes the user's read state for a conversation. Deliberately unguarded: + * marking as unread moves the read marker backwards, so a user action always wins locally — + * and the next room list sync re-asserts the server state either way. + */ + @Query( + """ + UPDATE Conversations + SET lastReadMessage = :lastReadMessage, + unreadMessages = :unreadMessages + WHERE internalId = :internalId + """ + ) + suspend fun updateReadState(internalId: String, lastReadMessage: Int, unreadMessages: 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 new file mode 100644 index 0000000000..a10a748d4e --- /dev/null +++ b/app/src/main/java/com/nextcloud/talk/jobs/ReadMarkerSyncWorker.kt @@ -0,0 +1,138 @@ +/* + * 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.content.Context +import android.util.Log +import androidx.work.BackoffPolicy +import androidx.work.Constraints +import androidx.work.CoroutineWorker +import androidx.work.Data +import androidx.work.ExistingWorkPolicy +import androidx.work.NetworkType +import androidx.work.OneTimeWorkRequest +import androidx.work.WorkManager +import androidx.work.WorkRequest +import androidx.work.WorkerParameters +import autodagger.AutoInjector +import com.nextcloud.talk.application.NextcloudTalkApplication +import com.nextcloud.talk.application.NextcloudTalkApplication.Companion.sharedApplication +import com.nextcloud.talk.chat.data.network.ChatMessageSyncer +import com.nextcloud.talk.chat.data.network.ChatNetworkDataSource +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 java.util.concurrent.TimeUnit +import javax.inject.Inject + +/** + * Sends the user's read marker to the server, retrying transient failures with backoff. + * + * The caller updates the local conversation entry optimistically before enqueuing this worker, so + * the conversation list reflects the read state immediately. The server stays the authority: + * every room list sync re-asserts the server's read state over the local entry — guarded by the + * pending marker in [ChatMessageSyncer] only while the marker is provably not delivered yet. + * When all attempts fail, the pending marker is released so the client falls back to the server + * state at the next sync instead of staying diverged. Work is unique per room with + * [ExistingWorkPolicy.REPLACE], so the newest marker for a room always wins and retries can never + * ship a stale marker backwards. + */ +@AutoInjector(NextcloudTalkApplication::class) +class ReadMarkerSyncWorker(context: Context, workerParams: WorkerParameters) : + CoroutineWorker(context, workerParams) { + + @Inject + lateinit var userManager: UserManager + + @Inject + lateinit var chatNetworkDataSource: ChatNetworkDataSource + + @Inject + lateinit var chatMessageSyncer: ChatMessageSyncer + + override suspend fun doWork(): Result { + sharedApplication!!.componentApplication.inject(this) + + 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) + + return when { + userId < 0 || roomToken.isNullOrEmpty() || lastReadMessage < 0 -> { + Log.e(TAG, "Missing user id, room token or read marker, dropping read marker sync") + Result.failure() + } + + else -> sendReadMarker(userId, roomToken, lastReadMessage) + } + } + + private fun sendReadMarker(userId: Long, roomToken: String, lastReadMessage: Int): Result { + val user = userManager.getUserWithId(userId).blockingGet() + val credentials = user?.let { ApiUtils.getCredentials(it.username, it.token) } + if (user == null || credentials == null) { + Log.e(TAG, "No user or credentials found for user id $userId, dropping read marker sync") + return fail(userId, roomToken, lastReadMessage) + } + + val sent = runCatching { + val url = ApiUtils.getUrlForChatReadMarker( + ApiUtils.getChatApiVersion(user.capabilities!!.spreedCapability!!, intArrayOf(ApiUtils.API_V1)), + user.baseUrl!!, + roomToken + ) + chatNetworkDataSource.setChatReadMarker(credentials, url, lastReadMessage).blockingSingle() + }.isSuccess + + return if (sent) { + Log.d(TAG, "Read marker $lastReadMessage sent for room $roomToken") + Result.success() + } else { + Log.w(TAG, "Sending read marker for room $roomToken failed (attempt ${runAttemptCount + 1})") + retryOrFail(userId, roomToken, lastReadMessage) + } + } + + private fun retryOrFail(userId: Long, roomToken: String, lastReadMessage: Int): Result = + if (runAttemptCount < MAX_RUN_ATTEMPTS - 1) { + Result.retry() + } else { + fail(userId, roomToken, lastReadMessage) + } + + private fun fail(userId: Long, roomToken: String, lastReadMessage: Int): Result { + chatMessageSyncer.clearPendingReadMarker("$userId@$roomToken", lastReadMessage) + return Result.failure() + } + + companion object { + 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 + + fun enqueue(context: Context, userId: Long, roomToken: String, lastReadMessage: Int) { + val data = Data.Builder() + .putLong(KEY_INTERNAL_USER_ID, userId) + .putString(KEY_ROOM_TOKEN, roomToken) + .putInt(KEY_LAST_READ_MESSAGE, lastReadMessage) + .build() + + val readMarkerWork = OneTimeWorkRequest.Builder(ReadMarkerSyncWorker::class.java) + .setInputData(data) + .setConstraints(Constraints.Builder().setRequiredNetworkType(NetworkType.CONNECTED).build()) + .setBackoffCriteria(BackoffPolicy.EXPONENTIAL, WorkRequest.MIN_BACKOFF_MILLIS, TimeUnit.MILLISECONDS) + .build() + + WorkManager.getInstance(context).enqueueUniqueWork( + "read-marker-sync-$userId@$roomToken", + ExistingWorkPolicy.REPLACE, + readMarkerWork + ) + } + } +} 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 81337ae51b..d13ac5ca90 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 @@ -277,6 +277,10 @@ class DummyConversationDaoImpl : ConversationsDao { unreadMessages: Int ): Int = 0 + override suspend fun updateReadState(internalId: String, lastReadMessage: Int, unreadMessages: Int) { + /* */ + } + override fun insertConversation(conversation: ConversationEntity) { /* */ } From a25a4bae08b01c352ac7629098cb01dc032eb76e Mon Sep 17 00:00:00 2001 From: Andy Scherzinger Date: Fri, 14 Aug 2026 18:17:12 +0200 Subject: [PATCH 04/10] test(conversations): cover conversation list freshness end to end Covers, against a real in-memory database, that a background catch-up updates the conversation entry from the fetched messages (preview, activity, derived unread count with system messages excluded the way the server excludes them), that the guarded update never regresses a newer stored entry, that the local read state write-through resets and recounts the unread badge (own messages excluded), that neither a room list sync nor a single-room refresh can revert the read state while its marker is pending - with the server's authority restored once it confirms the marker, including a lower read state from marking unread on another device - and that the reactive room list flow delivers plain database writes without any fetch call. Assisted-by: Claude Code:claude-fable-5 Signed-off-by: Andy Scherzinger --- ...onversationListFreshnessIntegrationTest.kt | 356 ++++++++++++++++++ 1 file changed, 356 insertions(+) create mode 100644 app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListFreshnessIntegrationTest.kt 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 new file mode 100644 index 0000000000..60cca7b2b2 --- /dev/null +++ b/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListFreshnessIntegrationTest.kt @@ -0,0 +1,356 @@ +/* + * 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 android.app.Application +import android.content.Context +import androidx.room.Room +import androidx.test.core.app.ApplicationProvider +import com.bluelinelabs.logansquare.LoganSquare +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.mappers.asEntity +import com.nextcloud.talk.data.database.model.ConversationEntity +import com.nextcloud.talk.data.network.NetworkMonitor +import com.nextcloud.talk.data.source.local.TalkDatabase +import com.nextcloud.talk.data.user.model.User +import com.nextcloud.talk.data.user.model.UserEntity +import com.nextcloud.talk.models.domain.ConversationModel +import com.nextcloud.talk.models.json.capabilities.Capabilities +import com.nextcloud.talk.models.json.capabilities.SpreedCapability +import com.nextcloud.talk.models.json.chat.ChatMessageJson +import com.nextcloud.talk.models.json.chat.ChatOCS +import com.nextcloud.talk.models.json.chat.ChatOverall +import com.nextcloud.talk.models.json.conversations.Conversation +import com.nextcloud.talk.utils.ApiUtils +import io.reactivex.Observable +import io.reactivex.android.plugins.RxAndroidPlugins +import io.reactivex.schedulers.Schedulers +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.launch +import kotlinx.coroutines.runBlocking +import org.junit.After +import org.junit.Assert.assertEquals +import org.junit.Before +import org.junit.Test +import org.junit.runner.RunWith +import org.mockito.kotlin.any +import org.mockito.kotlin.mock +import org.mockito.kotlin.whenever +import org.mockito.kotlin.wheneverBlocking +import org.robolectric.RobolectricTestRunner +import org.robolectric.annotation.Config +import retrofit2.Response + +/** + * Integration tests for the conversation list freshness: background catch-ups and local read + * state changes must be visible in the conversation entries (and thus the reactive room list) + * without waiting for the next full room list sync — while a newer stored state must never be + * regressed by a catch-up. + */ +@RunWith(RobolectricTestRunner::class) +@Config(application = Application::class, sdk = [33]) +@Suppress("TooManyFunctions") +class ConversationListFreshnessIntegrationTest { + + private lateinit var db: TalkDatabase + private lateinit var syncer: ChatMessageSyncer + + private val chatNetwork: ChatNetworkDataSource = mock() + private val conversationsNetwork: ConversationsNetworkDataSource = mock() + private val networkMonitor: NetworkMonitor = mock() + + @Before + fun setUp() { + RxAndroidPlugins.setInitMainThreadSchedulerHandler { Schedulers.trampoline() } + RxAndroidPlugins.setMainThreadSchedulerHandler { Schedulers.trampoline() } + + val context = ApplicationProvider.getApplicationContext() + db = Room.inMemoryDatabaseBuilder(context, TalkDatabase::class.java) + .allowMainThreadQueries() + .build() + db.usersDao().saveUser(UserEntity(id = ACCOUNT_ID, userId = "me", username = "me", baseUrl = BASE_URL)) + + whenever(networkMonitor.isOnline).thenReturn(MutableStateFlow(true)) + + syncer = ChatMessageSyncer( + db.chatMessagesDao(), + db.chatBlocksDao(), + db.conversationsDao(), + chatNetwork, + networkMonitor + ) + } + + @After + fun tearDown() { + db.close() + RxAndroidPlugins.reset() + } + + @Test + fun `background catch-up updates the conversation entry from the fetched messages`() { + runBlocking { + seedConversation(lastActivity = 10, lastReadMessage = 10, unreadMessages = 0) + wheneverBlocking { chatNetwork.pullChatMessages(any(), any(), any()) } + .thenReturn( + Response.success( + overall( + message(10), + message(11), + message(12), + message( + id = 13, + messageType = "system", + systemMessageType = ChatMessage.SystemMessageType.USER_ADDED + ) + ) + ) + ) + + syncer.catchUpRoom(target(), lastReadMessage = 10, unreadMessages = 2) + + val conversation = conversationEntity() + assertEquals("activity must advance to the newest message", 13L, conversation.lastActivity) + assertEquals("system messages must not count as unread", 2, conversation.unreadMessages) + assertEquals("the read marker must not be touched", 10, conversation.lastReadMessage) + val lastMessage = LoganSquare.parse(conversation.lastMessage, ChatMessageJson::class.java) + assertEquals("the newest fetched message must become the preview", 13L, lastMessage.id) + } + } + + @Test + fun `a single-room refresh cannot revert the read state while its marker is pending`() { + val user = user(withKeepNotificationsCapability = false) + val repository = OfflineFirstConversationsRepository( + db.conversationsDao(), + conversationsNetwork, + chatNetwork, + networkMonitor, + syncer, + ApplicationProvider.getApplicationContext() + ) + whenever(chatNetwork.getRoom(any(), any())).thenReturn( + Observable.just( + ConversationModel.mapToConversationModel( + staleServerRoom(lastReadMessage = 10, unreadMessages = 2), + user + ) + ) + ) + + runBlocking { + seedConversation(lastActivity = 12, lastReadMessage = 10, unreadMessages = 2) + syncer.updateLocalReadState(target(), lastReadMessage = 12) + + repository.getRoom(user, ROOM_TOKEN).join() + awaitUntil { conversationEntity().lastActivity == 12L } + + assertEquals("stale room refresh must not revert the read marker", 12, conversationEntity().lastReadMessage) + assertEquals("stale room refresh must not revert the unread count", 0, conversationEntity().unreadMessages) + } + } + + @Test + fun `background catch-up never regresses a newer conversation entry`() { + runBlocking { + seedConversation(lastActivity = 20, lastReadMessage = 10, unreadMessages = 5) + wheneverBlocking { chatNetwork.pullChatMessages(any(), any(), any()) } + .thenReturn( + Response.success(overall(message(10), message(11), message(12))) + ) + + syncer.catchUpRoom(target(), lastReadMessage = 10, unreadMessages = 2) + + val conversation = conversationEntity() + assertEquals("a newer stored activity must win", 20L, conversation.lastActivity) + assertEquals("the stored unread count must be kept", 5, conversation.unreadMessages) + } + } + + @Test + fun `local read state write-through updates the conversation entry immediately`() { + runBlocking { + seedConversation(lastActivity = 12, lastReadMessage = 10, unreadMessages = 2) + wheneverBlocking { chatNetwork.pullChatMessages(any(), any(), any()) } + .thenReturn( + Response.success(overall(message(10), message(11), message(12, actorId = "me"))) + ) + syncer.catchUpRoom(target(), lastReadMessage = 10, unreadMessages = 2) + + syncer.updateLocalReadState(target(), lastReadMessage = 12) + assertEquals(12, conversationEntity().lastReadMessage) + assertEquals("everything below the marker is read", 0, conversationEntity().unreadMessages) + + syncer.updateLocalReadState(target(), lastReadMessage = 10) + assertEquals(10, conversationEntity().lastReadMessage) + assertEquals("own messages must not count as unread", 1, conversationEntity().unreadMessages) + } + } + + @Test + fun `a sync cannot revert the read state while its marker is pending`() { + val user = user(withKeepNotificationsCapability = false) + val repository = OfflineFirstConversationsRepository( + db.conversationsDao(), + conversationsNetwork, + chatNetwork, + networkMonitor, + syncer, + ApplicationProvider.getApplicationContext() + ) + whenever(conversationsNetwork.getRooms(any(), any(), any())).thenReturn( + Observable.just(listOf(staleServerRoom(lastReadMessage = 10, unreadMessages = 2))), + Observable.just(listOf(staleServerRoom(lastReadMessage = 12, unreadMessages = 0))), + Observable.just(listOf(staleServerRoom(lastReadMessage = 8, unreadMessages = 4))) + ) + + runBlocking { + seedConversation(lastActivity = 12, lastReadMessage = 10, unreadMessages = 2) + syncer.updateLocalReadState(target(), lastReadMessage = 12) + assertEquals(12, conversationEntity().lastReadMessage) + assertEquals(0, conversationEntity().unreadMessages) + + repository.getRooms(user).join() + assertEquals("stale sync must not revert the read marker", 12, conversationEntity().lastReadMessage) + assertEquals("stale sync must not revert the unread count", 0, conversationEntity().unreadMessages) + + repository.getRooms(user).join() + assertEquals(12, conversationEntity().lastReadMessage) + assertEquals(0, conversationEntity().unreadMessages) + + repository.getRooms(user).join() + assertEquals("server authority must be restored", 8, conversationEntity().lastReadMessage) + assertEquals("marking unread on another device must apply", 4, conversationEntity().unreadMessages) + } + } + + @Test + fun `room list flow reflects database writes reactively`() { + val user = user(withKeepNotificationsCapability = false) + val repository = OfflineFirstConversationsRepository( + db.conversationsDao(), + conversationsNetwork, + chatNetwork, + networkMonitor, + syncer, + ApplicationProvider.getApplicationContext() + ) + whenever(conversationsNetwork.getRooms(any(), any(), any())).thenReturn( + Observable.just(listOf(Conversation(token = ROOM_TOKEN, lastActivity = 10, unreadMessages = 1))) + ) + + runBlocking { + val emissions = mutableListOf>() + val collector = launch(Dispatchers.IO) { + repository.roomListFlow.collect { emissions.add(it) } + } + + repository.getRooms(user).join() + awaitUntil { emissions.lastOrNull()?.firstOrNull()?.unreadMessages == 1 } + + db.conversationsDao().updateReadState(INTERNAL_CONVERSATION_ID, lastReadMessage = 99, unreadMessages = 7) + awaitUntil { emissions.lastOrNull()?.firstOrNull()?.unreadMessages == 7 } + + collector.cancel() + } + } + + private fun staleServerRoom(lastReadMessage: Int, unreadMessages: Int): Conversation = + Conversation( + token = ROOM_TOKEN, + lastActivity = 12, + lastReadMessage = lastReadMessage, + unreadMessages = unreadMessages + ) + + private suspend fun seedConversation(lastActivity: Long, lastReadMessage: Int, unreadMessages: Int) { + val entity = Conversation( + token = ROOM_TOKEN, + lastActivity = lastActivity, + lastReadMessage = lastReadMessage, + unreadMessages = unreadMessages, + lastMessage = message(lastReadMessage.toLong()) + ).asEntity(ACCOUNT_ID) + db.conversationsDao().upsertConversations(ACCOUNT_ID, listOf(entity)) + } + + private suspend fun conversationEntity(): ConversationEntity = + db.conversationsDao().getConversationForUser(ACCOUNT_ID, ROOM_TOKEN).first()!! + + private suspend fun awaitUntil(timeoutMillis: Long = TIMEOUT_MILLIS, condition: suspend () -> Boolean) { + val start = System.currentTimeMillis() + while (!condition()) { + if (System.currentTimeMillis() - start > timeoutMillis) { + throw AssertionError("Condition not met within $timeoutMillis ms") + } + delay(POLL_INTERVAL_MILLIS) + } + } + + private fun user(withKeepNotificationsCapability: Boolean = true): User { + val features = if (withKeepNotificationsCapability) { + listOf("chat-keep-notifications") + } else { + emptyList() + } + return User( + id = ACCOUNT_ID, + userId = "me", + username = "me", + baseUrl = BASE_URL, + token = "app-password", + capabilities = Capabilities().apply { + spreedCapability = SpreedCapability().apply { this.features = features } + } + ) + } + + private fun target(): ChatMessageSyncer.SyncTarget = + ChatMessageSyncer.SyncTarget( + user = user(), + roomToken = ROOM_TOKEN, + threadId = null, + credentials = "credentials", + urlForChatting = ApiUtils.getUrlForChat(1, BASE_URL, ROOM_TOKEN) + ) + + private fun message( + id: Long, + actorId: String = "other", + messageType: String = "comment", + systemMessageType: ChatMessage.SystemMessageType = ChatMessage.SystemMessageType.DUMMY + ): ChatMessageJson = + ChatMessageJson( + id = id, + token = ROOM_TOKEN, + actorType = "users", + actorId = actorId, + actorDisplayName = "Actor $actorId", + timestamp = id, + message = "message $id", + messageType = messageType, + systemMessageType = systemMessageType + ) + + private fun overall(vararg messages: ChatMessageJson): ChatOverall = + ChatOverall(ocs = ChatOCS(meta = null, data = messages.toList())) + + companion object { + private const val ACCOUNT_ID = 1L + private const val BASE_URL = "https://server.example.com" + private const val ROOM_TOKEN = "room1" + private const val INTERNAL_CONVERSATION_ID = "$ACCOUNT_ID@$ROOM_TOKEN" + private const val TIMEOUT_MILLIS = 10_000L + private const val POLL_INTERVAL_MILLIS = 50L + } +} From 3f57cc3a29d803576c01567ff48c22b082d2394f Mon Sep 17 00:00:00 2001 From: Andy Scherzinger Date: Fri, 14 Aug 2026 18:31:21 +0200 Subject: [PATCH 05/10] refactor(conversations): extract ConversationListUpdater from the syncer ChatMessageSyncer had grown into owning message fetching, chat blocks, catch-up coalescing, HTTP sync anchors and - since the conversation list freshness work - conversation entry updates, the local read state and the pending read markers. Move the conversation-list concerns into a dedicated ConversationListUpdater: reflecting catch-ups in the conversation entries, the optimistic read state write-through, and the pending markers with their stale-server-response guard. Behavior is unchanged; the syncer delegates the catch-up reflection, and the chat repository, the read marker worker and the conversations repository now use the updater directly instead of going through the syncer. Assisted-by: Claude Code:claude-fable-5 Signed-off-by: Andy Scherzinger --- .../chat/data/network/ChatMessageSyncer.kt | 135 +----------- .../network/OfflineFirstChatRepository.kt | 6 +- .../data/network/ConversationListUpdater.kt | 203 ++++++++++++++++++ .../OfflineFirstConversationsRepository.kt | 48 +---- .../talk/dagger/modules/RepositoryModule.kt | 30 ++- .../talk/jobs/ReadMarkerSyncWorker.kt | 8 +- .../talk/utils/preview/ComposePreviewUtils.kt | 16 +- .../data/network/ChatMessageSyncerTest.kt | 4 +- .../network/OfflineFirstChatRepositoryTest.kt | 10 +- ...onversationListFreshnessIntegrationTest.kt | 18 +- .../RoomListMessagePrefetchIntegrationTest.kt | 7 +- 11 files changed, 291 insertions(+), 194 deletions(-) create mode 100644 app/src/main/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdater.kt diff --git a/app/src/main/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncer.kt b/app/src/main/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncer.kt index 240ff5e500..99566ff554 100644 --- a/app/src/main/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncer.kt +++ b/app/src/main/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncer.kt @@ -10,12 +10,11 @@ package com.nextcloud.talk.chat.data.network import android.database.sqlite.SQLiteConstraintException import android.os.SystemClock import android.util.Log -import com.bluelinelabs.logansquare.LoganSquare import com.nextcloud.talk.chat.data.model.ChatMessage import com.nextcloud.talk.chat.domain.ChatPullResult +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 com.nextcloud.talk.data.database.mappers.asEntity import com.nextcloud.talk.data.database.model.ChatBlockEntity import com.nextcloud.talk.data.database.model.ChatMessageEntity @@ -49,9 +48,9 @@ import javax.inject.Inject class ChatMessageSyncer @Inject constructor( private val chatDao: ChatMessagesDao, private val chatBlocksDao: ChatBlocksDao, - private val conversationsDao: ConversationsDao, private val network: ChatNetworkDataSource, - private val networkMonitor: NetworkMonitor + private val networkMonitor: NetworkMonitor, + private val conversationListUpdater: ConversationListUpdater ) { /** @@ -365,117 +364,11 @@ class ChatMessageSyncer @Inject constructor( Log.d(TAG, "Background catch-up for room ${target.roomToken}: no new messages") } - updateConversationFromCatchUp(target, outcome) + conversationListUpdater.updateConversationFromCatchUp(target, outcome) return outcome } - /** - * Reflects a background catch-up in the conversation list: the room's cached conversation - * entry is updated with the newest persisted message, its activity timestamp and a locally - * derived unread count, so the list is up to date the moment it is opened — before the room - * list sync (which stays in place as the authority and self-healing safeguard) has answered. - * - * Concurrency: runs inside the per-room catch-up mutex and writes via a single guarded UPDATE - * that only applies when the derived state is newer than the stored one, so a concurrently - * finishing room list sync can neither be overwritten with older data nor interleave with a - * read-modify-write. Thread catch-ups don't describe the room itself and are skipped. - */ - private suspend fun updateConversationFromCatchUp(target: SyncTarget, outcome: SyncOutcome) { - val newestMessage = outcome.newestPersistedMessage - if (target.threadId != null || !outcome.persistedNewMessages || newestMessage == null) { - return - } - val conversation = - conversationsDao.getConversationForUser(target.accountId, target.roomToken).first() ?: return - - val unreadMessages = deriveUnreadMessagesCount(target, conversation.lastReadMessage) - val updatedRows = conversationsDao.updateConversationFromCatchUp( - internalId = conversation.internalId, - lastMessageJson = LoganSquare.serialize(newestMessage), - lastActivity = newestMessage.timestamp, - unreadMessages = unreadMessages - ) - Log.d( - TAG, - "Conversation list update for room ${target.roomToken} from catch-up " + - "(lastActivity=${newestMessage.timestamp}, unread=$unreadMessages): " + - if (updatedRows > 0) "applied" else "skipped, stored state is newer" - ) - } - - /** - * Read markers that were written locally but whose delivery the server has not confirmed yet. - * - * A room list sync request can be answered before a marker sent at the same time reaches the - * server, so the sync response carries a provably stale read state — applying it would revert - * the conversation entry to unread until the next sync. While a marker is pending, such stale - * responses are kept out of the merge; the entry is only released once a sync confirms the - * marker (server read state caught up) or sending it ultimately failed, so the server's - * authority is restored either way and marking as unread from another device stays possible. - */ - private val pendingReadMarkers = ConcurrentHashMap() - - /** - * The locally written but not yet server-confirmed read marker of the conversation, or null. - */ - fun pendingReadMarker(internalConversationId: String): Int? = pendingReadMarkers[internalConversationId] - - /** - * Releases a pending read marker: called when a room list sync confirmed it or when sending it - * ultimately failed. Only removes [lastReadMessage] itself, so a newer marker written in the - * meantime stays pending. - */ - fun clearPendingReadMarker(internalConversationId: String, lastReadMessage: Int) { - pendingReadMarkers.remove(internalConversationId, lastReadMessage) - } - - /** - * Optimistically writes the user's read state into the conversation entry, so the - * conversation list reflects it immediately — before (and independent of) the read marker - * reaching the server. The unread count is recounted from the cached messages above the new - * marker. The server stays the authority: the next room list sync re-asserts its state, with - * [pendingReadMarkers] bridging the window until the marker's delivery is confirmed. - */ - suspend fun updateLocalReadState(target: SyncTarget, lastReadMessage: Int) { - val conversation = - conversationsDao.getConversationForUser(target.accountId, target.roomToken).first() ?: return - val unreadMessages = chatDao.countMessagesNewerThan( - internalConversationId = target.internalConversationId, - messageId = lastReadMessage.toLong(), - excludedActorId = target.user.userId!! - ) - pendingReadMarkers[target.internalConversationId] = lastReadMessage - conversationsDao.updateReadState(conversation.internalId, lastReadMessage, unreadMessages) - Log.d(TAG, "Local read state for room ${target.roomToken}: lastRead=$lastReadMessage, unread=$unreadMessages") - } - - /** - * Derives the room's unread count from the cached messages, or [UNREAD_COUNT_UNKNOWN] when it - * cannot be derived — the count is only trustworthy when the latest chat block reaches back to - * the last read message. Own messages don't count: sending from another device advances the - * server-side read marker, which the locally cached [lastReadMessage] may lag behind. - */ - private suspend fun deriveUnreadMessagesCount(target: SyncTarget, lastReadMessage: Int): Int { - val latestBlock = if (lastReadMessage > 0) { - chatBlocksDao.getLatestChatBlock(target.internalConversationId, null).first() - } else { - null - } - val blockReachesUnreadBoundary = latestBlock != null && - (latestBlock.oldestMessageId <= lastReadMessage || !latestBlock.hasHistory) - - return if (blockReachesUnreadBoundary) { - chatDao.countMessagesNewerThan( - internalConversationId = target.internalConversationId, - messageId = lastReadMessage.toLong(), - excludedActorId = target.user.userId!! - ) - } else { - UNREAD_COUNT_UNKNOWN - } - } - /** * First fetch for a conversation whose cached messages do not reach the unread boundary yet * (typically a conversation without any chat block). @@ -804,7 +697,7 @@ class ChatMessageSyncer @Inject constructor( persistedMessages.maxOfOrNull { it.id }?.let { recordHttpSyncedMessageId(target, it) } val newestPersistedMessage = if (persistedMessages.isNotEmpty()) { result.messages - .filter { it.systemMessageType !in LAST_MESSAGE_HIDDEN_SYSTEM_TYPES } + .filter { it.systemMessageType !in ConversationListUpdater.LAST_MESSAGE_HIDDEN_SYSTEM_TYPES } .maxByOrNull { it.id } } else { null @@ -1097,24 +990,6 @@ class ChatMessageSyncer @Inject constructor( private const val DEFAULT_MESSAGES_LIMIT = 100 - /** - * Marks that the unread count could not be derived locally; the guarded conversation - * update keeps the stored count in that case. - */ - private const val UNREAD_COUNT_UNKNOWN = -1 - - /** - * System messages that never become a conversation's last message on the server, so a - * locally derived conversation preview must skip them as well. - */ - private val LAST_MESSAGE_HIDDEN_SYSTEM_TYPES = setOf( - ChatMessage.SystemMessageType.REACTION, - ChatMessage.SystemMessageType.REACTION_REVOKED, - ChatMessage.SystemMessageType.REACTION_DELETED, - ChatMessage.SystemMessageType.MESSAGE_DELETED, - ChatMessage.SystemMessageType.MESSAGE_EDITED, - ChatMessage.SystemMessageType.POLL_VOTED - ) private const val MILLIS_PER_SECOND = 1000L private const val ROOM_REFRESH_MAX_AGE_MILLIS = 3 * 60 * 60 * 1000L // 3 hours private const val CATCH_UP_COOLDOWN_MILLIS = 5_000L diff --git a/app/src/main/java/com/nextcloud/talk/chat/data/network/OfflineFirstChatRepository.kt b/app/src/main/java/com/nextcloud/talk/chat/data/network/OfflineFirstChatRepository.kt index 57d06b98e6..c8d6f57a67 100644 --- a/app/src/main/java/com/nextcloud/talk/chat/data/network/OfflineFirstChatRepository.kt +++ b/app/src/main/java/com/nextcloud/talk/chat/data/network/OfflineFirstChatRepository.kt @@ -12,6 +12,7 @@ import android.os.Bundle import android.util.Log import com.nextcloud.talk.chat.data.ChatMessageRepository import com.nextcloud.talk.chat.data.model.ChatMessage +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.mappers.asEntity @@ -56,7 +57,8 @@ class OfflineFirstChatRepository @Inject constructor( private val chatBlocksDao: ChatBlocksDao, private val network: ChatNetworkDataSource, private val networkMonitor: NetworkMonitor, - private val syncer: ChatMessageSyncer + private val syncer: ChatMessageSyncer, + private val conversationListUpdater: ConversationListUpdater ) : ChatMessageRepository { lateinit var currentUser: User @@ -364,7 +366,7 @@ class OfflineFirstChatRepository @Inject constructor( } override suspend fun updateLocalReadState(lastReadMessage: Int) { - syncer.updateLocalReadState(syncTarget, lastReadMessage) + conversationListUpdater.updateLocalReadState(syncTarget, lastReadMessage) } override suspend fun loadMoreMessages( 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 new file mode 100644 index 0000000000..49fa4bfe76 --- /dev/null +++ b/app/src/main/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdater.kt @@ -0,0 +1,203 @@ +/* + * 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 android.util.Log +import com.bluelinelabs.logansquare.LoganSquare +import com.nextcloud.talk.chat.data.model.ChatMessage +import com.nextcloud.talk.chat.data.network.ChatMessageSyncer +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.model.ConversationEntity +import kotlinx.coroutines.flow.first +import java.util.concurrent.ConcurrentHashMap +import javax.inject.Inject + +/** + * Keeps the conversation list entries fresh between room list syncs. + * + * Owns the two write paths into the conversations table that don't come from the server's room + * list: reflecting background message catch-ups (preview, activity, derived unread count) and the + * optimistic local read state — plus the pending read markers that keep provably stale server + * responses from reverting that read state. The room list sync remains the authority and + * self-healing safeguard for everything else. + */ +class ConversationListUpdater @Inject constructor( + private val chatDao: ChatMessagesDao, + private val chatBlocksDao: ChatBlocksDao, + private val conversationsDao: ConversationsDao +) { + + /** + * Read markers that were written locally but whose delivery the server has not confirmed yet. + * + * A room list sync request can be answered before a marker sent at the same time reaches the + * server, so the sync response carries a provably stale read state — applying it would revert + * the conversation entry to unread until the next sync. While a marker is pending, such stale + * responses are kept out of the merge; the entry is only released once a sync confirms the + * marker (server read state caught up) or sending it ultimately failed, so the server's + * authority is restored either way and marking as unread from another device stays possible. + */ + private val pendingReadMarkers = ConcurrentHashMap() + + /** + * The locally written but not yet server-confirmed read marker of the conversation, or null. + */ + fun pendingReadMarker(internalConversationId: String): Int? = pendingReadMarkers[internalConversationId] + + /** + * Releases a pending read marker: called when a room list sync confirmed it or when sending it + * ultimately failed. Only removes [lastReadMessage] itself, so a newer marker written in the + * meantime stays pending. + */ + fun clearPendingReadMarker(internalConversationId: String, lastReadMessage: Int) { + pendingReadMarkers.remove(internalConversationId, lastReadMessage) + } + + /** + * Optimistically writes the user's read state into the conversation entry, so the + * conversation list reflects it immediately — before (and independent of) the read marker + * reaching the server. The unread count is recounted from the cached messages above the new + * marker. The server stays the authority: the next room list sync re-asserts its state, with + * [pendingReadMarkers] bridging the window until the marker's delivery is confirmed. + */ + suspend fun updateLocalReadState(target: ChatMessageSyncer.SyncTarget, lastReadMessage: Int) { + val conversation = + conversationsDao.getConversationForUser(target.accountId, target.roomToken).first() ?: return + val unreadMessages = chatDao.countMessagesNewerThan( + internalConversationId = target.internalConversationId, + messageId = lastReadMessage.toLong(), + excludedActorId = target.user.userId!! + ) + pendingReadMarkers[target.internalConversationId] = lastReadMessage + conversationsDao.updateReadState(conversation.internalId, lastReadMessage, unreadMessages) + Log.d(TAG, "Local read state for room ${target.roomToken}: lastRead=$lastReadMessage, unread=$unreadMessages") + } + + /** + * Keeps provably stale read states out of the merge of server responses (the room list sync + * and the single-room refresh): a response computed before a concurrently sent read marker + * reached the server still reports the room as unread, and applying it would revert the + * conversation entry until the next sync. While a marker for the room is pending and the + * server's read state is still behind it, the local read state is kept; once the server has + * caught up (or moved past it, e.g. read further on another device) the marker is released + * and the server state applies unchanged — including a lower one, so marking as unread from + * another device keeps working. + */ + fun preserveReadStateOfPendingMarkers( + previousConversations: Map, + conversationsFromServer: List + ): List = + conversationsFromServer.map { serverItem -> + val pendingMarker = pendingReadMarker(serverItem.internalId) + ?: return@map serverItem + val previous = previousConversations[serverItem.internalId] + ?: return@map serverItem + + if (serverItem.lastReadMessage < pendingMarker) { + Log.d( + TAG, + "Keeping local read state for room ${serverItem.token}: server lastRead=" + + "${serverItem.lastReadMessage} is behind the pending read marker $pendingMarker" + ) + serverItem.copy( + lastReadMessage = previous.lastReadMessage, + unreadMessages = previous.unreadMessages + ) + } else { + clearPendingReadMarker(serverItem.internalId, pendingMarker) + serverItem + } + } + + /** + * Reflects a background catch-up in the conversation list: the room's cached conversation + * entry is updated with the newest persisted message, its activity timestamp and a locally + * derived unread count, so the list is up to date the moment it is opened — before the room + * list sync (which stays in place as the authority and self-healing safeguard) has answered. + * + * Concurrency: runs inside the per-room catch-up mutex and writes via a single guarded UPDATE + * that only applies when the derived state is newer than the stored one, so a concurrently + * finishing room list sync can neither be overwritten with older data nor interleave with a + * read-modify-write. Thread catch-ups don't describe the room itself and are skipped. + */ + suspend fun updateConversationFromCatchUp( + target: ChatMessageSyncer.SyncTarget, + outcome: ChatMessageSyncer.SyncOutcome + ) { + val newestMessage = outcome.newestPersistedMessage + if (target.threadId != null || !outcome.persistedNewMessages || newestMessage == null) { + return + } + val conversation = + conversationsDao.getConversationForUser(target.accountId, target.roomToken).first() ?: return + + val unreadMessages = deriveUnreadMessagesCount(target, conversation.lastReadMessage) + val updatedRows = conversationsDao.updateConversationFromCatchUp( + internalId = conversation.internalId, + lastMessageJson = LoganSquare.serialize(newestMessage), + lastActivity = newestMessage.timestamp, + unreadMessages = unreadMessages + ) + Log.d( + TAG, + "Conversation list update for room ${target.roomToken} from catch-up " + + "(lastActivity=${newestMessage.timestamp}, unread=$unreadMessages): " + + if (updatedRows > 0) "applied" else "skipped, stored state is newer" + ) + } + + /** + * Derives the room's unread count from the cached messages, or [UNREAD_COUNT_UNKNOWN] when it + * cannot be derived — the count is only trustworthy when the latest chat block reaches back to + * the last read message. Own messages don't count: sending from another device advances the + * server-side read marker, which the locally cached [lastReadMessage] may lag behind. + */ + private suspend fun deriveUnreadMessagesCount(target: ChatMessageSyncer.SyncTarget, lastReadMessage: Int): Int { + val latestBlock = if (lastReadMessage > 0) { + chatBlocksDao.getLatestChatBlock(target.internalConversationId, null).first() + } else { + null + } + val blockReachesUnreadBoundary = latestBlock != null && + (latestBlock.oldestMessageId <= lastReadMessage || !latestBlock.hasHistory) + + return if (blockReachesUnreadBoundary) { + chatDao.countMessagesNewerThan( + internalConversationId = target.internalConversationId, + messageId = lastReadMessage.toLong(), + excludedActorId = target.user.userId!! + ) + } else { + UNREAD_COUNT_UNKNOWN + } + } + + companion object { + private val TAG: String = ConversationListUpdater::class.java.simpleName + + /** + * Marks that the unread count could not be derived locally; the guarded conversation + * update keeps the stored count in that case. + */ + private const val UNREAD_COUNT_UNKNOWN = -1 + + /** + * System messages that never become a conversation's last message on the server, so a + * locally derived conversation preview must skip them as well. + */ + val LAST_MESSAGE_HIDDEN_SYSTEM_TYPES = setOf( + ChatMessage.SystemMessageType.REACTION, + ChatMessage.SystemMessageType.REACTION_REVOKED, + ChatMessage.SystemMessageType.REACTION_DELETED, + ChatMessage.SystemMessageType.MESSAGE_DELETED, + ChatMessage.SystemMessageType.MESSAGE_EDITED, + ChatMessage.SystemMessageType.POLL_VOTED + ) + } +} 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 d153639888..7612f35eb7 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 @@ -55,6 +55,7 @@ class OfflineFirstConversationsRepository @Inject constructor( private val chatNetworkDataSource: ChatNetworkDataSource, private val networkMonitor: NetworkMonitor, private val chatMessageSyncer: ChatMessageSyncer, + private val conversationListUpdater: ConversationListUpdater, private val context: Context ) : OfflineConversationsRepository { private val observedAccountId = MutableStateFlow(null) @@ -143,8 +144,10 @@ class OfflineFirstConversationsRepository @Inject constructor( model.hiddenUpcomingEvent = existingEntity?.hiddenUpcomingEvent _conversationFlow.emit(model) val previous = existingEntity?.let { mapOf(it.internalId to it) }.orEmpty() - val entityList = - preserveReadStateOfPendingMarkers(previous, listOf(model.asEntity())) + val entityList = conversationListUpdater.preserveReadStateOfPendingMarkers( + previous, + listOf(model.asEntity()) + ) dao.upsertConversations(user.id!!, entityList) } } @@ -187,7 +190,10 @@ class OfflineFirstConversationsRepository @Inject constructor( dao.syncConversationsForUser( accountId = user.id!!, - serverItems = preserveReadStateOfPendingMarkers(previousConversations, conversationsFromSync), + serverItems = conversationListUpdater.preserveReadStateOfPendingMarkers( + previousConversations, + conversationsFromSync + ), conversationIdsToDelete = determineLeftConversationIds(previousConversations, conversationsFromSync) ) @@ -297,42 +303,6 @@ class OfflineFirstConversationsRepository @Inject constructor( connectivityManager.restrictBackgroundStatus == ConnectivityManager.RESTRICT_BACKGROUND_STATUS_ENABLED } - /** - * Keeps provably stale read states out of the merge of server responses (the room list sync - * and the single-room refresh of [getRoom]): a response computed before a concurrently sent - * read marker reached the server still reports the room as unread, and applying it would - * revert the conversation entry until the next sync. While a marker for the room is pending - * and the server's read state is still behind it, the local read state is kept; once the - * server has caught up (or moved past it, e.g. read further on another device) the marker is - * released and the server state applies unchanged — including a lower one, so marking as - * unread from another device keeps working. - */ - private fun preserveReadStateOfPendingMarkers( - previousConversations: Map, - conversationsFromSync: List - ): List = - conversationsFromSync.map { serverItem -> - val pendingMarker = chatMessageSyncer.pendingReadMarker(serverItem.internalId) - ?: return@map serverItem - val previous = previousConversations[serverItem.internalId] - ?: return@map serverItem - - if (serverItem.lastReadMessage < pendingMarker) { - Log.d( - TAG, - "Keeping local read state for room ${serverItem.token}: server lastRead=" + - "${serverItem.lastReadMessage} is behind the pending read marker $pendingMarker" - ) - serverItem.copy( - lastReadMessage = previous.lastReadMessage, - unreadMessages = previous.unreadMessages - ) - } else { - chatMessageSyncer.clearPendingReadMarker(serverItem.internalId, pendingMarker) - serverItem - } - } - private fun determineLeftConversationIds( previousConversations: Map, conversationsFromSync: List diff --git a/app/src/main/java/com/nextcloud/talk/dagger/modules/RepositoryModule.kt b/app/src/main/java/com/nextcloud/talk/dagger/modules/RepositoryModule.kt index e2c4e106ce..dbbd3ef400 100644 --- a/app/src/main/java/com/nextcloud/talk/dagger/modules/RepositoryModule.kt +++ b/app/src/main/java/com/nextcloud/talk/dagger/modules/RepositoryModule.kt @@ -30,6 +30,7 @@ import com.nextcloud.talk.conversationcreation.data.ConversationCreationReposito import com.nextcloud.talk.conversationinfoedit.data.ConversationInfoEditRepository import com.nextcloud.talk.conversationinfoedit.data.ConversationInfoEditRepositoryImpl import com.nextcloud.talk.conversationlist.data.OfflineConversationsRepository +import com.nextcloud.talk.conversationlist.data.network.ConversationListUpdater import com.nextcloud.talk.conversationlist.data.network.ConversationsNetworkDataSource import com.nextcloud.talk.conversationlist.data.network.OfflineFirstConversationsRepository import com.nextcloud.talk.conversationlist.data.network.RetrofitConversationsNetwork @@ -145,16 +146,29 @@ class RepositoryModule { fun provideChatMessageSyncer( chatMessagesDao: ChatMessagesDao, chatBlocksDao: ChatBlocksDao, - conversationsDao: ConversationsDao, dataSource: ChatNetworkDataSource, - networkMonitor: NetworkMonitor + networkMonitor: NetworkMonitor, + conversationListUpdater: ConversationListUpdater ): ChatMessageSyncer = ChatMessageSyncer( chatMessagesDao, chatBlocksDao, - conversationsDao, dataSource, - networkMonitor + networkMonitor, + conversationListUpdater + ) + + @Provides + @Singleton + fun provideConversationListUpdater( + chatMessagesDao: ChatMessagesDao, + chatBlocksDao: ChatBlocksDao, + conversationsDao: ConversationsDao + ): ConversationListUpdater = + ConversationListUpdater( + chatMessagesDao, + chatBlocksDao, + conversationsDao ) @Provides @@ -165,7 +179,8 @@ class RepositoryModule { chatBlocksDao: ChatBlocksDao, dataSource: ChatNetworkDataSource, networkMonitor: NetworkMonitor, - syncer: ChatMessageSyncer + syncer: ChatMessageSyncer, + conversationListUpdater: ConversationListUpdater ): ChatMessageRepository = OfflineFirstChatRepository( logger, @@ -173,7 +188,8 @@ class RepositoryModule { chatBlocksDao, dataSource, networkMonitor, - syncer + syncer, + conversationListUpdater ) @Provides @@ -185,6 +201,7 @@ class RepositoryModule { chatNetworkDataSource: ChatNetworkDataSource, networkMonitor: NetworkMonitor, chatMessageSyncer: ChatMessageSyncer, + conversationListUpdater: ConversationListUpdater, context: Context ): OfflineConversationsRepository = OfflineFirstConversationsRepository( @@ -193,6 +210,7 @@ class RepositoryModule { chatNetworkDataSource, networkMonitor, chatMessageSyncer, + conversationListUpdater, context ) 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 a10a748d4e..15e033ec70 100644 --- a/app/src/main/java/com/nextcloud/talk/jobs/ReadMarkerSyncWorker.kt +++ b/app/src/main/java/com/nextcloud/talk/jobs/ReadMarkerSyncWorker.kt @@ -21,8 +21,8 @@ import androidx.work.WorkerParameters import autodagger.AutoInjector import com.nextcloud.talk.application.NextcloudTalkApplication import com.nextcloud.talk.application.NextcloudTalkApplication.Companion.sharedApplication -import com.nextcloud.talk.chat.data.network.ChatMessageSyncer import com.nextcloud.talk.chat.data.network.ChatNetworkDataSource +import com.nextcloud.talk.conversationlist.data.network.ConversationListUpdater import com.nextcloud.talk.users.UserManager import com.nextcloud.talk.utils.ApiUtils import com.nextcloud.talk.utils.bundle.BundleKeys.KEY_INTERNAL_USER_ID @@ -36,7 +36,7 @@ import javax.inject.Inject * The caller updates the local conversation entry optimistically before enqueuing this worker, so * the conversation list reflects the read state immediately. The server stays the authority: * every room list sync re-asserts the server's read state over the local entry — guarded by the - * pending marker in [ChatMessageSyncer] only while the marker is provably not delivered yet. + * pending marker in [ConversationListUpdater] only while the marker is provably not delivered yet. * When all attempts fail, the pending marker is released so the client falls back to the server * state at the next sync instead of staying diverged. Work is unique per room with * [ExistingWorkPolicy.REPLACE], so the newest marker for a room always wins and retries can never @@ -53,7 +53,7 @@ class ReadMarkerSyncWorker(context: Context, workerParams: WorkerParameters) : lateinit var chatNetworkDataSource: ChatNetworkDataSource @Inject - lateinit var chatMessageSyncer: ChatMessageSyncer + lateinit var conversationListUpdater: ConversationListUpdater override suspend fun doWork(): Result { sharedApplication!!.componentApplication.inject(this) @@ -106,7 +106,7 @@ class ReadMarkerSyncWorker(context: Context, workerParams: WorkerParameters) : } private fun fail(userId: Long, roomToken: String, lastReadMessage: Int): Result { - chatMessageSyncer.clearPendingReadMarker("$userId@$roomToken", lastReadMessage) + conversationListUpdater.clearPendingReadMarker("$userId@$roomToken", lastReadMessage) return Result.failure() } diff --git a/app/src/main/java/com/nextcloud/talk/utils/preview/ComposePreviewUtils.kt b/app/src/main/java/com/nextcloud/talk/utils/preview/ComposePreviewUtils.kt index 31d963bbad..54b6a1098c 100644 --- a/app/src/main/java/com/nextcloud/talk/utils/preview/ComposePreviewUtils.kt +++ b/app/src/main/java/com/nextcloud/talk/utils/preview/ComposePreviewUtils.kt @@ -21,6 +21,7 @@ import com.nextcloud.talk.chat.data.ChatMessageRepository import com.nextcloud.talk.chat.data.io.AudioFocusRequestManager import com.nextcloud.talk.chat.data.io.MediaRecorderManager import com.nextcloud.talk.chat.data.network.ChatMessageSyncer +import com.nextcloud.talk.conversationlist.data.network.ConversationListUpdater import com.nextcloud.talk.chat.data.network.ChatNetworkDataSource import com.nextcloud.talk.chat.data.network.OfflineFirstChatRepository import com.nextcloud.talk.chat.data.network.RetrofitChatNetwork @@ -149,13 +150,20 @@ class ComposePreviewUtils private constructor(context: Context) { val logger: TestLogger get() = TestLogger + val conversationListUpdater: ConversationListUpdater + get() = ConversationListUpdater( + chatMessagesDao, + chatBlocksDao, + conversationsDao + ) + val chatMessageSyncer: ChatMessageSyncer get() = ChatMessageSyncer( chatMessagesDao, chatBlocksDao, - conversationsDao, chatNetworkDataSource, - networkMonitor + networkMonitor, + conversationListUpdater ) val chatRepository: ChatMessageRepository @@ -165,7 +173,8 @@ class ComposePreviewUtils private constructor(context: Context) { chatBlocksDao, chatNetworkDataSource, networkMonitor, - chatMessageSyncer + chatMessageSyncer, + conversationListUpdater ) val threadsRepository: ThreadsRepository @@ -181,6 +190,7 @@ class ComposePreviewUtils private constructor(context: Context) { chatNetworkDataSource, networkMonitor, chatMessageSyncer, + conversationListUpdater, mContext ) diff --git a/app/src/test/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncerTest.kt b/app/src/test/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncerTest.kt index a346da40e3..6dde4504db 100644 --- a/app/src/test/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncerTest.kt +++ b/app/src/test/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncerTest.kt @@ -9,6 +9,7 @@ package com.nextcloud.talk.chat.data.network import android.database.sqlite.SQLiteConstraintException import com.nextcloud.talk.chat.data.model.ChatMessage +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 @@ -61,7 +62,8 @@ class ChatMessageSyncerTest { @Before fun setUp() { - syncer = ChatMessageSyncer(chatDao, chatBlocksDao, conversationsDao, network, networkMonitor) + val conversationListUpdater = ConversationListUpdater(chatDao, chatBlocksDao, conversationsDao) + syncer = ChatMessageSyncer(chatDao, chatBlocksDao, network, networkMonitor, conversationListUpdater) whenever(networkMonitor.isOnline).thenReturn(MutableStateFlow(true)) whenever(conversationsDao.getConversationForUser(any(), any())).thenReturn(flowOf(null)) } diff --git a/app/src/test/java/com/nextcloud/talk/chat/data/network/OfflineFirstChatRepositoryTest.kt b/app/src/test/java/com/nextcloud/talk/chat/data/network/OfflineFirstChatRepositoryTest.kt index 1bc74d7f11..199b81431b 100644 --- a/app/src/test/java/com/nextcloud/talk/chat/data/network/OfflineFirstChatRepositoryTest.kt +++ b/app/src/test/java/com/nextcloud/talk/chat/data/network/OfflineFirstChatRepositoryTest.kt @@ -9,6 +9,7 @@ package com.nextcloud.talk.chat.data.network import android.os.Bundle import com.nextcloud.talk.chat.data.model.ChatMessage +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 @@ -70,7 +71,14 @@ class OfflineFirstChatRepositoryTest { chatBlocksDao, network, networkMonitor, - ChatMessageSyncer(chatDao, chatBlocksDao, conversationsDao, network, networkMonitor) + ChatMessageSyncer( + chatDao, + chatBlocksDao, + network, + networkMonitor, + ConversationListUpdater(chatDao, chatBlocksDao, conversationsDao) + ), + ConversationListUpdater(chatDao, chatBlocksDao, conversationsDao) ) repository.initData(user(), CREDENTIALS, CHAT_URL, ROOM_TOKEN, null) } 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 60cca7b2b2..ad66166370 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 @@ -64,6 +64,7 @@ class ConversationListFreshnessIntegrationTest { private lateinit var db: TalkDatabase private lateinit var syncer: ChatMessageSyncer + private lateinit var conversationListUpdater: ConversationListUpdater private val chatNetwork: ChatNetworkDataSource = mock() private val conversationsNetwork: ConversationsNetworkDataSource = mock() @@ -82,12 +83,14 @@ class ConversationListFreshnessIntegrationTest { whenever(networkMonitor.isOnline).thenReturn(MutableStateFlow(true)) + conversationListUpdater = + ConversationListUpdater(db.chatMessagesDao(), db.chatBlocksDao(), db.conversationsDao()) syncer = ChatMessageSyncer( db.chatMessagesDao(), db.chatBlocksDao(), - db.conversationsDao(), chatNetwork, - networkMonitor + networkMonitor, + conversationListUpdater ) } @@ -137,6 +140,7 @@ class ConversationListFreshnessIntegrationTest { chatNetwork, networkMonitor, syncer, + conversationListUpdater, ApplicationProvider.getApplicationContext() ) whenever(chatNetwork.getRoom(any(), any())).thenReturn( @@ -150,7 +154,7 @@ class ConversationListFreshnessIntegrationTest { runBlocking { seedConversation(lastActivity = 12, lastReadMessage = 10, unreadMessages = 2) - syncer.updateLocalReadState(target(), lastReadMessage = 12) + conversationListUpdater.updateLocalReadState(target(), lastReadMessage = 12) repository.getRoom(user, ROOM_TOKEN).join() awaitUntil { conversationEntity().lastActivity == 12L } @@ -187,11 +191,11 @@ class ConversationListFreshnessIntegrationTest { ) syncer.catchUpRoom(target(), lastReadMessage = 10, unreadMessages = 2) - syncer.updateLocalReadState(target(), lastReadMessage = 12) + conversationListUpdater.updateLocalReadState(target(), lastReadMessage = 12) assertEquals(12, conversationEntity().lastReadMessage) assertEquals("everything below the marker is read", 0, conversationEntity().unreadMessages) - syncer.updateLocalReadState(target(), lastReadMessage = 10) + conversationListUpdater.updateLocalReadState(target(), lastReadMessage = 10) assertEquals(10, conversationEntity().lastReadMessage) assertEquals("own messages must not count as unread", 1, conversationEntity().unreadMessages) } @@ -206,6 +210,7 @@ class ConversationListFreshnessIntegrationTest { chatNetwork, networkMonitor, syncer, + conversationListUpdater, ApplicationProvider.getApplicationContext() ) whenever(conversationsNetwork.getRooms(any(), any(), any())).thenReturn( @@ -216,7 +221,7 @@ class ConversationListFreshnessIntegrationTest { runBlocking { seedConversation(lastActivity = 12, lastReadMessage = 10, unreadMessages = 2) - syncer.updateLocalReadState(target(), lastReadMessage = 12) + conversationListUpdater.updateLocalReadState(target(), lastReadMessage = 12) assertEquals(12, conversationEntity().lastReadMessage) assertEquals(0, conversationEntity().unreadMessages) @@ -243,6 +248,7 @@ class ConversationListFreshnessIntegrationTest { chatNetwork, networkMonitor, syncer, + conversationListUpdater, ApplicationProvider.getApplicationContext() ) whenever(conversationsNetwork.getRooms(any(), any(), any())).thenReturn( diff --git a/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/RoomListMessagePrefetchIntegrationTest.kt b/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/RoomListMessagePrefetchIntegrationTest.kt index 9d49f2d4c6..20ec0e4890 100644 --- a/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/RoomListMessagePrefetchIntegrationTest.kt +++ b/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/RoomListMessagePrefetchIntegrationTest.kt @@ -75,12 +75,14 @@ class RoomListMessagePrefetchIntegrationTest { whenever(networkMonitor.isOnline).thenReturn(MutableStateFlow(true)) + val conversationListUpdater = + ConversationListUpdater(db.chatMessagesDao(), db.chatBlocksDao(), db.conversationsDao()) syncer = ChatMessageSyncer( db.chatMessagesDao(), db.chatBlocksDao(), - db.conversationsDao(), chatNetwork, - networkMonitor + networkMonitor, + conversationListUpdater ) repository = OfflineFirstConversationsRepository( db.conversationsDao(), @@ -88,6 +90,7 @@ class RoomListMessagePrefetchIntegrationTest { chatNetwork, networkMonitor, syncer, + conversationListUpdater, context ) } From dbf433010dd6369f47a898fdbb117051759b920a Mon Sep 17 00:00:00 2001 From: Andy Scherzinger Date: Fri, 14 Aug 2026 19:28:48 +0200 Subject: [PATCH 06/10] fix(conversations): don't re-restore the scroll position on every resume Returning from a chat to the conversation list made the list visibly jump shortly after rendering. The saved scroll position was re-armed on every resume and restored on the next room list emission - which used to be an immediate local snapshot, making the restore an invisible no-op. With the list derived reactively from the database, the state current at resume never re-emits, so the first post-resume emission is the server sync response seconds later: scrollToItem then re-anchored an already rendered (and possibly slightly changed) list under the user's eyes. Restore the position only once per activity lifetime: after a recreation the LazyListState is genuinely lost and the first emission - which the reactive flow now delivers immediately on subscription - restores it, while a plain resume keeps the retained state's position naturally, without any late scrollToItem. Assisted-by: Claude Code:claude-fable-5 Signed-off-by: Andy Scherzinger --- .../talk/conversationlist/ConversationsListActivity.kt | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/app/src/main/java/com/nextcloud/talk/conversationlist/ConversationsListActivity.kt b/app/src/main/java/com/nextcloud/talk/conversationlist/ConversationsListActivity.kt index 5ace6396dc..1881004dbe 100644 --- a/app/src/main/java/com/nextcloud/talk/conversationlist/ConversationsListActivity.kt +++ b/app/src/main/java/com/nextcloud/talk/conversationlist/ConversationsListActivity.kt @@ -160,7 +160,7 @@ class ConversationsListActivity : BaseActivity() { // Lazy list state – set from inside setContent, read from onPause private var conversationListLazyListState: androidx.compose.foundation.lazy.LazyListState? = null - // Ensures saved scroll position is restored only once per resume cycle, not on every room-list refresh. + // restored once per activity lifetime; a late scrollToItem on resume would jump the rendered list private var scrollPositionRestored = false private var nextUnreadConversationScrollPosition = 0 @@ -385,7 +385,6 @@ class ConversationsListActivity : BaseActivity() { override fun onResume() { super.onResume() - scrollPositionRestored = false showNotificationWarningState.value = shouldShowNotificationWarning() showShareToScreenState.value = hasActivityActionSendIntent() From acfb9608a186b467deb2d6278b1c30079c4febdc Mon Sep 17 00:00:00 2001 From: Andy Scherzinger Date: Fri, 14 Aug 2026 20:01:14 +0200 Subject: [PATCH 07/10] fix(conversations): guard optimistic list actions against stale syncs The conversation list actions - mark as read/unread (menu and swipe), add to or remove from favorites, and tag assignment - write the local conversation entry optimistically and revert on server failure, but had no protection against the same race the read marker had: a room list sync (or single-room refresh) whose response was computed before the action reached the server reverts the entry to the old state until the next sync, flipping badges and jumping favorite-sorted rows. Generalize the pending read marker mechanism in ConversationListUpdater to these actions: each optimistic write registers a pending change, and while the server response provably doesn't reflect it yet, the local state is kept in the merge. Once the server confirms the change it is released and the server state applies unchanged - so changes made on other devices keep flowing - and a failed action releases its pending change when reverting, restoring the server's authority. Mark-as-read reuses the pending read marker itself; mark-as-unread, favorites and tags get their own pending state with value-based confirmation. Assisted-by: Claude Code:claude-fable-5 Signed-off-by: Andy Scherzinger --- .../data/network/ConversationListUpdater.kt | 147 +++++++++++++++--- .../OfflineFirstConversationsRepository.kt | 4 +- .../viewmodels/ConversationsListViewModel.kt | 12 +- .../viewmodels/ConversationTagsViewModel.kt | 6 +- ...onversationListFreshnessIntegrationTest.kt | 66 +++++++- 5 files changed, 204 insertions(+), 31 deletions(-) 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 49fa4bfe76..cf6185939a 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 @@ -27,6 +27,7 @@ import javax.inject.Inject * responses from reverting that read state. The room list sync remains the authority and * self-healing safeguard for everything else. */ +@Suppress("TooManyFunctions") class ConversationListUpdater @Inject constructor( private val chatDao: ChatMessagesDao, private val chatBlocksDao: ChatBlocksDao, @@ -45,11 +46,59 @@ class ConversationListUpdater @Inject constructor( */ private val pendingReadMarkers = ConcurrentHashMap() + /** + * Favorite flags applied locally but not yet confirmed by a server response; guarded in the + * merge like the pending read markers. + */ + private val pendingFavorites = ConcurrentHashMap() + + /** + * Rooms marked as unread from the conversation list whose server state doesn't reflect it yet. + */ + private val pendingUnreadFlags = ConcurrentHashMap() + + /** + * Tag assignments applied locally but not yet confirmed by a server response. + */ + private val pendingTagIds = ConcurrentHashMap>() + /** * The locally written but not yet server-confirmed read marker of the conversation, or null. */ fun pendingReadMarker(internalConversationId: String): Int? = pendingReadMarkers[internalConversationId] + /** + * Registers a read marker sent from the conversation list (mark as read), so the merge guards + * it exactly like a marker sent from an open chat. + */ + fun markPendingReadMarker(internalConversationId: String, lastReadMessage: Int) { + pendingReadMarkers[internalConversationId] = lastReadMessage + } + + fun markPendingFavorite(internalConversationId: String, favorite: Boolean) { + pendingFavorites[internalConversationId] = favorite + } + + fun clearPendingFavorite(internalConversationId: String, favorite: Boolean) { + pendingFavorites.remove(internalConversationId, favorite) + } + + fun markPendingUnread(internalConversationId: String) { + pendingUnreadFlags[internalConversationId] = true + } + + fun clearPendingUnread(internalConversationId: String) { + pendingUnreadFlags.remove(internalConversationId) + } + + fun markPendingTags(internalConversationId: String, tagIds: List) { + pendingTagIds[internalConversationId] = tagIds + } + + fun clearPendingTags(internalConversationId: String, tagIds: List) { + pendingTagIds.remove(internalConversationId, tagIds) + } + /** * Releases a pending read marker: called when a room list sync confirmed it or when sending it * ultimately failed. Only removes [lastReadMessage] itself, so a newer marker written in the @@ -80,40 +129,88 @@ class ConversationListUpdater @Inject constructor( } /** - * Keeps provably stale read states out of the merge of server responses (the room list sync - * and the single-room refresh): a response computed before a concurrently sent read marker - * reached the server still reports the room as unread, and applying it would revert the - * conversation entry until the next sync. While a marker for the room is pending and the - * server's read state is still behind it, the local read state is kept; once the server has - * caught up (or moved past it, e.g. read further on another device) the marker is released - * and the server state applies unchanged — including a lower one, so marking as unread from - * another device keeps working. + * Keeps provably stale local-action state out of the merge of server responses (the room + * list sync and the single-room refresh): a response computed before a concurrently sent + * change — read marker, mark as unread, favorite flag, tag assignment — reached the server + * still reports the old state, and applying it would revert the conversation entry until the + * next sync. While a change for the room is pending and the server hasn't reflected it yet, + * 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. */ - fun preserveReadStateOfPendingMarkers( + fun preservePendingLocalState( previousConversations: Map, conversationsFromServer: List ): List = conversationsFromServer.map { serverItem -> - val pendingMarker = pendingReadMarker(serverItem.internalId) - ?: return@map serverItem val previous = previousConversations[serverItem.internalId] ?: return@map serverItem + var guarded = guardPendingReadMarker(serverItem, previous) + guarded = guardPendingUnread(guarded, previous) + guarded = guardPendingFavorite(guarded, previous) + guarded = guardPendingTags(guarded, previous) + guarded + } + + private fun guardPendingReadMarker( + serverItem: ConversationEntity, + previous: ConversationEntity + ): ConversationEntity { + val pendingMarker = pendingReadMarker(serverItem.internalId) ?: return serverItem + + return if (serverItem.lastReadMessage < pendingMarker) { + Log.d( + TAG, + "Keeping local read state for room ${serverItem.token}: server lastRead=" + + "${serverItem.lastReadMessage} is behind the pending read marker $pendingMarker" + ) + serverItem.copy( + lastReadMessage = previous.lastReadMessage, + unreadMessages = previous.unreadMessages + ) + } else { + clearPendingReadMarker(serverItem.internalId, pendingMarker) + serverItem + } + } + + private fun guardPendingUnread(serverItem: ConversationEntity, previous: ConversationEntity): ConversationEntity { + if (!pendingUnreadFlags.containsKey(serverItem.internalId)) { + return serverItem + } + + return if (serverItem.unreadMessages > 0) { + clearPendingUnread(serverItem.internalId) + serverItem + } else { + serverItem.copy( + lastReadMessage = previous.lastReadMessage, + unreadMessages = previous.unreadMessages + ) + } + } + + private fun guardPendingFavorite(serverItem: ConversationEntity, previous: ConversationEntity): ConversationEntity { + val desired = pendingFavorites[serverItem.internalId] ?: return serverItem + + return if (serverItem.favorite == desired) { + clearPendingFavorite(serverItem.internalId, desired) + serverItem + } else { + serverItem.copy(favorite = previous.favorite) + } + } - if (serverItem.lastReadMessage < pendingMarker) { - Log.d( - TAG, - "Keeping local read state for room ${serverItem.token}: server lastRead=" + - "${serverItem.lastReadMessage} is behind the pending read marker $pendingMarker" - ) - serverItem.copy( - lastReadMessage = previous.lastReadMessage, - unreadMessages = previous.unreadMessages - ) - } else { - clearPendingReadMarker(serverItem.internalId, pendingMarker) - serverItem - } + private fun guardPendingTags(serverItem: ConversationEntity, previous: ConversationEntity): ConversationEntity { + val desired = pendingTagIds[serverItem.internalId] ?: return serverItem + + return if (serverItem.tagIds == desired) { + clearPendingTags(serverItem.internalId, desired) + serverItem + } else { + serverItem.copy(tagIds = previous.tagIds) } + } /** * Reflects a background catch-up in the conversation list: the room's cached conversation 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 7612f35eb7..055f248b76 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 @@ -144,7 +144,7 @@ class OfflineFirstConversationsRepository @Inject constructor( model.hiddenUpcomingEvent = existingEntity?.hiddenUpcomingEvent _conversationFlow.emit(model) val previous = existingEntity?.let { mapOf(it.internalId to it) }.orEmpty() - val entityList = conversationListUpdater.preserveReadStateOfPendingMarkers( + val entityList = conversationListUpdater.preservePendingLocalState( previous, listOf(model.asEntity()) ) @@ -190,7 +190,7 @@ class OfflineFirstConversationsRepository @Inject constructor( dao.syncConversationsForUser( accountId = user.id!!, - serverItems = conversationListUpdater.preserveReadStateOfPendingMarkers( + serverItems = conversationListUpdater.preservePendingLocalState( previousConversations, conversationsFromSync ), diff --git a/app/src/main/java/com/nextcloud/talk/conversationlist/viewmodels/ConversationsListViewModel.kt b/app/src/main/java/com/nextcloud/talk/conversationlist/viewmodels/ConversationsListViewModel.kt index 4b2e6ab334..8a0ba3c352 100644 --- a/app/src/main/java/com/nextcloud/talk/conversationlist/viewmodels/ConversationsListViewModel.kt +++ b/app/src/main/java/com/nextcloud/talk/conversationlist/viewmodels/ConversationsListViewModel.kt @@ -16,6 +16,7 @@ import com.nextcloud.talk.R import com.nextcloud.talk.arbitrarystorage.ArbitraryStorageManager import com.nextcloud.talk.contacts.ContactsRepository import com.nextcloud.talk.conversationlist.data.OfflineConversationsRepository +import com.nextcloud.talk.conversationlist.data.network.ConversationListUpdater import com.nextcloud.talk.conversationlist.ui.ConversationListEntry import com.nextcloud.talk.data.user.model.User import com.nextcloud.talk.invitation.data.InvitationsModel @@ -77,7 +78,8 @@ class ConversationsListViewModel @Inject constructor( private val invitationsRepository: InvitationsRepository, private val arbitraryStorageManager: ArbitraryStorageManager, var userManager: UserManager, - private val conversationsRepository: ConversationsRepository + private val conversationsRepository: ConversationsRepository, + private val conversationListUpdater: ConversationListUpdater ) : ViewModel() { private val _currentUser = currentUserProvider.currentUser.blockingGet() @@ -714,6 +716,7 @@ class ConversationsListViewModel @Inject constructor( ) val url = ApiUtils.getUrlForChatReadMarker(apiVersion, currentUser.baseUrl, conversation.token) viewModelScope.launch { + messageId?.let { conversationListUpdater.markPendingReadMarker(conversation.internalId, it) } withContext(Dispatchers.IO) { repository.updateConversation(optimistic) } @@ -723,6 +726,7 @@ class ConversationsListViewModel @Inject constructor( } _readUnreadState.value = ConversationReadUnreadUiState.Success } catch (e: Exception) { + messageId?.let { conversationListUpdater.clearPendingReadMarker(conversation.internalId, it) } withContext(Dispatchers.IO) { repository.updateConversation(original) } @@ -741,6 +745,7 @@ class ConversationsListViewModel @Inject constructor( ) val url = ApiUtils.getUrlForChatReadMarker(apiVersion, currentUser.baseUrl, conversation.token) viewModelScope.launch { + conversationListUpdater.markPendingUnread(conversation.internalId) withContext(Dispatchers.IO) { repository.updateConversation(optimistic) } @@ -750,6 +755,7 @@ class ConversationsListViewModel @Inject constructor( } _readUnreadState.value = ConversationReadUnreadUiState.Success } catch (e: Exception) { + conversationListUpdater.clearPendingUnread(conversation.internalId) withContext(Dispatchers.IO) { repository.updateConversation(original) } @@ -769,6 +775,7 @@ class ConversationsListViewModel @Inject constructor( val apiVersion = ApiUtils.getConversationApiVersion(currentUser, intArrayOf(ApiUtils.API_V4, ApiUtils.API_V1)) val url = ApiUtils.getUrlForRoomFavorite(apiVersion, currentUser.baseUrl, conversation.token) viewModelScope.launch { + conversationListUpdater.markPendingFavorite(conversation.internalId, favorite = true) withContext(Dispatchers.IO) { repository.updateConversation(optimistic) } @@ -778,6 +785,7 @@ class ConversationsListViewModel @Inject constructor( } _favoriteState.value = FavoriteUiState.Success } catch (e: Exception) { + conversationListUpdater.clearPendingFavorite(conversation.internalId, favorite = true) withContext(Dispatchers.IO) { repository.updateConversation(original) } @@ -793,6 +801,7 @@ class ConversationsListViewModel @Inject constructor( val apiVersion = ApiUtils.getConversationApiVersion(currentUser, intArrayOf(ApiUtils.API_V4, ApiUtils.API_V1)) val url = ApiUtils.getUrlForRoomFavorite(apiVersion, currentUser.baseUrl, conversation.token) viewModelScope.launch { + conversationListUpdater.markPendingFavorite(conversation.internalId, favorite = false) withContext(Dispatchers.IO) { repository.updateConversation(optimistic) } @@ -802,6 +811,7 @@ class ConversationsListViewModel @Inject constructor( } _favoriteState.value = FavoriteUiState.Success } catch (e: Exception) { + conversationListUpdater.clearPendingFavorite(conversation.internalId, favorite = false) withContext(Dispatchers.IO) { repository.updateConversation(original) } diff --git a/app/src/main/java/com/nextcloud/talk/conversationtags/viewmodels/ConversationTagsViewModel.kt b/app/src/main/java/com/nextcloud/talk/conversationtags/viewmodels/ConversationTagsViewModel.kt index e3a70ec636..032ce4965f 100644 --- a/app/src/main/java/com/nextcloud/talk/conversationtags/viewmodels/ConversationTagsViewModel.kt +++ b/app/src/main/java/com/nextcloud/talk/conversationtags/viewmodels/ConversationTagsViewModel.kt @@ -12,6 +12,7 @@ import androidx.lifecycle.ViewModel import androidx.lifecycle.viewModelScope import com.bluelinelabs.logansquare.LoganSquare import com.nextcloud.talk.conversationlist.data.OfflineConversationsRepository +import com.nextcloud.talk.conversationlist.data.network.ConversationListUpdater import com.nextcloud.talk.conversationtags.data.ConversationTagsRepository import com.nextcloud.talk.data.user.model.User import com.nextcloud.talk.models.domain.ConversationModel @@ -34,7 +35,8 @@ import javax.inject.Inject class ConversationTagsViewModel @Inject constructor( private val conversationTagsRepository: ConversationTagsRepository, private val repository: OfflineConversationsRepository, - private val currentUserProvider: CurrentUserProviderOld + private val currentUserProvider: CurrentUserProviderOld, + private val conversationListUpdater: ConversationListUpdater ) : ViewModel() { private val currentUser: User = currentUserProvider.currentUser.blockingGet() @@ -164,6 +166,7 @@ class ConversationTagsViewModel @Inject constructor( val optimistic = conversation.copy(tagIds = tagIds) replaceConversationForTagAssignment(conversation.token, optimistic) viewModelScope.launch { + conversationListUpdater.markPendingTags(conversation.internalId, tagIds) withContext(Dispatchers.IO) { repository.updateConversation(optimistic) } @@ -177,6 +180,7 @@ class ConversationTagsViewModel @Inject constructor( ) } } catch (e: Exception) { + conversationListUpdater.clearPendingTags(conversation.internalId, tagIds) replaceConversationForTagAssignment(conversation.token, original) withContext(Dispatchers.IO) { repository.updateConversation(original) 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 ad66166370..95d88d0858 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 @@ -40,6 +40,8 @@ import kotlinx.coroutines.launch import kotlinx.coroutines.runBlocking import org.junit.After import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertTrue import org.junit.Before import org.junit.Test import org.junit.runner.RunWith @@ -239,6 +241,54 @@ class ConversationListFreshnessIntegrationTest { } } + @Test + fun `a sync cannot revert a pending favorite until the server confirms it`() { + val user = user(withKeepNotificationsCapability = false) + val repository = repository() + whenever(conversationsNetwork.getRooms(any(), any(), any())).thenReturn( + Observable.just(listOf(staleServerRoom(lastReadMessage = 10, unreadMessages = 0))), + Observable.just(listOf(staleServerRoom(lastReadMessage = 10, unreadMessages = 0, favorite = true))), + Observable.just(listOf(staleServerRoom(lastReadMessage = 10, unreadMessages = 0))) + ) + + runBlocking { + seedConversation(lastActivity = 12, lastReadMessage = 10, unreadMessages = 0) + conversationListUpdater.markPendingFavorite(INTERNAL_CONVERSATION_ID, favorite = true) + db.conversationsDao().updateConversation(conversationEntity().copy(favorite = true)) + + repository.getRooms(user).join() + assertTrue("stale sync must not revert the favorite", conversationEntity().favorite) + + repository.getRooms(user).join() + assertTrue("confirming sync must keep the favorite", conversationEntity().favorite) + + repository.getRooms(user).join() + assertFalse("server authority must be restored", conversationEntity().favorite) + } + } + + @Test + fun `a sync cannot revert a pending mark as unread until the server confirms it`() { + val user = user(withKeepNotificationsCapability = false) + val repository = repository() + whenever(conversationsNetwork.getRooms(any(), any(), any())).thenReturn( + Observable.just(listOf(staleServerRoom(lastReadMessage = 12, unreadMessages = 0))), + Observable.just(listOf(staleServerRoom(lastReadMessage = 9, unreadMessages = 3))) + ) + + runBlocking { + seedConversation(lastActivity = 12, lastReadMessage = 12, unreadMessages = 0) + conversationListUpdater.markPendingUnread(INTERNAL_CONVERSATION_ID) + db.conversationsDao().updateConversation(conversationEntity().copy(unreadMessages = 1)) + + repository.getRooms(user).join() + assertEquals("stale sync must not clear the unread state", 1, conversationEntity().unreadMessages) + + repository.getRooms(user).join() + assertEquals("confirming sync must apply the server state", 3, conversationEntity().unreadMessages) + } + } + @Test fun `room list flow reflects database writes reactively`() { val user = user(withKeepNotificationsCapability = false) @@ -271,12 +321,24 @@ class ConversationListFreshnessIntegrationTest { } } - private fun staleServerRoom(lastReadMessage: Int, unreadMessages: Int): Conversation = + private fun repository(): OfflineFirstConversationsRepository = + OfflineFirstConversationsRepository( + db.conversationsDao(), + conversationsNetwork, + chatNetwork, + networkMonitor, + syncer, + conversationListUpdater, + ApplicationProvider.getApplicationContext() + ) + + private fun staleServerRoom(lastReadMessage: Int, unreadMessages: Int, favorite: Boolean = false): Conversation = Conversation( token = ROOM_TOKEN, lastActivity = 12, lastReadMessage = lastReadMessage, - unreadMessages = unreadMessages + unreadMessages = unreadMessages, + favorite = favorite ) private suspend fun seedConversation(lastActivity: Long, lastReadMessage: Int, unreadMessages: Int) { From 7701c500b3ffaae91252dac0aed7923757188963 Mon Sep 17 00:00:00 2001 From: Andy Scherzinger Date: Fri, 14 Aug 2026 20:25:18 +0200 Subject: [PATCH 08/10] feat(conversations): make archiving optimistic and guard it like the other list actions Archiving from the conversation list was the last list action without an optimistic write: it awaited the server call and a full room list sync before the row moved, and a concurrent sync whose response was computed before the archive reached the server could still flip the row back until the next refresh. Move the action into ConversationsListViewModel following the favorite pattern: the archived flag is written locally first (the reactive list moves the row instantly, no extra sync round-trip), the server call is retried once and reverted on failure, and the change is registered as a pending archived flag in ConversationListUpdater so stale server responses cannot revert it before the server confirms. Assisted-by: Claude Code:claude-fable-5 Signed-off-by: Andy Scherzinger --- .../ConversationsListActivity.kt | 46 +++++++++---------- .../data/network/ConversationListUpdater.kt | 25 ++++++++++ .../viewmodels/ConversationsListViewModel.kt | 46 +++++++++++++++++++ ...onversationListFreshnessIntegrationTest.kt | 36 ++++++++++++++- 4 files changed, 127 insertions(+), 26 deletions(-) diff --git a/app/src/main/java/com/nextcloud/talk/conversationlist/ConversationsListActivity.kt b/app/src/main/java/com/nextcloud/talk/conversationlist/ConversationsListActivity.kt index 1881004dbe..9adba33eb6 100644 --- a/app/src/main/java/com/nextcloud/talk/conversationlist/ConversationsListActivity.kt +++ b/app/src/main/java/com/nextcloud/talk/conversationlist/ConversationsListActivity.kt @@ -109,12 +109,10 @@ import com.nextcloud.talk.utils.bundle.BundleKeys.KEY_SHARED_TEXT import com.nextcloud.talk.utils.permissions.PlatformPermissionUtil import com.nextcloud.talk.utils.power.PowerManagerUtils import com.nextcloud.talk.utils.singletons.ApplicationWideCurrentRoomHolder -import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.collect import kotlinx.coroutines.flow.onEach import kotlinx.coroutines.launch -import kotlinx.coroutines.withContext import org.greenrobot.eventbus.Subscribe import org.greenrobot.eventbus.ThreadMode import retrofit2.HttpException @@ -559,6 +557,27 @@ class ConversationsListActivity : BaseActivity() { } } } + + lifecycleScope.launch { + conversationsListViewModel.archiveState.collect { state -> + when (state) { + is ConversationsListViewModel.ArchiveUiState.Success -> { + val messageRes = if (state.archived) { + R.string.archived_conversation + } else { + R.string.unarchived_conversation + } + showSnackbar(String.format(resources.getString(messageRes), state.conversationDisplayName)) + conversationsListViewModel.resetArchiveState() + } + is ConversationsListViewModel.ArchiveUiState.Error -> { + showSnackbar(resources.getString(R.string.nc_common_error_sorry)) + conversationsListViewModel.resetArchiveState() + } + ConversationsListViewModel.ArchiveUiState.None -> { /* no-op */ } + } + } + } } private fun handleNoteToSelfShortcut(noteToSelfAvailable: Boolean, noteToSelfToken: String) { @@ -1173,29 +1192,8 @@ class ConversationsListActivity : BaseActivity() { @Suppress("Detekt.TooGenericExceptionCaught", "TooGenericExceptionCaught") private fun handleArchiving(conversation: ConversationModel) { - val apiVersion = ApiUtils.getConversationApiVersion(currentUser!!, intArrayOf(ApiUtils.API_V4, ApiUtils.API_V1)) - val url = ApiUtils.getUrlForArchive(apiVersion, currentUser?.baseUrl, conversation.token) - lifecycleScope.launch { - try { - if (conversation.hasArchived) { - withContext(Dispatchers.IO) { ncApiCoroutines.unarchiveConversation(credentials!!, url) } - fetchRooms() - showSnackbar( - String.format(resources.getString(R.string.unarchived_conversation), conversation.displayName) - ) - } else { - withContext(Dispatchers.IO) { ncApiCoroutines.archiveConversation(credentials!!, url) } - fetchRooms() - showSnackbar( - String.format(resources.getString(R.string.archived_conversation), conversation.displayName) - ) - } - } catch (e: Exception) { - showSnackbar(resources.getString(R.string.nc_common_error_sorry)) - } - } + conversationsListViewModel.toggleConversationArchive(conversation) } - private fun addConversationToFavorites(conversation: ConversationModel) { conversationsListViewModel.addConversationToFavorites(conversation) } 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 cf6185939a..5a197c458c 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 @@ -57,6 +57,11 @@ class ConversationListUpdater @Inject constructor( */ private val pendingUnreadFlags = ConcurrentHashMap() + /** + * Archived flags applied locally but not yet confirmed by a server response. + */ + private val pendingArchivedFlags = ConcurrentHashMap() + /** * Tag assignments applied locally but not yet confirmed by a server response. */ @@ -91,6 +96,14 @@ class ConversationListUpdater @Inject constructor( pendingUnreadFlags.remove(internalConversationId) } + fun markPendingArchived(internalConversationId: String, archived: Boolean) { + pendingArchivedFlags[internalConversationId] = archived + } + + fun clearPendingArchived(internalConversationId: String, archived: Boolean) { + pendingArchivedFlags.remove(internalConversationId, archived) + } + fun markPendingTags(internalConversationId: String, tagIds: List) { pendingTagIds[internalConversationId] = tagIds } @@ -148,6 +161,7 @@ class ConversationListUpdater @Inject constructor( var guarded = guardPendingReadMarker(serverItem, previous) guarded = guardPendingUnread(guarded, previous) guarded = guardPendingFavorite(guarded, previous) + guarded = guardPendingArchived(guarded, previous) guarded = guardPendingTags(guarded, previous) guarded } @@ -201,6 +215,17 @@ class ConversationListUpdater @Inject constructor( } } + private fun guardPendingArchived(serverItem: ConversationEntity, previous: ConversationEntity): ConversationEntity { + val desired = pendingArchivedFlags[serverItem.internalId] ?: return serverItem + + return if (serverItem.hasArchived == desired) { + clearPendingArchived(serverItem.internalId, desired) + serverItem + } else { + serverItem.copy(hasArchived = previous.hasArchived) + } + } + private fun guardPendingTags(serverItem: ConversationEntity, previous: ConversationEntity): ConversationEntity { val desired = pendingTagIds[serverItem.internalId] ?: return serverItem diff --git a/app/src/main/java/com/nextcloud/talk/conversationlist/viewmodels/ConversationsListViewModel.kt b/app/src/main/java/com/nextcloud/talk/conversationlist/viewmodels/ConversationsListViewModel.kt index 8a0ba3c352..fb03272fa8 100644 --- a/app/src/main/java/com/nextcloud/talk/conversationlist/viewmodels/ConversationsListViewModel.kt +++ b/app/src/main/java/com/nextcloud/talk/conversationlist/viewmodels/ConversationsListViewModel.kt @@ -126,6 +126,15 @@ class ConversationsListViewModel @Inject constructor( private val _favoriteState = MutableStateFlow(FavoriteUiState.None) val favoriteState: StateFlow = _favoriteState.asStateFlow() + sealed class ArchiveUiState { + data object None : ArchiveUiState() + data class Success(val archived: Boolean, val conversationDisplayName: String) : ArchiveUiState() + data object Error : ArchiveUiState() + } + + private val _archiveState = MutableStateFlow(ArchiveUiState.None) + val archiveState: StateFlow = _archiveState.asStateFlow() + object GetRoomsStartState : ViewState class GetRoomsErrorState(val throwable: Throwable) : ViewState open class GetRoomsSuccessState(val listIsNotEmpty: Boolean) : ViewState @@ -768,6 +777,43 @@ class ConversationsListViewModel @Inject constructor( _favoriteState.value = FavoriteUiState.None } + fun resetArchiveState() { + _archiveState.value = ArchiveUiState.None + } + + @Suppress("Detekt.TooGenericExceptionCaught") + fun toggleConversationArchive(conversation: ConversationModel) { + val original = conversation.copy() + val desiredArchived = !conversation.hasArchived + val optimistic = conversation.copy(hasArchived = desiredArchived) + val apiVersion = ApiUtils.getConversationApiVersion(currentUser, intArrayOf(ApiUtils.API_V4, ApiUtils.API_V1)) + val url = ApiUtils.getUrlForArchive(apiVersion, currentUser.baseUrl, conversation.token) + viewModelScope.launch { + conversationListUpdater.markPendingArchived(conversation.internalId, desiredArchived) + withContext(Dispatchers.IO) { + repository.updateConversation(optimistic) + } + try { + withContext(Dispatchers.IO) { + withRetry(1) { + if (desiredArchived) { + conversationsRepository.archiveConversation(credentials, url) + } else { + conversationsRepository.unarchiveConversation(credentials, url) + } + } + } + _archiveState.value = ArchiveUiState.Success(desiredArchived, conversation.displayName) + } catch (e: Exception) { + conversationListUpdater.clearPendingArchived(conversation.internalId, desiredArchived) + withContext(Dispatchers.IO) { + repository.updateConversation(original) + } + _archiveState.value = ArchiveUiState.Error + } + } + } + @Suppress("Detekt.TooGenericExceptionCaught") fun addConversationToFavorites(conversation: ConversationModel) { val original = conversation.copy() 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 95d88d0858..b24b0fda41 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 @@ -289,6 +289,32 @@ class ConversationListFreshnessIntegrationTest { } } + @Test + fun `a sync cannot revert a pending archive until the server confirms it`() { + val user = user(withKeepNotificationsCapability = false) + val repository = repository() + whenever(conversationsNetwork.getRooms(any(), any(), any())).thenReturn( + Observable.just(listOf(staleServerRoom(lastReadMessage = 10, unreadMessages = 0))), + Observable.just(listOf(staleServerRoom(lastReadMessage = 10, unreadMessages = 0, hasArchived = true))), + Observable.just(listOf(staleServerRoom(lastReadMessage = 10, unreadMessages = 0))) + ) + + runBlocking { + seedConversation(lastActivity = 12, lastReadMessage = 10, unreadMessages = 0) + conversationListUpdater.markPendingArchived(INTERNAL_CONVERSATION_ID, archived = true) + db.conversationsDao().updateConversation(conversationEntity().copy(hasArchived = true)) + + repository.getRooms(user).join() + assertTrue("stale sync must not revert the archive", conversationEntity().hasArchived) + + repository.getRooms(user).join() + assertTrue("confirming sync must keep the archive", conversationEntity().hasArchived) + + repository.getRooms(user).join() + assertFalse("server authority must be restored", conversationEntity().hasArchived) + } + } + @Test fun `room list flow reflects database writes reactively`() { val user = user(withKeepNotificationsCapability = false) @@ -332,13 +358,19 @@ class ConversationListFreshnessIntegrationTest { ApplicationProvider.getApplicationContext() ) - private fun staleServerRoom(lastReadMessage: Int, unreadMessages: Int, favorite: Boolean = false): Conversation = + private fun staleServerRoom( + lastReadMessage: Int, + unreadMessages: Int, + favorite: Boolean = false, + hasArchived: Boolean = false + ): Conversation = Conversation( token = ROOM_TOKEN, lastActivity = 12, lastReadMessage = lastReadMessage, unreadMessages = unreadMessages, - favorite = favorite + favorite = favorite, + hasArchived = hasArchived ) private suspend fun seedConversation(lastActivity: Long, lastReadMessage: Int, unreadMessages: Int) { From d76285ac4ac90fd6409692b155bce16db882535d Mon Sep 17 00:00:00 2001 From: Marcel Hibbe Date: Wed, 19 Aug 2026 15:00:57 +0200 Subject: [PATCH 09/10] fix(conversations): don't regress unread while a read marker is pending A background catch-up (room-list-triggered or push-notification-driven) derives its unread count from conversation.lastReadMessage read fresh from the database. A read marker just registered as pending - e.g. from leaving a chat - may not have committed its own updateReadState write to that same row yet by the time a concurrently running catch-up reads it, so the count gets derived from a stale, not-yet-advanced read position. Unlike the room list sync and single-room refresh merge, this write path never consulted the pending marker at all, so nothing caught the spurious positive count before it landed in the conversations table: the conversation flashed as read and then flipped back to unread, staying that way until the next full sync self-healed it. Take the higher of the stored lastReadMessage and the pending marker as the derivation baseline - it's always at least as advanced as what's actually stored, so it closes the race without suppressing unread detection for a message that is genuinely newer than the marker. Assisted-by: Claude Sonnet 5 Signed-off-by: Marcel Hibbe --- .../data/network/ConversationListUpdater.kt | 16 +++++++- ...onversationListFreshnessIntegrationTest.kt | 41 +++++++++++++++++++ 2 files changed, 56 insertions(+), 1 deletion(-) 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 5a197c458c..04c645e6ae 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 @@ -247,6 +247,16 @@ class ConversationListUpdater @Inject constructor( * that only applies when the derived state is newer than the stored one, so a concurrently * finishing room list sync can neither be overwritten with older data nor interleave with a * read-modify-write. Thread catch-ups don't describe the room itself and are skipped. + * + * Unlike [preservePendingLocalState], this path doesn't merge against a server response, so a + * pending read marker can't be guarded the same way — but the marker is still consulted as the + * unread count's baseline: a read marker registered here (e.g. from leaving the chat) may not + * have committed its own [ConversationsDao.updateReadState] write to this row yet by the time a + * concurrently running catch-up reads [ConversationEntity.lastReadMessage], which would + * otherwise derive the count from a stale, not-yet-advanced marker and regress an + * already-read conversation back to unread. The pending marker is always at least as advanced + * as what's actually stored, so taking the higher of the two closes that race without + * suppressing unread detection for a message that is genuinely newer than the marker. */ suspend fun updateConversationFromCatchUp( target: ChatMessageSyncer.SyncTarget, @@ -259,7 +269,11 @@ class ConversationListUpdater @Inject constructor( val conversation = conversationsDao.getConversationForUser(target.accountId, target.roomToken).first() ?: return - val unreadMessages = deriveUnreadMessagesCount(target, conversation.lastReadMessage) + val lastReadMessage = maxOf( + conversation.lastReadMessage, + pendingReadMarker(target.internalConversationId) ?: 0 + ) + val unreadMessages = deriveUnreadMessagesCount(target, lastReadMessage) val updatedRows = conversationsDao.updateConversationFromCatchUp( internalId = conversation.internalId, lastMessageJson = LoganSquare.serialize(newestMessage), 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 b24b0fda41..f7014dd820 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 @@ -133,6 +133,47 @@ class ConversationListFreshnessIntegrationTest { } } + @Test + fun `background catch-up does not regress unread while a read marker is pending but not yet committed`() { + runBlocking { + // Simulates leaving the chat racing a concurrent catch-up: the read marker is + // registered as pending, but its own updateReadState write hasn't landed on this row + // yet by the time the catch-up reads lastReadMessage. + seedConversation(lastActivity = 10, lastReadMessage = 5, unreadMessages = 3) + conversationListUpdater.markPendingReadMarker(INTERNAL_CONVERSATION_ID, lastReadMessage = 12) + + wheneverBlocking { chatNetwork.pullChatMessages(any(), any(), any()) } + .thenReturn(Response.success(overall(message(10), message(11), message(12)))) + + syncer.catchUpRoom(target(), lastReadMessage = 5, unreadMessages = 3) + + assertEquals( + "catch-up must not derive unread below the pending read marker", + 0, + conversationEntity().unreadMessages + ) + } + } + + @Test + fun `background catch-up still reports a message newer than the pending read marker as unread`() { + runBlocking { + seedConversation(lastActivity = 10, lastReadMessage = 5, unreadMessages = 3) + conversationListUpdater.markPendingReadMarker(INTERNAL_CONVERSATION_ID, lastReadMessage = 12) + + wheneverBlocking { chatNetwork.pullChatMessages(any(), any(), any()) } + .thenReturn(Response.success(overall(message(10), message(11), message(12), message(13)))) + + syncer.catchUpRoom(target(), lastReadMessage = 5, unreadMessages = 3) + + assertEquals( + "a message newer than the pending marker must still count as unread", + 1, + conversationEntity().unreadMessages + ) + } + } + @Test fun `a single-room refresh cannot revert the read state while its marker is pending`() { val user = user(withKeepNotificationsCapability = false) From 3d3dafbca681edd59d522123550a60fa6261ede7 Mon Sep 17 00:00:00 2001 From: Marcel Hibbe Date: Wed, 19 Aug 2026 15:01:21 +0200 Subject: [PATCH 10/10] fix(chat): register the pending read marker synchronously Leaving a chat can race the conversation list's own resume-triggered sync: both fire at nearly the same moment, and the sync's response is only guarded correctly against a stale server state if the pending read marker already exists by the time that response gets merged. The marker was only registered inside updateLocalReadState, after two suspending database reads, itself launched fire-and-forget from onPause() - a real device trace showed the list's sync firing 230ms before the marker was actually registered, a gap that widens further under main-thread contention. During that window the sync sees no pending marker yet and applies the server's outdated read state unguarded, flipping the conversation back to unread until the next sync. Split registration out into its own synchronous, non-suspending call and run it before anything is launched, so the marker is armed the instant setChatReadMessage returns - no coroutine dispatch and no database read stand between deciding to mark as read and the guard being in place. Assisted-by: Claude Sonnet 5 Signed-off-by: Marcel Hibbe --- .../talk/chat/data/ChatMessageRepository.kt | 11 +++++++++++ .../data/network/OfflineFirstChatRepository.kt | 4 ++++ .../talk/chat/viewmodels/ChatViewModel.kt | 9 +++++++++ .../network/OfflineFirstChatRepositoryTest.kt | 16 +++++++++++++++- 4 files changed, 39 insertions(+), 1 deletion(-) diff --git a/app/src/main/java/com/nextcloud/talk/chat/data/ChatMessageRepository.kt b/app/src/main/java/com/nextcloud/talk/chat/data/ChatMessageRepository.kt index 1751f73dd6..aa111de3f0 100644 --- a/app/src/main/java/com/nextcloud/talk/chat/data/ChatMessageRepository.kt +++ b/app/src/main/java/com/nextcloud/talk/chat/data/ChatMessageRepository.kt @@ -79,6 +79,17 @@ interface ChatMessageRepository : LifecycleAwareManager { */ suspend fun updateLocalReadState(lastReadMessage: Int) + /** + * Registers [lastReadMessage] as a pending read marker synchronously, without going through + * [updateLocalReadState]'s suspending database reads first. Call this before launching the + * actual local write: leaving the chat can race a room list sync triggered by the conversation + * list resuming at (almost) the same time, and that sync's response must find the marker + * already pending to guard against a server state it computed before the marker was sent - + * a gap of even a couple hundred milliseconds while registration waits on database reads is + * enough for that race to lose. + */ + fun markPendingReadMarker(lastReadMessage: Int) + /** * Loads messages from local storage. If the messages are not found, then it * synchronizes the database with the server, before retrying exactly once. Only diff --git a/app/src/main/java/com/nextcloud/talk/chat/data/network/OfflineFirstChatRepository.kt b/app/src/main/java/com/nextcloud/talk/chat/data/network/OfflineFirstChatRepository.kt index c8d6f57a67..79cbe4171f 100644 --- a/app/src/main/java/com/nextcloud/talk/chat/data/network/OfflineFirstChatRepository.kt +++ b/app/src/main/java/com/nextcloud/talk/chat/data/network/OfflineFirstChatRepository.kt @@ -369,6 +369,10 @@ class OfflineFirstChatRepository @Inject constructor( conversationListUpdater.updateLocalReadState(syncTarget, lastReadMessage) } + override fun markPendingReadMarker(lastReadMessage: Int) { + conversationListUpdater.markPendingReadMarker(syncTarget.internalConversationId, lastReadMessage) + } + override suspend fun loadMoreMessages( anchorMessageId: Long, direction: ChatMessageRepository.LoadMoreDirection, diff --git a/app/src/main/java/com/nextcloud/talk/chat/viewmodels/ChatViewModel.kt b/app/src/main/java/com/nextcloud/talk/chat/viewmodels/ChatViewModel.kt index 4130e6caed..45a45998e6 100644 --- a/app/src/main/java/com/nextcloud/talk/chat/viewmodels/ChatViewModel.kt +++ b/app/src/main/java/com/nextcloud/talk/chat/viewmodels/ChatViewModel.kt @@ -1819,11 +1819,20 @@ class ChatViewModel @AssistedInject constructor( * failures with backoff. The server stays the authority — every room list sync re-asserts its * read state, so a marker that ultimately could not be sent falls back to the server state * instead of leaving the client diverged. + * + * [markPendingReadMarker] runs synchronously, before anything is launched, so the marker is + * armed the instant this returns: leaving the chat commonly races the conversation list's own + * resume-triggered sync, and that race is only guarded correctly if the pending marker already + * exists by the time the sync's response is merged. [updateLocalReadState] itself does two + * suspending database reads before writing - registering the marker only there, inside a + * launched coroutine, previously left a real gap (observed at ~250ms, more under main-thread + * contention) during which such a sync could see no pending marker yet and apply unguarded. */ fun setChatReadMessage(lastReadMessage: Int) { if (!this::currentUser.isInitialized) { return } + chatRepository.markPendingReadMarker(lastReadMessage) viewModelScope.launch { chatRepository.updateLocalReadState(lastReadMessage) } diff --git a/app/src/test/java/com/nextcloud/talk/chat/data/network/OfflineFirstChatRepositoryTest.kt b/app/src/test/java/com/nextcloud/talk/chat/data/network/OfflineFirstChatRepositoryTest.kt index 199b81431b..00343b1930 100644 --- a/app/src/test/java/com/nextcloud/talk/chat/data/network/OfflineFirstChatRepositoryTest.kt +++ b/app/src/test/java/com/nextcloud/talk/chat/data/network/OfflineFirstChatRepositoryTest.kt @@ -47,6 +47,7 @@ import retrofit2.Response * last read message, otherwise the initial window is re-fetched — anchored at the boundary when * the unread backlog calls for it. */ +@Suppress("TooManyFunctions") class OfflineFirstChatRepositoryTest { private val logger: Logger = mock() @@ -55,6 +56,7 @@ class OfflineFirstChatRepositoryTest { private val conversationsDao: ConversationsDao = mock() private val network: ChatNetworkDataSource = mock() private val networkMonitor: NetworkMonitor = mock() + private val conversationListUpdater = ConversationListUpdater(chatDao, chatBlocksDao, conversationsDao) private lateinit var repository: OfflineFirstChatRepository @@ -78,11 +80,23 @@ class OfflineFirstChatRepositoryTest { networkMonitor, ConversationListUpdater(chatDao, chatBlocksDao, conversationsDao) ), - ConversationListUpdater(chatDao, chatBlocksDao, conversationsDao) + conversationListUpdater ) repository.initData(user(), CREDENTIALS, CHAT_URL, ROOM_TOKEN, null) } + @Test + fun `markPendingReadMarker registers the marker synchronously, without any suspension`() { + // Leaving the chat can race a room list sync triggered by the conversation list resuming + // at (almost) the same time - that race is only guarded correctly if the pending marker + // already exists by the time the sync's response is merged, so registering it must not + // wait on updateLocalReadState's database reads. Calling it outside of runTest/any + // coroutine proves no suspension is involved. + repository.markPendingReadMarker(42) + + assertEquals(42, conversationListUpdater.pendingReadMarker(INTERNAL_CONVERSATION_ID)) + } + @Test fun `loadInitialMessages closes the backlog when the latest block reaches the last read message`() = runTest {