diff --git a/app/src/androidTest/java/com/nextcloud/talk/data/database/dao/ChatBlocksDaoTest.kt b/app/src/androidTest/java/com/nextcloud/talk/data/database/dao/ChatBlocksDaoTest.kt index a838e3cf7df..aed3fe96d7b 100644 --- a/app/src/androidTest/java/com/nextcloud/talk/data/database/dao/ChatBlocksDaoTest.kt +++ b/app/src/androidTest/java/com/nextcloud/talk/data/database/dao/ChatBlocksDaoTest.kt @@ -244,7 +244,7 @@ class ChatBlocksDaoTest { newestMessageId = searchedChatBlock.newestMessageId ) - assertEquals(5, results.first().size) + assertEquals(5, results.size) } @Test @@ -314,7 +314,7 @@ class ChatBlocksDaoTest { newestMessageId = searchedChatBlock.newestMessageId ) - assertEquals(1, resultsForThreadIdNull.first().size) + assertEquals(1, resultsForThreadIdNull.size) val resultsForThreadId123 = chatBlocksDao.getConnectedChatBlocks( internalConversationId = conversation1.internalId, @@ -323,7 +323,7 @@ class ChatBlocksDaoTest { newestMessageId = searchedChatBlock.newestMessageId ) - assertEquals(2, resultsForThreadId123.first().size) + assertEquals(2, resultsForThreadId123.size) } @Test @@ -384,7 +384,7 @@ class ChatBlocksDaoTest { threadId = null, oldestMessageId = 10, newestMessageId = 35 - ).first() + ) assertEquals(3, connectedBlocks.size) val mergedBlock = ChatBlockEntity( @@ -412,10 +412,121 @@ class ChatBlocksDaoTest { threadId = null, oldestMessageId = 10, newestMessageId = 40 - ).first().size + ).size ) } + @Test + fun testUpsertAndMergeConnectedChatBlocksMergesOverlappingBlocks() = + runTest { + val user = createUserEntity("account1", "Account 1") + usersDao.saveUser(user) + val account1 = usersDao.getUserWithUserId("account1").blockingGet() + + conversationsDao.upsertConversations( + account1.id, + listOf( + createConversationEntity( + accountId = account1.id, + token = "abc", + roomName = "Conversation One" + ) + ) + ) + + val conversation = conversationsDao.getConversationsForUser(account1.id).first()[0] + + chatBlocksDao.upsertChatBlock( + ChatBlockEntity( + internalConversationId = conversation.internalId, + accountId = conversation.accountId, + token = conversation.token, + threadId = null, + oldestMessageId = 10, + newestMessageId = 20, + hasHistory = false + ) + ) + chatBlocksDao.upsertChatBlock( + ChatBlockEntity( + internalConversationId = conversation.internalId, + accountId = conversation.accountId, + token = conversation.token, + threadId = null, + oldestMessageId = 25, + newestMessageId = 35, + hasHistory = true + ) + ) + + // the new block overlaps both existing blocks, so all three must merge into one + chatBlocksDao.upsertAndMergeConnectedChatBlocks( + ChatBlockEntity( + internalConversationId = conversation.internalId, + accountId = conversation.accountId, + token = conversation.token, + threadId = null, + oldestMessageId = 18, + newestMessageId = 27, + hasHistory = true + ) + ) + + val blocks = chatBlocksDao.getChatBlocksForConversation(conversation.internalId) + assertEquals(1, blocks.size) + assertEquals(10L, blocks[0].oldestMessageId) + assertEquals(35L, blocks[0].newestMessageId) + assertEquals(false, blocks[0].hasHistory) + } + + @Test + fun testUpsertAndMergeConnectedChatBlocksKeepsDisjointBlocksSeparate() = + runTest { + val user = createUserEntity("account1", "Account 1") + usersDao.saveUser(user) + val account1 = usersDao.getUserWithUserId("account1").blockingGet() + + conversationsDao.upsertConversations( + account1.id, + listOf( + createConversationEntity( + accountId = account1.id, + token = "abc", + roomName = "Conversation One" + ) + ) + ) + + val conversation = conversationsDao.getConversationsForUser(account1.id).first()[0] + + chatBlocksDao.upsertChatBlock( + ChatBlockEntity( + internalConversationId = conversation.internalId, + accountId = conversation.accountId, + token = conversation.token, + threadId = null, + oldestMessageId = 10, + newestMessageId = 20, + hasHistory = true + ) + ) + + chatBlocksDao.upsertAndMergeConnectedChatBlocks( + ChatBlockEntity( + internalConversationId = conversation.internalId, + accountId = conversation.accountId, + token = conversation.token, + threadId = null, + oldestMessageId = 30, + newestMessageId = 40, + hasHistory = true + ) + ) + + val blocks = chatBlocksDao.getChatBlocksForConversation(conversation.internalId) + assertEquals(2, blocks.size) + } + private fun createUserEntity(userId: String, userName: String) = UserEntity( userId = userId, 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 9a55fac9bd9..5cbb1fb120d 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 @@ -58,7 +58,7 @@ interface ChatMessageRepository : LifecycleAwareManager { fun updateConversation(conversationModel: ConversationModel) - suspend fun loadInitialMessages(withNetworkParams: Bundle, isChatRelaySupported: Boolean) + suspend fun loadInitialMessages(withNetworkParams: Bundle) suspend fun startMessagePolling(hasHighPerformanceBackend: Boolean) 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 new file mode 100644 index 00000000000..53509965d87 --- /dev/null +++ b/app/src/main/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncer.kt @@ -0,0 +1,851 @@ +/* + * Nextcloud Talk - Android Client + * + * SPDX-FileCopyrightText: 2026 Andy Scherzinger + * SPDX-License-Identifier: GPL-3.0-or-later + */ + +package com.nextcloud.talk.chat.data.network + +import android.database.sqlite.SQLiteConstraintException +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.data.database.dao.ChatBlocksDao +import com.nextcloud.talk.data.database.dao.ChatMessagesDao +import com.nextcloud.talk.data.database.mappers.asEntity +import com.nextcloud.talk.data.database.model.ChatBlockEntity +import com.nextcloud.talk.data.database.model.ChatMessageEntity +import com.nextcloud.talk.data.network.NetworkMonitor +import com.nextcloud.talk.data.user.model.User +import com.nextcloud.talk.models.json.chat.ChatMessageJson +import com.nextcloud.talk.utils.SpreedFeatures +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.flow +import kotlinx.coroutines.flow.flowOn +import kotlinx.coroutines.sync.Mutex +import retrofit2.HttpException +import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.atomic.AtomicBoolean +import javax.inject.Inject + +/** + * The chat message fetch-and-persist core, shared between the chat screen + * ([OfflineFirstChatRepository]) and background sync callers. + * + * Every operation takes a [SyncTarget] describing the account, room and thread to sync, so the + * syncer can be used for any room at any time — no open chat required. UI-bound side effects are + * reported through the optional [Events] listener. The only state kept between calls is the + * 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") +class ChatMessageSyncer @Inject constructor( + private val chatDao: ChatMessagesDao, + private val chatBlocksDao: ChatBlocksDao, + private val network: ChatNetworkDataSource, + private val networkMonitor: NetworkMonitor +) { + + /** + * Identifies the conversation (and optionally thread) a sync operation works on. + */ + data class SyncTarget( + val user: User, + val roomToken: String, + val threadId: Long?, + val credentials: String, + val urlForChatting: String + ) { + val internalConversationId: String = "${user.id}@$roomToken" + val accountId: Long + get() = user.id!! + } + + /** + * Side effects that only matter while a chat is on screen. Background callers can pass [NO_EVENTS]. + */ + interface Events { + suspend fun onLastCommonReadChanged(lastCommonRead: Int?) { + // no-op by default + } + + suspend fun onLoadingChanged(isLoading: Boolean) { + // no-op by default + } + + suspend fun onRoomRefreshNeeded() { + // no-op by default + } + + suspend fun onIncomingMessagesFromOthers() { + // no-op by default + } + } + + /** + * [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. + */ + data class SyncOutcome( + val persistedNewMessages: Boolean, + val newestPersistedMessageId: Long?, + val oldestPersistedMessageId: Long? = null, + val persistedMessageCount: Int = 0, + val syncFailed: Boolean = false + ) + + /** + * Builds the query parameters for a chat pull request. setReadMarker stays 0 so a sync never + * moves the user's read marker. Background fetches must additionally pass + * [markNotificationsAsRead] = false so the server keeps push notifications for the fetched + * messages — only supported when the server has the chat-keep-notifications capability. + */ + @Suppress("LongParameterList") + fun buildFieldMap( + lookIntoFuture: Boolean, + timeout: Int, + includeLastKnown: Boolean, + lastKnown: Int?, + limit: Int = DEFAULT_MESSAGES_LIMIT, + threadId: Long? = null, + lastCommonRead: Int? = null, + markNotificationsAsRead: Boolean = true + ): HashMap { + val fieldMap = HashMap() + + fieldMap["includeLastKnown"] = if (includeLastKnown) 1 else 0 + + if (lastKnown != null) { + fieldMap["lastKnownMessageId"] = lastKnown + } + + lastCommonRead?.let { + fieldMap["lastCommonReadId"] = it + } + + threadId?.let { fieldMap["threadId"] = it.toInt() } + + fieldMap["timeout"] = timeout + fieldMap["limit"] = limit + + fieldMap["lookIntoFuture"] = if (lookIntoFuture) 1 else 0 + fieldMap["setReadMarker"] = 0 + + if (!markNotificationsAsRead) { + fieldMap["markNotificationsAsRead"] = 0 + } + + return fieldMap + } + + /** + * Returns true when messages of the conversation (or thread) are cached locally and covered by + * a chat block, i.e. a catch-up only needs to fetch the delta. + */ + fun hasLocalChatBlock(internalConversationId: String, threadId: Long?): Boolean = + chatBlocksDao.getNewestMessageIdFromChatBlocks(internalConversationId, threadId) > 0 + + /** + * Deletes the expired messages of a conversation and reconciles its chat blocks afterwards: + * block boundaries are trimmed to the oldest/newest message that still exists and blocks whose + * messages are all gone are deleted. Without the reconciliation, block boundaries would point + * to deleted messages, e.g. faking coverage up to a message that is no longer cached. + */ + suspend fun cleanupExpiredMessages(internalConversationId: String) { + val deletedMessages = chatDao.deleteExpiredMessages( + internalConversationId, + System.currentTimeMillis() / MILLIS_PER_SECOND + ) + if (deletedMessages == 0) { + return + } + Log.d(TAG, "Deleted $deletedMessages expired messages for $internalConversationId, reconciling chat blocks") + + val blocks = chatBlocksDao.getChatBlocksForConversation(internalConversationId) + for (block in blocks) { + val newestExistingId = chatDao.getNewestMessageIdInRange( + internalConversationId = internalConversationId, + threadId = block.threadId, + oldestMessageId = block.oldestMessageId, + newestMessageId = block.newestMessageId + ) + + if (newestExistingId == null) { + Log.d(TAG, "Deleting chat block without any remaining messages ($internalConversationId)") + chatBlocksDao.deleteChatBlocks(listOf(block)) + continue + } + + val oldestExistingId = chatDao.getOldestMessageIdInRange( + internalConversationId = internalConversationId, + threadId = block.threadId, + oldestMessageId = block.oldestMessageId, + newestMessageId = block.newestMessageId + ) ?: continue + + if (block.oldestMessageId != oldestExistingId || block.newestMessageId != newestExistingId) { + block.oldestMessageId = oldestExistingId + block.newestMessageId = newestExistingId + chatBlocksDao.upsertChatBlock(block) + } + } + } + + /** + * Catches up a room with the server without requiring an open chat. + * + * If a chat block exists, only the delta since the newest locally known message is fetched and + * the block is extended. For rooms without any chat block (never-opened rooms) the newest + * messages are fetched and the initial chat block is created, so cached messages become + * visible to [ChatMessagesDao] block queries right away. + * + * Requires the chat-keep-notifications capability: without it, a background fetch would + * dismiss the user's push notifications for the fetched messages, so the catch-up is skipped + * entirely (same guard as on iOS). + * + * Bursts of catch-up requests for the same room (e.g. one push notification per message of an + * active group chat) are coalesced: while a catch-up runs, further requests only mark a rerun + * and return, and consecutive fetches are paced by [CATCH_UP_COOLDOWN_MILLIS]. + */ + suspend fun catchUpRoom(target: SyncTarget, limit: Int = DEFAULT_MESSAGES_LIMIT): SyncOutcome = + when { + !networkMonitor.isOnline.value -> { + Log.d(TAG, "Device is offline, skipping catch-up for ${target.internalConversationId}") + SYNC_FAILED + } + + !target.user.hasSpreedFeatureCapability(SpreedFeatures.CHAT_KEEP_NOTIFICATIONS.value) -> { + Log.d( + TAG, + "Server lacks ${SpreedFeatures.CHAT_KEEP_NOTIFICATIONS.value}, " + + "skipping catch-up for ${target.internalConversationId}" + ) + NOTHING_SYNCED + } + + else -> coalescedRoomCatchUp(target, limit) + } + + /** + * Runs at most one catch-up per room at a time. A request arriving while one is running only + * marks a rerun: the running catch-up re-fetches once more after finishing, so messages that + * arrived in between are still picked up without a parallel request. Consecutive fetches are + * paced by [CATCH_UP_COOLDOWN_MILLIS] and a single burst performs at most + * [MAX_CATCH_UP_RUNS_PER_BURST] fetches — later messages are covered by their own push or the + * next room list sync. + */ + private suspend fun coalescedRoomCatchUp(target: SyncTarget, limit: Int): SyncOutcome { + val stateKey = syncStateKey(target.internalConversationId, target.threadId) + val state = catchUpStates.getOrPut(stateKey) { RoomCatchUpState() } + + if (!state.mutex.tryLock()) { + state.rerunRequested.set(true) + Log.d(TAG, "Catch-up already running for $stateKey, coalescing into it") + return NOTHING_SYNCED + } + + try { + var outcome: SyncOutcome + var runs = 0 + do { + awaitCatchUpCooldown(state, stateKey) + state.rerunRequested.set(false) + outcome = fetchRoomCatchUp(target, limit) + state.lastCompletedAtMillis = SystemClock.elapsedRealtime() + runs++ + } while (state.rerunRequested.get() && runs < MAX_CATCH_UP_RUNS_PER_BURST) + return outcome + } finally { + state.mutex.unlock() + } + } + + private suspend fun awaitCatchUpCooldown(state: RoomCatchUpState, stateKey: String) { + val elapsedSinceLastCatchUp = SystemClock.elapsedRealtime() - state.lastCompletedAtMillis + val remainingCooldown = CATCH_UP_COOLDOWN_MILLIS - elapsedSinceLastCatchUp + if (state.lastCompletedAtMillis > 0 && remainingCooldown > 0) { + Log.d(TAG, "Catch-up cooldown for $stateKey, delaying fetch by $remainingCooldown ms") + delay(remainingCooldown) + } + } + + private class RoomCatchUpState { + val mutex = Mutex() + val rerunRequested = AtomicBoolean(false) + + @Volatile + var lastCompletedAtMillis = 0L + } + + private val catchUpStates = ConcurrentHashMap() + + /** + * Newest message id per conversation/thread that is known from HTTP syncs — deliberately + * EXCLUDING messages delivered via signaling. Signaling messages extend the latest chat block + * optimistically (assuming they are contiguous on top of it); the insurance request exists to + * verify that assumption and must therefore anchor on the last HTTP-synced message, otherwise + * messages that arrived between the last sync and the signaling delivery would never be + * fetched. Living in this singleton, the anchor survives reopening a chat. + */ + private val lastHttpSyncedMessageIds = ConcurrentHashMap() + + /** + * The newest message id of the conversation/thread confirmed via HTTP sync, or null when no + * sync happened yet since app start. + */ + fun lastHttpSyncedMessageId(internalConversationId: String, threadId: Long?): Long? = + lastHttpSyncedMessageIds[syncStateKey(internalConversationId, threadId)] + + private fun recordHttpSyncedMessageId(target: SyncTarget, messageId: Long) { + lastHttpSyncedMessageIds.merge(syncStateKey(target.internalConversationId, target.threadId), messageId, ::maxOf) + } + + private fun syncStateKey(internalConversationId: String, threadId: Long?): String = + "$internalConversationId#$threadId" + + private suspend fun fetchRoomCatchUp(target: SyncTarget, limit: Int): SyncOutcome { + val newestMessageIdFromDb = + chatBlocksDao.getNewestMessageIdFromChatBlocks(target.internalConversationId, target.threadId) + + val outcome = if (newestMessageIdFromDb > 0) { + tryCloseBacklog( + target = target, + fromMessageId = newestMessageIdFromDb, + limit = limit, + markNotificationsAsRead = false + ) + } else { + pullAndPersistMessages( + target, + buildFieldMap( + lookIntoFuture = false, + timeout = 0, + includeLastKnown = true, + lastKnown = null, + limit = limit, + threadId = target.threadId, + markNotificationsAsRead = false + ) + ) + } + + if (outcome.persistedNewMessages) { + Log.d( + TAG, + "Background catch-up for room ${target.roomToken}: fetched ${outcome.persistedMessageCount} " + + "message(s), ids ${outcome.oldestPersistedMessageId}..${outcome.newestPersistedMessageId}" + + if (newestMessageIdFromDb > 0) " (delta from $newestMessageIdFromDb)" else " (initial fetch)" + ) + } else { + Log.d(TAG, "Background catch-up for room ${target.roomToken}: no new messages") + } + + return outcome + } + + /** + * Tries to fetch the messages newer than [fromMessageId] until the backlog is fully closed — + * there is no guarantee it will be: see the [MAX_BACKLOG_ROUNDS] fallback below. + * + * A single fetch is capped by [limit], so one request only narrows a backlog larger than + * that — and on chat-relay servers the remaining gap would become permanent as soon as a + * signaling message extends the latest chat block over it. Fetches are therefore repeated + * until the server returns fewer messages than the limit. If [MAX_BACKLOG_ROUNDS] full pages + * were fetched and the backlog is still not closed, the newest messages are fetched with + * includeLastKnown instead: that creates a separate top chat block, so the remaining gap + * stays visible in the block structure (closable by scrolling up) rather than a block + * claiming ranges that were never fetched. + */ + @Suppress("LongParameterList", "LongMethod") + suspend fun tryCloseBacklog( + target: SyncTarget, + fromMessageId: Long, + limit: Int = DEFAULT_MESSAGES_LIMIT, + lastCommonRead: Int? = null, + markNotificationsAsRead: Boolean = true, + events: Events = NO_EVENTS + ): SyncOutcome { + var anchor = fromMessageId + var totalCount = 0 + var oldestPersisted: Long? = null + var newestPersisted: Long? = null + + repeat(MAX_BACKLOG_ROUNDS) { + val fieldMap = buildFieldMap( + lookIntoFuture = true, + timeout = 0, + includeLastKnown = false, + lastKnown = anchor.toInt(), + limit = limit, + threadId = target.threadId, + lastCommonRead = lastCommonRead, + markNotificationsAsRead = markNotificationsAsRead + ) + val roundOutcome = pullAndPersistMessages(target, fieldMap, events) + + if (roundOutcome.persistedNewMessages) { + totalCount += roundOutcome.persistedMessageCount + oldestPersisted = oldestPersisted ?: roundOutcome.oldestPersistedMessageId + newestPersisted = roundOutcome.newestPersistedMessageId ?: newestPersisted + } + + val caughtUp = !roundOutcome.persistedNewMessages || roundOutcome.persistedMessageCount < limit + val nextAnchor = roundOutcome.newestPersistedMessageId + if (caughtUp || nextAnchor == null) { + return SyncOutcome( + persistedNewMessages = totalCount > 0, + newestPersistedMessageId = newestPersisted, + oldestPersistedMessageId = oldestPersisted, + persistedMessageCount = totalCount, + syncFailed = roundOutcome.syncFailed + ) + } + anchor = nextAnchor + } + + Log.w( + TAG, + "Backlog above $fromMessageId in ${target.internalConversationId} still not closed after " + + "$MAX_BACKLOG_ROUNDS rounds (persisted $totalCount message(s), ids " + + "$oldestPersisted..$newestPersisted), fetching the newest messages instead" + ) + val fallbackOutcome = pullAndPersistMessages( + target, + buildFieldMap( + lookIntoFuture = false, + timeout = 0, + includeLastKnown = true, + lastKnown = null, + limit = limit, + threadId = target.threadId, + lastCommonRead = lastCommonRead, + markNotificationsAsRead = markNotificationsAsRead + ), + events + ) + return SyncOutcome( + persistedNewMessages = totalCount > 0 || fallbackOutcome.persistedNewMessages, + newestPersistedMessageId = fallbackOutcome.newestPersistedMessageId, + oldestPersistedMessageId = fallbackOutcome.oldestPersistedMessageId, + persistedMessageCount = fallbackOutcome.persistedMessageCount, + syncFailed = fallbackOutcome.syncFailed + ) + } + + fun pullMessagesFlow(target: SyncTarget, fieldMap: HashMap): Flow = + flow { + var attempts = 1 + + while (attempts < MAX_PULL_ATTEMPTS) { + runCatching { + network.pullChatMessages(target.credentials, target.urlForChatting, fieldMap) + }.fold( + onSuccess = { response -> + val result = when (response.code()) { + HTTP_CODE_OK -> ChatPullResult.Success( + messages = response.body()?.ocs?.data.orEmpty(), + lastCommonRead = response.headers()["X-Chat-Last-Common-Read"]?.toInt() + ) + HTTP_CODE_NOT_MODIFIED -> ChatPullResult.NotModified + HTTP_CODE_PRECONDITION_FAILED -> ChatPullResult.PreconditionFailed + else -> ChatPullResult.Error(HttpException(response)) + } + + emit(result) + return@flow + }, + onFailure = { e -> + Log.e(TAG, "Attempt $attempts failed", e) + attempts++ + fieldMap["limit"] = when (attempts) { + 2 -> RETRY_LIMIT_SECOND_ATTEMPT + 3 -> RETRY_LIMIT_THIRD_ATTEMPT + else -> RETRY_LIMIT_FALLBACK_ATTEMPT + } + } + ) + } + + emit(ChatPullResult.Error(IllegalStateException("All attempts failed"))) + }.flowOn(Dispatchers.IO) + + /** + * Pulls messages from the server as described by [fieldMap], persists them and updates the + * chat blocks of [target]. + */ + suspend fun pullAndPersistMessages( + target: SyncTarget, + fieldMap: HashMap, + events: Events = NO_EVENTS + ): SyncOutcome { + val isLongPoll = (fieldMap["timeout"] ?: 0) > 0 + if (!isLongPoll) events.onLoadingChanged(true) + try { + if (!networkMonitor.isOnline.value) { + Log.d(TAG, "Device is offline, can't load chat messages from server") + } + + val queriedMessageId = fieldMap["lastKnownMessageId"] + val lookIntoFuture = fieldMap["lookIntoFuture"] == 1 + + return when (val result = pullMessagesFlow(target, fieldMap).first()) { + is ChatPullResult.Success -> + handleSuccessfulPull(target, result, queriedMessageId, lookIntoFuture, events) + + is ChatPullResult.NotModified -> { + Log.d(TAG, "Server returned NOT_MODIFIED, nothing to update") + if (lookIntoFuture && queriedMessageId != null) { + // the server confirmed there is nothing newer than the queried message, so + // the queried message is a valid HTTP-synced anchor + recordHttpSyncedMessageId(target, queriedMessageId.toLong()) + } + NOTHING_SYNCED + } + + is ChatPullResult.PreconditionFailed -> { + Log.d(TAG, "Server returned PRECONDITION_FAILED, nothing to update") + NOTHING_SYNCED + } + + is ChatPullResult.Error -> { + Log.e(TAG, "Error pulling messages from server", result.throwable) + SYNC_FAILED + } + } + } finally { + if (!isLongPoll) events.onLoadingChanged(false) + } + } + + private suspend fun handleSuccessfulPull( + target: SyncTarget, + result: ChatPullResult.Success, + queriedMessageId: Int?, + lookIntoFuture: Boolean, + events: Events + ): SyncOutcome { + events.onLastCommonReadChanged(result.lastCommonRead) + + val hasHistory = getHasHistory(HTTP_CODE_OK, lookIntoFuture) + + Log.d( + TAG, + "internalConv=${target.internalConversationId} statusCode=$HTTP_CODE_OK " + + "lookIntoFuture=$lookIntoFuture hasHistory=$hasHistory " + + "queriedMessageId=$queriedMessageId" + ) + + val blockContainingQueriedMessage: ChatBlockEntity? = getBlockOfMessage(target, queriedMessageId) + + blockContainingQueriedMessage?.takeIf { !hasHistory }?.apply { + this.hasHistory = false + chatBlocksDao.upsertChatBlock(this) + Log.d(TAG, "End of chat reached, set hasHistory=false") + } + + return if (result.messages.isNotEmpty()) { + val persistedMessages = updateMessagesData( + target, + result.messages, + blockContainingQueriedMessage, + lookIntoFuture, + hasHistory, + events + ) + persistedMessages.maxOfOrNull { it.id }?.let { recordHttpSyncedMessageId(target, it) } + SyncOutcome( + persistedNewMessages = persistedMessages.isNotEmpty(), + newestPersistedMessageId = persistedMessages.maxOfOrNull { it.id }, + oldestPersistedMessageId = persistedMessages.minOfOrNull { it.id }, + persistedMessageCount = persistedMessages.size + ) + } else { + Log.d(TAG, "No new messages to update") + if (lookIntoFuture && queriedMessageId != null) { + // an empty response to a lookIntoFuture request confirms the queried message as + // the newest one, so it is a valid HTTP-synced anchor + recordHttpSyncedMessageId(target, queriedMessageId.toLong()) + } + NOTHING_SYNCED + } + } + + @Suppress("LongParameterList") + private suspend fun updateMessagesData( + target: SyncTarget, + chatMessagesJson: List, + blockContainingQueriedMessage: ChatBlockEntity?, + lookIntoFuture: Boolean, + hasHistory: Boolean, + events: Events + ): List { + val chatMessageEntities = persistChatMessagesAndHandleSystemMessages( + target, + chatMessagesJson, + emitOnIncoming = lookIntoFuture, + events = events + ) + + if (chatMessageEntities.isEmpty()) { + // Persisting was skipped because the conversation is not in the DB yet (see + // persistChatMessagesAndHandleSystemMessages). Without persisted messages there must be + // no chat block update either, otherwise a block would reference missing messages. + Log.w(TAG, "No messages were persisted for ${target.internalConversationId}, skipping chat block update") + return emptyList() + } + + val oldestIdFromSync = chatMessageEntities.minByOrNull { it.id }!!.id + val newestIdFromSync = chatMessageEntities.maxByOrNull { it.id }!!.id + Log.d(TAG, "oldestIdFromSync: $oldestIdFromSync") + Log.d(TAG, "newestIdFromSync: $newestIdFromSync") + + var oldestMessageIdForNewChatBlock = oldestIdFromSync + var newestMessageIdForNewChatBlock = newestIdFromSync + + if (blockContainingQueriedMessage != null) { + if (lookIntoFuture) { + val oldestMessageIdFromBlockOfQueriedMessage = blockContainingQueriedMessage.oldestMessageId + Log.d(TAG, "oldestMessageIdFromBlockOfQueriedMessage: $oldestMessageIdFromBlockOfQueriedMessage") + oldestMessageIdForNewChatBlock = oldestMessageIdFromBlockOfQueriedMessage + } else { + val newestMessageIdFromBlockOfQueriedMessage = blockContainingQueriedMessage.newestMessageId + Log.d(TAG, "newestMessageIdFromBlockOfQueriedMessage: $newestMessageIdFromBlockOfQueriedMessage") + newestMessageIdForNewChatBlock = newestMessageIdFromBlockOfQueriedMessage + } + } + + Log.d(TAG, "oldestMessageIdForNewChatBlock: $oldestMessageIdForNewChatBlock") + Log.d(TAG, "newestMessageIdForNewChatBlock: $newestMessageIdForNewChatBlock") + + val newChatBlock = ChatBlockEntity( + internalConversationId = target.internalConversationId, + accountId = target.accountId, + token = target.roomToken, + threadId = target.threadId, + oldestMessageId = oldestMessageIdForNewChatBlock, + newestMessageId = newestMessageIdForNewChatBlock, + hasHistory = hasHistory + ) + chatBlocksDao.upsertAndMergeConnectedChatBlocks(newChatBlock) + + return chatMessageEntities + } + + suspend fun persistChatMessagesAndHandleSystemMessages( + target: SyncTarget, + chatMessages: List, + emitOnIncoming: Boolean = false, + events: Events = NO_EVENTS + ): List { + handleSystemMessagesThatAffectDatabase(target, chatMessages, events) + + val chatMessageEntities = chatMessages.map { + it.asEntity(target.accountId) + } + + try { + chatDao.upsertChatMessagesAndDeleteTemp(target.internalConversationId, chatMessageEntities) + } catch (e: SQLiteConstraintException) { + // Skipped persisting messages: conversation $internalConversationId not in DB yet. + // This avoids "SQLiteConstraintException: FOREIGN KEY constraint failed". + // It may happen when a notification for a newly created conversation is opened. The websocket was just + // faster than the API request so no conversation for the message exists yet. Just swallow the exception + // and let the insurance request handle it. + Log.w(TAG, "Skip persisting messages from signaling: conversation not in DB yet. Swallowed exception: $e") + return emptyList() + } + + if (emitOnIncoming) { + val hasIncomingFromOther = chatMessages.any { msg -> + msg.systemMessageType == ChatMessage.SystemMessageType.DUMMY && + msg.actorId != target.user.userId + } + if (hasIncomingFromOther) { + events.onIncomingMessagesFromOthers() + } + } + + return chatMessageEntities + } + + /** + * Returns true if all system messages do not require translation. + * Ignores other message types. + */ + fun isUntranslatedSystemMessage(messagesJson: List): Boolean = + messagesJson.all { + it.systemMessageType == ChatMessage.SystemMessageType.DUMMY || + it.systemMessageType in ChatMessage.SYSTEM_MESSAGE_TYPE_UNTRANSLATED + } + + private suspend fun handleSystemMessagesThatAffectDatabase( + target: SyncTarget, + messagesJson: List, + events: Events + ) { + var needsRoomRefresh = false + messagesJson.forEach { messageJson -> + when (messageJson.systemMessageType) { + ChatMessage.SystemMessageType.REACTION, + ChatMessage.SystemMessageType.REACTION_REVOKED, + ChatMessage.SystemMessageType.REACTION_DELETED -> + // Signaling does not include reactionsSelf; derive it so the self-reaction + // border stays correct regardless of whether signaling or the API response lands first. + upsertParentMessage(target, messageJson, deriveReactions = true) + + ChatMessage.SystemMessageType.MESSAGE_DELETED, + ChatMessage.SystemMessageType.POLL_VOTED, + ChatMessage.SystemMessageType.MESSAGE_EDITED -> + upsertParentMessage(target, messageJson) + + ChatMessage.SystemMessageType.LOBBY_NONE, + ChatMessage.SystemMessageType.LOBBY_NON_MODERATORS, + ChatMessage.SystemMessageType.LOBBY_OPEN_TO_EVERYONE -> needsRoomRefresh = true + + ChatMessage.SystemMessageType.CLEARED_CHAT -> { + // for lookIntoFuture just deleting everything would be fine. + // But lets say we did not open the chat for a while and in between it was cleared. + // We just load the last messages but this don't contain the system message. + // We scroll up and load the system message. Deleting everything is not an option as we + // would loose the messages that we want to keep. We only want to + // delete the messages and chatBlocks older than the system message. + chatDao.deleteMessagesOlderThan(target.internalConversationId, messageJson.id) + chatBlocksDao.deleteChatBlocksOlderThan(target.internalConversationId, messageJson.id) + } + + ChatMessage.SystemMessageType.MESSAGE_PINNED, + ChatMessage.SystemMessageType.MESSAGE_UNPINNED -> needsRoomRefresh = true + + else -> {} + } + } + if (needsRoomRefresh) events.onRoomRefreshNeeded() + } + + // the parent message is always the newest state, no matter how old the system message is. + // that's why we can just take the parent, update it in DB and update the UI + private suspend fun upsertParentMessage( + target: SyncTarget, + messageJson: ChatMessageJson, + deriveReactions: Boolean = false + ) { + val parentMessageJson = messageJson.parentMessage ?: return + parentMessageJson.message ?: return + val parentMessageEntity = parentMessageJson.asEntity(target.accountId) + + // Preserve parentMessageId if missing in server response but present in local DB + val existingEntity = chatDao.getChatMessageEntity(target.internalConversationId, parentMessageJson.id) + if (existingEntity != null && parentMessageEntity.parentMessageId == null) { + parentMessageEntity.parentMessageId = existingEntity.parentMessageId + } + + if (deriveReactions) { + parentMessageEntity.reactionsSelf = + deriveReactionsSelf(target, messageJson, existingEntity) + } + + chatDao.upsertChatMessage(parentMessageEntity) + } + + /** + * Derives the correct reactionsSelf list for a parent message when a reaction system message arrives via + * signaling. The signaling payload does not include reactionsSelf, so we preserve the existing DB state and + * apply a targeted update only if the actor of the system message is the current user. + * + * The emoji is always in the message field (messageParameters is empty for all reaction system messages). + * + * This handles the race condition where signaling may arrive before or after + * ReactionsRepositoryImpl has written the optimistic local update. + */ + @Suppress("NestedBlockDepth") + private fun deriveReactionsSelf( + target: SyncTarget, + systemMessageJson: ChatMessageJson, + existingEntity: ChatMessageEntity? + ): ArrayList { + val reactionsSelf = ArrayList(existingEntity?.reactionsSelf ?: emptyList()) + val isCurrentUserActor = systemMessageJson.actorId == target.user.userId && + systemMessageJson.actorType == "users" + + if (isCurrentUserActor) { + val emoji = systemMessageJson.message + if (emoji != null) { + when (systemMessageJson.systemMessageType) { + ChatMessage.SystemMessageType.REACTION -> { + if (!reactionsSelf.contains(emoji)) reactionsSelf.add(emoji) + } + + ChatMessage.SystemMessageType.REACTION_REVOKED, + ChatMessage.SystemMessageType.REACTION_DELETED -> reactionsSelf.remove(emoji) + else -> {} + } + } + } + + return reactionsSelf + } + + /** + * 304 is returned when oldest message of chat was queried or when long polling request returned with no + * modification. hasHistory is only set to false, when 304 was returned for the the oldest message + */ + private fun getHasHistory(statusCode: Int, lookIntoFuture: Boolean): Boolean = + if (statusCode == HTTP_CODE_NOT_MODIFIED) { + lookIntoFuture + } else { + true + } + + suspend fun getBlockOfMessage(target: SyncTarget, queriedMessageId: Int?): ChatBlockEntity? { + var blockContainingQueriedMessage: ChatBlockEntity? = null + if (queriedMessageId != null) { + val blocksContainingQueriedMessage = + chatBlocksDao.getChatBlocksContainingMessageId( + internalConversationId = target.internalConversationId, + threadId = target.threadId, + messageId = queriedMessageId.toLong() + ) + + val chatBlocks = blocksContainingQueriedMessage.first() + if (chatBlocks.size > 1) { + Log.w(TAG, "multiple chat blocks with messageId $queriedMessageId were found") + } + + blockContainingQueriedMessage = if (chatBlocks.isNotEmpty()) { + chatBlocks.first() + } else { + null + } + } + return blockContainingQueriedMessage + } + + companion object { + val TAG: String = ChatMessageSyncer::class.java.simpleName + + val NO_EVENTS: Events = object : Events {} + + private val NOTHING_SYNCED = SyncOutcome(persistedNewMessages = false, newestPersistedMessageId = null) + private val SYNC_FAILED = + SyncOutcome(persistedNewMessages = false, newestPersistedMessageId = null, syncFailed = true) + + private const val DEFAULT_MESSAGES_LIMIT = 100 + private const val MILLIS_PER_SECOND = 1000L + private const val CATCH_UP_COOLDOWN_MILLIS = 5_000L + private const val MAX_CATCH_UP_RUNS_PER_BURST = 3 + private const val MAX_BACKLOG_ROUNDS = 5 + private const val HTTP_CODE_OK: Int = 200 + private const val HTTP_CODE_NOT_MODIFIED = 304 + private const val HTTP_CODE_PRECONDITION_FAILED = 412 + private const val MAX_PULL_ATTEMPTS = 5 + private const val RETRY_LIMIT_SECOND_ATTEMPT = 50 + private const val RETRY_LIMIT_THIRD_ATTEMPT = 10 + private const val RETRY_LIMIT_FALLBACK_ATTEMPT = 5 + } +} 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 ef2f10ead0c..98594e6d856 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 @@ -8,12 +8,10 @@ package com.nextcloud.talk.chat.data.network -import android.database.sqlite.SQLiteConstraintException 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.chat.domain.ChatPullResult import com.nextcloud.talk.data.database.dao.ChatBlocksDao import com.nextcloud.talk.data.database.dao.ChatMessagesDao import com.nextcloud.talk.data.database.mappers.asEntity @@ -33,7 +31,6 @@ import com.nextcloud.talk.models.json.generic.GenericOverall import com.nextcloud.talk.models.json.participants.Participant import com.nextcloud.talk.utils.bundle.BundleKeys import com.nextcloud.talk.utils.message.SendMessageUtils -import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.delay import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.MutableSharedFlow @@ -47,10 +44,8 @@ import kotlinx.coroutines.flow.firstOrNull import kotlinx.coroutines.flow.flatMapLatest import kotlinx.coroutines.flow.flow import kotlinx.coroutines.flow.flowOf -import kotlinx.coroutines.flow.flowOn import kotlinx.coroutines.flow.map import kotlinx.coroutines.flow.take -import retrofit2.HttpException import java.io.IOException import javax.inject.Inject @@ -60,7 +55,8 @@ class OfflineFirstChatRepository @Inject constructor( private val chatDao: ChatMessagesDao, private val chatBlocksDao: ChatBlocksDao, private val network: ChatNetworkDataSource, - private val networkMonitor: NetworkMonitor + private val networkMonitor: NetworkMonitor, + private val syncer: ChatMessageSyncer ) : ChatMessageRepository { lateinit var currentUser: User @@ -94,7 +90,7 @@ class OfflineFirstChatRepository @Inject constructor( get() = _lastCommonReadFlow private val _lastCommonReadFlow: - MutableSharedFlow = MutableSharedFlow() + MutableSharedFlow = MutableSharedFlow(replay = 1) override val lastReadMessageFlow: Flow get() = _lastReadMessageFlow @@ -127,8 +123,6 @@ class OfflineFirstChatRepository @Inject constructor( private lateinit var urlForChatting: String private var threadId: Long? = null - private var latestKnownMessageIdFromSync: Long = 0 - private val requestedParentIds = mutableSetOf() override fun initData( @@ -151,10 +145,39 @@ class OfflineFirstChatRepository @Inject constructor( this.conversationModel = conversationModel } - override suspend fun loadInitialMessages(withNetworkParams: Bundle, isChatRelaySupported: Boolean) { + private val syncTarget: ChatMessageSyncer.SyncTarget + get() = ChatMessageSyncer.SyncTarget( + user = currentUser, + roomToken = roomToken, + threadId = threadId, + credentials = credentials, + urlForChatting = urlForChatting + ) + + private val syncEvents = object : ChatMessageSyncer.Events { + override suspend fun onLastCommonReadChanged(lastCommonRead: Int?) { + newXChatLastCommonRead = lastCommonRead ?: newXChatLastCommonRead + updateUiForLastCommonRead() + } + + override suspend fun onLoadingChanged(isLoading: Boolean) { + _isLoadingFlow.value = isLoading + } + + override suspend fun onRoomRefreshNeeded() { + _roomRefreshFlow.emit(Unit) + } + + override suspend fun onIncomingMessagesFromOthers() { + _incomingMessageFlow.emit(Unit) + } + } + + override suspend fun loadInitialMessages(withNetworkParams: Bundle) { logger.d(TAG, "---- loadInitialMessages ------------") cleanupExpiredMessages() newXChatLastCommonRead = conversationModel.lastCommonReadMessage + updateUiForLastCommonRead() Log.d(TAG, "conversationModel.internalId: " + conversationModel.internalId) Log.d(TAG, "conversationModel.lastReadMessage:" + conversationModel.lastReadMessage) @@ -163,56 +186,68 @@ class OfflineFirstChatRepository @Inject constructor( Log.d(TAG, "newestMessageIdFromDb: $newestMessageIdFromDb") val weAlreadyHaveSomeOfflineMessages = newestMessageIdFromDb > 0 - val weHaveAtLeastTheLastReadMessage = newestMessageIdFromDb >= conversationModel.lastReadMessage.toLong() + val weLikelyOnlyHaveASmallBacklog = weAlreadyHaveSomeOfflineMessages && weHaveAtLeastTheLastReadMessage + Log.d(TAG, "weAlreadyHaveSomeOfflineMessages:$weAlreadyHaveSomeOfflineMessages") Log.d(TAG, "weHaveAtLeastTheLastReadMessage:$weHaveAtLeastTheLastReadMessage") - Log.d(TAG, "isChatRelaySupported:$isChatRelaySupported") + Log.d(TAG, "weLikelyOnlyHaveASmallBacklog:$weLikelyOnlyHaveASmallBacklog") - if (weAlreadyHaveSomeOfflineMessages && weHaveAtLeastTheLastReadMessage && !isChatRelaySupported) { - Log.d( - TAG, - "Initial online request is skipped because offline messages are up to date" + - " until lastReadMessage" + if (weLikelyOnlyHaveASmallBacklog) { + tryCloseBacklogFromNewestOfflineMessage(newestMessageIdFromDb) + } else { + fetchNewestMessagesForInitialLoad( + withNetworkParams, + weAlreadyHaveSomeOfflineMessages, + weHaveAtLeastTheLastReadMessage ) + } + } - // For messages newer than lastRead, lookIntoFuture will load them. - // We must only end up here when NO HPB is used! - // If a HPB is used, longPolling is not available to handle loading of newer messages. - // When a HPB is used the initial request must be made. - } else { - if (isChatRelaySupported) { - Log.d( - TAG, - "An online request for newest 100 messages is made because chatRelay is supported (No long " + - "polling available to catch up with messages newer than last read.)" - ) - } else if (!weAlreadyHaveSomeOfflineMessages) { - Log.d(TAG, "An online request for newest 100 messages is made because offline chat is empty") - if (networkMonitor.isOnline.value.not()) { - // _generalUIFlow.emit(ChatActivity.NO_OFFLINE_MESSAGES_FOUND) - } - } else { - Log.d( - TAG, - "An online request for newest 100 messages is made because we don't have the lastReadMessage " + - "(gaps could be closed by scrolling up to merge the chatblocks)" - ) - } + /** + * Tries to close the backlog since the newest offline message. + */ + private suspend fun tryCloseBacklogFromNewestOfflineMessage(newestMessageIdFromDb: Long) { + Log.d(TAG, "Try to close the backlog from the newest offline message for initial loading") + + syncer.tryCloseBacklog( + target = syncTarget, + fromMessageId = newestMessageIdFromDb, + lastCommonRead = newXChatLastCommonRead, + events = syncEvents + ) + } - // set up field map to load the newest messages - val fieldMap = getFieldMap( - lookIntoFuture = false, - timeout = 0, - includeLastKnown = true, - lastKnown = null + private suspend fun fetchNewestMessagesForInitialLoad( + withNetworkParams: Bundle, + weAlreadyHaveSomeOfflineMessages: Boolean, + weHaveAtLeastTheLastReadMessage: Boolean + ) { + if (!weAlreadyHaveSomeOfflineMessages) { + Log.d(TAG, "An online request for newest 100 messages is made because offline chat is empty") + if (networkMonitor.isOnline.value.not()) { + // _generalUIFlow.emit(ChatActivity.NO_OFFLINE_MESSAGES_FOUND) + } + } else if (!weHaveAtLeastTheLastReadMessage) { + Log.d( + TAG, + "An online request for newest 100 messages is made because we don't have the " + + "lastReadMessage (gaps could be closed by scrolling up to merge the chatblocks)" ) - withNetworkParams.putSerializable(BundleKeys.KEY_FIELD_MAP, fieldMap) - withNetworkParams.putString(BundleKeys.KEY_ROOM_TOKEN, conversationModel.token) - - Log.d(TAG, "Starting online request for initial loading") - getAndPersistMessages(withNetworkParams) } + + // set up field map to load the newest messages + val fieldMap = getFieldMap( + lookIntoFuture = false, + timeout = 0, + includeLastKnown = true, + lastKnown = null + ) + withNetworkParams.putSerializable(BundleKeys.KEY_FIELD_MAP, fieldMap) + withNetworkParams.putString(BundleKeys.KEY_ROOM_TOKEN, conversationModel.token) + + Log.d(TAG, "Starting online request for initial loading") + getAndPersistMessages(withNetworkParams) } override suspend fun startMessagePolling(hasHighPerformanceBackend: Boolean) { @@ -278,37 +313,43 @@ class OfflineFirstChatRepository @Inject constructor( while (true) { delay(INSURANCE_REQUEST_DELAY) - Log.d(TAG, "execute insurance request with latestKnownMessageIdFromSync: $latestKnownMessageIdFromSync") + Log.d(TAG, "execute insurance request") fetchNewMessages() } } private suspend fun cleanupExpiredMessages() { - // For now, only the messages are deleted without adapting the chatBlocks. It may turn out that there must be - // solutions to delete empty chatBlocks and trim chatBlocks if a first or last messages of it does not exist - // anymore - chatDao.deleteExpiredMessages(internalConversationId, System.currentTimeMillis() / MILLIES) + syncer.cleanupExpiredMessages(internalConversationId) } /** - * Fetches messages newer than latest known message. + * Fetches messages newer than the newest message known from HTTP syncs. + * + * The anchor deliberately EXCLUDES messages that were delivered via signaling: those extend + * the latest chat block optimistically (assuming they are contiguous on top of it), and this + * request is the insurance that verifies the assumption — anchoring on the chat block's + * newest message would skip exactly the range it has to check. The anchor lives in the + * singleton [ChatMessageSyncer], so it survives reopening the chat. * * @return `true` if at least one new message was received and persisted. */ override suspend fun fetchNewMessages(): Boolean { cleanupExpiredMessages() - val fieldMap = getFieldMap( - lookIntoFuture = true, - timeout = 0, - includeLastKnown = false, - lastKnown = latestKnownMessageIdFromSync.toInt(), - limit = 200 - ) - val networkParams = Bundle() - networkParams.putSerializable(BundleKeys.KEY_FIELD_MAP, fieldMap) - return getAndPersistMessages(networkParams) + val lastHttpSyncedMessageId = syncer.lastHttpSyncedMessageId(internalConversationId, threadId) ?: 0L + + // tryCloseBacklog loops until the backlog is fully closed, so a backlog larger than the + // request limit is caught up within one insurance cycle instead of narrowing it by one + // page every cycle. + val outcome = syncer.tryCloseBacklog( + target = syncTarget, + fromMessageId = lastHttpSyncedMessageId, + limit = 200, + lastCommonRead = newXChatLastCommonRead, + events = syncEvents + ) + return outcome.persistedNewMessages } override suspend fun loadMoreMessages( @@ -334,7 +375,7 @@ class OfflineFirstChatRepository @Inject constructor( Log.d(TAG, "Starting online request for loadMoreMessages") getAndPersistMessages(withNetworkParams) - return getBlockOfMessage(anchorMessageId.toInt())?.let { + return syncer.getBlockOfMessage(syncTarget, anchorMessageId.toInt())?.let { ChatMessageRepository.MessagesRange( oldestMessageId = it.oldestMessageId, newestMessageId = it.newestMessageId @@ -342,36 +383,22 @@ class OfflineFirstChatRepository @Inject constructor( } } - @Suppress("LongParameterList") private fun getFieldMap( lookIntoFuture: Boolean, timeout: Int, includeLastKnown: Boolean, lastKnown: Int?, limit: Int = DEFAULT_MESSAGES_LIMIT - ): HashMap { - val fieldMap = HashMap() - - fieldMap["includeLastKnown"] = if (includeLastKnown) 1 else 0 - - if (lastKnown != null) { - fieldMap["lastKnownMessageId"] = lastKnown - } - - newXChatLastCommonRead?.let { - fieldMap["lastCommonReadId"] = it - } - - threadId?.let { fieldMap["threadId"] = it.toInt() } - - fieldMap["timeout"] = timeout - fieldMap["limit"] = limit - - fieldMap["lookIntoFuture"] = if (lookIntoFuture) 1 else 0 - fieldMap["setReadMarker"] = 0 - - return fieldMap - } + ): HashMap = + syncer.buildFieldMap( + lookIntoFuture = lookIntoFuture, + timeout = timeout, + includeLastKnown = includeLastKnown, + lastKnown = lastKnown, + limit = limit, + threadId = threadId, + lastCommonRead = newXChatLastCommonRead + ) override suspend fun getNumberOfThreadReplies(threadId: Long): Int = chatDao.getNumberOfThreadReplies(internalConversationId, threadId) @@ -431,7 +458,7 @@ class OfflineFirstChatRepository @Inject constructor( newestMessageId = newestId, hasHistory = true ) - updateBlocks(block) + chatBlocksDao.upsertAndMergeConnectedChatBlocks(block) ChatMessageRepository.MessagesRange( oldestMessageId = oldestId, @@ -479,351 +506,14 @@ class OfflineFirstChatRepository @Inject constructor( } } - fun pullMessagesFlow(bundle: Bundle): Flow = - flow { - val fieldMap = bundle.getSerializable(BundleKeys.KEY_FIELD_MAP) as HashMap - var attempts = 1 - - while (attempts < MAX_PULL_ATTEMPTS) { - runCatching { - network.pullChatMessages(credentials, urlForChatting, fieldMap) - }.fold( - onSuccess = { response -> - val result = when (response.code()) { - HTTP_CODE_OK -> ChatPullResult.Success( - messages = response.body()?.ocs?.data.orEmpty(), - lastCommonRead = response.headers()["X-Chat-Last-Common-Read"]?.toInt() - ) - HTTP_CODE_NOT_MODIFIED -> ChatPullResult.NotModified - HTTP_CODE_PRECONDITION_FAILED -> ChatPullResult.PreconditionFailed - else -> ChatPullResult.Error(HttpException(response)) - } - - emit(result) - return@flow - }, - onFailure = { e -> - Log.e(TAG, "Attempt $attempts failed", e) - attempts++ - fieldMap["limit"] = when (attempts) { - 2 -> RETRY_LIMIT_SECOND_ATTEMPT - 3 -> RETRY_LIMIT_THIRD_ATTEMPT - else -> RETRY_LIMIT_FALLBACK_ATTEMPT - } - } - ) - } - - emit(ChatPullResult.Error(IllegalStateException("All attempts failed"))) - }.flowOn(Dispatchers.IO) - private suspend fun getAndPersistMessages(bundle: Bundle): Boolean { val fieldMap = bundle.getSerializable(BundleKeys.KEY_FIELD_MAP) as HashMap - val isLongPoll = (fieldMap["timeout"] ?: 0) > 0 - if (!isLongPoll) _isLoadingFlow.value = true - try { - if (!networkMonitor.isOnline.value) { - Log.d(TAG, "Device is offline, can't load chat messages from server") - } - - val queriedMessageId = fieldMap["lastKnownMessageId"] - val lookIntoFuture = fieldMap["lookIntoFuture"] == 1 - - val result = pullMessagesFlow(bundle).first() - - when (result) { - is ChatPullResult.Success -> { - newXChatLastCommonRead = result.lastCommonRead - updateUiForLastCommonRead() - - val hasHistory = getHasHistory(HTTP_CODE_OK, lookIntoFuture) - - Log.d( - TAG, - "internalConv=$internalConversationId statusCode=${HTTP_CODE_OK} " + - "lookIntoFuture=$lookIntoFuture hasHistory=$hasHistory " + - "queriedMessageId=$queriedMessageId" - ) - - val blockContainingQueriedMessage: ChatBlockEntity? = getBlockOfMessage(queriedMessageId) - - blockContainingQueriedMessage?.takeIf { !hasHistory }?.apply { - this.hasHistory = false - chatBlocksDao.upsertChatBlock(this) - Log.d(TAG, "End of chat reached, set hasHistory=false") - } - - if (result.messages.isNotEmpty()) { - updateMessagesData( - result.messages, - blockContainingQueriedMessage, - lookIntoFuture, - hasHistory - ) - return true - } else { - Log.d(TAG, "No new messages to update") - return false - } - } - - is ChatPullResult.NotModified -> { - Log.d(TAG, "Server returned NOT_MODIFIED, nothing to update") - return false - } - - is ChatPullResult.PreconditionFailed -> { - Log.d(TAG, "Server returned PRECONDITION_FAILED, nothing to update") - return false - } - - is ChatPullResult.Error -> { - Log.e(TAG, "Error pulling messages from server", result.throwable) - return false - } - } - } finally { - if (!isLongPoll) _isLoadingFlow.value = false - } + val outcome = syncer.pullAndPersistMessages(syncTarget, fieldMap, syncEvents) + return outcome.persistedNewMessages } - private suspend fun OfflineFirstChatRepository.updateMessagesData( - chatMessagesJson: List, - blockContainingQueriedMessage: ChatBlockEntity?, - lookIntoFuture: Boolean, - hasHistory: Boolean - ) { - val chatMessageEntities = - persistChatMessagesAndHandleSystemMessages(chatMessagesJson, emitOnIncoming = lookIntoFuture) - - if (chatMessageEntities.isEmpty()) { - Log.w(TAG, "No messages were persisted, skipping chat block update") - return - } - - val oldestIdFromSync = chatMessageEntities.minByOrNull { it.id }!!.id - val newestIdFromSync = chatMessageEntities.maxByOrNull { it.id }!!.id - Log.d(TAG, "oldestIdFromSync: $oldestIdFromSync") - Log.d(TAG, "newestIdFromSync: $newestIdFromSync") - - latestKnownMessageIdFromSync = maxOf(latestKnownMessageIdFromSync, newestIdFromSync) - - var oldestMessageIdForNewChatBlock = oldestIdFromSync - var newestMessageIdForNewChatBlock = newestIdFromSync - - if (blockContainingQueriedMessage != null) { - if (lookIntoFuture) { - val oldestMessageIdFromBlockOfQueriedMessage = blockContainingQueriedMessage.oldestMessageId - Log.d(TAG, "oldestMessageIdFromBlockOfQueriedMessage: $oldestMessageIdFromBlockOfQueriedMessage") - oldestMessageIdForNewChatBlock = oldestMessageIdFromBlockOfQueriedMessage - } else { - val newestMessageIdFromBlockOfQueriedMessage = blockContainingQueriedMessage.newestMessageId - Log.d(TAG, "newestMessageIdFromBlockOfQueriedMessage: $newestMessageIdFromBlockOfQueriedMessage") - newestMessageIdForNewChatBlock = newestMessageIdFromBlockOfQueriedMessage - } - } - - Log.d(TAG, "oldestMessageIdForNewChatBlock: $oldestMessageIdForNewChatBlock") - Log.d(TAG, "newestMessageIdForNewChatBlock: $newestMessageIdForNewChatBlock") - - val newChatBlock = ChatBlockEntity( - internalConversationId = internalConversationId, - accountId = conversationModel.accountId, - token = conversationModel.token, - threadId = threadId, - oldestMessageId = oldestMessageIdForNewChatBlock, - newestMessageId = newestMessageIdForNewChatBlock, - hasHistory = hasHistory - ) - updateBlocks(newChatBlock) - } - - /** - * Returns true if all system messages do not require translation. - * Ignores other message types. - */ private fun isUntranslatedSystemMessage(messagesJson: List): Boolean = - messagesJson.all { - it.systemMessageType == ChatMessage.SystemMessageType.DUMMY || - it.systemMessageType in ChatMessage.SYSTEM_MESSAGE_TYPE_UNTRANSLATED - } - - private suspend fun handleSystemMessagesThatAffectDatabase(messagesJson: List) { - var needsRoomRefresh = false - messagesJson.forEach { messageJson -> - when (messageJson.systemMessageType) { - ChatMessage.SystemMessageType.REACTION, - ChatMessage.SystemMessageType.REACTION_REVOKED, - ChatMessage.SystemMessageType.REACTION_DELETED -> - // Signaling does not include reactionsSelf; derive it so the self-reaction - // border stays correct regardless of whether signaling or the API response lands first. - upsertParentMessage(messageJson, deriveReactions = true) - - ChatMessage.SystemMessageType.MESSAGE_DELETED, - ChatMessage.SystemMessageType.POLL_VOTED, - ChatMessage.SystemMessageType.MESSAGE_EDITED -> - upsertParentMessage(messageJson) - - ChatMessage.SystemMessageType.LOBBY_NONE, - ChatMessage.SystemMessageType.LOBBY_NON_MODERATORS, - ChatMessage.SystemMessageType.LOBBY_OPEN_TO_EVERYONE -> needsRoomRefresh = true - - ChatMessage.SystemMessageType.CLEARED_CHAT -> { - // for lookIntoFuture just deleting everything would be fine. - // But lets say we did not open the chat for a while and in between it was cleared. - // We just load the last messages but this don't contain the system message. - // We scroll up and load the system message. Deleting everything is not an option as we - // would loose the messages that we want to keep. We only want to - // delete the messages and chatBlocks older than the system message. - chatDao.deleteMessagesOlderThan(internalConversationId, messageJson.id) - chatBlocksDao.deleteChatBlocksOlderThan(internalConversationId, messageJson.id) - } - - ChatMessage.SystemMessageType.MESSAGE_PINNED, - ChatMessage.SystemMessageType.MESSAGE_UNPINNED -> needsRoomRefresh = true - - else -> {} - } - } - if (needsRoomRefresh) _roomRefreshFlow.emit(Unit) - } - - // the parent message is always the newest state, no matter how old the system message is. - // that's why we can just take the parent, update it in DB and update the UI - private suspend fun upsertParentMessage(messageJson: ChatMessageJson, deriveReactions: Boolean = false) { - val parentMessageJson = messageJson.parentMessage ?: return - parentMessageJson.message ?: return - val parentMessageEntity = parentMessageJson.asEntity(currentUser.id!!) - - // Preserve parentMessageId if missing in server response but present in local DB - val existingEntity = chatDao.getChatMessageEntity(internalConversationId, parentMessageJson.id) - if (existingEntity != null && parentMessageEntity.parentMessageId == null) { - parentMessageEntity.parentMessageId = existingEntity.parentMessageId - } - - if (deriveReactions) { - parentMessageEntity.reactionsSelf = - deriveReactionsSelf(messageJson, existingEntity) - } - - chatDao.upsertChatMessage(parentMessageEntity) - } - - /** - * Derives the correct reactionsSelf list for a parent message when a reaction system message arrives via - * signaling. The signaling payload does not include reactionsSelf, so we preserve the existing DB state and - * apply a targeted update only if the actor of the system message is the current user. - * - * The emoji is always in the message field (messageParameters is empty for all reaction system messages). - * - * This handles the race condition where signaling may arrive before or after - * ReactionsRepositoryImpl has written the optimistic local update. - */ - private fun deriveReactionsSelf( - systemMessageJson: ChatMessageJson, - existingEntity: ChatMessageEntity? - ): ArrayList { - val reactionsSelf = ArrayList(existingEntity?.reactionsSelf ?: emptyList()) - val isCurrentUserActor = systemMessageJson.actorId == currentUser.userId && - systemMessageJson.actorType == "users" - - if (isCurrentUserActor) { - val emoji = systemMessageJson.message - if (emoji != null) { - when (systemMessageJson.systemMessageType) { - ChatMessage.SystemMessageType.REACTION -> { - if (!reactionsSelf.contains(emoji)) reactionsSelf.add(emoji) - } - - ChatMessage.SystemMessageType.REACTION_REVOKED, - ChatMessage.SystemMessageType.REACTION_DELETED -> reactionsSelf.remove(emoji) - else -> {} - } - } - } - - return reactionsSelf - } - - /** - * 304 is returned when oldest message of chat was queried or when long polling request returned with no - * modification. hasHistory is only set to false, when 304 was returned for the the oldest message - */ - private fun getHasHistory(statusCode: Int, lookIntoFuture: Boolean): Boolean = - if (statusCode == HTTP_CODE_NOT_MODIFIED) { - lookIntoFuture - } else { - true - } - - private suspend fun getBlockOfMessage(queriedMessageId: Int?): ChatBlockEntity? { - var blockContainingQueriedMessage: ChatBlockEntity? = null - if (queriedMessageId != null) { - val blocksContainingQueriedMessage = - chatBlocksDao.getChatBlocksContainingMessageId( - internalConversationId = internalConversationId, - threadId = threadId, - messageId = queriedMessageId.toLong() - ) - - val chatBlocks = blocksContainingQueriedMessage.first() - if (chatBlocks.size > 1) { - Log.w(TAG, "multiple chat blocks with messageId $queriedMessageId were found") - } - - blockContainingQueriedMessage = if (chatBlocks.isNotEmpty()) { - chatBlocks.first() - } else { - null - } - } - return blockContainingQueriedMessage - } - - private suspend fun updateBlocks(chatBlock: ChatBlockEntity) { - chatBlocksDao.upsertChatBlock(chatBlock) - - val connectedChatBlocks = - chatBlocksDao.getConnectedChatBlocks( - internalConversationId = internalConversationId, - threadId = threadId, - oldestMessageId = chatBlock.oldestMessageId, - newestMessageId = chatBlock.newestMessageId - ).first() - - if (connectedChatBlocks.size == 1) { - Log.d(TAG, "This chatBlock is not connected to others") - val chatBlockFromDb = connectedChatBlocks[0] - Log.d(TAG, "chatBlockFromDb.oldestMessageId: " + chatBlockFromDb.oldestMessageId) - Log.d(TAG, "chatBlockFromDb.newestMessageId: " + chatBlockFromDb.newestMessageId) - } else if (connectedChatBlocks.size > 1) { - Log.d(TAG, "Found " + connectedChatBlocks.size + " chat blocks that are connected") - val oldestIdFromDbChatBlocks = - connectedChatBlocks.minByOrNull { it.oldestMessageId }!!.oldestMessageId - val newestIdFromDbChatBlocks = - connectedChatBlocks.maxByOrNull { it.newestMessageId }!!.newestMessageId - - val hasNoHistory = connectedChatBlocks.any { !it.hasHistory } - val hasHistory = !hasNoHistory - Log.d(TAG, "hasHistory = $hasHistory") - - val newChatBlock = ChatBlockEntity( - internalConversationId = internalConversationId, - accountId = conversationModel.accountId, - token = conversationModel.token, - threadId = threadId, - oldestMessageId = oldestIdFromDbChatBlocks, - newestMessageId = newestIdFromDbChatBlocks, - hasHistory = hasHistory - ) - chatBlocksDao.replaceConnectedChatBlocks(connectedChatBlocks, newChatBlock) - Log.d(TAG, "A new chat block was created that covers all the range of the found chatblocks") - Log.d(TAG, "new chatBlock - oldest MessageId: $oldestIdFromDbChatBlocks") - Log.d(TAG, "new chatBlock - newest MessageId: $newestIdFromDbChatBlocks") - } else { - Log.d(TAG, "No chat block found ....") - } - } + syncer.isUntranslatedSystemMessage(messagesJson) override fun handleOnPause() { itIsPaused = true @@ -1092,37 +782,8 @@ class OfflineFirstChatRepository @Inject constructor( suspend fun persistChatMessagesAndHandleSystemMessages( chatMessages: List, emitOnIncoming: Boolean = false - ): List { - handleSystemMessagesThatAffectDatabase(chatMessages) - - val chatMessageEntities = chatMessages.map { - it.asEntity(currentUser.id!!) - } - - try { - chatDao.upsertChatMessagesAndDeleteTemp(internalConversationId, chatMessageEntities) - } catch (e: SQLiteConstraintException) { - // Skipped persisting messages: conversation $internalConversationId not in DB yet. - // This avoids "SQLiteConstraintException: FOREIGN KEY constraint failed". - // It may happen when a notification for a newly created conversation is opened. The websocket was just - // faster than the API request so no conversation for the message exists yet. Just swallow the exception - // and let the insurance request handle it. - Log.w(TAG, "Skip persisting messages from signaling: conversation not in DB yet. Swallowed exception: $e") - return emptyList() - } - - if (emitOnIncoming) { - val hasIncomingFromOther = chatMessages.any { msg -> - msg.systemMessageType == ChatMessage.SystemMessageType.DUMMY && - msg.actorId != currentUser.userId - } - if (hasIncomingFromOther) { - _incomingMessageFlow.emit(Unit) - } - } - - return chatMessageEntities - } + ): List = + syncer.persistChatMessagesAndHandleSystemMessages(syncTarget, chatMessages, emitOnIncoming, syncEvents) override fun observeLatestMessages(internalConversationId: String): Flow> = chatBlocksDao @@ -1298,15 +959,8 @@ class OfflineFirstChatRepository @Inject constructor( companion object { val TAG: String = OfflineFirstChatRepository::class.java.simpleName - private const val HTTP_CODE_OK: Int = 200 - private const val HTTP_CODE_NOT_MODIFIED = 304 - private const val HTTP_CODE_PRECONDITION_FAILED = 412 private const val HALF_SECOND = 500L private const val DEFAULT_MESSAGES_LIMIT = 100 - private const val MAX_PULL_ATTEMPTS = 5 - private const val RETRY_LIMIT_SECOND_ATTEMPT = 50 - private const val RETRY_LIMIT_THIRD_ATTEMPT = 10 - private const val RETRY_LIMIT_FALLBACK_ATTEMPT = 5 private const val MILLIES = 1000L private const val INSURANCE_REQUEST_DELAY = 2 * 60 * MILLIES 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 c71dc15d05d..996346fe5d3 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 @@ -483,10 +483,7 @@ class ChatViewModel @AssistedInject constructor( null } } - .distinctUntilChangedBy { it.lastReadMessage } - .onEach { - println("Conversation changed: lastRead=${it.lastReadMessage}") - } + .distinctUntilChangedBy { it.lastReadMessage to it.lastCommonReadMessage } private val conversationAndUserFlow = combine(conversationFlow, nonNullUserFlow) { c, u -> c to u } @@ -1057,7 +1054,7 @@ class ChatViewModel @AssistedInject constructor( val (conversation, capabilities) = convAndCaps CombinedInput( messages, - lastCommonRead, + maxOf(lastCommonRead, conversation.lastCommonReadMessage), parentMap, conversation.lastReadMessage, expandedParents, @@ -1277,20 +1274,20 @@ class ChatViewModel @AssistedInject constructor( chatRepository.updateConversation(conversation) - val isChatRelaySupported = withTimeoutOrNull(WEBSOCKET_CONNECT_TIMEOUT_MS) { - awaitChatRelaySupport(user) - } ?: false + // The live-update mode (chat relay vs long polling) is decided in parallel: the + // initial load must never wait for the websocket to connect. + viewModelScope.launch { + val isChatRelaySupported = withTimeoutOrNull(WEBSOCKET_CONNECT_TIMEOUT_MS) { + awaitChatRelaySupport(user) + } ?: false + startMessagePolling(isChatRelaySupported) + } loadInitialMessages( withCredentials = credentials, - withUrl = url, - isChatRelaySupported = isChatRelaySupported + withUrl = url ) - viewModelScope.launch { - startMessagePolling(isChatRelaySupported) - } - getCapabilities(user, chatRoomToken, conversation) } .launchIn(viewModelScope) @@ -1712,13 +1709,12 @@ class ChatViewModel @AssistedInject constructor( } } - suspend fun loadInitialMessages(withCredentials: String, withUrl: String, isChatRelaySupported: Boolean) { + suspend fun loadInitialMessages(withCredentials: String, withUrl: String) { val bundle = Bundle() bundle.putString(BundleKeys.KEY_CHAT_URL, withUrl) bundle.putString(BundleKeys.KEY_CREDENTIALS, withCredentials) chatRepository.loadInitialMessages( - withNetworkParams = bundle, - isChatRelaySupported = isChatRelaySupported + withNetworkParams = bundle ) } 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 182a2469603..80440e73acb 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 @@ -8,7 +8,11 @@ package com.nextcloud.talk.conversationlist.data.network +import android.content.Context +import android.net.ConnectivityManager +import android.os.PowerManager import android.util.Log +import com.nextcloud.talk.chat.data.network.ChatMessageSyncer import com.nextcloud.talk.chat.data.network.ChatNetworkDataSource import com.nextcloud.talk.conversationlist.data.OfflineConversationsRepository import com.nextcloud.talk.data.database.dao.ConversationsDao @@ -18,7 +22,9 @@ import com.nextcloud.talk.data.database.model.ConversationEntity import com.nextcloud.talk.data.network.NetworkMonitor import com.nextcloud.talk.data.user.model.User import com.nextcloud.talk.models.domain.ConversationModel +import com.nextcloud.talk.utils.ApiUtils import com.nextcloud.talk.utils.CapabilitiesUtil.isUserStatusAvailable +import com.nextcloud.talk.utils.SpreedFeatures import io.reactivex.Observer import io.reactivex.android.schedulers.AndroidSchedulers import io.reactivex.disposables.Disposable @@ -30,8 +36,11 @@ import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.first import kotlinx.coroutines.flow.map +import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.launch import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.sync.Semaphore +import kotlinx.coroutines.sync.withPermit import javax.inject.Inject import kotlin.collections.map @@ -39,7 +48,9 @@ class OfflineFirstConversationsRepository @Inject constructor( private val dao: ConversationsDao, private val network: ConversationsNetworkDataSource, private val chatNetworkDataSource: ChatNetworkDataSource, - private val networkMonitor: NetworkMonitor + private val networkMonitor: NetworkMonitor, + private val chatMessageSyncer: ChatMessageSyncer, + private val context: Context ) : OfflineConversationsRepository { override val roomListFlow: Flow> get() = _roomListFlow @@ -159,21 +170,133 @@ class OfflineFirstConversationsRepository @Inject constructor( it.asEntity(user.id!!) } + val previousConversations = dao.getConversationsForUser(user.id!!).first() + .associateBy { it.internalId } + deleteLeftConversations( user, conversationsFromSync ) dao.upsertConversations(user.id!!, conversationsFromSync) + + val roomsWithNewMessages = getRoomsWithNewMessages(conversationsFromSync, previousConversations) + scope.launch { catchUpRoomsWithNewMessages(user, roomsWithNewMessages) } } catch (e: Exception) { Log.e(TAG, "Something went wrong when fetching conversations", e) } return conversationsFromSync } + /** + * Determines the rooms whose messages should be caught up in the background: rooms with + * activity newer than the last synced state (matching the iOS behavior) plus unread rooms that + * have no cached messages yet (never-opened rooms). + */ + private fun getRoomsWithNewMessages( + conversationsFromSync: List, + previousConversations: Map + ): List = + conversationsFromSync.filter { room -> + val previous = previousConversations[room.internalId] + val activityAdvanced = previous == null || room.lastActivity > previous.lastActivity + activityAdvanced || + (room.unreadMessages > 0 && !chatMessageSyncer.hasLocalChatBlock(room.internalId, null)) + } + + /** + * Prefetches the messages of [rooms] into the local database so they are instantly visible + * when a chat is opened. Runs after the room list sync; failures are logged and never affect + * the conversation list itself. + * + * The catch-up is skipped in battery saver mode and when background data is restricted on a + * metered network (mirroring the Low Power Mode guard on iOS), and is bounded to the + * [MAX_ROOMS_TO_CATCH_UP] most recently active rooms with [MAX_CONCURRENT_CATCH_UPS] parallel + * requests, so a fresh install with many rooms cannot cause an unbounded request burst. + */ + private suspend fun catchUpRoomsWithNewMessages(user: User, rooms: List) { + if (rooms.isEmpty() || !isCatchUpAllowed(user)) { + return + } + + val credentials = ApiUtils.getCredentials(user.username, user.token) ?: return + + val cappedRooms = rooms + .sortedByDescending { it.lastActivity } + .take(MAX_ROOMS_TO_CATCH_UP) + if (cappedRooms.size < rooms.size) { + Log.w(TAG, "Capping message catch-up to ${cappedRooms.size} of ${rooms.size} rooms") + } + + Log.d(TAG, "Catching up messages for ${cappedRooms.size} rooms") + coroutineScope { + val semaphore = Semaphore(MAX_CONCURRENT_CATCH_UPS) + cappedRooms.forEach { room -> + launch { + semaphore.withPermit { + val target = ChatMessageSyncer.SyncTarget( + user = user, + roomToken = room.token, + threadId = null, + credentials = credentials, + urlForChatting = ApiUtils.getUrlForChat(CHAT_API_VERSION, user.baseUrl!!, room.token) + ) + runCatching { chatMessageSyncer.catchUpRoom(target) } + .onFailure { Log.e(TAG, "Message catch-up failed for room ${room.token}", it) } + } + } + } + } + } + + private fun isCatchUpAllowed(user: User): Boolean = + when { + !user.hasSpreedFeatureCapability(SpreedFeatures.CHAT_KEEP_NOTIFICATIONS.value) -> { + Log.d(TAG, "Server lacks ${SpreedFeatures.CHAT_KEEP_NOTIFICATIONS.value}, skipping message catch-up") + false + } + + isPowerSaveMode() -> { + Log.d(TAG, "Battery saver is active, skipping message catch-up") + false + } + + isBackgroundDataRestricted() -> { + Log.d(TAG, "Background data is restricted on a metered network, skipping message catch-up") + false + } + + else -> true + } + + private fun isPowerSaveMode(): Boolean { + val powerManager = context.getSystemService(Context.POWER_SERVICE) as PowerManager + return powerManager.isPowerSaveMode + } + + private fun isBackgroundDataRestricted(): Boolean { + val connectivityManager = context.getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManager + return connectivityManager.isActiveNetworkMetered && + connectivityManager.restrictBackgroundStatus == ConnectivityManager.RESTRICT_BACKGROUND_STATUS_ENABLED + } + private suspend fun deleteLeftConversations(user: User, conversationsFromSync: List) { - val conversationsFromSyncIds = conversationsFromSync.map { it.internalId }.toSet() val oldConversationsFromDb = dao.getConversationsForUser(user.id!!).first() + if (conversationsFromSync.isEmpty() && oldConversationsFromDb.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, " + + "skipping deletion of left conversations" + ) + return + } + + val conversationsFromSyncIds = conversationsFromSync.map { it.internalId }.toSet() + val conversationIdsToDelete = oldConversationsFromDb .map { it.internalId } .filterNot { it in conversationsFromSyncIds } @@ -193,5 +316,8 @@ class OfflineFirstConversationsRepository @Inject constructor( companion object { val TAG = OfflineFirstConversationsRepository::class.simpleName + private const val CHAT_API_VERSION = 1 + private const val MAX_ROOMS_TO_CATCH_UP = 20 + private const val MAX_CONCURRENT_CATCH_UPS = 3 } } 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 e1c91d7f403..800315dfea5 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 @@ -16,6 +16,7 @@ import com.nextcloud.talk.account.data.network.NetworkLoginDataSource import com.nextcloud.talk.api.NcApi import com.nextcloud.talk.api.NcApiCoroutines import com.nextcloud.talk.chat.data.ChatMessageRepository +import com.nextcloud.talk.chat.data.network.ChatMessageSyncer import com.nextcloud.talk.chat.data.network.ChatNetworkDataSource import com.nextcloud.talk.chat.data.network.OfflineFirstChatRepository import com.nextcloud.talk.logger.Logger @@ -140,34 +141,57 @@ class RepositoryModule { InvitationsRepositoryImpl(ncApi, ncApiCoroutines) @Provides + @Singleton + fun provideChatMessageSyncer( + chatMessagesDao: ChatMessagesDao, + chatBlocksDao: ChatBlocksDao, + dataSource: ChatNetworkDataSource, + networkMonitor: NetworkMonitor + ): ChatMessageSyncer = + ChatMessageSyncer( + chatMessagesDao, + chatBlocksDao, + dataSource, + networkMonitor + ) + + @Provides + @Suppress("LongParameterList") fun provideOfflineFirstChatRepository( logger: Logger, chatMessagesDao: ChatMessagesDao, chatBlocksDao: ChatBlocksDao, dataSource: ChatNetworkDataSource, - networkMonitor: NetworkMonitor + networkMonitor: NetworkMonitor, + syncer: ChatMessageSyncer ): ChatMessageRepository = OfflineFirstChatRepository( logger, chatMessagesDao, chatBlocksDao, dataSource, - networkMonitor + networkMonitor, + syncer ) @Provides @Singleton + @Suppress("LongParameterList") fun provideOfflineFirstConversationsRepository( dao: ConversationsDao, dataSource: ConversationsNetworkDataSource, chatNetworkDataSource: ChatNetworkDataSource, - networkMonitor: NetworkMonitor + networkMonitor: NetworkMonitor, + chatMessageSyncer: ChatMessageSyncer, + context: Context ): OfflineConversationsRepository = OfflineFirstConversationsRepository( dao, dataSource, chatNetworkDataSource, - networkMonitor + networkMonitor, + chatMessageSyncer, + context ) @Provides diff --git a/app/src/main/java/com/nextcloud/talk/data/database/dao/ChatBlocksDao.kt b/app/src/main/java/com/nextcloud/talk/data/database/dao/ChatBlocksDao.kt index 313ee4d3ccb..91d4730732e 100644 --- a/app/src/main/java/com/nextcloud/talk/data/database/dao/ChatBlocksDao.kt +++ b/app/src/main/java/com/nextcloud/talk/data/database/dao/ChatBlocksDao.kt @@ -7,6 +7,7 @@ package com.nextcloud.talk.data.database.dao +import android.util.Log import androidx.room.Dao import androidx.room.Delete import androidx.room.Insert @@ -19,7 +20,7 @@ import kotlinx.coroutines.flow.Flow @Dao interface ChatBlocksDao { @Delete - fun deleteChatBlocks(blocks: List) + suspend fun deleteChatBlocks(blocks: List) @Query( """ @@ -54,12 +55,12 @@ interface ChatBlocksDao { ORDER BY newestMessageId ASC """ ) - fun getConnectedChatBlocks( + suspend fun getConnectedChatBlocks( internalConversationId: String, threadId: Long?, oldestMessageId: Long, newestMessageId: Long - ): Flow> + ): List @Query( """ @@ -74,6 +75,41 @@ interface ChatBlocksDao { @Insert(onConflict = OnConflictStrategy.REPLACE) suspend fun upsertChatBlock(chatBlock: ChatBlockEntity) + /** + * Upserts [chatBlock] and merges all chat blocks it overlaps into one covering their combined + * range, as a single atomic operation. + * + * The open-path delta fetch, long polling, the insurance request, signaling and the background + * catch-up can all update the blocks of the same conversation concurrently. Without one + * transaction around upsert, connectivity query and merge, two callers could each upsert + * their block and query connectivity before seeing the other's write, leaving overlapping + * blocks behind. + */ + @Transaction + suspend fun upsertAndMergeConnectedChatBlocks(chatBlock: ChatBlockEntity) { + upsertChatBlock(chatBlock) + + val connectedChatBlocks = getConnectedChatBlocks( + internalConversationId = chatBlock.internalConversationId, + threadId = chatBlock.threadId, + oldestMessageId = chatBlock.oldestMessageId, + newestMessageId = chatBlock.newestMessageId + ) + if (connectedChatBlocks.size > 1) { + val mergedBlock = chatBlock.copy( + oldestMessageId = connectedChatBlocks.minOf { it.oldestMessageId }, + newestMessageId = connectedChatBlocks.maxOf { it.newestMessageId }, + hasHistory = connectedChatBlocks.all { it.hasHistory } + ) + replaceConnectedChatBlocks(connectedChatBlocks, mergedBlock) + Log.d( + TAG, + "Merged ${connectedChatBlocks.size} connected chat blocks into " + + "${mergedBlock.oldestMessageId}..${mergedBlock.newestMessageId}" + ) + } + } + @Transaction suspend fun replaceConnectedChatBlocks(connectedBlocks: List, mergedBlock: ChatBlockEntity) { val newestConnectedBlock = connectedBlocks.maxByOrNull { it.newestMessageId } @@ -121,4 +157,18 @@ interface ChatBlocksDao { """ ) fun getLatestChatBlock(internalConversationId: String, threadId: Long?): Flow + + @Query( + """ + SELECT * + FROM ChatBlocks + WHERE internalConversationId = :internalConversationId + ORDER BY newestMessageId ASC + """ + ) + suspend fun getChatBlocksForConversation(internalConversationId: String): List + + companion object { + private const val TAG = "ChatBlocksDao" + } } 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 9c7a289f0fa..66a743da2bf 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 @@ -297,6 +297,40 @@ interface ChatMessagesDao { ) suspend fun deleteExpiredMessages(internalConversationId: String, currentTimeSecs: Long): Int + @Query( + """ + SELECT MIN(id) + FROM ChatMessages + WHERE internalConversationId = :internalConversationId + AND isTemporary = 0 + AND (:threadId IS NULL OR threadId = :threadId) + AND id BETWEEN :oldestMessageId AND :newestMessageId + """ + ) + suspend fun getOldestMessageIdInRange( + internalConversationId: String, + threadId: Long?, + oldestMessageId: Long, + newestMessageId: Long + ): Long? + + @Query( + """ + SELECT MAX(id) + FROM ChatMessages + WHERE internalConversationId = :internalConversationId + AND isTemporary = 0 + AND (:threadId IS NULL OR threadId = :threadId) + AND id BETWEEN :oldestMessageId AND :newestMessageId + """ + ) + suspend fun getNewestMessageIdInRange( + internalConversationId: String, + threadId: Long?, + oldestMessageId: Long, + newestMessageId: Long + ): Long? + @Query( """ DELETE FROM chatmessages diff --git a/app/src/main/java/com/nextcloud/talk/jobs/ChatMessageCatchUpWorker.kt b/app/src/main/java/com/nextcloud/talk/jobs/ChatMessageCatchUpWorker.kt new file mode 100644 index 00000000000..97314055021 --- /dev/null +++ b/app/src/main/java/com/nextcloud/talk/jobs/ChatMessageCatchUpWorker.kt @@ -0,0 +1,134 @@ +/* + * Nextcloud Talk - Android Client + * + * SPDX-FileCopyrightText: 2026 Andy Scherzinger + * SPDX-License-Identifier: GPL-3.0-or-later + */ +package com.nextcloud.talk.jobs + +import android.content.Context +import android.os.PowerManager +import android.util.Log +import androidx.work.BackoffPolicy +import androidx.work.Constraints +import androidx.work.CoroutineWorker +import androidx.work.Data +import androidx.work.NetworkType +import androidx.work.OneTimeWorkRequest +import androidx.work.WorkManager +import androidx.work.WorkRequest +import androidx.work.WorkerParameters +import autodagger.AutoInjector +import com.nextcloud.talk.application.NextcloudTalkApplication +import com.nextcloud.talk.application.NextcloudTalkApplication.Companion.sharedApplication +import com.nextcloud.talk.chat.data.network.ChatMessageSyncer +import com.nextcloud.talk.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 com.nextcloud.talk.utils.bundle.BundleKeys.KEY_THREAD_ID +import java.util.concurrent.TimeUnit +import javax.inject.Inject + +/** + * Prefetches a pushed room's messages into the local database so they are instantly visible when + * the chat is opened from the notification (or later). Enqueued by [NotificationWorker] after the + * notification is displayed, so the fetch never delays or suppresses the notification and + * transient failures are retried with backoff instead of being lost with the notification worker. + * + * Push bursts for the same room enqueue one worker each, but the per-room coalescing of the + * [ChatMessageSyncer] singleton collapses overlapping catch-ups into few actual fetches. Skipped + * in battery saver mode; the chat-keep-notifications capability gate and the offline check are + * handled inside [ChatMessageSyncer.catchUpRoom]. + */ +@AutoInjector(NextcloudTalkApplication::class) +class ChatMessageCatchUpWorker(context: Context, workerParams: WorkerParameters) : + CoroutineWorker(context, workerParams) { + + @Inject + lateinit var userManager: UserManager + + @Inject + lateinit var chatMessageSyncer: ChatMessageSyncer + + override suspend fun doWork(): Result { + sharedApplication!!.componentApplication.inject(this) + + val userId = inputData.getLong(KEY_INTERNAL_USER_ID, -1) + val roomToken = inputData.getString(KEY_ROOM_TOKEN) + val threadId = inputData.getLong(KEY_THREAD_ID, NO_THREAD).takeIf { it != NO_THREAD } + + return when { + userId < 0 || roomToken.isNullOrEmpty() -> { + Log.e(TAG, "Missing user id or room token, dropping message catch-up") + Result.failure() + } + + isPowerSaveMode() -> { + Log.d(TAG, "Battery saver is active, skipping message catch-up for room $roomToken") + Result.success() + } + + else -> catchUpRoom(userId, roomToken, threadId) + } + } + + private suspend fun catchUpRoom(userId: Long, roomToken: String, threadId: Long?): 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 message catch-up") + return Result.failure() + } + + val target = ChatMessageSyncer.SyncTarget( + user = user, + roomToken = roomToken, + threadId = threadId, + credentials = credentials, + urlForChatting = ApiUtils.getUrlForChat(CHAT_API_VERSION, user.baseUrl!!, roomToken) + ) + + val outcome = runCatching { chatMessageSyncer.catchUpRoom(target) }.getOrElse { throwable -> + Log.e(TAG, "Message catch-up failed for room $roomToken", throwable) + null + } + + return if (outcome == null || outcome.syncFailed) { + Log.w(TAG, "Message catch-up for room $roomToken did not complete (attempt ${runAttemptCount + 1})") + retryOrFail() + } else { + Result.success() + } + } + + private fun retryOrFail(): Result = if (runAttemptCount < MAX_RUN_ATTEMPTS - 1) Result.retry() else Result.failure() + + private fun isPowerSaveMode(): Boolean { + val powerManager = applicationContext.getSystemService(Context.POWER_SERVICE) as PowerManager + return powerManager.isPowerSaveMode + } + + companion object { + private val TAG: String = ChatMessageCatchUpWorker::class.java.simpleName + private const val CHAT_API_VERSION = 1 + private const val NO_THREAD = -1L + private const val MAX_RUN_ATTEMPTS = 3 + + fun enqueue(context: Context, userId: Long, roomToken: String, threadId: Long?) { + val data = Data.Builder() + .putLong(KEY_INTERNAL_USER_ID, userId) + .putString(KEY_ROOM_TOKEN, roomToken) + .apply { threadId?.let { putLong(KEY_THREAD_ID, it) } } + .build() + + val catchUpWork = OneTimeWorkRequest.Builder(ChatMessageCatchUpWorker::class.java) + .setInputData(data) + .setConstraints(Constraints.Builder().setRequiredNetworkType(NetworkType.CONNECTED).build()) + .setBackoffCriteria(BackoffPolicy.EXPONENTIAL, WorkRequest.MIN_BACKOFF_MILLIS, TimeUnit.MILLISECONDS) + .build() + + WorkManager.getInstance(context).enqueue(catchUpWork) + } + } +} diff --git a/app/src/main/java/com/nextcloud/talk/jobs/NotificationWorker.kt b/app/src/main/java/com/nextcloud/talk/jobs/NotificationWorker.kt index 5256272b7a3..6ea912878cd 100644 --- a/app/src/main/java/com/nextcloud/talk/jobs/NotificationWorker.kt +++ b/app/src/main/java/com/nextcloud/talk/jobs/NotificationWorker.kt @@ -213,6 +213,22 @@ class NotificationWorker(context: Context, workerParams: WorkerParameters) : Wor getNcDataAndShowNotification(mainActivityIntent) } + /** + * Enqueues a [ChatMessageCatchUpWorker] that prefetches the pushed room's messages into the + * local database so they are instantly visible when the chat is opened from the notification + * (or later). For messages in a thread, [threadId] targets the thread so its chat block is + * extended. The worker runs independently of this one: the notification is displayed + * beforehand and a slow or failing fetch (retried there with backoff) can never delay it. + */ + private fun catchUpPushedRoom(threadId: Long?) { + val roomToken = pushMessage.id + if (pushMessage.type != TYPE_CHAT || roomToken == null) { + logger.d(TAG, "Skipping message catch-up for pushed room (not a chat push)") + return + } + ChatMessageCatchUpWorker.enqueue(applicationContext, user.id!!, roomToken, threadId) + } + private fun handleRemoteTalkSharePushMessage() { val mainActivityIntent = Intent(context, MainActivity::class.java) mainActivityIntent.flags = getIntentFlags() @@ -490,6 +506,7 @@ class NotificationWorker(context: Context, workerParams: WorkerParameters) : Wor threadId?.let { intent.putExtra(KEY_THREAD_ID, it) } showNotification(intent, ncNotification) + catchUpPushedRoom(threadId) } } @@ -504,6 +521,10 @@ class NotificationWorker(context: Context, workerParams: WorkerParameters) : Wor setContentsFromPushNotificationSubject() showNotification(intent, null) + // without the server notification the thread id is unknown — still catch up + // the room itself so the pushed message is cached for the main chat + catchUpPushedRoom(threadId = null) + Log.e(TAG, "Failed to get NC notification. Using decrypted data from push notification itself", e) if (BuildConfig.DEBUG) { Handler(Looper.getMainLooper()).post { diff --git a/app/src/main/java/com/nextcloud/talk/utils/CapabilitiesUtil.kt b/app/src/main/java/com/nextcloud/talk/utils/CapabilitiesUtil.kt index 6e11eed629c..ca41c1279fe 100644 --- a/app/src/main/java/com/nextcloud/talk/utils/CapabilitiesUtil.kt +++ b/app/src/main/java/com/nextcloud/talk/utils/CapabilitiesUtil.kt @@ -40,6 +40,7 @@ enum class SpreedFeatures(val value: String) { GEO_LOCATION_SHARING("geo-location-sharing"), TALK_POLLS("talk-polls"), FAVORITES("favorites"), + CHAT_KEEP_NOTIFICATIONS("chat-keep-notifications"), CHAT_READ_MARKER("chat-read-marker"), CHAT_UNREAD("chat-unread"), EDIT_MESSAGES("edit-messages"), 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 aaa755a46d5..0da2a13a4c6 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 @@ -20,6 +20,7 @@ import com.nextcloud.talk.api.NcApiCoroutines 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.chat.data.network.ChatNetworkDataSource import com.nextcloud.talk.chat.data.network.OfflineFirstChatRepository import com.nextcloud.talk.chat.data.network.RetrofitChatNetwork @@ -148,13 +149,22 @@ class ComposePreviewUtils private constructor(context: Context) { val logger: TestLogger get() = TestLogger + val chatMessageSyncer: ChatMessageSyncer + get() = ChatMessageSyncer( + chatMessagesDao, + chatBlocksDao, + chatNetworkDataSource, + networkMonitor + ) + val chatRepository: ChatMessageRepository get() = OfflineFirstChatRepository( logger, chatMessagesDao, chatBlocksDao, chatNetworkDataSource, - networkMonitor + networkMonitor, + chatMessageSyncer ) val threadsRepository: ThreadsRepository @@ -168,7 +178,9 @@ class ComposePreviewUtils private constructor(context: Context) { conversationsDao, conversationNetworkDataSource, chatNetworkDataSource, - networkMonitor + networkMonitor, + chatMessageSyncer, + mContext ) val reactionsRepository: ReactionsRepository 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 47d621e6318..65dc4524f51 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 @@ -134,6 +134,20 @@ class DummyChatMessagesDaoImpl : ChatMessagesDao { override suspend fun deleteExpiredMessages(internalConversationId: String, currentTimeSecs: Long): Int = 0 + override suspend fun getOldestMessageIdInRange( + internalConversationId: String, + threadId: Long?, + oldestMessageId: Long, + newestMessageId: Long + ): Long? = null + + override suspend fun getNewestMessageIdInRange( + internalConversationId: String, + threadId: Long?, + oldestMessageId: Long, + newestMessageId: Long + ): Long? = null + override fun getNumberOfThreadReplies(internalConversationId: String, threadId: Long): Int = 0 } @@ -260,7 +274,7 @@ class DummyConversationDaoImpl : ConversationsDao { } class DummyChatBlocksDaoImpl : ChatBlocksDao { - override fun deleteChatBlocks(blocks: List) { + override suspend fun deleteChatBlocks(blocks: List) { /* */ } @@ -270,12 +284,12 @@ class DummyChatBlocksDaoImpl : ChatBlocksDao { messageId: Long ): Flow> = flowOf() - override fun getConnectedChatBlocks( + override suspend fun getConnectedChatBlocks( internalConversationId: String, threadId: Long?, oldestMessageId: Long, newestMessageId: Long - ): Flow> = flowOf() + ): List = emptyList() override fun getNewestMessageIdFromChatBlocks(internalConversationId: String, threadId: Long?): Long = 0L @@ -288,4 +302,7 @@ class DummyChatBlocksDaoImpl : ChatBlocksDao { } override fun getLatestChatBlock(internalConversationId: String, threadId: Long?): Flow = flowOf() + + override suspend fun getChatBlocksForConversation(internalConversationId: String): List = + emptyList() } diff --git a/app/src/test/java/android/util/Log.kt b/app/src/test/java/android/util/Log.kt index aa69a10fa32..716ce32a373 100644 --- a/app/src/test/java/android/util/Log.kt +++ b/app/src/test/java/android/util/Log.kt @@ -30,6 +30,13 @@ object Log { return 1 } + @JvmStatic + fun e(tag: String, msg: String, tr: Throwable): Int { + println("ERROR: $tag: $msg: $tr") + + return 1 + } + @JvmStatic fun i(tag: String, msg: String): Int { println("INFO: $tag: $msg") 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 new file mode 100644 index 00000000000..1ffedb5b0a6 --- /dev/null +++ b/app/src/test/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncerTest.kt @@ -0,0 +1,509 @@ +/* + * Nextcloud Talk - Android Client + * + * SPDX-FileCopyrightText: 2026 Andy Scherzinger + * SPDX-License-Identifier: GPL-3.0-or-later + */ + +package com.nextcloud.talk.chat.data.network + +import android.database.sqlite.SQLiteConstraintException +import com.nextcloud.talk.chat.data.model.ChatMessage +import com.nextcloud.talk.data.database.dao.ChatBlocksDao +import com.nextcloud.talk.data.database.dao.ChatMessagesDao +import com.nextcloud.talk.data.database.model.ChatBlockEntity +import com.nextcloud.talk.data.network.NetworkMonitor +import com.nextcloud.talk.data.user.model.User +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 kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.flowOf +import kotlinx.coroutines.launch +import kotlinx.coroutines.test.runTest +import okhttp3.Protocol +import okhttp3.Request +import okhttp3.ResponseBody.Companion.toResponseBody +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertNull +import org.junit.Assert.assertTrue +import org.junit.Before +import org.junit.Test +import org.mockito.kotlin.any +import org.mockito.kotlin.argumentCaptor +import org.mockito.kotlin.doSuspendableAnswer +import org.mockito.kotlin.eq +import org.mockito.kotlin.mock +import org.mockito.kotlin.never +import org.mockito.kotlin.times +import org.mockito.kotlin.verify +import org.mockito.kotlin.verifyBlocking +import org.mockito.kotlin.verifyNoInteractions +import org.mockito.kotlin.whenever +import org.mockito.kotlin.wheneverBlocking +import retrofit2.Response + +@Suppress("TooManyFunctions") +class ChatMessageSyncerTest { + + private val chatDao: ChatMessagesDao = mock() + private val chatBlocksDao: ChatBlocksDao = mock() + private val network: ChatNetworkDataSource = mock() + private val networkMonitor: NetworkMonitor = mock() + + private lateinit var syncer: ChatMessageSyncer + + @Before + fun setUp() { + syncer = ChatMessageSyncer(chatDao, chatBlocksDao, network, networkMonitor) + whenever(networkMonitor.isOnline).thenReturn(MutableStateFlow(true)) + } + + @Test + fun `buildFieldMap keeps push notifications when markNotificationsAsRead is false`() { + val fieldMap = syncer.buildFieldMap( + lookIntoFuture = true, + timeout = 0, + includeLastKnown = false, + lastKnown = 42, + markNotificationsAsRead = false + ) + + assertEquals(0, fieldMap["markNotificationsAsRead"]) + assertEquals(0, fieldMap["setReadMarker"]) + assertEquals(1, fieldMap["lookIntoFuture"]) + assertEquals(42, fieldMap["lastKnownMessageId"]) + } + + @Test + fun `buildFieldMap omits markNotificationsAsRead by default`() { + val fieldMap = syncer.buildFieldMap( + lookIntoFuture = false, + timeout = 0, + includeLastKnown = true, + lastKnown = null + ) + + assertFalse(fieldMap.containsKey("markNotificationsAsRead")) + assertFalse(fieldMap.containsKey("lastKnownMessageId")) + assertEquals(0, fieldMap["setReadMarker"]) + assertEquals(1, fieldMap["includeLastKnown"]) + } + + @Test + fun `catchUpRoom skips when offline and marks the sync as failed`() = + runTest { + whenever(networkMonitor.isOnline).thenReturn(MutableStateFlow(false)) + + val outcome = syncer.catchUpRoom(target()) + + assertFalse(outcome.persistedNewMessages) + assertTrue(outcome.syncFailed) + verifyNoInteractions(network) + } + + @Test + fun `catchUpRoom skips without chat-keep-notifications capability`() = + runTest { + val outcome = syncer.catchUpRoom(target(user(withKeepNotificationsCapability = false))) + + assertFalse(outcome.persistedNewMessages) + // a missing capability is not retryable, so the skip must not count as failure + assertFalse(outcome.syncFailed) + verifyNoInteractions(network) + } + + @Test + fun `catchUpRoom marks the sync as failed when the server request errors`() = + runTest { + whenever(chatBlocksDao.getNewestMessageIdFromChatBlocks(INTERNAL_CONVERSATION_ID, null)) + .thenReturn(42L) + wheneverBlocking { network.pullChatMessages(any(), any(), any()) } + .thenReturn(Response.error(HTTP_INTERNAL_SERVER_ERROR, "".toResponseBody())) + + val outcome = syncer.catchUpRoom(target()) + + assertFalse(outcome.persistedNewMessages) + assertTrue(outcome.syncFailed) + } + + @Test + fun `catchUpRoom delta-fetches from the newest cached message when a chat block exists`() = + runTest { + val existingBlock = block(oldest = 10, newest = 42) + whenever(chatBlocksDao.getNewestMessageIdFromChatBlocks(INTERNAL_CONVERSATION_ID, null)) + .thenReturn(42L) + whenever(chatBlocksDao.getChatBlocksContainingMessageId(INTERNAL_CONVERSATION_ID, null, 42L)) + .thenReturn(flowOf(listOf(existingBlock))) + wheneverBlocking { network.pullChatMessages(any(), any(), any()) } + .thenReturn(Response.success(overall(message(43), message(44)))) + + val outcome = syncer.catchUpRoom(target()) + + assertTrue(outcome.persistedNewMessages) + assertEquals(44L, outcome.newestPersistedMessageId) + assertEquals(43L, outcome.oldestPersistedMessageId) + assertEquals(2, outcome.persistedMessageCount) + + val fieldMapCaptor = argumentCaptor>() + verifyBlocking(network) { pullChatMessages(eq(CREDENTIALS), eq(CHAT_URL), fieldMapCaptor.capture()) } + assertEquals(1, fieldMapCaptor.firstValue["lookIntoFuture"]) + assertEquals(42, fieldMapCaptor.firstValue["lastKnownMessageId"]) + assertEquals(0, fieldMapCaptor.firstValue["markNotificationsAsRead"]) + + val blockCaptor = argumentCaptor() + verifyBlocking(chatBlocksDao) { upsertAndMergeConnectedChatBlocks(blockCaptor.capture()) } + assertEquals(10L, blockCaptor.firstValue.oldestMessageId) + assertEquals(44L, blockCaptor.firstValue.newestMessageId) + } + + @Test + fun `catchUpRoom fetches the newest messages and creates the first block for a never-opened room`() = + runTest { + whenever(chatBlocksDao.getNewestMessageIdFromChatBlocks(INTERNAL_CONVERSATION_ID, null)) + .thenReturn(0L) + wheneverBlocking { network.pullChatMessages(any(), any(), any()) } + .thenReturn(Response.success(overall(message(1), message(2), message(3)))) + + val outcome = syncer.catchUpRoom(target()) + + assertTrue(outcome.persistedNewMessages) + assertEquals(3L, outcome.newestPersistedMessageId) + assertEquals(1L, outcome.oldestPersistedMessageId) + assertEquals(3, outcome.persistedMessageCount) + + val fieldMapCaptor = argumentCaptor>() + verifyBlocking(network) { pullChatMessages(eq(CREDENTIALS), eq(CHAT_URL), fieldMapCaptor.capture()) } + assertEquals(0, fieldMapCaptor.firstValue["lookIntoFuture"]) + assertEquals(1, fieldMapCaptor.firstValue["includeLastKnown"]) + assertFalse(fieldMapCaptor.firstValue.containsKey("lastKnownMessageId")) + assertEquals(0, fieldMapCaptor.firstValue["markNotificationsAsRead"]) + + val blockCaptor = argumentCaptor() + verifyBlocking(chatBlocksDao) { upsertAndMergeConnectedChatBlocks(blockCaptor.capture()) } + assertEquals(1L, blockCaptor.firstValue.oldestMessageId) + assertEquals(3L, blockCaptor.firstValue.newestMessageId) + } + + @Test + fun `catchUpRoom coalesces a burst of calls for the same room into two fetches`() = + runTest { + val existingBlock = block(oldest = 10, newest = 42) + whenever(chatBlocksDao.getNewestMessageIdFromChatBlocks(INTERNAL_CONVERSATION_ID, null)) + .thenReturn(42L) + whenever(chatBlocksDao.getChatBlocksContainingMessageId(INTERNAL_CONVERSATION_ID, null, 42L)) + .thenReturn(flowOf(listOf(existingBlock))) + + val firstFetchStarted = CompletableDeferred() + val firstFetchReleased = CompletableDeferred() + var pullCount = 0 + wheneverBlocking { network.pullChatMessages(any(), any(), any()) }.doSuspendableAnswer { + pullCount++ + if (pullCount == 1) { + firstFetchStarted.complete(Unit) + firstFetchReleased.await() + } + Response.success(overall(message(43), message(44))) + } + + val firstCall = launch { syncer.catchUpRoom(target()) } + firstFetchStarted.await() + assertEquals("first fetch must be in flight", 1, pullCount) + + // a second and third call while the first fetch is in flight must not fetch in + // parallel — they only mark a rerun for the running catch-up + val secondOutcome = syncer.catchUpRoom(target()) + val thirdOutcome = syncer.catchUpRoom(target()) + assertFalse(secondOutcome.persistedNewMessages) + assertFalse(thirdOutcome.persistedNewMessages) + assertEquals(1, pullCount) + + firstFetchReleased.complete(Unit) + firstCall.join() + + // the running catch-up re-fetched exactly once for the whole burst + verifyBlocking(network, times(2)) { pullChatMessages(any(), any(), any()) } + } + + @Test + fun `pullAndPersistMessages extends the queried block and delegates the merge to the dao`() = + runTest { + val blockOfQueriedMessage = block(oldest = 3, newest = 44) + whenever(chatBlocksDao.getChatBlocksContainingMessageId(INTERNAL_CONVERSATION_ID, null, 42L)) + .thenReturn(flowOf(listOf(blockOfQueriedMessage))) + wheneverBlocking { network.pullChatMessages(any(), any(), any()) } + .thenReturn(Response.success(overall(message(43), message(44)))) + + val fieldMap = syncer.buildFieldMap( + lookIntoFuture = true, + timeout = 0, + includeLastKnown = false, + lastKnown = 42 + ) + syncer.pullAndPersistMessages(target(), fieldMap) + + // merging with other overlapping blocks happens atomically inside the dao, the syncer + // only hands over the fetched range extended down to the queried block's oldest id + val blockCaptor = argumentCaptor() + verifyBlocking(chatBlocksDao) { upsertAndMergeConnectedChatBlocks(blockCaptor.capture()) } + assertEquals(3L, blockCaptor.firstValue.oldestMessageId) + assertEquals(44L, blockCaptor.firstValue.newestMessageId) + } + + @Test + fun `pullAndPersistMessages skips the block update when nothing was persisted`() = + runTest { + whenever(chatBlocksDao.getChatBlocksContainingMessageId(INTERNAL_CONVERSATION_ID, null, 42L)) + .thenReturn(flowOf(emptyList())) + wheneverBlocking { network.pullChatMessages(any(), any(), any()) } + .thenReturn(Response.success(overall(message(43)))) + wheneverBlocking { chatDao.upsertChatMessagesAndDeleteTemp(any(), any()) } + .thenThrow(SQLiteConstraintException()) + + val fieldMap = syncer.buildFieldMap( + lookIntoFuture = true, + timeout = 0, + includeLastKnown = false, + lastKnown = 42 + ) + val outcome = syncer.pullAndPersistMessages(target(), fieldMap) + + assertFalse(outcome.persistedNewMessages) + assertNull(outcome.newestPersistedMessageId) + verifyBlocking(chatBlocksDao, never()) { upsertAndMergeConnectedChatBlocks(any()) } + } + + @Test + fun `pullAndPersistMessages treats not modified as no new messages`() = + runTest { + val rawResponse = okhttp3.Response.Builder() + .request(Request.Builder().url(CHAT_URL).build()) + .protocol(Protocol.HTTP_1_1) + .code(HTTP_NOT_MODIFIED) + .message("Not Modified") + .build() + wheneverBlocking { network.pullChatMessages(any(), any(), any()) } + .thenReturn(Response.error("".toResponseBody(), rawResponse)) + whenever(chatBlocksDao.getChatBlocksContainingMessageId(INTERNAL_CONVERSATION_ID, null, 42L)) + .thenReturn(flowOf(emptyList())) + + val fieldMap = syncer.buildFieldMap( + lookIntoFuture = true, + timeout = 0, + includeLastKnown = false, + lastKnown = 42 + ) + val outcome = syncer.pullAndPersistMessages(target(), fieldMap) + + assertFalse(outcome.persistedNewMessages) + verifyBlocking(chatDao, never()) { upsertChatMessagesAndDeleteTemp(any(), any()) } + } + + @Test + fun `tryCloseBacklog loops until the server returns fewer messages than the limit`() = + runTest { + val existingBlock = block(oldest = 10, newest = 42) + whenever(chatBlocksDao.getChatBlocksContainingMessageId(eq(INTERNAL_CONVERSATION_ID), eq(null), any())) + .thenReturn(flowOf(listOf(existingBlock))) + wheneverBlocking { network.pullChatMessages(any(), any(), any()) } + .thenReturn( + Response.success(overall(message(43), message(44))), + Response.success(overall(message(45))) + ) + + val outcome = syncer.tryCloseBacklog(target(), fromMessageId = 42, limit = 2) + + assertTrue(outcome.persistedNewMessages) + assertEquals(3, outcome.persistedMessageCount) + assertEquals(43L, outcome.oldestPersistedMessageId) + assertEquals(45L, outcome.newestPersistedMessageId) + + val fieldMapCaptor = argumentCaptor>() + verifyBlocking(network, times(2)) { pullChatMessages(any(), any(), fieldMapCaptor.capture()) } + assertEquals(42, fieldMapCaptor.firstValue["lastKnownMessageId"]) + assertEquals(44, fieldMapCaptor.secondValue["lastKnownMessageId"]) + } + + @Test + fun `tryCloseBacklog falls back to the newest messages when the backlog persists`() = + runTest { + val existingBlock = block(oldest = 10, newest = 42) + whenever(chatBlocksDao.getChatBlocksContainingMessageId(eq(INTERNAL_CONVERSATION_ID), eq(null), any())) + .thenReturn(flowOf(listOf(existingBlock))) + wheneverBlocking { network.pullChatMessages(any(), any(), any()) } + .thenReturn( + Response.success(overall(message(43))), + Response.success(overall(message(44))), + Response.success(overall(message(45))), + Response.success(overall(message(46))), + Response.success(overall(message(47))), + Response.success(overall(message(100))) + ) + + val outcome = syncer.tryCloseBacklog(target(), fromMessageId = 42, limit = 1) + + // the reported range is the fallback's own — it must not be merged with the backlog + // rounds' range, since the fallback lands in a separate, disconnected chat block + assertTrue(outcome.persistedNewMessages) + assertEquals(1, outcome.persistedMessageCount) + assertEquals(100L, outcome.oldestPersistedMessageId) + assertEquals(100L, outcome.newestPersistedMessageId) + + val fieldMapCaptor = argumentCaptor>() + verifyBlocking(network, times(6)) { pullChatMessages(any(), any(), fieldMapCaptor.capture()) } + // the last request is the fallback: newest messages with includeLastKnown, no anchor + val fallbackFieldMap = fieldMapCaptor.allValues.last() + assertEquals(0, fallbackFieldMap["lookIntoFuture"]) + assertEquals(1, fallbackFieldMap["includeLastKnown"]) + assertFalse(fallbackFieldMap.containsKey("lastKnownMessageId")) + } + + @Test + fun `http pulls record the insurance anchor but signaling persists do not`() = + runTest { + val existingBlock = block(oldest = 10, newest = 42) + whenever(chatBlocksDao.getChatBlocksContainingMessageId(INTERNAL_CONVERSATION_ID, null, 42L)) + .thenReturn(flowOf(listOf(existingBlock))) + wheneverBlocking { network.pullChatMessages(any(), any(), any()) } + .thenReturn(Response.success(overall(message(43), message(44)))) + + val fieldMap = syncer.buildFieldMap( + lookIntoFuture = true, + timeout = 0, + includeLastKnown = false, + lastKnown = 42 + ) + syncer.pullAndPersistMessages(target(), fieldMap) + + assertEquals(44L, syncer.lastHttpSyncedMessageId(INTERNAL_CONVERSATION_ID, null)) + + // a message delivered via signaling must NOT move the anchor: the insurance request + // has to verify the signaling assumption and needs the last HTTP-synced id for that + syncer.persistChatMessagesAndHandleSystemMessages(target(), listOf(message(50))) + + assertEquals(44L, syncer.lastHttpSyncedMessageId(INTERNAL_CONVERSATION_ID, null)) + } + + @Test + fun `an empty lookIntoFuture response confirms the queried anchor`() = + runTest { + whenever(chatBlocksDao.getChatBlocksContainingMessageId(INTERNAL_CONVERSATION_ID, null, 42L)) + .thenReturn(flowOf(emptyList())) + wheneverBlocking { network.pullChatMessages(any(), any(), any()) } + .thenReturn(Response.success(overall())) + + val fieldMap = syncer.buildFieldMap( + lookIntoFuture = true, + timeout = 0, + includeLastKnown = false, + lastKnown = 42 + ) + syncer.pullAndPersistMessages(target(), fieldMap) + + assertEquals(42L, syncer.lastHttpSyncedMessageId(INTERNAL_CONVERSATION_ID, null)) + } + + @Test + fun `cleanupExpiredMessages trims block boundaries and deletes empty blocks`() = + runTest { + val partiallyExpiredBlock = block(oldest = 1, newest = 10) + val fullyExpiredBlock = block(oldest = 20, newest = 30) + wheneverBlocking { chatDao.deleteExpiredMessages(eq(INTERNAL_CONVERSATION_ID), any()) } + .thenReturn(2) + wheneverBlocking { chatBlocksDao.getChatBlocksForConversation(INTERNAL_CONVERSATION_ID) } + .thenReturn(listOf(partiallyExpiredBlock, fullyExpiredBlock)) + wheneverBlocking { chatDao.getNewestMessageIdInRange(INTERNAL_CONVERSATION_ID, null, 1L, 10L) } + .thenReturn(8L) + wheneverBlocking { chatDao.getOldestMessageIdInRange(INTERNAL_CONVERSATION_ID, null, 1L, 10L) } + .thenReturn(2L) + wheneverBlocking { chatDao.getNewestMessageIdInRange(INTERNAL_CONVERSATION_ID, null, 20L, 30L) } + .thenReturn(null) + + syncer.cleanupExpiredMessages(INTERNAL_CONVERSATION_ID) + + val blockCaptor = argumentCaptor() + verifyBlocking(chatBlocksDao) { upsertChatBlock(blockCaptor.capture()) } + assertEquals(2L, blockCaptor.firstValue.oldestMessageId) + assertEquals(8L, blockCaptor.firstValue.newestMessageId) + + verify(chatBlocksDao).deleteChatBlocks(listOf(fullyExpiredBlock)) + } + + @Test + fun `cleanupExpiredMessages leaves blocks untouched when nothing expired`() = + runTest { + wheneverBlocking { chatDao.deleteExpiredMessages(eq(INTERNAL_CONVERSATION_ID), any()) } + .thenReturn(0) + + syncer.cleanupExpiredMessages(INTERNAL_CONVERSATION_ID) + + verifyNoInteractions(chatBlocksDao) + } + + 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 = "https://server.example.com", + capabilities = Capabilities().apply { + spreedCapability = SpreedCapability().apply { this.features = features } + } + ) + } + + private fun target(user: User = user()): ChatMessageSyncer.SyncTarget = + ChatMessageSyncer.SyncTarget( + user = user, + roomToken = ROOM_TOKEN, + threadId = null, + credentials = CREDENTIALS, + urlForChatting = CHAT_URL + ) + + private fun block(oldest: Long, newest: Long): ChatBlockEntity = + ChatBlockEntity( + internalConversationId = INTERNAL_CONVERSATION_ID, + accountId = ACCOUNT_ID, + token = ROOM_TOKEN, + threadId = null, + oldestMessageId = oldest, + newestMessageId = newest, + hasHistory = true + ) + + private fun message(id: Long): ChatMessageJson = + ChatMessageJson( + id = id, + token = ROOM_TOKEN, + actorType = "users", + actorId = "other", + actorDisplayName = "Other User", + timestamp = id, + message = "message $id", + messageType = "comment", + systemMessageType = ChatMessage.SystemMessageType.DUMMY + ) + + 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 ROOM_TOKEN = "room1" + private const val INTERNAL_CONVERSATION_ID = "$ACCOUNT_ID@$ROOM_TOKEN" + private const val CREDENTIALS = "credentials" + private const val CHAT_URL = "https://server.example.com/ocs/v2.php/apps/spreed/api/v1/chat/$ROOM_TOKEN" + private const val HTTP_NOT_MODIFIED = 304 + private const val HTTP_INTERNAL_SERVER_ERROR = 500 + } +} 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 new file mode 100644 index 00000000000..ee64de6ea60 --- /dev/null +++ b/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/RoomListMessagePrefetchIntegrationTest.kt @@ -0,0 +1,196 @@ +/* + * Nextcloud Talk - Android Client + * + * SPDX-FileCopyrightText: 2026 Andy Scherzinger + * 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.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.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.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.delay +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.runBlocking +import org.junit.After +import org.junit.Assert.assertEquals +import org.junit.Before +import org.junit.Test +import org.junit.runner.RunWith +import org.mockito.kotlin.any +import org.mockito.kotlin.eq +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 test for the room list message prefetch: a room list sync with unread rooms must + * leave the chat messages of those rooms in the local database, covered by a contiguous chat + * block reaching the conversation's last message — so opening the chat needs no network request. + */ +@RunWith(RobolectricTestRunner::class) +@Config(application = Application::class, sdk = [33]) +class RoomListMessagePrefetchIntegrationTest { + + private lateinit var db: TalkDatabase + private lateinit var syncer: ChatMessageSyncer + private lateinit var repository: OfflineFirstConversationsRepository + + private val conversationsNetwork: ConversationsNetworkDataSource = mock() + private val chatNetwork: ChatNetworkDataSource = mock() + private val networkMonitor: NetworkMonitor = mock() + + @Before + fun setUp() { + RxAndroidPlugins.setInitMainThreadSchedulerHandler { Schedulers.trampoline() } + RxAndroidPlugins.setMainThreadSchedulerHandler { Schedulers.trampoline() } + + val context = ApplicationProvider.getApplicationContext() + db = Room.inMemoryDatabaseBuilder(context, TalkDatabase::class.java) + .allowMainThreadQueries() + .build() + db.usersDao().saveUser(UserEntity(id = ACCOUNT_ID, userId = "me", username = "me", baseUrl = BASE_URL)) + + whenever(networkMonitor.isOnline).thenReturn(MutableStateFlow(true)) + + syncer = ChatMessageSyncer(db.chatMessagesDao(), db.chatBlocksDao(), chatNetwork, networkMonitor) + repository = OfflineFirstConversationsRepository( + db.conversationsDao(), + conversationsNetwork, + chatNetwork, + networkMonitor, + syncer, + context + ) + } + + @After + fun tearDown() { + db.close() + RxAndroidPlugins.reset() + } + + @Test + fun `room list sync prefetches unread messages into contiguous chat blocks`() { + val rooms = listOf( + conversation(ROOM_A, unreadMessages = 2, lastMessageId = 12), + conversation(ROOM_B, unreadMessages = 1, lastMessageId = 7) + ) + whenever(conversationsNetwork.getRooms(any(), any(), any())).thenReturn(Observable.just(rooms)) + wheneverBlocking { chatNetwork.pullChatMessages(any(), eq(chatUrl(ROOM_A)), any()) } + .thenReturn(Response.success(overall(message(10, ROOM_A), message(11, ROOM_A), message(12, ROOM_A)))) + wheneverBlocking { chatNetwork.pullChatMessages(any(), eq(chatUrl(ROOM_B)), any()) } + .thenReturn(Response.success(overall(message(6, ROOM_B), message(7, ROOM_B)))) + + runBlocking { + repository.getRooms(user()).join() + + // the message catch-up runs as a fire-and-forget coroutine after the room list sync + awaitUntil { + newestBlockMessageId(ROOM_A) == 12L && newestBlockMessageId(ROOM_B) == 7L + } + + assertMessagesUpToLastMessage(ROOM_A, oldestFetched = 10L, lastMessageId = 12L) + assertMessagesUpToLastMessage(ROOM_B, oldestFetched = 6L, lastMessageId = 7L) + } + } + + private suspend fun assertMessagesUpToLastMessage(roomToken: String, oldestFetched: Long, lastMessageId: Long) { + val internalConversationId = "$ACCOUNT_ID@$roomToken" + + val newestCachedMessage = db.chatMessagesDao().getNewestMessageIdInRange( + internalConversationId = internalConversationId, + threadId = null, + oldestMessageId = 0, + newestMessageId = Long.MAX_VALUE + ) + assertEquals("newest cached message must reach lastMessage.id", lastMessageId, newestCachedMessage) + + val blocks = db.chatBlocksDao().getChatBlocksForConversation(internalConversationId) + assertEquals("exactly one contiguous chat block expected", 1, blocks.size) + assertEquals(oldestFetched, blocks[0].oldestMessageId) + assertEquals(lastMessageId, blocks[0].newestMessageId) + } + + private fun newestBlockMessageId(roomToken: String): Long = + db.chatBlocksDao().getNewestMessageIdFromChatBlocks("$ACCOUNT_ID@$roomToken", null) + + private suspend fun awaitUntil(timeoutMillis: Long = TIMEOUT_MILLIS, condition: () -> 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(): User = + User( + id = ACCOUNT_ID, + userId = "me", + username = "me", + baseUrl = BASE_URL, + token = "app-password", + capabilities = Capabilities().apply { + spreedCapability = SpreedCapability().apply { features = listOf("chat-keep-notifications") } + } + ) + + private fun conversation(roomToken: String, unreadMessages: Int, lastMessageId: Long): Conversation = + Conversation( + token = roomToken, + lastActivity = lastMessageId, + unreadMessages = unreadMessages, + lastMessage = message(lastMessageId, roomToken) + ) + + private fun message(id: Long, roomToken: String): ChatMessageJson = + ChatMessageJson( + id = id, + token = roomToken, + actorType = "users", + actorId = "other", + actorDisplayName = "Other User", + timestamp = id, + message = "message $id", + messageType = "comment", + systemMessageType = ChatMessage.SystemMessageType.DUMMY + ) + + private fun overall(vararg messages: ChatMessageJson): ChatOverall = + ChatOverall(ocs = ChatOCS(meta = null, data = messages.toList())) + + private fun chatUrl(roomToken: String): String = ApiUtils.getUrlForChat(1, BASE_URL, roomToken) + + companion object { + private const val ACCOUNT_ID = 1L + private const val BASE_URL = "https://server.example.com" + private const val ROOM_A = "roomA" + private const val ROOM_B = "roomB" + private const val TIMEOUT_MILLIS = 10_000L + private const val POLL_INTERVAL_MILLIS = 50L + } +}