Ask the server only for the conversations that changed

app/src/main/java/com/nextcloud/talk/api/NcApi.java

@@ -95,9 +95,10 @@ Observable<ResponseBody> getContactsWithSearchParam(@Header("Authorization") Str
         Server URL is: baseUrl + ocsApiVersion + spreedApiVersion + /room
      */
     @GET
-    Observable<RoomsOverall> getRooms(@Header("Authorization") String authorization,
-                                      @Url String url,
-                                      @Nullable @Query("includeStatus") Boolean includeStatus);
+    Observable<Response<RoomsOverall>> getRooms(@Header("Authorization") String authorization,
+                                                @Url String url,
+                                                @Nullable @Query("includeStatus") Boolean includeStatus,
+                                                @Nullable @Query("modifiedSince") Long modifiedSince);
 
     /*
         Server URL is: baseUrl + ocsApiVersion + spreedApiVersion + /room/roomToken

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

@@ -297,7 +297,7 @@ class ConversationsListActivity : BaseActivity() {
                 isMaintenanceModeState.value = false
                 isRefreshingState.value = true
                 appPreferences.setConversationListPositionAndOffset(0, 0)
-                fetchRooms()
+                fetchRooms(forceFullSync = true)
                 fetchPendingInvitations()
             },
             onFabClick = {
@@ -714,8 +714,8 @@ class ConversationsListActivity : BaseActivity() {
         lifecycleScope.launch { snackbarHostState.showSnackbar(text) }
     }
 
-    fun fetchRooms() {
-        conversationsListViewModel.getRooms(currentUser)
+    fun fetchRooms(forceFullSync: Boolean = false) {
+        conversationsListViewModel.getRooms(currentUser, forceFullSync)
     }
 
     private fun fetchPendingInvitations() {
@@ -1324,8 +1324,7 @@ class ConversationsListActivity : BaseActivity() {
                             resources.getString(R.string.nc_shortcut_conversation_deleted)
                         )
                     }
-                    conversationsListViewModel.clearConversationPendingLeave(token)
-                    fetchRooms()
+                    conversationsListViewModel.onConversationLeft(currentUser, token)
                 }
                 WorkInfo.State.FAILED -> {
                     logger.e(TAG, "LeaveConversationWorker failed for token $token")
@@ -1584,6 +1583,7 @@ class ConversationsListActivity : BaseActivity() {
                                     conversation.displayName
                                 )
                             )
+                            fetchRooms(forceFullSync = true)
                         }
 
                         WorkInfo.State.FAILED -> {

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

@@ -41,9 +41,14 @@ 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.
      */
     @Deprecated("use observeConversation")
-    fun getRooms(user: User): Job
+    fun getRooms(user: User, forceFullSync: Boolean = false): Job
 
     /**
      * Called once onStart to emit a conversation to [conversationFlow]
@@ -61,5 +66,15 @@ interface OfflineConversationsRepository {
     @Deprecated("use observeConversation")
     suspend fun getLocallyStoredConversation(user: User, roomToken: String): ConversationModel?
 
+    /**
+     * Makes the next conversation list sync of the account with the internal id [accountId] fetch
+     * the whole list.
+     *
+     * A conversation this device left or deleted is absent from the server's answer rather than
+     * marked as gone, which only a full response can be read as a removal. Without this the row
+     * survives locally until the next full sync falls due.
+     */
+    fun requireFullSync(accountId: Long)
+
     fun observeConversation(accountId: Long, roomToken: String): Flow<ConversationResult>
 }

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

@@ -11,6 +11,29 @@ import com.nextcloud.talk.data.user.model.User
 import com.nextcloud.talk.models.json.conversations.ConversationDto
 import io.reactivex.Observable
 
+/**
+ * The outcome of a conversation list request.
+ *
+ * [conversations] are the rooms the server returned, [modifiedBefore] the timestamp it reported in
+ * `X-Nextcloud-Talk-Modified-Before`, or null when it sent none. [wasDelta] is true when the
+ * request asked for only the conversations changed since a timestamp. A response carries no marker
+ * saying whether the server honoured that, so a filtered request reports a delta either way and
+ * removals stay unreadable from it until a full request is made.
+ */
+data class RoomListResult(val conversations: List<ConversationDto>, val modifiedBefore: Long?, val wasDelta: Boolean)
+
 interface ConversationsNetworkDataSource {
-    fun getRooms(user: User, url: String, includeStatus: Boolean): Observable<List<ConversationDto>>
+    /**
+     * Loads [user]'s conversation list.
+     *
+     * With [includeStatus] the response carries the user status of one-to-one rooms. With
+     * [modifiedSince] set, and on a server that supports it, the response covers only the
+     * conversations changed since that timestamp.
+     */
+    fun getRooms(
+        user: User,
+        url: String,
+        includeStatus: Boolean,
+        modifiedSince: Long? = null
+    ): Observable<RoomListResult>
 }

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

@@ -13,6 +13,7 @@ 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
 import com.nextcloud.talk.chat.data.network.ChatNetworkDataSource
 import com.nextcloud.talk.conversationlist.data.OfflineConversationsRepository
