Pre-process media during the attachment preview

features/messages/impl/src/main/kotlin/io/element/android/features/messages/impl/attachments/preview/AttachmentsPreviewPresenter.kt

@@ -8,6 +8,7 @@
 package io.element.android.features.messages.impl.attachments.preview
 
 import androidx.compose.runtime.Composable
+import androidx.compose.runtime.LaunchedEffect
 import androidx.compose.runtime.MutableState
 import androidx.compose.runtime.getValue
 import androidx.compose.runtime.mutableStateOf
@@ -19,26 +20,39 @@ import dagger.assisted.AssistedFactory
 import dagger.assisted.AssistedInject
 import io.element.android.features.messages.impl.attachments.Attachment
 import io.element.android.libraries.androidutils.file.TemporaryUriDeleter
+import io.element.android.libraries.androidutils.file.safeDelete
+import io.element.android.libraries.architecture.AsyncData
 import io.element.android.libraries.architecture.Presenter
+import io.element.android.libraries.featureflag.api.FeatureFlagService
+import io.element.android.libraries.featureflag.api.FeatureFlags
 import io.element.android.libraries.matrix.api.core.ProgressCallback
 import io.element.android.libraries.matrix.api.permalink.PermalinkBuilder
 import io.element.android.libraries.mediaupload.api.MediaSender
+import io.element.android.libraries.mediaupload.api.MediaUploadInfo
+import io.element.android.libraries.mediaupload.api.allFiles
 import io.element.android.libraries.textcomposer.model.TextEditorState
 import io.element.android.libraries.textcomposer.model.rememberMarkdownTextEditorState
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.Job
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.distinctUntilChanged
+import kotlinx.coroutines.flow.filter
+import kotlinx.coroutines.flow.flatMapConcat
 import kotlinx.coroutines.isActive
 import kotlinx.coroutines.launch
 import timber.log.Timber
 import kotlin.coroutines.coroutineContext
 
