Refresh the conversation list while it is open

app/src/main/java/com/nextcloud/talk/conversationlist/ConversationsListActivity.kt

@@ -36,8 +36,6 @@ import androidx.lifecycle.lifecycleScope
 import androidx.lifecycle.repeatOnLifecycle
 import androidx.work.Data
 import androidx.work.OneTimeWorkRequest
-import com.nextcloud.talk.dagger.modules.assistedViewModels
-import com.nextcloud.talk.utils.setExpeditedIfSupported
 import androidx.work.WorkInfo
 import androidx.work.WorkManager
 import autodagger.AutoInjector
@@ -62,6 +60,7 @@ import com.nextcloud.talk.conversationlist.ui.ConversationsListScreenCallbacks
 import com.nextcloud.talk.conversationlist.ui.ConversationsListScreenState
 import com.nextcloud.talk.conversationlist.viewmodels.ConversationsListViewModel
 import com.nextcloud.talk.conversationtags.viewmodels.ConversationTagsViewModel
+import com.nextcloud.talk.dagger.modules.assistedViewModels
 import com.nextcloud.talk.data.network.NetworkMonitor
 import com.nextcloud.talk.data.user.model.User
 import com.nextcloud.talk.events.ConversationsListFetchDataEvent
@@ -115,17 +114,20 @@ import com.nextcloud.talk.utils.bundle.BundleKeys.KEY_SCROLL_TO_NOTIFICATION_CAT
 import com.nextcloud.talk.utils.bundle.BundleKeys.KEY_SHARED_TEXT
 import com.nextcloud.talk.utils.permissions.PlatformPermissionUtil
 import com.nextcloud.talk.utils.power.PowerManagerUtils
+import com.nextcloud.talk.utils.setExpeditedIfSupported
 import com.nextcloud.talk.utils.singletons.ApplicationWideCurrentRoomHolder
+import java.util.concurrent.TimeUnit
+import javax.inject.Inject
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.isActive
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
 import org.greenrobot.eventbus.Subscribe
 import org.greenrobot.eventbus.ThreadMode
 import retrofit2.HttpException
-import java.util.concurrent.TimeUnit
-import javax.inject.Inject
 
 @SuppressLint("StringFormatInvalid")
 @AutoInjector(NextcloudTalkApplication::class)
@@ -215,6 +217,8 @@ class ConversationsListActivity : BaseActivity() {
             lifecycleScope.launch { userManager.setUserAsActive(currentUser) }
         }
 
