feat: background asset upload (WPB-20906)

cells/src/commonMain/kotlin/com/wire/kalium/cells/data/MessageAttachmentDraftDataSource.kt

@@ -22,6 +22,7 @@ import com.wire.kalium.cells.domain.model.AttachmentDraft
 import com.wire.kalium.cells.domain.model.AttachmentUploadStatus
 import com.wire.kalium.cells.domain.model.CellNode
 import com.wire.kalium.common.error.wrapStorageRequest
+import com.wire.kalium.common.functional.mapRight
 import com.wire.kalium.logic.data.id.ConversationId
 import com.wire.kalium.logic.data.id.QualifiedID
 import com.wire.kalium.logic.data.message.AssetContent
@@ -36,6 +37,7 @@ import com.wire.kalium.util.KaliumDispatcherImpl
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.withContext
+import kotlin.collections.map
 
 internal class MessageAttachmentDraftDataSource internal constructor(
     private val messageAttachmentDao: MessageAttachmentDraftDao,
@@ -88,6 +90,12 @@ internal class MessageAttachmentDraftDataSource internal constructor(
         messageAttachmentDao.deleteAttachments(QualifiedIDEntity(conversationId.value, conversationId.domain))
     }
 
+    override suspend fun observeAllDrafts() = withContext(dispatchers.io) {
+        messageAttachmentDao.observeAttachments()
+            .wrapStorageRequest()
+            .mapRight { list -> list.map { it.toModel() } }
+    }
+
     override suspend fun get(uuid: String) = withContext(dispatchers.io) {
         wrapStorageRequest {
             messageAttachmentDao.getAttachment(uuid)?.toModel()

cells/src/commonMain/kotlin/com/wire/kalium/cells/domain/MessageAttachmentDraftRepository.kt

@@ -21,6 +21,7 @@ import com.wire.kalium.cells.domain.model.AttachmentDraft
 import com.wire.kalium.cells.domain.model.AttachmentUploadStatus
 import com.wire.kalium.cells.domain.model.CellNode
 import com.wire.kalium.common.error.CoreFailure
+import com.wire.kalium.common.error.StorageFailure
 import com.wire.kalium.common.functional.Either
 import com.wire.kalium.logic.data.id.ConversationId
 import com.wire.kalium.logic.data.id.QualifiedID
@@ -45,4 +46,5 @@ public interface MessageAttachmentDraftRepository {
     public suspend fun updateStatus(uuid: String, status: AttachmentUploadStatus): Either<CoreFailure, Unit>
     public suspend fun remove(uuid: String): Either<CoreFailure, Unit>
     public suspend fun removeAttachmentDrafts(conversationId: ConversationId)
+    public suspend fun observeAllDrafts(): Flow<Either<StorageFailure, List<AttachmentDraft>>>
 }

logic/src/commonMain/kotlin/com/wire/kalium/logic/data/asset/AssetRepository.kt

@@ -46,6 +46,7 @@ import okio.IOException
 import okio.Path
 import okio.Path.Companion.toPath
 import okio.Sink
+import okio.use
 
 @Mockable
 interface AssetRepository {
@@ -265,11 +266,10 @@ internal class AssetDataSource(
     ): Either<CoreFailure, UploadedAssetId> =
         assetMapper.toMetadataApiModel(uploadAssetData, kaliumFileSystem).let { metaData ->
             wrapApiRequest {
-                val dataSource = kaliumFileSystem.source(uploadAssetData.tempEncryptedDataPath)
-
-                // we should also consider for avatar images, the compression for preview vs complete picture
-                assetApi.uploadAsset(metaData, { dataSource }, uploadAssetData.dataSize)
-                    .also { dataSource.close() }
+                kaliumFileSystem.source(uploadAssetData.tempEncryptedDataPath).use { dataSource ->
+                    // we should also consider for avatar images, the compression for preview vs complete picture
+                    assetApi.uploadAsset(metaData, { dataSource }, uploadAssetData.dataSize)
+                }
             }
         }
             .flatMap { assetResponse ->

logic/src/commonMain/kotlin/com/wire/kalium/logic/data/message/MessageRepository.kt

@@ -276,6 +276,8 @@ internal interface MessageRepository {
         newMessageId: String,
         editInstant: Instant
     ): Either<StorageFailure, Unit>
+
+    suspend fun observeAssetStatuses(): Flow<Either<StorageFailure, List<AssetTransferStatus>>>
 }
 
 // TODO: suppress TooManyFunctions for now, something we need to fix in the future
@@ -727,6 +729,10 @@ internal class MessageDataSource internal constructor(
         .wrapStorageRequest()
         .mapRight { assetStatusEntities -> assetStatusEntities.map { it.toModel() } }
 
+    override suspend fun observeAssetStatuses() = messageDAO.observeAssetStatuses()
+        .wrapStorageRequest()
+        .mapRight { assetStatusEntities -> assetStatusEntities.map { it.transfer_status.toModel() } }
+
     override suspend fun getMessageAssetTransferStatus(
         messageId: String,
         conversationId: ConversationId

logic/src/commonMain/kotlin/com/wire/kalium/logic/feature/asset/ObserveAssetUploadStateUseCase.kt

@@ -0,0 +1,72 @@
+/*
+ * Wire
+ * Copyright (C) 2024 Wire Swiss GmbH
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see http://www.gnu.org/licenses/.
+ */
+
+package com.wire.kalium.logic.feature.asset
+
+import com.wire.kalium.cells.domain.MessageAttachmentDraftRepository
+import com.wire.kalium.cells.domain.model.AttachmentUploadStatus
+import com.wire.kalium.common.functional.fold
+import com.wire.kalium.logic.data.asset.AssetTransferStatus
+import com.wire.kalium.logic.data.message.MessageRepository
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.combine
+import kotlinx.coroutines.flow.distinctUntilChanged
+import kotlinx.coroutines.flow.map
+
+/**
+ * Use case observing statuses of assets when uploading and downloading.
+ */
+interface ObserveAssetUploadStateUseCase {
+    suspend operator fun invoke(): Flow<Boolean>
+}
+
+internal class ObserveAssetUploadStateUseCaseImpl(
+    private val messageRepository: MessageRepository,
+    private val attachmentDraftRepository: MessageAttachmentDraftRepository,
+) : ObserveAssetUploadStateUseCase {
+    override suspend fun invoke(): Flow<Boolean> {
+        return combine(
+            observeRegularAssetsUpload(),
+            observeAttachmentDraftsUpload()
+        ) { regularAssetsUploading, draftAttachmentsUploading ->
+            regularAssetsUploading || draftAttachmentsUploading
+        }.distinctUntilChanged()
+    }
+
+    private suspend fun observeRegularAssetsUpload() =
+        messageRepository.observeAssetStatuses()
+            .map { result ->
+                result.fold(
+                    { false },
+                    { statuses ->
+                        statuses.any { status -> status == AssetTransferStatus.UPLOAD_IN_PROGRESS }
+                    }
+                )
+            }
+
+    private suspend fun observeAttachmentDraftsUpload() =
+        attachmentDraftRepository.observeAllDrafts()
+            .map { result ->
+                result.fold(
+                    { false },
+                    { drafts ->
+                        drafts.any { draft -> draft.uploadStatus == AttachmentUploadStatus.UPLOADING }
+                    }
+                )
+            }
+}