fix: cancelling an upload could still share the file

Cancel relied solely on WorkManager's isStopped, but
cancelUniqueWork() dispatches asynchronously through WorkManager's
own executor and Room DB - a small/compressed image could finish
uploading and get shared to the room before that cancellation ever
became visible to doWork(). Track cancelled referenceIds in an
in-memory set instead, set synchronously by the UI before
cancelUniqueWork() is called, so the worker can see it immediately.

Also close a second gap: once the upload succeeded, the local
placeholder stayed flagged as temporary indefinitely, so a "too late"
cancel could still delete it locally - making an already-sent message
disappear from the UI until the next sync brought it back. Guard the
placeholder deletion to only fire while the message is still PENDING.

Assisted-by: Claude:claude-sonnet-5

Signed-off-by: Marcel Hibbe <dev@mhibbe.de>

Author: mahibi

app/src/main/java/com/nextcloud/talk/chat/data/ChatMessageRepository.kt

@@ -160,7 +160,12 @@ interface ChatMessageRepository : LifecycleAwareManager {
         referenceId: String
     ): Flow<Result<ChatMessage?>>
 
-    suspend fun deleteTempMessageByReferenceId(referenceId: String)
+    /**
+     * Deletes the local upload placeholder for [referenceId], but only while it's still PENDING.
+     * Returns false without deleting anything if the upload already finished (and was shared) by
+     * the time this is called, so a late cancel can't hide an already-sent message.
+     */
+    suspend fun deleteTempMessageByReferenceId(referenceId: String): Boolean
 
     suspend fun editChatMessage(credentials: String, url: String, text: String): Flow<Result<ChatOverallSingleMessage>>
 

app/src/main/java/com/nextcloud/talk/chat/data/network/OfflineFirstChatRepository.kt

@@ -755,9 +755,8 @@ class OfflineFirstChatRepository @Inject constructor(
             }
         }
 