+@OptIn(ExperimentalCoroutinesApi::class)
 class AttachmentsPreviewPresenter @AssistedInject constructor(
     @Assisted private val attachment: Attachment,
     @Assisted private val onDoneListener: OnDoneListener,
     private val mediaSender: MediaSender,
     private val permalinkBuilder: PermalinkBuilder,
     private val temporaryUriDeleter: TemporaryUriDeleter,
+    private val featureFlagsService: FeatureFlagService,
 ) : Presenter<AttachmentsPreviewState> {
     @AssistedFactory
     interface Factory {
@@ -63,19 +77,61 @@ class AttachmentsPreviewPresenter @AssistedInject constructor(
 
         val ongoingSendAttachmentJob = remember { mutableStateOf<Job?>(null) }
 
+        val userSentAttachment = remember {
+            MutableStateFlow(false)
+        }
+
+        val mediaUploadInfoStateFlow = remember { MutableStateFlow<AsyncData<MediaUploadInfo>>(AsyncData.Uninitialized) }
+        var prePropressingJob: Job? = null
+        LaunchedEffect(Unit) {
+            prePropressingJob = preProcessAttachment(
+                attachment,
+                mediaUploadInfoStateFlow,
+            )
+        }
+
+        LaunchedEffect(Unit) {
+            userSentAttachment.filter { it }
+                .flatMapConcat {
+                    mediaUploadInfoStateFlow.filter { it.isReady() }
+                }
+                .distinctUntilChanged()
+                .collect { mediaUploadInfo ->
+                    if (mediaUploadInfo is AsyncData.Success) {
+                        val caption = markdownTextEditorState.getMessageMarkdown(permalinkBuilder)
+                            .takeIf { it.isNotEmpty() }
+                        ongoingSendAttachmentJob.value = coroutineScope.launch {
+                            sendPreProcessedMedia(
+                                mediaUploadInfo = mediaUploadInfo.data,
+                                caption = caption,
+                                sendActionState = sendActionState,
+                            )
+                        }
+                    } else if (mediaUploadInfo is AsyncData.Failure) {
+                        sendActionState.value = SendActionState.Failure(mediaUploadInfo.error)
+                    }
+                    // else: cannot happen since we filtered with isReady()
+                }
+        }
+
         fun handleEvents(attachmentsPreviewEvents: AttachmentsPreviewEvents) {
             when (attachmentsPreviewEvents) {
-                is AttachmentsPreviewEvents.SendAttachment -> {
+                is AttachmentsPreviewEvents.SendAttachment -> coroutineScope.launch {
-                    val caption = markdownTextEditorState.getMessageMarkdown(permalinkBuilder)
+                    val useSendQueue = featureFlagsService.isFeatureEnabled(FeatureFlags.MediaUploadOnSendQueue)
-                        .takeIf { it.isNotEmpty() }
+                    userSentAttachment.value = true
-                    ongoingSendAttachmentJob.value = coroutineScope.sendAttachment(
+                    val instantSending = mediaUploadInfoStateFlow.value.isReady() && useSendQueue
-                        attachment = attachment,
+                    sendActionState.value = if (instantSending) {
-                        caption = caption,
+                        SendActionState.Sending.InstantSending
-                        sendActionState = sendActionState,
+                    } else {
-                    )
+                        SendActionState.Sending.Processing
+                    }
                 }
                 AttachmentsPreviewEvents.Cancel -> {
-                    coroutineScope.cancel(attachment)
+                    coroutineScope.cancel(
+                        attachment,
+                        prePropressingJob,
+                        mediaUploadInfoStateFlow.value,
+                    )
                 }
                 AttachmentsPreviewEvents.ClearSendState -> {
                     ongoingSendAttachmentJob.value?.let {
@@ -95,56 +151,91 @@ class AttachmentsPreviewPresenter @AssistedInject constructor(
         )
     }
 
-    private fun CoroutineScope.sendAttachment(
+    private fun CoroutineScope.preProcessAttachment(
         attachment: Attachment,
-        caption: String?,
+        mediaUploadInfoState: MutableStateFlow<AsyncData<MediaUploadInfo>>,
-        sendActionState: MutableState<SendActionState>,
     ) = launch {
         when (attachment) {
             is Attachment.Media -> {
-                sendMedia(
+                preProcessMedia(
                     mediaAttachment = attachment,
-                    caption = caption,
+                    mediaUploadInfoState = mediaUploadInfoState,
-                    sendActionState = sendActionState,
                 )
             }
         }
     }
 
+    private suspend fun preProcessMedia(
+        mediaAttachment: Attachment.Media,
+        mediaUploadInfoState: MutableStateFlow<AsyncData<MediaUploadInfo>>,
+    ) {
+        mediaUploadInfoState.emit(AsyncData.Loading())
+        mediaSender.preProcessMedia(
+            uri = mediaAttachment.localMedia.uri,
+            mimeType = mediaAttachment.localMedia.info.mimeType,
+        ).fold(
+            onSuccess = { mediaUploadInfo ->
+                mediaUploadInfoState.emit(AsyncData.Success(mediaUploadInfo))
+            },
+            onFailure = {
+                Timber.e(it, "Failed to pre-process media")
+                if (it is CancellationException) {
+                    throw it
+                } else {
+                    mediaUploadInfoState.emit(AsyncData.Failure(it))
+                }
+            }
+        )
+    }
+
     private fun CoroutineScope.cancel(
         attachment: Attachment,
+        preProcessingJob: Job?,
+        mediaUploadInfo: AsyncData<MediaUploadInfo>,
     ) = launch {
         // Delete the temporary file
         when (attachment) {
             is Attachment.Media -> {
                 temporaryUriDeleter.delete(attachment.localMedia.uri)
+                preProcessingJob?.cancel()
+                mediaUploadInfo.dataOrNull()?.let { data ->
+                    cleanUp(data)
+                }
             }
         }
         onDoneListener()
     }
 
-    private suspend fun sendMedia(
+    private fun cleanUp(
-        mediaAttachment: Attachment.Media,
+        mediaUploadInfo: MediaUploadInfo,
+    ) {
+        mediaUploadInfo.allFiles().forEach { file ->
+            file.safeDelete()
+        }
+    }
+
+    private suspend fun sendPreProcessedMedia(
+        mediaUploadInfo: MediaUploadInfo,
         caption: String?,
         sendActionState: MutableState<SendActionState>,
     ) = runCatching {
         val context = coroutineContext
         val progressCallback = object : ProgressCallback {
             override fun onProgress(current: Long, total: Long) {
+                // Note will not happen if useSendQueue is true
                 if (context.isActive) {
                     sendActionState.value = SendActionState.Sending.Uploading(current.toFloat() / total.toFloat())
                 }
             }
         }
-        sendActionState.value = SendActionState.Sending.Processing
+        mediaSender.sendPreProcessedMedia(
-        mediaSender.sendMedia(
+            mediaUploadInfo = mediaUploadInfo,
-            uri = mediaAttachment.localMedia.uri,
-            mimeType = mediaAttachment.localMedia.info.mimeType,
             caption = caption,
             progressCallback = progressCallback
         ).getOrThrow()
     }.fold(
         onSuccess = {
+            cleanUp(mediaUploadInfo)
             onDoneListener()
         },
         onFailure = { error ->

features/messages/impl/src/main/kotlin/io/element/android/features/messages/impl/attachments/preview/AttachmentsPreviewState.kt

@@ -27,6 +27,7 @@ sealed interface SendActionState {
 
     @Immutable
     sealed interface Sending : SendActionState {
+        data object InstantSending : Sending
         data object Processing : Sending
         data class Uploading(val progress: Float) : Sending
     }

features/messages/impl/src/main/kotlin/io/element/android/features/messages/impl/attachments/preview/AttachmentsPreviewView.kt

@@ -99,12 +99,17 @@ private fun AttachmentSendStateView(
     onRetryClick: () -> Unit
 ) {
     when (sendActionState) {
-        is SendActionState.Sending -> {
+        is SendActionState.Sending.Processing -> {
             ProgressDialog(
-                type = when (sendActionState) {
+                type = ProgressDialogType.Indeterminate,
-                    is SendActionState.Sending.Uploading -> ProgressDialogType.Determinate(sendActionState.progress)
+                text = stringResource(id = CommonStrings.common_sending),
-                    SendActionState.Sending.Processing -> ProgressDialogType.Indeterminate
+                showCancelButton = true,
-                },
+                onDismissRequest = onDismissClick,
+            )
+        }
+        is SendActionState.Sending.Uploading -> {
+            ProgressDialog(
+                type = ProgressDialogType.Determinate(sendActionState.progress),
                 text = stringResource(id = CommonStrings.common_sending),
                 showCancelButton = true,
                 onDismissRequest = onDismissClick,

features/messages/impl/src/test/kotlin/io/element/android/features/messages/impl/attachments/AttachmentsPreviewPresenterTest.kt

@@ -20,6 +20,8 @@ import io.element.android.features.messages.impl.attachments.preview.OnDoneListe
 import io.element.android.features.messages.impl.attachments.preview.SendActionState
 import io.element.android.features.messages.impl.fixtures.aMediaAttachment
 import io.element.android.libraries.androidutils.file.TemporaryUriDeleter
+import io.element.android.libraries.featureflag.api.FeatureFlags
+import io.element.android.libraries.featureflag.test.FakeFeatureFlagService
 import io.element.android.libraries.matrix.api.core.ProgressCallback
 import io.element.android.libraries.matrix.api.media.AudioInfo
 import io.element.android.libraries.matrix.api.media.FileInfo
@@ -40,10 +42,12 @@ import io.element.android.libraries.preferences.test.InMemorySessionPreferencesS
 import io.element.android.tests.testutils.WarmUpRule
 import io.element.android.tests.testutils.fake.FakeTemporaryUriDeleter
 import io.element.android.tests.testutils.lambda.any
+import io.element.android.tests.testutils.lambda.lambdaError
 import io.element.android.tests.testutils.lambda.lambdaRecorder
 import io.element.android.tests.testutils.lambda.value
 import io.element.android.tests.testutils.test
 import io.mockk.mockk
+import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.test.advanceUntilIdle
 import kotlinx.coroutines.test.runTest
@@ -94,6 +98,121 @@ class AttachmentsPreviewPresenterTest {
         }
     }
 
+    @Test
+    fun `present - send media after pre-processing success scenario`() = runTest {
+        val sendFileResult = lambdaRecorder<File, FileInfo, String?, String?, ProgressCallback?, Result<FakeMediaUploadHandler>> { _, _, _, _, _ ->
+            Result.success(FakeMediaUploadHandler())
+        }
+        val room = FakeMatrixRoom(
+            sendFileResult = sendFileResult,
+        )
+        val onDoneListener = lambdaRecorder<Unit> { }
+        val processLatch = CompletableDeferred<Unit>()
+        val presenter = createAttachmentsPreviewPresenter(
+            room = room,
+            mediaPreProcessor = FakeMediaPreProcessor(
+                processLatch = processLatch,
+            ),
+            onDoneListener = { onDoneListener() },
+        )
+        moleculeFlow(RecompositionMode.Immediate) {
+            presenter.present()
+        }.test {
+            val initialState = awaitItem()
+            assertThat(initialState.sendActionState).isEqualTo(SendActionState.Idle)
+            // Pre-processing finishes
+            processLatch.complete(Unit)
+            advanceUntilIdle()
+            initialState.eventSink(AttachmentsPreviewEvents.SendAttachment)
+            assertThat(awaitItem().sendActionState).isEqualTo(SendActionState.Sending.InstantSending)
+            advanceUntilIdle()
+            sendFileResult.assertions().isCalledOnce()
+            onDoneListener.assertions().isCalledOnce()
+        }
+    }
+
+    @Test
+    fun `present - send media before pre-processing success scenario`() = runTest {
+        val sendFileResult = lambdaRecorder<File, FileInfo, String?, String?, ProgressCallback?, Result<FakeMediaUploadHandler>> { _, _, _, _, _ ->
+            Result.success(FakeMediaUploadHandler())
+        }
+        val room = FakeMatrixRoom(
+            sendFileResult = sendFileResult,
+        )
+        val onDoneListener = lambdaRecorder<Unit> { }
+        val processLatch = CompletableDeferred<Unit>()
+        val presenter = createAttachmentsPreviewPresenter(
+            room = room,
+            mediaPreProcessor = FakeMediaPreProcessor(
+                processLatch = processLatch,
+            ),
+            onDoneListener = { onDoneListener() },
+        )
+        moleculeFlow(RecompositionMode.Immediate) {
+            presenter.present()
+        }.test {
+            val initialState = awaitItem()
+            assertThat(initialState.sendActionState).isEqualTo(SendActionState.Idle)
+            initialState.eventSink(AttachmentsPreviewEvents.SendAttachment)
+            assertThat(awaitItem().sendActionState).isEqualTo(SendActionState.Sending.Processing)
+            // Pre-processing finishes
+            processLatch.complete(Unit)
+            advanceUntilIdle()
+            sendFileResult.assertions().isCalledOnce()
+            onDoneListener.assertions().isCalledOnce()
+        }
+    }
+
+    @Test
+    fun `present - send media with pre-processing failure after user sends media`() = runTest {
+        val room = FakeMatrixRoom()
+        val onDoneListener = lambdaRecorder<Unit> { }
+        val processLatch = CompletableDeferred<Unit>()
+        val presenter = createAttachmentsPreviewPresenter(
+            room = room,
+            mediaPreProcessor = FakeMediaPreProcessor().apply {
+                givenResult(Result.failure(Exception()))
+            },
+            onDoneListener = { onDoneListener() },
+        )
+        moleculeFlow(RecompositionMode.Immediate) {
+            presenter.present()
+        }.test {
+            val initialState = awaitItem()
+            assertThat(initialState.sendActionState).isEqualTo(SendActionState.Idle)
+            initialState.eventSink(AttachmentsPreviewEvents.SendAttachment)
+            assertThat(awaitItem().sendActionState).isEqualTo(SendActionState.Sending.Processing)
+            // Pre-processing finishes
+            processLatch.complete(Unit)
+            assertThat(awaitItem().sendActionState).isInstanceOf(SendActionState.Failure::class.java)
+        }
+    }
+
+    @Test
+    fun `present - send media with pre-processing failure before user sends media`() = runTest {
+        val room = FakeMatrixRoom()
+        val onDoneListener = lambdaRecorder<Unit> { }
+        val processLatch = CompletableDeferred<Unit>()
+        val presenter = createAttachmentsPreviewPresenter(
+            room = room,
+            mediaPreProcessor = FakeMediaPreProcessor().apply {
+                givenResult(Result.failure(Exception()))
+            },
+            onDoneListener = { onDoneListener() },
+        )
+        moleculeFlow(RecompositionMode.Immediate) {
+            presenter.present()
+        }.test {
+            val initialState = awaitItem()
+            assertThat(initialState.sendActionState).isEqualTo(SendActionState.Idle)
+            // Pre-processing finishes
+            processLatch.complete(Unit)
+            advanceUntilIdle()
+            initialState.eventSink(AttachmentsPreviewEvents.SendAttachment)
+            assertThat(awaitItem().sendActionState).isInstanceOf(SendActionState.Failure::class.java)
+        }
+    }
+
     @Test
     fun `present - cancel scenario`() = runTest {
         val onDoneListener = lambdaRecorder<Unit> { }
@@ -277,7 +396,8 @@ class AttachmentsPreviewPresenterTest {
         permalinkBuilder: PermalinkBuilder = FakePermalinkBuilder(),
         mediaPreProcessor: MediaPreProcessor = FakeMediaPreProcessor(),
         temporaryUriDeleter: TemporaryUriDeleter = FakeTemporaryUriDeleter(),
-        onDoneListener: OnDoneListener = OnDoneListener {},
+        onDoneListener: OnDoneListener = OnDoneListener { lambdaError() },
+        mediaUploadOnSendQueueEnabled: Boolean = true,
     ): AttachmentsPreviewPresenter {
         return AttachmentsPreviewPresenter(
             attachment = aMediaAttachment(localMedia),
@@ -285,6 +405,9 @@ class AttachmentsPreviewPresenterTest {
             mediaSender = MediaSender(mediaPreProcessor, room, InMemorySessionPreferencesStore()),
             permalinkBuilder = permalinkBuilder,
             temporaryUriDeleter = temporaryUriDeleter,
+            featureFlagsService = FakeFeatureFlagService(
+                initialState = mapOf(FeatureFlags.MediaUploadOnSendQueue.key to mediaUploadOnSendQueueEnabled),
+            )
         )
     }
 }

libraries/mediaupload/api/src/main/kotlin/io/element/android/libraries/mediaupload/api/MediaSender.kt

@@ -27,6 +27,35 @@ class MediaSender @Inject constructor(
     private val ongoingUploadJobs = ConcurrentHashMap<Job.Key, MediaUploadHandler>()
     val hasOngoingMediaUploads get() = ongoingUploadJobs.isNotEmpty()
 
+    suspend fun preProcessMedia(
+        uri: Uri,
+        mimeType: String,
+    ): Result<MediaUploadInfo> {
+        val compressIfPossible = sessionPreferencesStore.doesCompressMedia().first()
+        return preProcessor
+            .process(
+                uri = uri,
+                mimeType = mimeType,
+                deleteOriginal = false,
+                compressIfPossible = compressIfPossible,
+            )
+    }
+
+    suspend fun sendPreProcessedMedia(
+        mediaUploadInfo: MediaUploadInfo,
+        caption: String? = null,
+        formattedCaption: String? = null,
+        progressCallback: ProgressCallback? = null
+    ): Result<Unit> {
+        return room.sendMedia(
+            uploadInfo = mediaUploadInfo,
+            progressCallback = progressCallback,
+            caption = caption,
+            formattedCaption = formattedCaption
+        )
+            .handleSendResult()
+    }
+
     suspend fun sendMedia(
         uri: Uri,
         mimeType: String,

libraries/mediaupload/api/src/main/kotlin/io/element/android/libraries/mediaupload/api/MediaUploadInfo.kt

@@ -22,3 +22,12 @@ sealed interface MediaUploadInfo {
     data class VoiceMessage(override val file: File, val audioInfo: AudioInfo, val waveform: List<Float>) : MediaUploadInfo
     data class AnyFile(override val file: File, val fileInfo: FileInfo) : MediaUploadInfo
 }
+
+fun MediaUploadInfo.allFiles(): List<File> {
+    return listOfNotNull(
+        file,
+        (this@allFiles as? MediaUploadInfo.Image)?.thumbnailFile,
+        (this@allFiles as? MediaUploadInfo.Video)?.thumbnailFile,
+    )
+}
+

libraries/mediaupload/test/src/main/kotlin/io/element/android/libraries/mediaupload/test/FakeMediaPreProcessor.kt

@@ -16,10 +16,13 @@ import io.element.android.libraries.matrix.api.media.VideoInfo
 import io.element.android.libraries.mediaupload.api.MediaPreProcessor
 import io.element.android.libraries.mediaupload.api.MediaUploadInfo
 import io.element.android.tests.testutils.simulateLongTask
+import kotlinx.coroutines.CompletableDeferred
 import java.io.File
 import kotlin.time.Duration.Companion.seconds
 
-class FakeMediaPreProcessor : MediaPreProcessor {
+class FakeMediaPreProcessor(
+    private val processLatch: CompletableDeferred<Unit>? = null,
+) : MediaPreProcessor {
     var processCallCount = 0
         private set
 
@@ -41,6 +44,7 @@ class FakeMediaPreProcessor : MediaPreProcessor {
         deleteOriginal: Boolean,
         compressIfPossible: Boolean
     ): Result<MediaUploadInfo> = simulateLongTask {
+        processLatch?.await()
         processCallCount++
         result
     }