@@ -30,6 +31,7 @@ import com.nextcloud.talk.utils.SpreedFeatures
 import com.nextcloud.talk.utils.withRetry
 import io.reactivex.android.schedulers.AndroidSchedulers
 import io.reactivex.schedulers.Schedulers
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
@@ -45,13 +47,15 @@ import kotlinx.coroutines.sync.withPermit
 import javax.inject.Inject
 import kotlin.collections.map
 
+@Suppress("LongParameterList")
 class OfflineFirstConversationsRepository @Inject constructor(
     private val dao: ConversationsDao,
     private val network: ConversationsNetworkDataSource,
     private val chatNetworkDataSource: ChatNetworkDataSource,
     private val networkMonitor: NetworkMonitor,
     private val chatMessageSyncer: ChatMessageSyncer,
     private val conversationListUpdater: ConversationListUpdater,
+    private val arbitraryStorageManager: ArbitraryStorageManager,
     private val context: Context,
     private val logger: Logger
 ) : OfflineConversationsRepository {
@@ -94,10 +98,10 @@ class OfflineFirstConversationsRepository @Inject constructor(
                 }
             }
 
-    override fun getRooms(user: User): Job =
+    override fun getRooms(user: User, forceFullSync: Boolean): Job =
         scope.launch {
             if (networkMonitor.isOnline.value) {
-                getRoomsFromServer(user)
+                getRoomsFromServer(user, forceFullSync = forceFullSync)
             }
         }
 
@@ -148,56 +152,124 @@ class OfflineFirstConversationsRepository @Inject constructor(
     }
 
     @Suppress("Detekt.TooGenericExceptionCaught")
-    private suspend fun getRoomsFromServer(user: User): List<ConversationEntity>? {
+    private suspend fun getRoomsFromServer(user: User, forceFullSync: Boolean = false): List<ConversationEntity>? {
         var conversationsFromSync: List<ConversationEntity>? = null
 
         if (!networkMonitor.isOnline.value) {
             Log.d(TAG, "Device is offline, can't load conversations from server")
             return null
         }
 
-        val includeStatus = isUserStatusAvailable(user)
+        val accountId = user.id!!
+        val modifiedSince = modifiedSinceFor(user, forceFullSync)
+
+        val includeStatus = modifiedSince == null && isUserStatusAvailable(user)
 
         try {
-            val conversationsList = withRetry(
+            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)
+                network.getRooms(user, user.baseUrl!!, includeStatus, modifiedSince)
                     .subscribeOn(Schedulers.io())
                     .observeOn(AndroidSchedulers.mainThread())
                     .blockingSingle()
             }
 
-            conversationsFromSync = conversationsList.map {
-                it.asEntity(user.id!!)
+            conversationsFromSync = roomList.conversations.map {
+                it.asEntity(accountId)
             }
 
-            val previousConversations = dao.getConversationsForUser(user.id!!).first()
+            val previousConversations = dao.getConversationsForUser(accountId).first()
                 .associateBy { it.internalId }
 
+            val serverItems = if (includeStatus) {
+                conversationsFromSync
+            } else {
+                keepCachedStatus(conversationsFromSync, previousConversations)
+            }
+
             dao.syncConversationsForUser(
-                accountId = user.id!!,
+                accountId = accountId,
                 serverItems = conversationListUpdater.preservePendingLocalState(
                     previousConversations,
-                    conversationsFromSync
+                    serverItems
                 ),
-                conversationIdsToDelete = determineLeftConversationIds(previousConversations, conversationsFromSync)
+                conversationIdsToDelete = if (roomList.wasDelta) {
+                    emptyList()
+                } else {
+                    determineLeftConversationIds(previousConversations, conversationsFromSync)
+                }
             )
 
+            rememberSyncedState(accountId, roomList)
+
             val roomsWithNewMessages = getRoomsWithNewMessages(conversationsFromSync, previousConversations)
             scope.launch { catchUpRoomsWithNewMessages(user, roomsWithNewMessages) }
+        } catch (e: CancellationException) {
+            throw e
         } catch (e: Exception) {
             Log.e(TAG, "Something went wrong when fetching conversations", e)
-            val hasCachedConversations = dao.getConversationsForUser(user.id!!).first().isNotEmpty()
+            storeTimestamp(accountId, KEY_MODIFIED_SINCE, null)
+            val hasCachedConversations = dao.getConversationsForUser(accountId).first().isNotEmpty()
             if (!hasCachedConversations) {
                 _syncErrorFlow.emit(e)
             }
         }
         return conversationsFromSync
     }
 
+    override fun requireFullSync(accountId: Long) {
+        storeTimestamp(accountId, KEY_LAST_FULL_SYNC_AT, null)
+    }
+
+    /**
+     * The value to send as `modifiedSince`, or null when this sync has to be a full one.
+     *
+     * 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.
+     */
+    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) {
+            readTimestamp(accountId, KEY_MODIFIED_SINCE)
+        } else {
+            null
+        }
+    }
+
+    /**
+     * 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.
+     */
+    private fun rememberSyncedState(accountId: Long, roomList: RoomListResult) {
+        storeTimestamp(accountId, KEY_MODIFIED_SINCE, roomList.modifiedBefore)
+        if (!roomList.wasDelta) {
+            storeTimestamp(accountId, KEY_LAST_FULL_SYNC_AT, System.currentTimeMillis())
+        }
+    }
+
+    /** Null for anything that is not a plausible timestamp, so a bad value asks for a full sync. */
+    private fun readTimestamp(accountId: Long, key: String): Long? =
+        arbitraryStorageManager.getStorageSetting(accountId, key, "")
+            .blockingGet()
+            ?.value
+            ?.toLongOrNull()
+            ?.takeIf { it > 0 }
+
+    private fun storeTimestamp(accountId: Long, key: String, value: Long?) {
+        arbitraryStorageManager.storeStorageSetting(accountId, key, value?.toString(), "")
+    }
+
     /**
      * 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
@@ -258,7 +330,10 @@ class OfflineFirstConversationsRepository @Inject constructor(
                                 unreadMessages = room.unreadMessages
                             )
                         }
-                            .onFailure { Log.e(TAG, "Message catch-up failed for room ${room.token}", it) }
+                            .onFailure {
+                                if (it is CancellationException) throw it
+                                Log.e(TAG, "Message catch-up failed for room ${room.token}", it)
+                            }
                     }
                 }
             }
@@ -296,6 +371,25 @@ class OfflineFirstConversationsRepository @Inject constructor(
             connectivityManager.restrictBackgroundStatus == ConnectivityManager.RESTRICT_BACKGROUND_STATUS_ENABLED
     }
 
+    /**
+     * Returns [conversations] with the stored user status of each, for a response that did not
+     * carry one. The upsert replaces the row it writes, so a response fetched without
+     * `includeStatus` would otherwise clear the status of every conversation it covers.
+     */
+    private fun keepCachedStatus(
+        conversations: List<ConversationEntity>,
+        previousConversations: Map<String, ConversationEntity>
+    ): List<ConversationEntity> =
+        conversations.map { item ->
+            val previous = previousConversations[item.internalId] ?: return@map item
+            item.copy(
+                status = previous.status,
+                statusClearAt = previous.statusClearAt,
+                statusIcon = previous.statusIcon,
+                statusMessage = previous.statusMessage
+            )
+        }
+
     private fun determineLeftConversationIds(
         previousConversations: Map<String, ConversationEntity>,
         conversationsFromSync: List<ConversationEntity>
@@ -331,5 +425,8 @@ class OfflineFirstConversationsRepository @Inject constructor(
         private const val NETWORK_FETCH_RETRIES = 3
         private const val NETWORK_FETCH_RETRY_INITIAL_DELAY_MS = 1000L
         private const val NETWORK_FETCH_RETRY_MAX_DELAY_MS = 8000L
+        private const val FULL_SYNC_INTERVAL_MILLIS = 5 * 60 * 1000L
+        private const val KEY_MODIFIED_SINCE = "conversation_list_modified_since"
+        private const val KEY_LAST_FULL_SYNC_AT = "conversation_list_last_full_sync_at"
     }
 }

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

