diff --git a/app/build.gradle.kts b/app/build.gradle.kts index bf70a90953..031159c1ec 100644 --- a/app/build.gradle.kts +++ b/app/build.gradle.kts @@ -373,6 +373,7 @@ dependencies { androidTestImplementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:1.11.0") androidTestImplementation("androidx.test:core-ktx:1.7.0") androidTestImplementation("org.mockito:mockito-android:5.22.0") + testImplementation("androidx.work:work-testing:$workVersion") androidTestImplementation("androidx.work:work-testing:$workVersion") androidTestImplementation("androidx.test.espresso:espresso-core:$espressoVersion") { exclude(group = "com.android.support", module = "support-annotations") diff --git a/app/src/main/java/com/nextcloud/talk/application/NextcloudTalkApplication.kt b/app/src/main/java/com/nextcloud/talk/application/NextcloudTalkApplication.kt index cec49766a6..2b59296113 100644 --- a/app/src/main/java/com/nextcloud/talk/application/NextcloudTalkApplication.kt +++ b/app/src/main/java/com/nextcloud/talk/application/NextcloudTalkApplication.kt @@ -51,6 +51,7 @@ import com.nextcloud.talk.dagger.modules.UtilsModule import com.nextcloud.talk.dagger.modules.ViewModelModule import com.nextcloud.talk.filebrowser.webdav.DavUtils import com.nextcloud.talk.jobs.AccountRemovalWorker +import com.nextcloud.talk.jobs.ConversationsSyncWorker import com.nextcloud.talk.jobs.CapabilitiesSyncWorker import com.nextcloud.talk.jobs.SignalingSettingsWorker import com.nextcloud.talk.jobs.WebsocketConnectionsWorker @@ -271,6 +272,8 @@ class NextcloudTalkApplication : ExistingPeriodicWorkPolicy.REPLACE, periodicCapabilitiesUpdateWork ) + + ConversationsSyncWorker.schedule(applicationContext) } override fun onTerminate() { diff --git a/app/src/main/java/com/nextcloud/talk/conversationlist/data/OfflineConversationsRepository.kt b/app/src/main/java/com/nextcloud/talk/conversationlist/data/OfflineConversationsRepository.kt index 5ac4e201ea..d5aa33dfe3 100644 --- a/app/src/main/java/com/nextcloud/talk/conversationlist/data/OfflineConversationsRepository.kt +++ b/app/src/main/java/com/nextcloud/talk/conversationlist/data/OfflineConversationsRepository.kt @@ -8,6 +8,7 @@ package com.nextcloud.talk.conversationlist.data import com.nextcloud.talk.conversationlist.data.network.OfflineFirstConversationsRepository.ConversationResult +import com.nextcloud.talk.data.database.model.ConversationEntity import com.nextcloud.talk.data.user.model.User import com.nextcloud.talk.models.domain.ConversationModel import kotlinx.coroutines.Job @@ -50,6 +51,41 @@ interface OfflineConversationsRepository { @Deprecated("use observeConversation") fun getRooms(user: User, forceFullSync: Boolean = false): Job + /** + * Synchronizes [user]'s conversations with the server and returns once they are stored, + * handing back the rooms whose messages should be caught up, or null when the sync failed. + * The catch-up itself is left to the caller, through [catchUpRooms]. + * + * [getRooms] launches into the repository's own scope and returns immediately, which is what + * the conversation list wants and what a background worker cannot use: WorkManager tears the + * process down once the worker returns, mid-request. This does not select the observed account + * either - that is what the conversation list screen shows, and a worker walking several + * accounts must not move it. Leaving the catch-up to the caller lets a worker sync the room + * lists of all accounts before it spends time on any account's messages. + * + * With [roomListTimeoutMillis] set, the sync is given at most that long and reports failure + * when it runs out. + * + * Connectivity is not pre-checked: the request is sent and a failure is reported like any + * other, so a network that is connected but not yet validated still gets its chance. + * + * A failure is reported only through the return value, never on [syncErrorFlow]: that flow + * feeds the conversation list, which must not show an error of an account it does not show. + * + * Safe to call from any thread: the blocking work runs on the IO dispatcher. + */ + suspend fun syncRooms( + user: User, + forceFullSync: Boolean = false, + roomListTimeoutMillis: Long? = null + ): List? + + /** + * Prefetches the messages of [rooms], as returned by [syncRooms], into the local database and + * returns once that is done. Failures for single rooms are logged, not thrown. + */ + suspend fun catchUpRooms(user: User, rooms: List) + /** * Called once onStart to emit a conversation to [conversationFlow] * to be handled asynchronously. @@ -76,5 +112,11 @@ interface OfflineConversationsRepository { */ fun requireFullSync(accountId: Long) + /** + * When the conversation list of the account with the internal id [accountId] was last fetched + * in full, or null when it never was. Safe to call from any thread. + */ + suspend fun lastFullSyncAt(accountId: Long): Long? + fun observeConversation(accountId: Long, roomToken: String): Flow } 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 00a4744693..28261526d7 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 @@ -35,6 +35,7 @@ import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job +import kotlinx.coroutines.TimeoutCancellationException import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.distinctUntilChanged @@ -44,10 +45,13 @@ import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.launch import kotlinx.coroutines.sync.Semaphore import kotlinx.coroutines.sync.withPermit +import kotlinx.coroutines.withContext +import kotlinx.coroutines.withTimeout import javax.inject.Inject import kotlin.collections.map +import kotlin.time.Duration.Companion.milliseconds -@Suppress("LongParameterList") +@Suppress("LongParameterList", "TooManyFunctions") class OfflineFirstConversationsRepository @Inject constructor( private val dao: ConversationsDao, private val network: ConversationsNetworkDataSource, @@ -101,10 +105,31 @@ class OfflineFirstConversationsRepository @Inject constructor( override fun getRooms(user: User, forceFullSync: Boolean): Job = scope.launch { if (networkMonitor.isOnline.value) { - getRoomsFromServer(user, forceFullSync = forceFullSync) + val roomsWithNewMessages = getRoomsFromServer( + user, + forceFullSync = forceFullSync, + reportSyncError = true + ) ?: return@launch + // a launch of its own, so the returned job completes without waiting for the catch-up + scope.launch { catchUpRooms(user, roomsWithNewMessages) } } } + override suspend fun syncRooms( + user: User, + forceFullSync: Boolean, + roomListTimeoutMillis: Long? + ): List? = + // the sync blocks its thread on the request and on stored state, so it must not run on the caller's + withContext(Dispatchers.IO) { + getRoomsFromServer( + user, + forceFullSync = forceFullSync, + roomListTimeoutMillis = roomListTimeoutMillis, + reportSyncError = false + ) + } + @Suppress("Detekt.TooGenericExceptionCaught") override fun getRoom(user: User, roomToken: String): Job = scope.launch { @@ -151,79 +176,103 @@ class OfflineFirstConversationsRepository @Inject constructor( return getConversation(id, roomToken) } + /** + * Syncs [user]'s room list, returning the rooms whose messages should be caught up, or null + * when the sync failed. + * + * [reportSyncError] lets a failure reach [syncErrorFlow]. That flow does not say which account + * failed and the conversation list shows whatever arrives there, so only a sync of the account + * the list shows may report to it. + */ @Suppress("Detekt.TooGenericExceptionCaught") - private suspend fun getRoomsFromServer(user: User, forceFullSync: Boolean = false): List? { - var conversationsFromSync: List? = null - - if (!networkMonitor.isOnline.value) { - Log.d(TAG, "Device is offline, can't load conversations from server") - return null - } - + private suspend fun getRoomsFromServer( + user: User, + forceFullSync: Boolean = false, + roomListTimeoutMillis: Long? = null, + reportSyncError: Boolean + ): List? { val accountId = user.id!! val modifiedSince = modifiedSinceFor(user, forceFullSync) - val includeStatus = modifiedSince == null && isUserStatusAvailable(user) - - try { - val roomList = withRetry( - retries = NETWORK_FETCH_RETRIES, - initialDelayMillis = NETWORK_FETCH_RETRY_INITIAL_DELAY_MS, - maxDelayMillis = NETWORK_FETCH_RETRY_MAX_DELAY_MS - ) { - network.getRooms(user, user.baseUrl!!, includeStatus, modifiedSince) - .subscribeOn(Schedulers.io()) - .observeOn(AndroidSchedulers.mainThread()) - .blockingSingle() - } - - conversationsFromSync = roomList.conversations.map { - it.asEntity(accountId) - } - - val previousConversations = dao.getConversationsForUser(accountId).first() - .associateBy { it.internalId } - - val serverItems = if (includeStatus) { - conversationsFromSync + return try { + if (roomListTimeoutMillis == null) { + syncRoomList(user, modifiedSince) } else { - keepCachedStatus(conversationsFromSync, previousConversations) + withTimeout(roomListTimeoutMillis.milliseconds) { syncRoomList(user, modifiedSince) } } - - dao.syncConversationsForUser( - accountId = accountId, - serverItems = conversationListUpdater.preservePendingLocalState( - previousConversations, - serverItems - ), - conversationIdsToDelete = if (roomList.wasDelta) { - emptyList() - } else { - determineLeftConversationIds(previousConversations, conversationsFromSync) - } - ) - - rememberSyncedState(accountId, roomList) - - val roomsWithNewMessages = getRoomsWithNewMessages(conversationsFromSync, previousConversations) - scope.launch { catchUpRoomsWithNewMessages(user, roomsWithNewMessages) } + } catch (e: TimeoutCancellationException) { + // the caller's run goes on, so this is a failed sync rather than a cancellation + Log.w(TAG, "Room list sync for account $accountId ran out of its time budget") + null } catch (e: CancellationException) { throw e } catch (e: Exception) { Log.e(TAG, "Something went wrong when fetching conversations", e) storeTimestamp(accountId, KEY_MODIFIED_SINCE, null) - val hasCachedConversations = dao.getConversationsForUser(accountId).first().isNotEmpty() - if (!hasCachedConversations) { + if (reportSyncError && dao.getConversationsForUser(accountId).first().isEmpty()) { _syncErrorFlow.emit(e) } + null + } + } + + /** + * Fetches [user]'s room list, filtered by [modifiedSince] when set, and stores it, returning the + * rooms whose messages should be caught up. + */ + private suspend fun syncRoomList(user: User, modifiedSince: Long?): List { + val accountId = user.id!! + val includeStatus = modifiedSince == null && isUserStatusAvailable(user) + + val roomList = withRetry( + retries = NETWORK_FETCH_RETRIES, + initialDelayMillis = NETWORK_FETCH_RETRY_INITIAL_DELAY_MS, + maxDelayMillis = NETWORK_FETCH_RETRY_MAX_DELAY_MS + ) { + network.getRooms(user, user.baseUrl!!, includeStatus, modifiedSince) + .subscribeOn(Schedulers.io()) + .observeOn(AndroidSchedulers.mainThread()) + .blockingSingle() + } + + val conversationsFromSync = roomList.conversations.map { + it.asEntity(accountId) } - return conversationsFromSync + + val previousConversations = dao.getConversationsForUser(accountId).first() + .associateBy { it.internalId } + + val serverItems = if (includeStatus) { + conversationsFromSync + } else { + keepCachedStatus(conversationsFromSync, previousConversations) + } + + dao.syncConversationsForUser( + accountId = accountId, + serverItems = conversationListUpdater.preservePendingLocalState( + previousConversations, + serverItems + ), + conversationIdsToDelete = if (roomList.wasDelta) { + emptyList() + } else { + determineLeftConversationIds(previousConversations, conversationsFromSync) + } + ) + + rememberSyncedState(accountId, roomList) + + return getRoomsWithNewMessages(conversationsFromSync, previousConversations) } override fun requireFullSync(accountId: Long) { storeTimestamp(accountId, KEY_LAST_FULL_SYNC_AT, null) } + override suspend fun lastFullSyncAt(accountId: Long): Long? = + withContext(Dispatchers.IO) { readTimestamp(accountId, KEY_LAST_FULL_SYNC_AT) } + /** * The value to send as `modifiedSince`, or null when this sync has to be a full one. * @@ -289,14 +338,14 @@ class OfflineFirstConversationsRepository @Inject constructor( /** * 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 conversation list itself, and cancellation propagates. * * 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) { + override suspend fun catchUpRooms(user: User, rooms: List) { if (rooms.isEmpty() || !isCatchUpAllowed(user)) { return } diff --git a/app/src/main/java/com/nextcloud/talk/extensions/ContextExtensions.kt b/app/src/main/java/com/nextcloud/talk/extensions/ContextExtensions.kt new file mode 100644 index 0000000000..7cda514f0d --- /dev/null +++ b/app/src/main/java/com/nextcloud/talk/extensions/ContextExtensions.kt @@ -0,0 +1,13 @@ +/* + * Nextcloud Talk - Android Client + * + * SPDX-FileCopyrightText: 2026 Andy Scherzinger + * SPDX-License-Identifier: GPL-3.0-or-later + */ +package com.nextcloud.talk.extensions + +import android.content.Context +import android.os.PowerManager + +/** Whether the device is currently in battery saver mode. */ +fun Context.isPowerSaveMode(): Boolean = (getSystemService(Context.POWER_SERVICE) as PowerManager).isPowerSaveMode diff --git a/app/src/main/java/com/nextcloud/talk/jobs/ChatMessageCatchUpWorker.kt b/app/src/main/java/com/nextcloud/talk/jobs/ChatMessageCatchUpWorker.kt index cb57651cfa..a3acb4fc61 100644 --- a/app/src/main/java/com/nextcloud/talk/jobs/ChatMessageCatchUpWorker.kt +++ b/app/src/main/java/com/nextcloud/talk/jobs/ChatMessageCatchUpWorker.kt @@ -7,7 +7,6 @@ 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 @@ -22,6 +21,7 @@ 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.extensions.isPowerSaveMode import com.nextcloud.talk.chat.data.network.ChatMessageSyncer import com.nextcloud.talk.data.database.dao.ConversationsDao import com.nextcloud.talk.users.UserManager @@ -71,7 +71,7 @@ class ChatMessageCatchUpWorker(context: Context, workerParams: WorkerParameters) Result.failure() } - isPowerSaveMode() -> { + applicationContext.isPowerSaveMode() -> { Log.d(TAG, "Battery saver is active, skipping message catch-up for room $roomToken") Result.success() } @@ -127,11 +127,6 @@ class ChatMessageCatchUpWorker(context: Context, workerParams: WorkerParameters) 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 diff --git a/app/src/main/java/com/nextcloud/talk/jobs/ConversationsSyncWorker.kt b/app/src/main/java/com/nextcloud/talk/jobs/ConversationsSyncWorker.kt new file mode 100644 index 0000000000..f72068d5b6 --- /dev/null +++ b/app/src/main/java/com/nextcloud/talk/jobs/ConversationsSyncWorker.kt @@ -0,0 +1,237 @@ +/* + * 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.util.Log +import androidx.annotation.VisibleForTesting +import androidx.lifecycle.Lifecycle +import androidx.lifecycle.ProcessLifecycleOwner +import androidx.work.Constraints +import androidx.work.CoroutineWorker +import androidx.work.ExistingPeriodicWorkPolicy +import androidx.work.NetworkType +import androidx.work.PeriodicWorkRequest +import androidx.work.WorkManager +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.conversationlist.data.OfflineConversationsRepository +import com.nextcloud.talk.data.database.model.ConversationEntity +import com.nextcloud.talk.data.user.model.User +import com.nextcloud.talk.extensions.isPowerSaveMode +import com.nextcloud.talk.users.UserManager +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.withContext +import java.util.concurrent.TimeUnit +import javax.inject.Inject + +/** + * Periodic worker that syncs the conversation list, and the messages that sync prefetches, for + * every configured account. + * + * Accounts are synced one after another, in two passes: first the room list of every account, + * then the messages of each. A slow message prefetch therefore cannot keep another account's + * room list from being synced. The run is skipped in battery saver mode and while the app is in + * the foreground, stops syncing room lists once the app comes to the foreground, and a run in which + * any account failed is retried up to [MAX_RUN_ATTEMPTS] times. Network availability is enforced by the + * [NetworkType.CONNECTED] constraint on the request rather than checked here. + */ +@AutoInjector(NextcloudTalkApplication::class) +class ConversationsSyncWorker(context: Context, workerParams: WorkerParameters) : + CoroutineWorker(context, workerParams) { + + @Inject + lateinit var userManager: UserManager + + @Inject + lateinit var conversationsRepository: OfflineConversationsRepository + + override suspend fun doWork(): Result { + sharedApplication!!.componentApplication.inject(this) + return sync() + } + + /** Runs the sync, or reports success without syncing when a guard stands the run down. */ + @VisibleForTesting + internal suspend fun sync(): Result = + when { + applicationContext.isPowerSaveMode() -> { + Log.d(TAG, "Battery saver is active, skipping the background conversation sync") + Result.success() + } + + isAppInForeground() -> { + Log.d(TAG, "App is in the foreground, skipping the background conversation sync") + Result.success() + } + + else -> syncAccounts() + } + + private suspend fun syncAccounts(): Result { + val accounts = readAccounts() + return when { + accounts == null -> retryOrFail() + + accounts.isEmpty() -> { + Log.d(TAG, "No account to sync") + Result.success() + } + + else -> syncAccounts(longestWaitingFirst(accounts)) + } + } + + /** The accounts to sync, or null when they could not be read. */ + private suspend fun readAccounts(): List? = + runCatching { userManager.getUsers() }.getOrElse { throwable -> + if (throwable is CancellationException) throw throwable + Log.e(TAG, "Could not read the accounts to sync", throwable) + null + } + + /** + * [accounts] ordered by the time of their last full sync, oldest first. WorkManager stops a run + * after ten minutes, so a fixed order would let slow accounts at the front keep the ones behind + * them from ever being reached. The order is only an optimisation, so when the sync times + * cannot be read the accounts keep their own order. + */ + private suspend fun longestWaitingFirst(accounts: List): List = + runCatching { + accounts + .map { it to (conversationsRepository.lastFullSyncAt(it.id!!) ?: 0L) } + .sortedBy { (_, lastFullSyncAt) -> lastFullSyncAt } + .map { (user, _) -> user } + }.getOrElse { throwable -> + if (throwable is CancellationException) throw throwable + Log.w(TAG, "Could not read the sync timestamps to order the accounts", throwable) + accounts + } + + private suspend fun syncAccounts(accounts: List): Result { + // room lists first, for every account: they are what the conversation list shows. Messages + // are prefetched only after that, with the time the run has left + val syncedAccounts = syncRoomLists(accounts) + catchUpMessages(syncedAccounts) + + val failed = syncedAccounts.count { (_, roomsWithNewMessages) -> roomsWithNewMessages == null } + + return if (failed == 0) { + Result.success() + } else { + Log.w(TAG, "$failed of ${syncedAccounts.size} tried accounts did not sync (attempt ${runAttemptCount + 1})") + retryOrFail() + } + } + + /** + * Syncs the room list of each of [accounts] in turn, pairing each account with the rooms to + * catch up, or null when its sync failed. Accounts the run did not get to are left out. + */ + private suspend fun syncRoomLists(accounts: List): List?>> { + val syncedAccounts = mutableListOf?>>() + for (user in accounts) { + if (cameToForeground()) break + syncedAccounts += user to syncAccount(user) + } + return syncedAccounts + } + + /** + * Catches up the messages of every account whose room list was synced, even once the app is in + * the foreground: the list stored those rooms' latest activity already, so its own sync no + * longer counts them as having new messages and would not catch them up in the run's place. + */ + private suspend fun catchUpMessages(syncedAccounts: List?>>) { + for ((user, roomsWithNewMessages) in syncedAccounts) { + roomsWithNewMessages?.let { catchUpAccount(user, it) } + } + } + + /** + * Whether the app came to the foreground since the run started. The conversation list then + * syncs on its own, and a run going on alongside it would race it for the same rows, so the + * remaining room lists are left to the next run. + */ + private suspend fun cameToForeground(): Boolean = + isAppInForeground().also { inForeground -> + if (inForeground) Log.d(TAG, "App came to the foreground, leaving the remaining room lists to the next run") + } + + /** + * Syncs a single account's room list, returning the rooms whose messages should be caught up, + * or null when it failed. Failures are logged; the run being cancelled propagates. + * + * The sync is given at most [ROOM_LIST_TIMEOUT_MINUTES] minutes. An account whose list cannot be + * synced would otherwise be able to spend the whole execution window WorkManager grants the run, + * and since its stored timestamp never advances it would keep sorting first and keep the + * accounts behind it from ever being reached. + */ + private suspend fun syncAccount(user: User): List? = + runCatching { + conversationsRepository.syncRooms( + user, + roomListTimeoutMillis = TimeUnit.MINUTES.toMillis(ROOM_LIST_TIMEOUT_MINUTES) + ) + }.getOrElse { throwable -> + if (throwable is CancellationException) throw throwable + Log.e(TAG, "Background conversation sync failed for account ${user.id}", throwable) + null + } + + /** + * Prefetches the messages of an account's [rooms]. A failure is logged and does not fail the + * run: the room list is stored already, and the messages load when the chat is opened. + */ + private suspend fun catchUpAccount(user: User, rooms: List) { + runCatching { conversationsRepository.catchUpRooms(user, rooms) }.onFailure { throwable -> + if (throwable is CancellationException) throw throwable + Log.e(TAG, "Background message catch-up failed for account ${user.id}", throwable) + } + } + + private fun retryOrFail(): Result = if (runAttemptCount < MAX_RUN_ATTEMPTS - 1) Result.retry() else Result.failure() + + private suspend fun isAppInForeground(): Boolean = + withContext(Dispatchers.Main.immediate) { + ProcessLifecycleOwner.get().lifecycle.currentState.isAtLeast(Lifecycle.State.STARTED) + } + + companion object { + private val TAG: String = ConversationsSyncWorker::class.java.simpleName + private const val MAX_RUN_ATTEMPTS = 3 + private const val ROOM_LIST_TIMEOUT_MINUTES = 2L + private const val REPEAT_INTERVAL_MINUTES = 15L + const val UNIQUE_WORK_NAME = "PeriodicConversationsSync" + + /** + * Schedules the worker to run every [REPEAT_INTERVAL_MINUTES] minutes while a network is + * available. An already scheduled run is updated to this request rather than replaced, so + * a changed interval or constraint reaches existing installs while the period keeps its + * timing - replacing it would restart the period on every app start, and an app opened + * often would never reach a run. + */ + fun schedule(context: Context) { + val work = PeriodicWorkRequest.Builder( + ConversationsSyncWorker::class.java, + REPEAT_INTERVAL_MINUTES, + TimeUnit.MINUTES + ).setConstraints( + Constraints.Builder().setRequiredNetworkType(NetworkType.CONNECTED).build() + ).build() + + WorkManager.getInstance(context).enqueueUniquePeriodicWork( + UNIQUE_WORK_NAME, + ExistingPeriodicWorkPolicy.UPDATE, + work + ) + } + } +} diff --git a/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/OfflineFirstConversationsRepositoryTest.kt b/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/OfflineFirstConversationsRepositoryTest.kt index 9e00d419a6..5c1a9717bd 100644 --- a/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/OfflineFirstConversationsRepositoryTest.kt +++ b/app/src/test/java/com/nextcloud/talk/conversationlist/data/network/OfflineFirstConversationsRepositoryTest.kt @@ -29,6 +29,7 @@ import io.reactivex.Observable import io.reactivex.android.plugins.RxAndroidPlugins import io.reactivex.schedulers.Schedulers import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.awaitCancellation import kotlinx.coroutines.delay import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.first @@ -47,9 +48,11 @@ import org.mockito.Mockito.timeout import org.mockito.kotlin.any import org.mockito.kotlin.anyOrNull 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.stub import org.mockito.kotlin.verify import org.mockito.kotlin.verifyBlocking import org.mockito.kotlin.wheneverBlocking @@ -131,6 +134,18 @@ class OfflineFirstConversationsRepositoryTest { verifyBlocking(network, never()) { getRooms(any(), any(), any(), anyOrNull()) } } + @Test + fun `syncRooms tries the server even when the device is not known to be online`() = + runBlocking { + whenever(networkMonitor.isOnline).thenReturn(MutableStateFlow(false)) + val room = conversation(token = ROOM_TOKEN, lastActivity = 5, unreadMessages = 0) + whenever(network.getRooms(any(), any(), any(), anyOrNull())).thenReturn(roomList(listOf(room))) + + val roomsWithNewMessages = repository.syncRooms(user()) + + assertEquals(listOf(ROOM_TOKEN), roomsWithNewMessages?.map { it.token }) + } + @Test fun `getRooms fetches conversations from the server and syncs them locally when online`() = runBlocking { @@ -145,6 +160,107 @@ class OfflineFirstConversationsRepositoryTest { assertEquals(ROOM_TOKEN, entityCaptor.firstValue.first().token) } + @Test + fun `syncRooms reports a sync whose database write failed as failed`() = + runBlocking { + val room = conversation(token = ROOM_TOKEN, lastActivity = 5, unreadMessages = 0) + whenever(network.getRooms(any(), any(), any(), anyOrNull())).thenReturn(roomList(listOf(room))) + wheneverBlocking { dao.syncConversationsForUser(any(), any(), any()) } + .thenThrow(IllegalStateException("database is gone")) + + assertNull(repository.syncRooms(user())) + } + + @Test + fun `syncRooms does its blocking work off the caller's thread`() = + runBlocking { + val room = conversation(token = ROOM_TOKEN, lastActivity = 5, unreadMessages = 0) + var requestThread: Thread? = null + whenever(network.getRooms(any(), any(), any(), anyOrNull())).thenAnswer { + requestThread = Thread.currentThread() + roomList(listOf(room)) + } + + repository.syncRooms(user()) + + assertTrue(requestThread != null && requestThread != Thread.currentThread()) + } + + @Test + fun `getRooms reports a failed sync of an account without cached conversations`() = + runBlocking { + val room = conversation(token = ROOM_TOKEN, lastActivity = 5, unreadMessages = 0) + whenever(network.getRooms(any(), any(), any(), anyOrNull())).thenReturn(roomList(listOf(room))) + wheneverBlocking { dao.syncConversationsForUser(any(), any(), any()) } + .thenThrow(IllegalStateException("database is gone")) + val errors = mutableListOf() + val collector = launch(Dispatchers.Unconfined) { repository.syncErrorFlow.collect { errors += it } } + + repository.getRooms(user()).join() + + awaitUntil { errors.isNotEmpty() } + collector.cancel() + } + + @Test + fun `syncRooms keeps a failed sync off the conversation list's error flow`() = + runBlocking { + val room = conversation(token = ROOM_TOKEN, lastActivity = 5, unreadMessages = 0) + whenever(network.getRooms(any(), any(), any(), anyOrNull())).thenReturn(roomList(listOf(room))) + wheneverBlocking { dao.syncConversationsForUser(any(), any(), any()) } + .thenThrow(IllegalStateException("database is gone")) + val errors = mutableListOf() + val collector = launch(Dispatchers.Unconfined) { repository.syncErrorFlow.collect { errors += it } } + + repository.syncRooms(user()) + + collector.cancel() + assertEquals(emptyList(), errors) + } + + @Test + fun `syncRooms reports a room list sync that runs out of its time budget as failed`() = + runTest { + val room = conversation(token = ROOM_TOKEN, lastActivity = 5, unreadMessages = 2) + whenever(network.getRooms(any(), any(), any(), anyOrNull())).thenReturn(roomList(listOf(room))) + dao.stub { + onBlocking { syncConversationsForUser(any(), any(), any()) } doSuspendableAnswer { awaitCancellation() } + } + + assertNull(repository.syncRooms(user(), roomListTimeoutMillis = ROOM_LIST_TIMEOUT_MILLIS)) + verifyBlocking(chatMessageSyncer, never()) { catchUpRoom(any(), any(), anyOrNull(), any()) } + } + + @Test + fun `syncRooms hands the rooms with new messages back instead of catching them up`() = + runBlocking { + stubCatchUpRoom() + val room = conversation(token = ROOM_TOKEN, lastActivity = 5, unreadMessages = 2) + whenever(network.getRooms(any(), any(), any(), anyOrNull())).thenReturn(roomList(listOf(room))) + + val roomsWithNewMessages = repository.syncRooms(user()) + + assertEquals(listOf(ROOM_TOKEN), roomsWithNewMessages?.map { it.token }) + verifyBlocking(chatMessageSyncer, after(AFTER_DELAY_MILLIS).never()) { + catchUpRoom(any(), any(), anyOrNull(), any()) + } + } + + @Test + fun `catchUpRooms prefetches the messages of the rooms syncRooms handed back`() = + runBlocking { + stubCatchUpRoom() + val room = conversation(token = ROOM_TOKEN, lastActivity = 5, unreadMessages = 2) + whenever(network.getRooms(any(), any(), any(), anyOrNull())).thenReturn(roomList(listOf(room))) + val roomsWithNewMessages = repository.syncRooms(user())!! + + repository.catchUpRooms(user(), roomsWithNewMessages) + + val targetCaptor = argumentCaptor() + verifyBlocking(chatMessageSyncer) { catchUpRoom(targetCaptor.capture(), any(), anyOrNull(), any()) } + assertEquals(ROOM_TOKEN, targetCaptor.firstValue.roomToken) + } + @Test fun `getRooms skips deleting local conversations when the server returns an empty list`() = runBlocking { @@ -459,6 +575,7 @@ class OfflineFirstConversationsRepositoryTest { private const val AWAIT_TIMEOUT_MILLIS = 2000L private const val POLL_INTERVAL_MILLIS = 50L private const val COLLECTOR_STARTUP_MILLIS = 100L + private const val ROOM_LIST_TIMEOUT_MILLIS = 1000L } } diff --git a/app/src/test/java/com/nextcloud/talk/jobs/ConversationsSyncWorkerTest.kt b/app/src/test/java/com/nextcloud/talk/jobs/ConversationsSyncWorkerTest.kt new file mode 100644 index 0000000000..62b30490fe --- /dev/null +++ b/app/src/test/java/com/nextcloud/talk/jobs/ConversationsSyncWorkerTest.kt @@ -0,0 +1,293 @@ +/* + * Nextcloud Talk - Android Client + * + * SPDX-FileCopyrightText: 2026 Andy Scherzinger + * SPDX-License-Identifier: GPL-3.0-or-later + */ + +package com.nextcloud.talk.jobs + +import android.app.Application +import android.content.Context +import android.os.PowerManager +import androidx.lifecycle.Lifecycle +import androidx.lifecycle.LifecycleRegistry +import androidx.lifecycle.ProcessLifecycleOwner +import androidx.test.core.app.ApplicationProvider +import androidx.work.ListenableWorker +import androidx.work.WorkInfo +import androidx.work.WorkManager +import androidx.work.testing.TestListenableWorkerBuilder +import androidx.work.testing.WorkManagerTestInitHelper +import com.nextcloud.talk.conversationlist.data.OfflineConversationsRepository +import com.nextcloud.talk.data.database.model.ConversationEntity +import com.nextcloud.talk.data.user.model.User +import com.nextcloud.talk.users.UserManager +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.runBlocking +import org.junit.After +import org.junit.Assert.assertEquals +import org.junit.Assert.assertThrows +import org.junit.Test +import org.junit.runner.RunWith +import org.mockito.kotlin.any +import org.mockito.kotlin.anyOrNull +import org.mockito.kotlin.argumentCaptor +import org.mockito.kotlin.eq +import org.mockito.kotlin.inOrder +import org.mockito.kotlin.mock +import org.mockito.kotlin.never +import org.mockito.kotlin.times +import org.mockito.kotlin.verifyBlocking +import org.mockito.kotlin.wheneverBlocking +import org.robolectric.RobolectricTestRunner +import org.robolectric.Shadows.shadowOf +import org.robolectric.annotation.Config +import java.util.concurrent.TimeUnit + +/** + * Tests for [ConversationsSyncWorker]: which accounts a run syncs, when a run stands down without + * syncing, and how a failed run reports itself. + */ +@Suppress("TooManyFunctions") +@RunWith(RobolectricTestRunner::class) +@Config(application = Application::class, sdk = [33]) +class ConversationsSyncWorkerTest { + + private val userManager: UserManager = mock() + private val repository: OfflineConversationsRepository = mock() + + @Test + fun `every account is synced, not just the current one`() { + val worker = worker() + wheneverBlocking { userManager.getUsers() }.thenReturn(listOf(user(1), user(2), user(3))) + wheneverBlocking { repository.syncRooms(any(), any(), anyOrNull()) }.thenReturn(NO_ROOMS) + + val result = runBlocking { worker.sync() } + + assertEquals(ListenableWorker.Result.success(), result) + verifyBlocking(repository) { syncRooms(user(1), false, ROOM_LIST_TIMEOUT_MILLIS) } + verifyBlocking(repository) { syncRooms(user(2), false, ROOM_LIST_TIMEOUT_MILLIS) } + verifyBlocking(repository) { syncRooms(user(3), false, ROOM_LIST_TIMEOUT_MILLIS) } + } + + @Test + fun `battery saver stands the sync down entirely`() { + val worker = worker() + shadowOf(applicationContext().getSystemService(Context.POWER_SERVICE) as PowerManager) + .setIsPowerSaveMode(true) + + val result = runBlocking { worker.sync() } + + assertEquals(ListenableWorker.Result.success(), result) + verifyBlocking(userManager, never()) { getUsers() } + verifyBlocking(repository, never()) { syncRooms(any(), any(), anyOrNull()) } + } + + @Test + fun `an app in the foreground stands the sync down entirely`() { + val worker = worker() + moveAppTo(Lifecycle.State.STARTED) + + val result = runBlocking { worker.sync() } + + assertEquals(ListenableWorker.Result.success(), result) + verifyBlocking(userManager, never()) { getUsers() } + verifyBlocking(repository, never()) { syncRooms(any(), any(), anyOrNull()) } + } + + @Test + fun `an account with nothing to sync is not an error`() { + val worker = worker() + wheneverBlocking { userManager.getUsers() }.thenReturn(emptyList()) + + val result = runBlocking { worker.sync() } + + assertEquals(ListenableWorker.Result.success(), result) + verifyBlocking(repository, never()) { syncRooms(any(), any(), anyOrNull()) } + } + + @Test + fun `a failed account asks for another attempt`() { + val worker = worker(runAttempt = 0) + wheneverBlocking { userManager.getUsers() }.thenReturn(listOf(user(1), user(2))) + wheneverBlocking { repository.syncRooms(user(1), false, ROOM_LIST_TIMEOUT_MILLIS) }.thenReturn(NO_ROOMS) + wheneverBlocking { repository.syncRooms(user(2), false, ROOM_LIST_TIMEOUT_MILLIS) }.thenReturn(null) + + val result = runBlocking { worker.sync() } + + assertEquals(ListenableWorker.Result.retry(), result) + verifyBlocking(repository) { syncRooms(user(1), false, ROOM_LIST_TIMEOUT_MILLIS) } + verifyBlocking(repository) { catchUpRooms(user(1), NO_ROOMS) } + verifyBlocking(repository, never()) { catchUpRooms(eq(user(2)), any()) } + } + + @Test + fun `every room list is synced before any messages are caught up`() { + val worker = worker() + wheneverBlocking { userManager.getUsers() }.thenReturn(listOf(user(1), user(2))) + wheneverBlocking { repository.syncRooms(any(), any(), anyOrNull()) }.thenReturn(NO_ROOMS) + + runBlocking { worker.sync() } + + val order = inOrder(repository) + runBlocking { + order.verify(repository).syncRooms(eq(user(1)), any(), anyOrNull()) + order.verify(repository).syncRooms(eq(user(2)), any(), anyOrNull()) + order.verify(repository).catchUpRooms(user(1), NO_ROOMS) + order.verify(repository).catchUpRooms(user(2), NO_ROOMS) + } + } + + @Test + fun `a failed message catch-up does not fail the run`() { + val worker = worker() + wheneverBlocking { userManager.getUsers() }.thenReturn(listOf(user(1), user(2))) + wheneverBlocking { repository.syncRooms(any(), any(), anyOrNull()) }.thenReturn(NO_ROOMS) + wheneverBlocking { repository.catchUpRooms(eq(user(1)), any()) } + .thenThrow(IllegalStateException("database is gone")) + + val result = runBlocking { worker.sync() } + + assertEquals(ListenableWorker.Result.success(), result) + verifyBlocking(repository) { catchUpRooms(user(2), NO_ROOMS) } + } + + @Test + fun `a sync that keeps failing gives up instead of retrying for ever`() { + val worker = worker(runAttempt = 2) + wheneverBlocking { userManager.getUsers() }.thenReturn(listOf(user(1))) + wheneverBlocking { repository.syncRooms(any(), any(), anyOrNull()) }.thenReturn(null) + + val result = runBlocking { worker.sync() } + + assertEquals(ListenableWorker.Result.failure(), result) + } + + @Test + fun `the account that waited longest is synced first`() { + val worker = worker() + wheneverBlocking { userManager.getUsers() }.thenReturn(listOf(user(1), user(2), user(3))) + wheneverBlocking { repository.lastFullSyncAt(1) }.thenReturn(300) + wheneverBlocking { repository.lastFullSyncAt(2) }.thenReturn(null) + wheneverBlocking { repository.lastFullSyncAt(3) }.thenReturn(100) + wheneverBlocking { repository.syncRooms(any(), any(), anyOrNull()) }.thenReturn(NO_ROOMS) + + runBlocking { worker.sync() } + + val synced = argumentCaptor() + verifyBlocking(repository, times(3)) { syncRooms(synced.capture(), any(), anyOrNull()) } + assertEquals(listOf(2L, 3L, 1L), synced.allValues.map { it.id }) + verifyBlocking(repository, times(1)) { lastFullSyncAt(1) } + verifyBlocking(repository, times(1)) { lastFullSyncAt(2) } + verifyBlocking(repository, times(1)) { lastFullSyncAt(3) } + } + + @Test + fun `a failed read of the sync times keeps the account order and still syncs every account`() { + val worker = worker() + wheneverBlocking { userManager.getUsers() }.thenReturn(listOf(user(1), user(2), user(3))) + wheneverBlocking { repository.lastFullSyncAt(any()) }.thenThrow(IllegalStateException("database is gone")) + wheneverBlocking { repository.syncRooms(any(), any(), anyOrNull()) }.thenReturn(NO_ROOMS) + + val result = runBlocking { worker.sync() } + + assertEquals(ListenableWorker.Result.success(), result) + val synced = argumentCaptor() + verifyBlocking(repository, times(3)) { syncRooms(synced.capture(), any(), anyOrNull()) } + assertEquals(listOf(1L, 2L, 3L), synced.allValues.map { it.id }) + } + + @Test + fun `the app coming to the foreground stops the room lists but not the messages already due`() { + val worker = worker() + wheneverBlocking { userManager.getUsers() }.thenReturn(listOf(user(1), user(2))) + wheneverBlocking { repository.syncRooms(eq(user(1)), any(), anyOrNull()) }.thenAnswer { + moveAppTo(Lifecycle.State.STARTED) + NO_ROOMS + } + + val result = runBlocking { worker.sync() } + + assertEquals(ListenableWorker.Result.success(), result) + verifyBlocking(repository, never()) { syncRooms(eq(user(2)), any(), anyOrNull()) } + verifyBlocking(repository) { catchUpRooms(user(1), NO_ROOMS) } + } + + @Test + fun `a failed account lookup is retried rather than reported as a sync`() { + val worker = worker() + wheneverBlocking { userManager.getUsers() }.thenThrow(IllegalStateException("database is gone")) + + val result = runBlocking { worker.sync() } + + assertEquals(ListenableWorker.Result.retry(), result) + verifyBlocking(repository, never()) { syncRooms(any(), any(), anyOrNull()) } + } + + @Test + fun `a last attempt whose account lookup fails reports failure`() { + val worker = worker(runAttempt = 2) + wheneverBlocking { userManager.getUsers() }.thenThrow(IllegalStateException("database is gone")) + + val result = runBlocking { worker.sync() } + + assertEquals(ListenableWorker.Result.failure(), result) + } + + @Test + fun `a cancelled run stops instead of syncing the remaining accounts`() { + val worker = worker() + wheneverBlocking { userManager.getUsers() }.thenReturn(listOf(user(1), user(2), user(3))) + wheneverBlocking { repository.syncRooms(any(), any(), anyOrNull()) } + .thenThrow(CancellationException("run stopped")) + + assertThrows(CancellationException::class.java) { runBlocking { worker.sync() } } + + verifyBlocking(repository, times(1)) { syncRooms(any(), any(), anyOrNull()) } + } + + @Test + fun `scheduling again updates the scheduled run instead of keeping or replacing it`() { + WorkManagerTestInitHelper.initializeTestWorkManager(applicationContext()) + val workManager = WorkManager.getInstance(applicationContext()) + + ConversationsSyncWorker.schedule(applicationContext()) + val first = workManager.getWorkInfosForUniqueWork(ConversationsSyncWorker.UNIQUE_WORK_NAME).get().single() + ConversationsSyncWorker.schedule(applicationContext()) + val second = workManager.getWorkInfosForUniqueWork(ConversationsSyncWorker.UNIQUE_WORK_NAME).get().single() + + assertEquals(first.id, second.id) + assertEquals(first.generation + 1, second.generation) + assertEquals(WorkInfo.State.ENQUEUED, second.state) + } + + @After + fun tearDown() { + moveAppTo(Lifecycle.State.CREATED) + } + + /** Moves the process lifecycle, which outlives a single test, to [state]. */ + private fun moveAppTo(state: Lifecycle.State) { + (ProcessLifecycleOwner.get().lifecycle as LifecycleRegistry).currentState = state + } + + private fun worker(runAttempt: Int = 0): ConversationsSyncWorker = + TestListenableWorkerBuilder(applicationContext()) + .setRunAttemptCount(runAttempt) + .build() + .also { + it.userManager = userManager + it.conversationsRepository = repository + } + + private fun applicationContext(): Context = ApplicationProvider.getApplicationContext() + + private fun user(id: Long): User = User(id = id, userId = "user$id", username = "user$id", baseUrl = BASE_URL) + + companion object { + private const val BASE_URL = "https://server.example.com" + private val ROOM_LIST_TIMEOUT_MILLIS = TimeUnit.MINUTES.toMillis(2) + private val NO_ROOMS = emptyList() + } +}