+        startForegroundRefreshLoop()
+
         setSupportActionBar(null)
         forwardMessageState.value = intent.getBooleanExtra(KEY_FORWARD_MSG_FLAG, false)
         if (savedInstanceState != null) {
@@ -714,6 +718,22 @@ class ConversationsListActivity : BaseActivity() {
         lifecycleScope.launch { snackbarHostState.showSnackbar(text) }
     }
 
+    /**
+     * Starts a loop that refreshes the conversation list every
+     * [FOREGROUND_REFRESH_INTERVAL_MILLIS] milliseconds while this screen is resumed, waiting one
+     * interval before the first refresh.
+     */
+    private fun startForegroundRefreshLoop() {
+        lifecycleScope.launch {
+            repeatOnLifecycle(Lifecycle.State.RESUMED) {
+                while (isActive) {
+                    delay(FOREGROUND_REFRESH_INTERVAL_MILLIS)
+                    currentUser?.let { conversationsListViewModel.refreshRoomsIfIdle(it) }
+                }
+            }
+        }
+    }
+
     fun fetchRooms(forceFullSync: Boolean = false) {
         conversationsListViewModel.getRooms(currentUser, forceFullSync)
     }
@@ -1630,6 +1650,7 @@ class ConversationsListActivity : BaseActivity() {
                 addFlags(Intent.FLAG_ACTIVITY_CLEAR_TOP)
             }
 
+        private const val FOREGROUND_REFRESH_INTERVAL_MILLIS = 30_000L
         const val BOTTOM_SHEET_DELAY: Long = 2500
         const val SEARCH_DEBOUNCE_INTERVAL_MS = 300
         const val HTTP_UNAUTHORIZED = 401

app/src/main/java/com/nextcloud/talk/conversationlist/data/OfflineConversationsRepository.kt

@@ -43,10 +43,8 @@ interface OfflineConversationsRepository {
      * Synchronizes the conversations of [user] with the server (when online). The synced changes
      * surface through [roomListFlow], which observes the database.
      *
-     * The sync asks the server only for what changed since the last one where it can. Set
-     * [forceFullSync] for the cases where that is not good enough and the whole list is wanted
-     * back - a pull to refresh, where a conversation the user left elsewhere should be gone by the
-     * time the indicator stops spinning, rather than within the next five minutes.
+     * The sync asks the server only for the conversations changed since the last one where it
+     * can. [forceFullSync] makes it fetch the whole list instead.
      */
     @Deprecated("use observeConversation")
     fun getRooms(user: User, forceFullSync: Boolean = false): Job
@@ -86,6 +84,15 @@ interface OfflineConversationsRepository {
      */
     suspend fun catchUpRooms(user: User, rooms: List<ConversationEntity>)
 
+    /**
+     * Whether a sync for [user] right now would either ask only for the conversations that changed,
+     * or be the full refresh that falls due every five minutes.
+     *
+     * False means the next sync would fetch the whole conversation list without the cadence calling
+     * for it.
+     */
+    suspend fun isPeriodicSyncDue(user: User): Boolean
+
     /**
      * Called once onStart to emit a conversation to [conversationFlow]
      * to be handled asynchronously.

app/src/main/java/com/nextcloud/talk/conversationlist/data/network/OfflineFirstConversationsRepository.kt

@@ -11,7 +11,6 @@ package com.nextcloud.talk.conversationlist.data.network
 import android.content.Context
 import android.database.sqlite.SQLiteConstraintException
 import android.net.ConnectivityManager
-import android.os.PowerManager
 import android.util.Log
 import com.nextcloud.talk.arbitrarystorage.ArbitraryStorageManager
 import com.nextcloud.talk.chat.data.network.ChatMessageSyncer
@@ -23,6 +22,7 @@ import com.nextcloud.talk.data.database.mappers.toDomainModel
 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.extensions.isPowerSaveMode
 import com.nextcloud.talk.logger.Logger
 import com.nextcloud.talk.models.domain.ConversationModel
 import com.nextcloud.talk.utils.ApiUtils
@@ -43,10 +43,13 @@ import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.Semaphore
+import kotlinx.coroutines.sync.withLock
 import kotlinx.coroutines.sync.withPermit
 import kotlinx.coroutines.withContext
 import kotlinx.coroutines.withTimeout
+import java.util.concurrent.ConcurrentHashMap
 import javax.inject.Inject
 import kotlin.collections.map
 import kotlin.time.Duration.Companion.milliseconds
@@ -84,6 +87,9 @@ class OfflineFirstConversationsRepository @Inject constructor(
 
     private val scope = CoroutineScope(Dispatchers.IO)
 
+    /** One lock per account, held for the whole of a room list sync. See [syncRoomListExclusively]. */
+    private val roomListSyncLocks = ConcurrentHashMap<Long, Mutex>()
+
     sealed interface ConversationResult {
         data class Found(val conversation: ConversationModel) : ConversationResult
         object NotFound : ConversationResult
@@ -192,13 +198,13 @@ class OfflineFirstConversationsRepository @Inject constructor(
         reportSyncError: Boolean
     ): List<ConversationEntity>? {
         val accountId = user.id!!
-        val modifiedSince = modifiedSinceFor(user, forceFullSync)
 
         return try {
             if (roomListTimeoutMillis == null) {
-                syncRoomList(user, modifiedSince)
+                syncRoomListExclusively(user, forceFullSync)
             } else {
-                withTimeout(roomListTimeoutMillis.milliseconds) { syncRoomList(user, modifiedSince) }
+                // waiting for another sync of the account counts against the budget too
+                withTimeout(roomListTimeoutMillis.milliseconds) { syncRoomListExclusively(user, forceFullSync) }
             }
         } catch (e: TimeoutCancellationException) {
             // the caller's run goes on, so this is a failed sync rather than a cancellation
@@ -216,6 +222,22 @@ class OfflineFirstConversationsRepository @Inject constructor(
         }
     }
 
+    /**
+     * Runs [syncRoomList] for [user] once no other sync of the same account is running.
+     *
+     * Each sync reads the stored conversations, merges its response into them and writes the
+     * result, and a full sync deletes what its response no longer has. Syncs of one account are
+     * started from several places (screen, pull to refresh, refresh tick, background worker), so
+     * they are run one after the other: interleaved, a delta response written after a full sync
+     * would put back a conversation the full sync had just deleted, and an older response would
+     * overwrite newer state. The waiting sync also picks its `modifiedSince` from what the previous
+     * one stored.
+     */
+    private suspend fun syncRoomListExclusively(user: User, forceFullSync: Boolean): List<ConversationEntity> =
+        roomListSyncLocks.getOrPut(user.id!!) { Mutex() }.withLock {
+            syncRoomList(user, modifiedSinceFor(user, forceFullSync))
+        }
+
     /**
      * Fetches [user]'s room list, filtered by [modifiedSince] when set, and stores it, returning the
      * rooms whose messages should be caught up.
@@ -274,31 +296,38 @@ class OfflineFirstConversationsRepository @Inject constructor(
         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.
+     * The stored timestamp to send as `modifiedSince`, or null when the sync has to fetch the whole
+     * list.
      *
-     * A filtered response cannot express a removal, so the server asks clients to refresh in full
-     * regularly (`docs/conversation.md`): at least every five minutes, and always when the internal
-     * signaling backend is in use, since there is no signaling server to announce a change out of
-     * band. On top of that a caller can demand one, for a pull to refresh or an account switch.
+     * Null is returned when [forceFullSync] is set, when the account uses the internal signaling
+     * backend, when the last full sync is older than [FULL_SYNC_INTERVAL_MILLIS], and when no
+     * timestamp is stored.
      */
     private fun modifiedSinceFor(user: User, forceFullSync: Boolean): Long? {
         val accountId = user.id!!
-        val lastFullSyncAt = readTimestamp(accountId, KEY_LAST_FULL_SYNC_AT)
-        val fullSyncIsRecent = lastFullSyncAt != null &&
-            System.currentTimeMillis() - lastFullSyncAt in 0 until FULL_SYNC_INTERVAL_MILLIS
         val usesExternalSignaling = !user.externalSignalingServer?.externalSignalingServer.isNullOrEmpty()
 
-        return if (!forceFullSync && usesExternalSignaling && fullSyncIsRecent) {
+        return if (!forceFullSync && usesExternalSignaling && fullSyncIsRecent(accountId)) {
             readTimestamp(accountId, KEY_MODIFIED_SINCE)
         } else {
             null
         }
     }
 
+    private fun fullSyncIsRecent(accountId: Long): Boolean {
+        val lastFullSyncAt = readTimestamp(accountId, KEY_LAST_FULL_SYNC_AT) ?: return false
+        return System.currentTimeMillis() - lastFullSyncAt in 0 until FULL_SYNC_INTERVAL_MILLIS
+    }
+
+    override suspend fun isPeriodicSyncDue(user: User): Boolean =
+        withContext(Dispatchers.IO) {
+            modifiedSinceFor(user, forceFullSync = false) != null || !fullSyncIsRecent(user.id!!)
+        }
+
     /**
-     * Stores what the next sync needs, and only once the response is safely in the database: a
-     * timestamp kept ahead of a write that then failed would permanently skip the conversations
-     * that write was carrying.
+     * Stores the timestamp [roomList] reported for the next request, and, when it was a full
+     * response, the time of this full sync. Called after the response has been written to the
+     * database.
      */
     private fun rememberSyncedState(accountId: Long, roomList: RoomListResult) {
         storeTimestamp(accountId, KEY_MODIFIED_SINCE, roomList.modifiedBefore)
@@ -307,7 +336,7 @@ class OfflineFirstConversationsRepository @Inject constructor(
         }
     }
 
-    /** Null for anything that is not a plausible timestamp, so a bad value asks for a full sync. */
+    /** The timestamp stored under [key] for [accountId], or null when none is stored or it is not a positive number. */
     private fun readTimestamp(accountId: Long, key: String): Long? =
         arbitraryStorageManager.getStorageSetting(accountId, key, "")
             .blockingGet()
@@ -396,7 +425,7 @@ class OfflineFirstConversationsRepository @Inject constructor(
                 false
             }
 
-            isPowerSaveMode() -> {
+            context.isPowerSaveMode() -> {
                 Log.d(TAG, "Battery saver is active, skipping message catch-up")
                 false
             }
@@ -409,11 +438,6 @@ class OfflineFirstConversationsRepository @Inject constructor(
             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 &&

app/src/main/java/com/nextcloud/talk/conversationlist/viewmodels/ConversationsListViewModel.kt

@@ -51,6 +51,7 @@ import io.reactivex.android.schedulers.AndroidSchedulers
 import io.reactivex.disposables.Disposable
 import io.reactivex.schedulers.Schedulers
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableStateFlow
@@ -160,8 +161,17 @@ class ConversationsListViewModel @AssistedInject constructor(
     private val _isLoadingRooms = MutableStateFlow(true)
 
     /**
-     * True while a [getRooms] call is in progress (from start until the fetch-Job completes,
-     * i.e. after both the local-DB and the network-sync emissions have been produced).
+     * Number of [getRooms] jobs that have not completed yet. Guarded by [roomSyncsLock], together
+     * with the matching update of [_isLoadingRooms]: [getRooms] is also called off the main thread,
+     * from a background event subscriber.
+     */
+    private var roomSyncsInFlight = 0
+    private val roomSyncsLock = Any()
+
+    /**
+     * True while any [getRooms] call is in progress (from start until the last overlapping
+     * fetch-Job completes, i.e. after both the local-DB and the network-sync emissions have been
+     * produced).
      */
     val isLoadingRooms: StateFlow<Boolean> = _isLoadingRooms.asStateFlow()
 
@@ -562,14 +572,64 @@ class ConversationsListViewModel @AssistedInject constructor(
         }
     }
 
+    /**
+     * Refreshes the conversation list, unless a room list sync is already running, a search is
+     * open, or the sync would fetch the whole list without the refresh cadence calling for it.
+     *
+     * The in-flight check covers the room list request only: [getRooms] completes once the list is
+     * stored and leaves the message catch-up running, so a tick can start while messages of the
+     * previous one are still being fetched. That is deliberate - the list is what this refreshes.
+     *
+     * Whether a sync is due is read from stored state, and a read that fails must not take the
+     * refresh loop with it: this runs every tick for as long as the screen is open, and an
+     * exception would escape the loop's scope rather than skip one refresh. That read suspends, so
+     * it is done before the other checks: a sync or a search started while it runs is still seen.
+     */
+    suspend fun refreshRoomsIfIdle(user: User) {
+        val syncDue = isPeriodicSyncDue(user)
+        when {
+            _isLoadingRooms.value ->
+                Log.d(TAG, "Foreground refresh skipped: a room list sync is already in flight")
+
+            _isSearchActiveFlow.value ->
+                Log.d(TAG, "Foreground refresh skipped: a search is active")
+
+            !syncDue ->
+                Log.d(TAG, "Foreground refresh skipped: no delta available and no full sync due")
+
+            else -> getRooms(user)
+        }
+    }
+
+    /** Whether the cadence calls for a sync, treating a failed read of the stored state as yes. */
+    private suspend fun isPeriodicSyncDue(user: User): Boolean =
+        runCatching { repository.isPeriodicSyncDue(user) }.getOrElse { throwable ->
+            if (throwable is CancellationException) throw throwable
+            Log.w(TAG, "Could not read when the last sync was, refreshing anyway", throwable)
+            true
+        }
+
     fun getRooms(user: User, forceFullSync: Boolean = false): Job {
         val startNanoTime = System.nanoTime()
         Log.d(TAG, "fetchData - getRooms - calling: $startNanoTime")
-        _isLoadingRooms.value = true
         val job = repository.getRooms(user, forceFullSync)
+        // overlapping syncs (resume, pull to refresh, a refresh tick) must not end the loading
+        // state while another one is still running, or pull to refresh stops spinning too early
+        synchronized(roomSyncsLock) {
+            roomSyncsInFlight++
+            _isLoadingRooms.value = true
+        }
         viewModelScope.launch {
-            job.join()
-            _isLoadingRooms.value = false
+            try {
+                job.join()
+            } finally {
+                synchronized(roomSyncsLock) {
+                    roomSyncsInFlight--
+                    if (roomSyncsInFlight == 0) {
+                        _isLoadingRooms.value = false
+                    }
+                }
+            }
         }
         return job
     }