@@ -8,21 +8,40 @@ package com.nextcloud.talk.conversationlist.data.network
 
 import com.nextcloud.talk.api.NcApi
 import com.nextcloud.talk.data.user.model.User
-import com.nextcloud.talk.models.json.conversations.ConversationDto
 import com.nextcloud.talk.utils.ApiUtils
 import io.reactivex.Observable
+import retrofit2.HttpException
 
 class RetrofitConversationsNetwork(private val ncApi: NcApi) : ConversationsNetworkDataSource {
-    override fun getRooms(user: User, url: String, includeStatus: Boolean): Observable<List<ConversationDto>> {
+    override fun getRooms(
+        user: User,
+        url: String,
+        includeStatus: Boolean,
+        modifiedSince: Long?
+    ): Observable<RoomListResult> {
         val credentials: String = ApiUtils.getCredentials(user.username, user.token)!!
         val apiVersion = ApiUtils.getConversationApiVersion(user, intArrayOf(ApiUtils.API_V4, ApiUtils.API_V3, 1))
 
+        val delta = modifiedSince?.takeIf { apiVersion >= ApiUtils.API_V4 && it > 0 }
+
         return ncApi.getRooms(
             credentials,
             ApiUtils.getUrlForRooms(apiVersion, user.baseUrl!!),
-            includeStatus
-        ).map { it ->
-            it.ocs?.data?.map { it } ?: listOf()
+            includeStatus,
+            delta
+        ).map { response ->
+            if (!response.isSuccessful) throw HttpException(response)
+
+            RoomListResult(
+                conversations = response.body()?.ocs?.data ?: emptyList(),
+                modifiedBefore = response.headers()[HEADER_MODIFIED_BEFORE]?.toLongOrNull()?.takeIf { it > 0 },
+                wasDelta = delta != null
+            )
         }
     }
+
+    companion object {
+        /** Taken by the server before it queries, to be sent as `modifiedSince` on the next request. */
+        private const val HEADER_MODIFIED_BEFORE = "X-Nextcloud-Talk-Modified-Before"
+    }
 }