Own composed message batches and receipts beyond Activity lifetime

tools/chatgui-android/app/src/main/java/com/utun/chat/data/ConversationSubmissions.kt

@@ -0,0 +1,54 @@
+package com.utun.chat.data
+
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.CoroutineStart
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.asStateFlow
+import kotlinx.coroutines.isActive
+import kotlinx.coroutines.launch
+
+/** Main: пакет «вложение + текст» принадлежит ViewModel, UI лишь читает подтверждения своего token.
+ * null означает ожидание; true — сохранение, false — отказ. Отсутствующая часть сразу true.
+ * Последнее подтверждение хранится до consume/следующей отправки этому адресату, включая поколение PM. */
+class ConversationSubmissions(private val scope: CoroutineScope,
+                              private val log: (String, String) -> Unit = { level, message -> LogManager.addLog(level, "Composer", message) }) {
+    data class State(val token: String, val attachmentSaved: Boolean?, val textSaved: Boolean?, val finished: Boolean = false)
+    private val _states = MutableStateFlow<Map<ConversationTarget, State>>(emptyMap())
+    val states = _states.asStateFlow()
+
+    /** false не принимает задачи и их файлы; true запускает обе части независимо от времени жизни UI caller. */
+    fun submit(target: ConversationTarget, token: String, attachment: (suspend () -> Boolean)?, text: (suspend () -> Boolean)?,
+               onSaved: () -> Unit, onFailure: () -> Unit): Boolean {
+        if (!scope.isActive || _states.value[target]?.finished == false) {
+            log("WARN", "Submission rejected target=${target.nativeId} ownerActive=${scope.isActive}")
+            return false
+        }
+        var state = State(token, if (attachment == null) true else null, if (text == null) true else null)
+        _states.value = _states.value + (target to state)
+        scope.launch(start = CoroutineStart.UNDISPATCHED) {
+            try {
+                state = state.copy(attachmentSaved = attachment?.invoke() ?: true)
+                _states.value = _states.value + (target to state)
+                state = state.copy(textSaved = text?.invoke() ?: true)
+                if (state.attachmentSaved == true && state.textSaved == true) onSaved() else onFailure()
+            } catch (e: CancellationException) {
+                log("WARN", "Submission owner cancelled target=${target.nativeId}")
+                throw e
+            } catch (e: Exception) {
+                log("ERROR", "Submission failed target=${target.nativeId}: $e")
+                onFailure()
+            } finally {
+                state = state.copy(attachmentSaved = state.attachmentSaved ?: false, textSaved = state.textSaved ?: false, finished = true)
+                _states.value = _states.value + (target to state)
+                log(if (state.attachmentSaved == true && state.textSaved == true) "INFO" else "WARN",
+                    "Submission finished target=${target.nativeId} attachment=${state.attachmentSaved} text=${state.textSaved}")
+            }
+        }
+        return true
+    }
+
+    fun consume(target: ConversationTarget, token: String) {
+        if (_states.value[target]?.let { it.token == token && it.finished } == true) _states.value = _states.value - target
+    }
+}

tools/chatgui-android/router-tests/build.gradle.kts

@@ -27,6 +27,7 @@ kotlin {
             "com/utun/chat/data/CallAudioTiming.kt", "com/utun/chat/data/CallAudioTrace.kt",
             "com/utun/chat/data/CallPlaybackBuffer.kt", "com/utun/chat/data/CallOutputRecovery.kt",
             "com/utun/chat/data/CallState.kt", "com/utun/chat/data/CallAudioHost.kt")
+        kotlin.include("com/utun/chat/data/ConversationSubmissions.kt", "com/utun/chat/data/ConversationTarget.kt")
     }
 }
 

tools/chatgui-android/router-tests/src/test/kotlin/com/utun/chat/data/ConversationSubmissionsTest.kt

@@ -0,0 +1,64 @@
+package com.utun.chat.data
+
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.runCurrent
+import kotlinx.coroutines.test.runTest
+import org.junit.Assert.*
+import org.junit.Test
+
+@OptIn(ExperimentalCoroutinesApi::class)
+class ConversationSubmissionsTest {
+    private val target = ConversationTarget.Pm("123", 2)
+
+    @Test fun losingTheUiWaiterCannotCancelTheBatchOrItsAcknowledgements() = runTest {
+        val owner = ConversationSubmissions(this) { _, _ -> }
+        val attachment = CompletableDeferred<Boolean>(); val text = CompletableDeferred<Boolean>()
+        var textStarted = false; var repliesFinished = 0
+        assertTrue(owner.submit(target, "a", { attachment.await() }, { textStarted = true; text.await() }, { repliesFinished++ }, {}))
+        val ui = launch { attachment.await() }; runCurrent(); ui.cancel()
+        assertFalse(owner.states.value.getValue(target).finished)
+        assertFalse(owner.submit(target, "duplicate", null, { true }, {}, {}))
+        attachment.complete(true); runCurrent()
+        assertTrue(textStarted); assertEquals(true, owner.states.value.getValue(target).attachmentSaved)
+        assertNull(owner.states.value.getValue(target).textSaved)
+        text.complete(true); runCurrent()
+        assertEquals(1, repliesFinished); assertTrue(owner.states.value.getValue(target).finished)
+        owner.consume(target, "stale"); assertTrue(owner.states.value.containsKey(target))
+        owner.consume(target, "a"); assertTrue(owner.states.value.isEmpty())
+    }
+
+    @Test fun failureKeepsTheUnsentPartAndDoesNotFinishReply() = runTest {
+        val logs = mutableListOf<String>()
+        val owner = ConversationSubmissions(this) { _, message -> logs.add(message) }
+        var errors = 0; var replied = false
+        owner.submit(target, "a", { true }, { throw IllegalStateException("database") }, { replied = true }, { errors++ })
+        runCurrent()
+        val result = owner.states.value.getValue(target)
+        assertTrue(result.finished); assertEquals(true, result.attachmentSaved); assertEquals(false, result.textSaved)
+        assertFalse(replied); assertEquals(1, errors); assertTrue(logs.any { "database" in it })
+    }
+
+    @Test fun oldGenerationAndOldReceiptCannotConsumeANewSubmission() = runTest {
+        val owner = ConversationSubmissions(this) { _, _ -> }
+        owner.submit(target, "a", null, { true }, {}, {})
+        val reopened = ConversationTarget.Pm("123", 3)
+        val pending = CompletableDeferred<Boolean>()
+        owner.submit(reopened, "b", null, { pending.await() }, {}, {})
+        owner.consume(reopened, "a"); owner.consume(target, "a")
+        assertFalse(owner.states.value.getValue(reopened).finished)
+        pending.complete(true); runCurrent(); owner.consume(reopened, "b")
+        assertTrue(owner.states.value.isEmpty())
+    }
+
+    @Test fun stoppedOwnerDoesNotAcceptFilesOrRunTasks() = runTest {
+        val job = Job().also { it.cancel() }
+        val owner = ConversationSubmissions(CoroutineScope(coroutineContext + job)) { _, _ -> }
+        var ran = false
+        assertFalse(owner.submit(target, "a", { ran = true; true }, null, {}, {}))
+        assertFalse(ran); assertTrue(owner.states.value.isEmpty())
+    }
+}