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..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 @@ -72,6 +72,24 @@ 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) + + /** + * 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/ChatMessageSyncer.kt b/app/src/main/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncer.kt index 641d10d581..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 @@ -12,6 +12,7 @@ import android.os.SystemClock import android.util.Log 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.mappers.asEntity @@ -43,12 +44,13 @@ 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 network: ChatNetworkDataSource, - private val networkMonitor: NetworkMonitor + private val networkMonitor: NetworkMonitor, + private val conversationListUpdater: ConversationListUpdater ) { /** @@ -91,13 +93,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,6 +364,8 @@ class ChatMessageSyncer @Inject constructor( Log.d(TAG, "Background catch-up for room ${target.roomToken}: no new messages") } + conversationListUpdater.updateConversationFromCatchUp(target, outcome) + return outcome } @@ -467,7 +476,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 +507,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 +526,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 +537,8 @@ class ChatMessageSyncer @Inject constructor( newestPersistedMessageId = newestPersisted, oldestPersistedMessageId = oldestPersisted, persistedMessageCount = totalCount, - syncFailed = roundOutcome.syncFailed + syncFailed = roundOutcome.syncFailed, + newestPersistedMessage = newestPersistedMessage ) } anchor = nextAnchor @@ -556,7 +569,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 +695,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 ConversationListUpdater.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 +989,7 @@ class ChatMessageSyncer @Inject constructor( SyncOutcome(persistedNewMessages = false, newestPersistedMessageId = null, syncFailed = true) private const val DEFAULT_MESSAGES_LIMIT = 100 + 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 9f54fddbd1..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 @@ -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 @@ -363,6 +365,14 @@ class OfflineFirstChatRepository @Inject constructor( return outcome.persistedNewMessages } + override suspend fun updateLocalReadState(lastReadMessage: Int) { + 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 3bbcfc3716..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 @@ -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,48 @@ 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. + * + * [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(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 + } + chatRepository.markPendingReadMarker(lastReadMessage) + 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/ConversationsListActivity.kt b/app/src/main/java/com/nextcloud/talk/conversationlist/ConversationsListActivity.kt index 5ace6396dc..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 @@ -160,7 +158,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 +383,6 @@ class ConversationsListActivity : BaseActivity() { override fun onResume() { super.onResume() - scrollPositionRestored = false showNotificationWarningState.value = shouldShowNotificationWarning() showShareToScreenState.value = hasActivityActionSendIntent() @@ -560,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) { @@ -1174,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/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/ConversationListUpdater.kt b/app/src/main/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdater.kt new file mode 100644 index 0000000000..04c645e6ae --- /dev/null +++ b/app/src/main/java/com/nextcloud/talk/conversationlist/data/network/ConversationListUpdater.kt @@ -0,0 +1,339 @@ +/* + * 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. + */ +@Suppress("TooManyFunctions") +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() + + /** + * 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() + + /** + * 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. + */ + 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 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 + } + + 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 + * 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 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 preservePendingLocalState( + previousConversations: Map, + conversationsFromServer: List + ): List = + conversationsFromServer.map { serverItem -> + val previous = previousConversations[serverItem.internalId] + ?: return@map serverItem + var guarded = guardPendingReadMarker(serverItem, previous) + guarded = guardPendingUnread(guarded, previous) + guarded = guardPendingFavorite(guarded, previous) + guarded = guardPendingArchived(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) + } + } + + 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 + + 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 + * 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. + * + * 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, + 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 lastReadMessage = maxOf( + conversation.lastReadMessage, + pendingReadMarker(target.internalConversationId) ?: 0 + ) + val unreadMessages = deriveUnreadMessagesCount(target, 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 32661535ee..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 @@ -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 @@ -50,11 +55,28 @@ 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 { - 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 +104,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) } } @@ -126,7 +143,11 @@ 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 = conversationListUpdater.preservePendingLocalState( + previous, + listOf(model.asEntity()) + ) dao.upsertConversations(user.id!!, entityList) } } @@ -138,12 +159,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 +188,14 @@ class OfflineFirstConversationsRepository @Inject constructor( val previousConversations = dao.getConversationsForUser(user.id!!).first() .associateBy { it.internalId } - deleteLeftConversations( - user, - conversationsFromSync + dao.syncConversationsForUser( + accountId = user.id!!, + serverItems = conversationListUpdater.preservePendingLocalState( + previousConversations, + conversationsFromSync + ), + conversationIdsToDelete = determineLeftConversationIds(previousConversations, conversationsFromSync) ) - dao.upsertConversations(user.id!!, conversationsFromSync) val roomsWithNewMessages = getRoomsWithNewMessages(conversationsFromSync, previousConversations) scope.launch { catchUpRoomsWithNewMessages(user, roomsWithNewMessages) } @@ -285,36 +303,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..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 @@ -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() @@ -124,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 @@ -149,6 +160,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()) /** @@ -710,8 +725,9 @@ 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.updateConversationLocallyAndEmit(currentUser, optimistic) + repository.updateConversation(optimistic) } try { withContext(Dispatchers.IO) { @@ -719,8 +735,9 @@ class ConversationsListViewModel @Inject constructor( } _readUnreadState.value = ConversationReadUnreadUiState.Success } catch (e: Exception) { + messageId?.let { conversationListUpdater.clearPendingReadMarker(conversation.internalId, it) } withContext(Dispatchers.IO) { - repository.updateConversationLocallyAndEmit(currentUser, original) + repository.updateConversation(original) } _readUnreadState.value = ConversationReadUnreadUiState.Error } @@ -737,8 +754,9 @@ class ConversationsListViewModel @Inject constructor( ) val url = ApiUtils.getUrlForChatReadMarker(apiVersion, currentUser.baseUrl, conversation.token) viewModelScope.launch { + conversationListUpdater.markPendingUnread(conversation.internalId) withContext(Dispatchers.IO) { - repository.updateConversationLocallyAndEmit(currentUser, optimistic) + repository.updateConversation(optimistic) } try { withContext(Dispatchers.IO) { @@ -746,8 +764,9 @@ class ConversationsListViewModel @Inject constructor( } _readUnreadState.value = ConversationReadUnreadUiState.Success } catch (e: Exception) { + conversationListUpdater.clearPendingUnread(conversation.internalId) withContext(Dispatchers.IO) { - repository.updateConversationLocallyAndEmit(currentUser, original) + repository.updateConversation(original) } _readUnreadState.value = ConversationReadUnreadUiState.Error } @@ -758,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() @@ -765,8 +821,9 @@ 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.updateConversationLocallyAndEmit(currentUser, optimistic) + repository.updateConversation(optimistic) } try { withContext(Dispatchers.IO) { @@ -774,8 +831,9 @@ class ConversationsListViewModel @Inject constructor( } _favoriteState.value = FavoriteUiState.Success } catch (e: Exception) { + conversationListUpdater.clearPendingFavorite(conversation.internalId, favorite = true) withContext(Dispatchers.IO) { - repository.updateConversationLocallyAndEmit(currentUser, original) + repository.updateConversation(original) } _favoriteState.value = FavoriteUiState.Error } @@ -789,8 +847,9 @@ 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.updateConversationLocallyAndEmit(currentUser, optimistic) + repository.updateConversation(optimistic) } try { withContext(Dispatchers.IO) { @@ -798,8 +857,9 @@ class ConversationsListViewModel @Inject constructor( } _favoriteState.value = FavoriteUiState.Success } catch (e: Exception) { + conversationListUpdater.clearPendingFavorite(conversation.internalId, favorite = false) 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..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,8 +166,9 @@ 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.updateConversationLocallyAndEmit(currentUser, optimistic) + repository.updateConversation(optimistic) } try { withContext(Dispatchers.IO) { @@ -177,9 +180,10 @@ class ConversationTagsViewModel @Inject constructor( ) } } catch (e: Exception) { + conversationListUpdater.clearPendingTags(conversation.internalId, tagIds) 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/dagger/modules/RepositoryModule.kt b/app/src/main/java/com/nextcloud/talk/dagger/modules/RepositoryModule.kt index 800315dfea..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 @@ -146,13 +147,28 @@ class RepositoryModule { chatMessagesDao: ChatMessagesDao, chatBlocksDao: ChatBlocksDao, dataSource: ChatNetworkDataSource, - networkMonitor: NetworkMonitor + networkMonitor: NetworkMonitor, + conversationListUpdater: ConversationListUpdater ): ChatMessageSyncer = ChatMessageSyncer( chatMessagesDao, chatBlocksDao, dataSource, - networkMonitor + networkMonitor, + conversationListUpdater + ) + + @Provides + @Singleton + fun provideConversationListUpdater( + chatMessagesDao: ChatMessagesDao, + chatBlocksDao: ChatBlocksDao, + conversationsDao: ConversationsDao + ): ConversationListUpdater = + ConversationListUpdater( + chatMessagesDao, + chatBlocksDao, + conversationsDao ) @Provides @@ -163,7 +179,8 @@ class RepositoryModule { chatBlocksDao: ChatBlocksDao, dataSource: ChatNetworkDataSource, networkMonitor: NetworkMonitor, - syncer: ChatMessageSyncer + syncer: ChatMessageSyncer, + conversationListUpdater: ConversationListUpdater ): ChatMessageRepository = OfflineFirstChatRepository( logger, @@ -171,7 +188,8 @@ class RepositoryModule { chatBlocksDao, dataSource, networkMonitor, - syncer + syncer, + conversationListUpdater ) @Provides @@ -183,6 +201,7 @@ class RepositoryModule { chatNetworkDataSource: ChatNetworkDataSource, networkMonitor: NetworkMonitor, chatMessageSyncer: ChatMessageSyncer, + conversationListUpdater: ConversationListUpdater, context: Context ): OfflineConversationsRepository = OfflineFirstConversationsRepository( @@ -191,6 +210,7 @@ class RepositoryModule { chatNetworkDataSource, networkMonitor, chatMessageSyncer, + conversationListUpdater, context ) 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 0e1604eed2..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 @@ -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 -> @@ -40,6 +57,47 @@ 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 + + /** + * 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..15e033ec70 --- /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.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 +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 [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 + * 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 conversationListUpdater: ConversationListUpdater + + 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 { + conversationListUpdater.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/ComposePreviewUtils.kt b/app/src/main/java/com/nextcloud/talk/utils/preview/ComposePreviewUtils.kt index 0da2a13a4c..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,12 +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, chatNetworkDataSource, - networkMonitor + networkMonitor, + conversationListUpdater ) val chatRepository: ChatMessageRepository @@ -164,7 +173,8 @@ class ComposePreviewUtils private constructor(context: Context) { chatBlocksDao, chatNetworkDataSource, networkMonitor, - chatMessageSyncer + chatMessageSyncer, + conversationListUpdater ) val threadsRepository: ThreadsRepository @@ -180,6 +190,7 @@ class ComposePreviewUtils private constructor(context: Context) { chatNetworkDataSource, networkMonitor, chatMessageSyncer, + conversationListUpdater, mContext ) 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..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 @@ -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,17 @@ class DummyConversationDaoImpl : ConversationsDao { /* */ } + override suspend fun updateConversationFromCatchUp( + internalId: String, + lastMessageJson: String, + lastActivity: Long, + unreadMessages: Int + ): Int = 0 + + override suspend fun updateReadState(internalId: String, lastReadMessage: Int, unreadMessages: Int) { + /* */ + } + 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..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,8 +9,10 @@ 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 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 +54,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 +62,10 @@ class ChatMessageSyncerTest { @Before fun setUp() { - syncer = ChatMessageSyncer(chatDao, chatBlocksDao, 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)) } @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..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 @@ -9,8 +9,10 @@ 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 import com.nextcloud.talk.data.database.model.ChatBlockEntity import com.nextcloud.talk.data.network.NetworkMonitor import com.nextcloud.talk.data.user.model.User @@ -45,13 +47,16 @@ 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() 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() + private val conversationListUpdater = ConversationListUpdater(chatDao, chatBlocksDao, conversationsDao) private lateinit var repository: OfflineFirstChatRepository @@ -68,11 +73,30 @@ class OfflineFirstChatRepositoryTest { chatBlocksDao, network, networkMonitor, - ChatMessageSyncer(chatDao, chatBlocksDao, network, networkMonitor) + ChatMessageSyncer( + chatDao, + chatBlocksDao, + network, + networkMonitor, + 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 { 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..f7014dd820 --- /dev/null +++ b/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/ConversationListFreshnessIntegrationTest.kt @@ -0,0 +1,497 @@ +/* + * 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.Assert.assertFalse +import org.junit.Assert.assertTrue +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 lateinit var conversationListUpdater: ConversationListUpdater + + 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)) + + conversationListUpdater = + ConversationListUpdater(db.chatMessagesDao(), db.chatBlocksDao(), db.conversationsDao()) + syncer = ChatMessageSyncer( + db.chatMessagesDao(), + db.chatBlocksDao(), + chatNetwork, + networkMonitor, + conversationListUpdater + ) + } + + @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 `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) + val repository = OfflineFirstConversationsRepository( + db.conversationsDao(), + conversationsNetwork, + chatNetwork, + networkMonitor, + syncer, + conversationListUpdater, + 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) + conversationListUpdater.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) + + conversationListUpdater.updateLocalReadState(target(), lastReadMessage = 12) + assertEquals(12, conversationEntity().lastReadMessage) + assertEquals("everything below the marker is read", 0, conversationEntity().unreadMessages) + + conversationListUpdater.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, + conversationListUpdater, + 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) + conversationListUpdater.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 `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 `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) + val repository = OfflineFirstConversationsRepository( + db.conversationsDao(), + conversationsNetwork, + chatNetwork, + networkMonitor, + syncer, + conversationListUpdater, + 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 repository(): OfflineFirstConversationsRepository = + OfflineFirstConversationsRepository( + db.conversationsDao(), + conversationsNetwork, + chatNetwork, + networkMonitor, + syncer, + conversationListUpdater, + ApplicationProvider.getApplicationContext() + ) + + 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, + hasArchived = hasArchived + ) + + 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 + } +} 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..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,13 +75,22 @@ class RoomListMessagePrefetchIntegrationTest { whenever(networkMonitor.isOnline).thenReturn(MutableStateFlow(true)) - syncer = ChatMessageSyncer(db.chatMessagesDao(), db.chatBlocksDao(), chatNetwork, networkMonitor) + val conversationListUpdater = + ConversationListUpdater(db.chatMessagesDao(), db.chatBlocksDao(), db.conversationsDao()) + syncer = ChatMessageSyncer( + db.chatMessagesDao(), + db.chatBlocksDao(), + chatNetwork, + networkMonitor, + conversationListUpdater + ) repository = OfflineFirstConversationsRepository( db.conversationsDao(), conversationsNetwork, chatNetwork, networkMonitor, syncer, + conversationListUpdater, context ) }