-    override suspend fun deleteTempMessageByReferenceId(referenceId: String) {
-        chatDao.deleteTempChatMessages(internalConversationId, listOf(referenceId))
-    }
+    override suspend fun deleteTempMessageByReferenceId(referenceId: String): Boolean =
+        chatDao.deleteTempChatMessageIfPending(internalConversationId, referenceId) > 0
 
     @Suppress("Detekt.TooGenericExceptionCaught")
     override suspend fun editChatMessage(

app/src/main/java/com/nextcloud/talk/chat/viewmodels/ChatViewModel.kt

@@ -235,7 +235,7 @@ class ChatViewModel @AssistedInject constructor(
 
     fun cancelUpload(referenceId: String) {
         val fileUri = uploadReferenceToUri.remove(referenceId) ?: return
-        WorkManager.getInstance(NextcloudTalkApplication.sharedApplication!!).cancelUniqueWork(fileUri)
+        UploadAndShareFilesWorker.cancelUpload(referenceId, fileUri)
         viewModelScope.launch {
             chatRepository.deleteTempMessageByReferenceId(referenceId)
         }

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

@@ -177,6 +177,23 @@ interface ChatMessagesDao {
     )
     fun deleteTempChatMessages(internalConversationId: String, referenceIds: List<String>)
 
+    /*
+     * Used when the user cancels an upload - unlike deleteTempChatMessages, this must NOT delete
+     * the placeholder once sendStatus has already moved past PENDING (i.e. the upload already
+     * finished and was shared), since by then the message has actually been sent and deleting the
+     * local placeholder would only hide it until the next sync brings it back.
+     */
+    @Query(
+        value = """
+            DELETE FROM ChatMessages
+            WHERE internalConversationId = :internalConversationId
+            AND referenceId = :referenceId
+            AND isTemporary = 1
+            AND sendStatus = 'PENDING'
+        """
+    )
+    fun deleteTempChatMessageIfPending(internalConversationId: String, referenceId: String): Int
+
     @Update
     fun updateChatMessage(message: ChatMessageEntity)
 

app/src/main/java/com/nextcloud/talk/jobs/UploadAndShareFilesWorker.kt

@@ -53,6 +53,8 @@ import com.nextcloud.talk.utils.VideoCompressor
 import com.nextcloud.talk.utils.database.user.CurrentUserProviderOld
 import com.nextcloud.talk.utils.permissions.PlatformPermissionUtil
 import com.nextcloud.talk.utils.preferences.AppPreferences
+import io.reactivex.Observable
+import io.reactivex.disposables.Disposable
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.SharedFlow
@@ -62,7 +64,10 @@ import okhttp3.MediaType.Companion.toMediaTypeOrNull
 import okhttp3.OkHttpClient
 import java.io.File
 import java.io.IOException
+import java.util.Collections
 import java.util.UUID
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.CountDownLatch
 import java.util.concurrent.TimeUnit
 import javax.inject.Inject
 
@@ -107,11 +112,35 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa
     private var chunkedFileUploader: ChunkedFileUploader? = null
     private var referenceId: String? = null
     private var internalConversationId: String? = null
+    private var uploadDisposable: Disposable? = null
+    private var uploadLatch: CountDownLatch? = null
+
+    /**
+     * WorkManager's own `isStopped`/`onStopped()` only becomes true/fires after
+     * cancelUniqueWork() round-trips through WorkManager's internal executor and Room DB - for a
+     * small/compressed image the whole upload+share can finish faster than that round-trip, so
+     * isStopped alone arrives too late. [cancelledReferenceIds] is set synchronously by the UI
+     * before cancelUniqueWork() is even called, so it's visible to doWork() immediately.
+     */
+    private fun isCancelled(): Boolean = referenceId?.let { cancelledReferenceIds.contains(it) } == true
 
-    @Suppress("Detekt.TooGenericExceptionCaught", "Detekt.LongMethod")
     override fun doWork(): Result {
         NextcloudTalkApplication.sharedApplication!!.componentApplication.inject(this)
 
+        try {
+            return doUpload()
+        } finally {
+            referenceId?.let { cancelledReferenceIds.remove(it) }
+        }
+    }
+
+    @Suppress(
+        "Detekt.TooGenericExceptionCaught",
+        "Detekt.LongMethod",
+        "Detekt.CyclomaticComplexMethod",
+        "Detekt.ReturnCount"
+    )
+    private fun doUpload(): Result {
         return try {
             currentUser = currentUserProvider.currentUser.blockingGet()
             val sourceFile = inputData.getString(DEVICE_SOURCE_FILE)
@@ -149,6 +178,11 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa
                 useConversationSubfolders = useConversationSubfolders
             )
 
+            if (uploadSuccess && (isStopped || isCancelled())) {
+                // Cancelled right as the upload finished - don't share a cancelled upload.
+                return Result.failure()
+            }
+
             if (uploadSuccess) {
                 // useConversationSubfolders already shares as part of uploadFile() via
                 // postConversationAttachment, so only share explicitly for the plain upload path.
@@ -160,7 +194,7 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa
                 }
                 Log.e(TAG, "Share operation failed after upload")
                 return failUpload()
-            } else if (isStopped) {
+            } else if (isStopped || isCancelled()) {
                 // since work is cancelled the result would be ignored anyways
                 return Result.failure()
             }
@@ -176,7 +210,9 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa
                 "Network error while uploading file (attempt ${runAttemptCount + 1}/$MAX_UPLOAD_ATTEMPTS)",
                 e
             )
-            if (runAttemptCount < MAX_UPLOAD_ATTEMPTS - 1) {
+            if (isStopped || isCancelled()) {
+                Result.failure()
+            } else if (runAttemptCount < MAX_UPLOAD_ATTEMPTS - 1) {
                 Result.retry()
             } else {
                 failUpload()
@@ -235,7 +271,7 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa
             chunkedFileUploader!!.upload(file!!, mimeType, remotePath)
         } else {
             Log.d(TAG, "starting normal upload (not chunked) of $fileName")
-            FileUploader(
+            val observable = FileUploader(
                 okHttpClient,
                 context,
                 currentUser,
@@ -245,9 +281,37 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa
                 ncApiCoroutines
             )
                 .upload(sourceFileUri, fileName, remotePath, null)
-                .blockingFirst()
+            blockingUpload(observable)
         }
 
+    // unlike .blockingFirst(), keeps a Disposable so onStopped() can cancel the underlying OkHttp call
+    private fun blockingUpload(observable: Observable<Boolean>): Boolean {
+        val latch = CountDownLatch(1)
+        uploadLatch = latch
+        var result = false
+        var error: Throwable? = null
+        uploadDisposable = observable.subscribe(
+            { success ->
+                result = success
+                latch.countDown()
+            },
+            { throwable ->
+                error = throwable
+                latch.countDown()
+            },
+            { latch.countDown() }
+        )
+        latch.await()
+        uploadDisposable = null
+        uploadLatch = null
+
+        if (isStopped || isCancelled()) {
+            return false
+        }
+        error?.let { throw it }
+        return result
+    }
+
     private fun uploadUsingConversationSubfolders(sourceFileUri: Uri, metaData: String?): Boolean =
         runBlocking {
             val credentials = ApiUtils.getCredentials(
@@ -287,7 +351,7 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa
                     .uploadToConversationSubfolder(sourceFileUri, tempRemotePath)
             }
 
-            if (!uploadSuccess) {
+            if (!uploadSuccess || isStopped || isCancelled()) {
                 return@runBlocking false
             }
 
@@ -352,6 +416,8 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa
         if (file != null && isChunkedUploading) {
             chunkedFileUploader?.abortUpload {}
         }
+        uploadDisposable?.dispose()
+        uploadLatch?.countDown()
         super.onStopped()
     }
 
@@ -399,6 +465,11 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa
         private const val MAX_UPLOAD_ATTEMPTS = 4
         const val REQUEST_PERMISSION = 3123
 
+        // referenceIds the user cancelled - set synchronously here, before cancelUniqueWork() is
+        // even called, so doWork() can see it immediately instead of waiting for isStopped, which
+        // only becomes true once WorkManager's own async cancellation dispatch completes.
+        private val cancelledReferenceIds: MutableSet<String> = Collections.newSetFromMap(ConcurrentHashMap())
+
         private val _uploadCompletedFlow: MutableSharedFlow<String> = MutableSharedFlow(
             replay = 1,
             extraBufferCapacity = 1
@@ -479,5 +550,10 @@ class UploadAndShareFilesWorker(val context: Context, workerParameters: WorkerPa
             WorkManager.getInstance().enqueueUniqueWork(fileUri, ExistingWorkPolicy.KEEP, uploadWorker)
             return uploadWorker.id
         }
+
+        fun cancelUpload(referenceId: String, fileUri: String) {
+            cancelledReferenceIds.add(referenceId)
+            WorkManager.getInstance().cancelUniqueWork(fileUri)
+        }
     }
 }

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

@@ -84,6 +84,8 @@ class DummyChatMessagesDaoImpl : ChatMessagesDao {
         /* */
     }
 
+    override fun deleteTempChatMessageIfPending(internalConversationId: String, referenceId: String): Int = 0
+
     override fun updateChatMessage(message: ChatMessageEntity) {
         /* */
     }