Don't delete conversations that arrive while a list sync is in flight

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

@@ -250,6 +250,10 @@ class OfflineFirstConversationsRepository @Inject constructor(
         val accountId = user.id!!
         val includeStatus = modifiedSince == null && isUserStatusAvailable(user)
 
+        // taken before the request: a conversation that is written locally while it is in flight is
+        // not something the response can be read as having left out
+        val knownBeforeRequest = dao.getConversationIdsForUser(accountId).toSet()
+
         val roomList = withRetry(
             retries = NETWORK_FETCH_RETRIES,
             initialDelayMillis = NETWORK_FETCH_RETRY_INITIAL_DELAY_MS,
@@ -285,6 +289,7 @@ class OfflineFirstConversationsRepository @Inject constructor(
                 emptyList()
             } else {
                 determineLeftConversationIds(previousConversations, conversationsFromSync)
+                    .filter { it in knownBeforeRequest }
             }
         )
 

app/src/main/java/com/nextcloud/talk/data/database/dao/ConversationsDao.kt

@@ -22,6 +22,10 @@ interface ConversationsDao {
     @Query("SELECT * FROM Conversations where accountId = :accountId")
     fun getConversationsForUser(accountId: Long): Flow<List<ConversationEntity>>
 
+    /** The internal ids of [accountId]'s conversations, without loading the conversations themselves. */
+    @Query("SELECT internalId FROM Conversations where accountId = :accountId")
+    suspend fun getConversationIdsForUser(accountId: Long): List<String>
+
     @Query("SELECT * FROM Conversations where accountId = :accountId AND token = :token")
     fun getConversationForUser(accountId: Long, token: String): Flow<ConversationEntity?>
 

app/src/main/java/com/nextcloud/talk/utils/preview/ComposePreviewUtilsDaos.kt

@@ -290,6 +290,8 @@ class DummyUserDaoImpl : UsersDao {
 class DummyConversationDaoImpl : ConversationsDao {
     override fun getConversationsForUser(accountId: Long): Flow<List<ConversationEntity>> = flowOf()
 
+    override suspend fun getConversationIdsForUser(accountId: Long): List<String> = emptyList()
+
     override fun getConversationForUser(accountId: Long, token: String): Flow<ConversationEntity?> = flowOf()
 
     override suspend fun upsertConversations(accountId: Long, serverItems: List<ConversationEntity>) {

app/src/test/java/com/nextcloud/talk/conversationlist/data/network/OfflineFirstConversationsRepositoryTest.kt

@@ -106,7 +106,9 @@ class OfflineFirstConversationsRepositoryTest {
         whenever(connectivityManager.restrictBackgroundStatus)
             .thenReturn(ConnectivityManager.RESTRICT_BACKGROUND_STATUS_DISABLED)
 
-        whenever(dao.getConversationsForUser(ACCOUNT_ID)).thenReturn(flowOf(emptyList()))
+        // every sync reads the stored ids before its request, for whichever account it syncs
+        wheneverBlocking { dao.getConversationIdsForUser(any()) }.thenReturn(emptyList())
+        stubStoredConversations()
         whenever(arbitraryStorageManager.getStorageSetting(any(), any(), any()))
             .thenReturn(Maybe.empty())
         whenever(conversationListUpdater.preservePendingLocalState(any(), any()))
@@ -271,7 +273,7 @@ class OfflineFirstConversationsRepositoryTest {
     fun `getRooms skips deleting local conversations when the server returns an empty list`() =
         runBlocking {
             val previous = conversation(token = ROOM_TOKEN, lastActivity = 5, unreadMessages = 0).asEntity(ACCOUNT_ID)
-            whenever(dao.getConversationsForUser(ACCOUNT_ID)).thenReturn(flowOf(listOf(previous)))
+            stubStoredConversations(previous)
             whenever(network.getRooms(any(), any(), any(), anyOrNull())).thenReturn(roomList(listOf()))
 
             repository.getRooms(user()).join()
@@ -286,7 +288,7 @@ class OfflineFirstConversationsRepositoryTest {
         runBlocking {
             val staying = conversation(token = "roomA", lastActivity = 5, unreadMessages = 0).asEntity(ACCOUNT_ID)
             val leaving = conversation(token = "roomB", lastActivity = 5, unreadMessages = 0).asEntity(ACCOUNT_ID)
-            whenever(dao.getConversationsForUser(ACCOUNT_ID)).thenReturn(flowOf(listOf(staying, leaving)))
+            stubStoredConversations(staying, leaving)
             val stayingRoom = conversation(token = "roomA", lastActivity = 5, unreadMessages = 0)
             whenever(network.getRooms(any(), any(), any(), anyOrNull())).thenReturn(roomList(listOf(stayingRoom)))
 
@@ -302,8 +304,7 @@ class OfflineFirstConversationsRepositoryTest {
         runBlocking {
             val staying = conversation(token = "roomA", lastActivity = 5, unreadMessages = 0)
             val leaving = conversation(token = "roomB", lastActivity = 5, unreadMessages = 0)
-            whenever(dao.getConversationsForUser(ACCOUNT_ID))
-                .thenReturn(flowOf(listOf(staying.asEntity(ACCOUNT_ID), leaving.asEntity(ACCOUNT_ID))))
+            stubStoredConversations(staying.asEntity(ACCOUNT_ID), leaving.asEntity(ACCOUNT_ID))
             val deltaResponseHeld = CountDownLatch(1)
             val deltaResponse = Observable.fromCallable {
                 deltaResponseHeld.await()
@@ -365,7 +366,7 @@ class OfflineFirstConversationsRepositoryTest {
     fun `getRooms merges the server response through the conversation list updater`() =
         runBlocking {
             val previous = conversation(token = ROOM_TOKEN, lastActivity = 5, unreadMessages = 0).asEntity(ACCOUNT_ID)
-            whenever(dao.getConversationsForUser(ACCOUNT_ID)).thenReturn(flowOf(listOf(previous)))
+            stubStoredConversations(previous)
             val serverRoom = conversation(token = ROOM_TOKEN, lastActivity = 6, unreadMessages = 1)
             whenever(network.getRooms(any(), any(), any(), anyOrNull())).thenReturn(roomList(listOf(serverRoom)))
 
@@ -445,8 +446,7 @@ class OfflineFirstConversationsRepositoryTest {
                 conversation(token = "unchanged", lastActivity = 10, unreadMessages = 0).asEntity(ACCOUNT_ID)
             val previousNoBlock =
                 conversation(token = "noBlockUnread", lastActivity = 10, unreadMessages = 3).asEntity(ACCOUNT_ID)
-            whenever(dao.getConversationsForUser(ACCOUNT_ID))
-                .thenReturn(flowOf(listOf(previousUnchanged, previousNoBlock)))
+            stubStoredConversations(previousUnchanged, previousNoBlock)
 
             val unchangedRoom = conversation(token = "unchanged", lastActivity = 10, unreadMessages = 0)
             val noBlockRoom = conversation(token = "noBlockUnread", lastActivity = 10, unreadMessages = 3)
@@ -629,6 +629,32 @@ class OfflineFirstConversationsRepositoryTest {
         }
     }
 
+    @Test
+    fun `a conversation that arrives while the request is in flight is not treated as left`() =
+        runBlocking {
+            val known = conversation(token = "known", lastActivity = 5, unreadMessages = 0).asEntity(ACCOUNT_ID)
+            val arrivedDuringRequest =
+                conversation(token = "arrived", lastActivity = 5, unreadMessages = 0).asEntity(ACCOUNT_ID)
+            wheneverBlocking { dao.getConversationIdsForUser(ACCOUNT_ID) }.thenReturn(listOf(known.internalId))
+            whenever(dao.getConversationsForUser(ACCOUNT_ID))
+                .thenReturn(flowOf(listOf(known, arrivedDuringRequest)))
+            val stillOnServer = conversation(token = "server", lastActivity = 5, unreadMessages = 0)
+            whenever(network.getRooms(any(), any(), any(), anyOrNull())).thenReturn(roomList(listOf(stillOnServer)))
+
+            repository.getRooms(user()).join()
+
+            verifyBlocking(dao) { syncConversationsForUser(eq(ACCOUNT_ID), any(), eq(listOf(known.internalId))) }
+        }
+
+    /**
+     * Stubs both reads the sync makes of the conversations table: the rows it merges against, and
+     * the ids it takes before the request to know which conversations the response can speak about.
+     */
+    private fun stubStoredConversations(vararg stored: ConversationEntity) {
+        whenever(dao.getConversationsForUser(ACCOUNT_ID)).thenReturn(flowOf(stored.toList()))
+        wheneverBlocking { dao.getConversationIdsForUser(ACCOUNT_ID) }.thenReturn(stored.map { it.internalId })
+    }
+
     private fun user(withKeepNotificationsCapability: Boolean = true): User {
         val features = if (withKeepNotificationsCapability) {
             listOf(SpreedFeatures.CHAT_KEEP_NOTIFICATIONS.value)