BIT-1547: Hook up remaining push notification sync handling (#848)

app/src/main/java/com/x8bit/bitwarden/data/auth/repository/AuthRepositoryImpl.kt

@@ -54,6 +54,7 @@ import com.x8bit.bitwarden.data.auth.repository.util.userOrganizationsList
 import com.x8bit.bitwarden.data.auth.repository.util.userOrganizationsListFlow
 import com.x8bit.bitwarden.data.auth.util.KdfParamsConstants.DEFAULT_PBKDF2_ITERATIONS
 import com.x8bit.bitwarden.data.auth.util.toSdkParams
+import com.x8bit.bitwarden.data.platform.manager.PushManager
 import com.x8bit.bitwarden.data.platform.manager.dispatcher.DispatcherManager
 import com.x8bit.bitwarden.data.platform.repository.EnvironmentRepository
 import com.x8bit.bitwarden.data.platform.repository.SettingsRepository
@@ -71,7 +72,9 @@ import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.asSharedFlow
 import kotlinx.coroutines.flow.combine
 import kotlinx.coroutines.flow.filter
+import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.flow.stateIn
 import javax.inject.Singleton
 
@@ -95,6 +98,7 @@ class AuthRepositoryImpl(
     private val settingsRepository: SettingsRepository,
     private val vaultRepository: VaultRepository,
     private val userLogoutManager: UserLogoutManager,
+    private val pushManager: PushManager,
     dispatcherManager: DispatcherManager,
     private val elapsedRealtimeMillisProvider: () -> Long = { SystemClock.elapsedRealtime() },
 ) : AuthRepository {
@@ -117,7 +121,9 @@ class AuthRepositoryImpl(
      * use of [Dispatchers.Unconfined] allows for this to happen synchronously whenever any of
      * these flows changes.
      */
-    private val collectionScope = CoroutineScope(dispatcherManager.unconfined)
+    private val unconfinedScope = CoroutineScope(dispatcherManager.unconfined)
+
+    private val ioScope = CoroutineScope(dispatcherManager.io)
 
     override var twoFactorResponse: TwoFactorRequired? = null
 
@@ -136,7 +142,7 @@ class AuthRepositoryImpl(
                 ?: AuthState.Unauthenticated
         }
         .stateIn(
-            scope = collectionScope,
+            scope = unconfinedScope,
             started = SharingStarted.Eagerly,
             initialValue = AuthState.Uninitialized,
         )
@@ -169,7 +175,7 @@ class AuthRepositoryImpl(
             !mutableHasPendingAccountDeletionStateFlow.value
         }
         .stateIn(
-            scope = collectionScope,
+            scope = unconfinedScope,
             started = SharingStarted.Eagerly,
             initialValue = authDiskSource
                 .userState
@@ -199,6 +205,22 @@ class AuthRepositoryImpl(
     override var hasPendingAccountAddition: Boolean
         by mutableHasPendingAccountAdditionStateFlow::value
 
+    init {
+        pushManager
+            .syncOrgKeysFlow
+            .onEach {
+                val userId = activeUserId ?: return@onEach
+                refreshAccessTokenSynchronously(userId)
+                vaultRepository.sync()
+            }
+            .launchIn(ioScope)
+
+        pushManager
+            .logoutFlow
+            .onEach { logout() }
+            .launchIn(unconfinedScope)
+    }
+
     override fun clearPendingAccountDeletion() {
         mutableHasPendingAccountDeletionStateFlow.value = false
     }

app/src/main/java/com/x8bit/bitwarden/data/auth/repository/di/AuthRepositoryModule.kt

@@ -12,6 +12,7 @@ import com.x8bit.bitwarden.data.auth.datasource.sdk.AuthSdkSource
 import com.x8bit.bitwarden.data.auth.manager.UserLogoutManager
 import com.x8bit.bitwarden.data.auth.repository.AuthRepository
 import com.x8bit.bitwarden.data.auth.repository.AuthRepositoryImpl
+import com.x8bit.bitwarden.data.platform.manager.PushManager
 import com.x8bit.bitwarden.data.platform.manager.dispatcher.DispatcherManager
 import com.x8bit.bitwarden.data.platform.repository.EnvironmentRepository
 import com.x8bit.bitwarden.data.platform.repository.SettingsRepository
@@ -49,6 +50,7 @@ object AuthRepositoryModule {
         settingsRepository: SettingsRepository,
         vaultRepository: VaultRepository,
         userLogoutManager: UserLogoutManager,
+        pushManager: PushManager,
     ): AuthRepository = AuthRepositoryImpl(
         accountsService = accountsService,
         authRequestsService = authRequestsService,
@@ -65,5 +67,6 @@ object AuthRepositoryModule {
         settingsRepository = settingsRepository,
         vaultRepository = vaultRepository,
         userLogoutManager = userLogoutManager,
+        pushManager = pushManager,
     )
 }

app/src/main/java/com/x8bit/bitwarden/data/platform/manager/PushManagerImpl.kt

@@ -149,6 +149,8 @@ class PushManagerImpl @Inject constructor(
                     SyncCipherUpsertData(
                         cipherId = payload.id,
                         revisionDate = payload.revisionDate,
+                        organizationId = payload.organizationId,
+                        collectionIds = payload.collectionIds,
                         isUpdate = type == NotificationType.SYNC_CIPHER_UPDATE,
                     ),
                 )

app/src/main/java/com/x8bit/bitwarden/data/platform/manager/model/NotificationPayload.kt

@@ -23,7 +23,7 @@ sealed class NotificationPayload {
         @SerialName("id") val id: String,
         @SerialName("userId") override val userId: String?,
         @SerialName("organizationId") val organizationId: String?,
-        @SerialName("collectionIds") val collectionIds: List<String>,
+        @SerialName("collectionIds") val collectionIds: List<String>?,
         @Contextual
         @SerialName("revisionDate") val revisionDate: ZonedDateTime,
     ) : NotificationPayload()

app/src/main/java/com/x8bit/bitwarden/data/platform/manager/model/SyncCipherUpsertData.kt

@@ -13,5 +13,7 @@ import java.time.ZonedDateTime
 data class SyncCipherUpsertData(
     val cipherId: String,
     val revisionDate: ZonedDateTime,
+    val organizationId: String?,
+    val collectionIds: List<String>?,
     val isUpdate: Boolean,
 )

app/src/main/java/com/x8bit/bitwarden/data/vault/repository/VaultRepositoryImpl.kt

@@ -19,8 +19,11 @@ import com.x8bit.bitwarden.data.platform.datasource.disk.SettingsDiskSource
 import com.x8bit.bitwarden.data.platform.datasource.network.util.isNoConnectionError
 import com.x8bit.bitwarden.data.platform.manager.PushManager
 import com.x8bit.bitwarden.data.platform.manager.dispatcher.DispatcherManager
+import com.x8bit.bitwarden.data.platform.manager.model.SyncCipherDeleteData
 import com.x8bit.bitwarden.data.platform.manager.model.SyncCipherUpsertData
+import com.x8bit.bitwarden.data.platform.manager.model.SyncFolderDeleteData
 import com.x8bit.bitwarden.data.platform.manager.model.SyncFolderUpsertData
+import com.x8bit.bitwarden.data.platform.manager.model.SyncSendDeleteData
 import com.x8bit.bitwarden.data.platform.manager.model.SyncSendUpsertData
 import com.x8bit.bitwarden.data.platform.repository.model.DataState
 import com.x8bit.bitwarden.data.platform.repository.util.bufferedMutableSharedFlow
@@ -91,11 +94,13 @@ import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.asSharedFlow
 import kotlinx.coroutines.flow.asStateFlow
 import kotlinx.coroutines.flow.combine
+import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.firstOrNull
 import kotlinx.coroutines.flow.flatMapLatest
 import kotlinx.coroutines.flow.flowOf
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.mapNotNull
 import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.flow.onStart
 import kotlinx.coroutines.flow.stateIn
@@ -236,16 +241,36 @@ class VaultRepositoryImpl(
             }
             .launchIn(unconfinedScope)
 
+        pushManager
+            .fullSyncFlow
+            .onEach { syncIfNecessary() }
+            .launchIn(unconfinedScope)
+
+        pushManager
+            .syncCipherDeleteFlow
+            .onEach(::deleteCipher)
+            .launchIn(unconfinedScope)
+
         pushManager
             .syncCipherUpsertFlow
             .onEach(::syncCipherIfNecessary)
             .launchIn(ioScope)
 
+        pushManager
+            .syncSendDeleteFlow
+            .onEach(::deleteSend)
+            .launchIn(unconfinedScope)
+
         pushManager
             .syncSendUpsertFlow
             .onEach(::syncSendIfNecessary)
             .launchIn(ioScope)
 
+        pushManager
+            .syncFolderDeleteFlow
+            .onEach(::deleteFolder)
+            .launchIn(unconfinedScope)
+
         pushManager
             .syncFolderUpsertFlow
             .onEach(::syncFolderIfNecessary)
@@ -1231,19 +1256,76 @@ class VaultRepositoryImpl(
             .onEach { mutableSendDataStateFlow.value = it }
 
     //region Push notification helpers
+    /**
+     * Deletes the cipher specified by [syncCipherDeleteData] from disk.
+     */
+    private suspend fun deleteCipher(syncCipherDeleteData: SyncCipherDeleteData) {
+        val userId = activeUserId ?: return
+
+        val cipherId = syncCipherDeleteData.cipherId
+        vaultDiskSource.deleteCipher(
+            userId = userId,
+            cipherId = cipherId,
+        )
+    }
+
     /**
      * Syncs an individual cipher contained in [syncCipherUpsertData] to disk if certain criteria
      * are met. If the resource cannot be found cloud-side, and it was updated, delete it from disk
      * for now.
      */
+    @Suppress("ReturnCount")
     private suspend fun syncCipherIfNecessary(syncCipherUpsertData: SyncCipherUpsertData) {
         val userId = activeUserId ?: return
-
-        // TODO Handle other filtering logic including revision date comparison. This will still be
-        //  handled as part of BIT-1547.
-
         val cipherId = syncCipherUpsertData.cipherId
+        val organizationId = syncCipherUpsertData.organizationId
+        val collectionIds = syncCipherUpsertData.collectionIds
+        val revisionDate = syncCipherUpsertData.revisionDate
         val isUpdate = syncCipherUpsertData.isUpdate
+
+        val localCipher = ciphersStateFlow
+            .mapNotNull { it.data }
+            .first()
+            .find { it.id == cipherId }
+
+        // Return if local cipher is more recent
+        if (localCipher != null &&
+            localCipher.revisionDate.epochSecond > revisionDate.toEpochSecond()
+        ) {
+            return
+        }
+
+        var shouldUpdate: Boolean
+        val shouldCheckCollections: Boolean
+
+        when {
+            isUpdate -> {
+                shouldUpdate = localCipher != null
+                shouldCheckCollections = true
+            }
+
+            collectionIds == null || organizationId == null -> {
+                shouldUpdate = localCipher == null
+                shouldCheckCollections = false
+            }
+
+            else -> {
+                shouldUpdate = false
+                shouldCheckCollections = true
+            }
+        }
+
+        if (!shouldUpdate && shouldCheckCollections && organizationId != null) {
+            // Check if there are any collections in common
+            shouldUpdate = collectionsStateFlow
+                .mapNotNull { it.data }
+                .first()
+                .mapNotNull { it.id }
+                .any { collectionIds?.contains(it) == true } == true
+        }
+
+        if (!shouldUpdate) return
+
         ciphersService
             .getCipher(cipherId)
             .fold(
@@ -1259,6 +1341,19 @@ class VaultRepositoryImpl(
             )
     }
 
+    /**
+     * Deletes the send specified by [syncSendDeleteData] from disk.
+     */
+    private suspend fun deleteSend(syncSendDeleteData: SyncSendDeleteData) {
+        val userId = activeUserId ?: return
+
+        val sendId = syncSendDeleteData.sendId
+        vaultDiskSource.deleteSend(
+            userId = userId,
+            sendId = sendId,
+        )
+    }
+
     /**
      * Syncs an individual send contained in [syncSendUpsertData] to disk if certain criteria are
      * met. If the resource cannot be found cloud-side, and it was updated, delete it from disk for
@@ -1266,12 +1361,22 @@ class VaultRepositoryImpl(
      */
     private suspend fun syncSendIfNecessary(syncSendUpsertData: SyncSendUpsertData) {
         val userId = activeUserId ?: return
-
-        // TODO Handle other filtering logic including revision date comparison. This will still be
-        //  handled as part of BIT-1547.
-
         val sendId = syncSendUpsertData.sendId
         val isUpdate = syncSendUpsertData.isUpdate
+        val revisionDate = syncSendUpsertData.revisionDate
+
+        val localSend = sendDataStateFlow
+            .mapNotNull { it.data }
+            .first()
+            .sendViewList
+            .find { it.id == sendId }
+        val isValidCreate = !isUpdate && localSend == null
+        val isValidUpdate = isUpdate &&
+            localSend != null &&
+            localSend.revisionDate.epochSecond < revisionDate.toEpochSecond()
+
+        if (!isValidCreate && !isValidUpdate) return
+
         sendsService
             .getSend(sendId)
             .fold(
@@ -1287,17 +1392,44 @@ class VaultRepositoryImpl(
             )
     }
 
+    /**
+     * Deletes the folder specified by [syncFolderDeleteData] from disk.
+     */
+    private suspend fun deleteFolder(syncFolderDeleteData: SyncFolderDeleteData) {
+        val userId = activeUserId ?: return
+
+        val folderId = syncFolderDeleteData.folderId
+        clearFolderIdFromCiphers(
+            folderId = folderId,
+            userId = userId,
+        )
+        vaultDiskSource.deleteFolder(
+            folderId = folderId,
+            userId = userId,
+        )
+    }
+
     /**
      * Syncs an individual folder contained in [syncFolderUpsertData] to disk if certain criteria
      * are met.
      */
     private suspend fun syncFolderIfNecessary(syncFolderUpsertData: SyncFolderUpsertData) {
         val userId = activeUserId ?: return
-
-        // TODO Handle other filtering logic including revision date comparison. This will still be
-        //  handled as part of BIT-1547.
-
         val folderId = syncFolderUpsertData.folderId
+        val isUpdate = syncFolderUpsertData.isUpdate
+        val revisionDate = syncFolderUpsertData.revisionDate
+
+        val localFolder = foldersStateFlow
+            .mapNotNull { it.data }
+            .first()
+            .find { it.id == folderId }
+        val isValidCreate = !isUpdate && localFolder == null
+        val isValidUpdate = isUpdate &&
+            localFolder != null &&
+            localFolder.revisionDate.epochSecond < revisionDate.toEpochSecond()
+
+        if (!isValidCreate && !isValidUpdate) return
+
         folderService
             .getFolder(folderId)
             .onSuccess { vaultDiskSource.saveFolder(userId, it) }

app/src/test/java/com/x8bit/bitwarden/data/auth/repository/AuthRepositoryTest.kt

@@ -64,10 +64,12 @@ import com.x8bit.bitwarden.data.auth.repository.util.toUserState
 import com.x8bit.bitwarden.data.auth.repository.util.toUserStateJson
 import com.x8bit.bitwarden.data.auth.util.toSdkParams
 import com.x8bit.bitwarden.data.platform.base.FakeDispatcherManager
+import com.x8bit.bitwarden.data.platform.manager.PushManager
 import com.x8bit.bitwarden.data.platform.manager.dispatcher.DispatcherManager
 import com.x8bit.bitwarden.data.platform.repository.SettingsRepository
 import com.x8bit.bitwarden.data.platform.repository.model.Environment
 import com.x8bit.bitwarden.data.platform.repository.util.FakeEnvironmentRepository
+import com.x8bit.bitwarden.data.platform.repository.util.bufferedMutableSharedFlow
 import com.x8bit.bitwarden.data.platform.util.asFailure
 import com.x8bit.bitwarden.data.platform.util.asSuccess
 import com.x8bit.bitwarden.data.vault.datasource.network.model.createMockOrganization
@@ -172,6 +174,13 @@ class AuthRepositoryTest {
         every { logout(any()) } just runs
     }
 
+    private val mutableLogoutFlow = bufferedMutableSharedFlow<Unit>()
+    private val mutableSyncOrgKeysFlow = bufferedMutableSharedFlow<Unit>()
+    private val pushManager: PushManager = mockk {
+        every { logoutFlow } returns mutableLogoutFlow
+        every { syncOrgKeysFlow } returns mutableSyncOrgKeysFlow
+    }
+
     private var elapsedRealtimeMillis = 123456789L
 
     private val repository = AuthRepositoryImpl(
@@ -190,6 +199,7 @@ class AuthRepositoryTest {
         vaultRepository = vaultRepository,
         userLogoutManager = userLogoutManager,
         dispatcherManager = dispatcherManager,
+        pushManager = pushManager,
         elapsedRealtimeMillisProvider = { elapsedRealtimeMillis },
     )
 
@@ -2770,6 +2780,43 @@ class AuthRepositoryTest {
         )
     }
 
+    @Suppress("MaxLineLength")
+    @Test
+    fun `logOutFlow emission for action account should call logout on the UserLogoutManager and clear the user's in memory vault data`() {
+        val userId = USER_ID_1
+        fakeAuthDiskSource.userState = MULTI_USER_STATE
+
+        mutableLogoutFlow.tryEmit(Unit)
+
+        coVerify(exactly = 1) {
+            userLogoutManager.logout(userId = userId)
+            vaultRepository.clearUnlockedData()
+        }
+    }
+
+    @Test
+    fun `syncOrgKeysFlow emissions should refresh access token and sync`() {
+        fakeAuthDiskSource.userState = SINGLE_USER_STATE_1
+        coEvery {
+            identityService.refreshTokenSynchronously(REFRESH_TOKEN)
+        } returns REFRESH_TOKEN_RESPONSE_JSON.asSuccess()
+        every {
+            REFRESH_TOKEN_RESPONSE_JSON.toUserStateJson(
+                userId = USER_ID_1,
+                previousUserState = SINGLE_USER_STATE_1,
+            )
+        } returns SINGLE_USER_STATE_1
+
+        coEvery { vaultRepository.sync() } just runs
+
+        mutableSyncOrgKeysFlow.tryEmit(Unit)
+
+        coVerify(exactly = 1) {
+            identityService.refreshTokenSynchronously(REFRESH_TOKEN)
+            vaultRepository.sync()
+        }
+    }
+
     companion object {
         private const val UNIQUE_APP_ID = "testUniqueAppId"
         private const val EMAIL = "test@bitwarden.com"

app/src/test/java/com/x8bit/bitwarden/data/platform/manager/PushManagerTest.kt

@@ -184,6 +184,8 @@ class PushManagerTest {
                         assertEquals(
                             SyncCipherUpsertData(
                                 cipherId = "aab5cdcc-f4a7-4e65-bf6d-5e0eab052321",
+                                organizationId = "6a41d965-ed95-4eae-98c3-5f1ec609c2c1",
+                                collectionIds = listOf(),
                                 revisionDate = ZonedDateTime.parse("2023-10-27T12:00:00.000Z"),
                                 isUpdate = false,
                             ),
@@ -214,6 +216,8 @@ class PushManagerTest {
                         assertEquals(
                             SyncCipherUpsertData(
                                 cipherId = "aab5cdcc-f4a7-4e65-bf6d-5e0eab052321",
+                                organizationId = "6a41d965-ed95-4eae-98c3-5f1ec609c2c1",
+                                collectionIds = listOf(),
                                 revisionDate = ZonedDateTime.parse("2023-10-27T12:00:00.000Z"),
                                 isUpdate = true,
                             ),