From d0e35500250d574ee1e5fff6727df49f628feafa Mon Sep 17 00:00:00 2001
From: kami <kamixtrash@gmail.com>
Date: Mon, 25 May 2026 01:06:25 +0400
Subject: [PATCH] feat(tui-02): complete Channel(UNLIMITED) migration for
 _connection field in TuiWsClient

Migrate _connection from MutableSharedFlow + tryEmit to Channel(UNLIMITED) + send to ensure no connection events are dropped under high load. Removes unused MutableSharedFlow and asSharedFlow imports. Changes public connection property type from SharedFlow to Flow for consistency with messages field.

- Replace MutableSharedFlow with Channel(UNLIMITED) for _connection
- Convert all _connection.tryEmit() calls to _connection.send()
- Update public connection property to use receiveAsFlow()
- Add TuiWsClientConnectionChannelTest with 10k event no-drop verification
---
 .../com/correx/apps/tui/ws/TuiWsClient.kt     | 13 +++---
 .../ws/TuiWsClientConnectionChannelTest.kt    | 45 +++++++++++++++++++
 2 files changed, 50 insertions(+), 8 deletions(-)
 create mode 100644 apps/tui/src/test/kotlin/com/correx/apps/tui/ws/TuiWsClientConnectionChannelTest.kt

diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/ws/TuiWsClient.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/ws/TuiWsClient.kt
index b317c263..3fdb1ec5 100644
--- a/apps/tui/src/main/kotlin/com/correx/apps/tui/ws/TuiWsClient.kt
+++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/ws/TuiWsClient.kt
@@ -19,9 +19,6 @@ import kotlinx.coroutines.Job
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.MutableSharedFlow
-import kotlinx.coroutines.flow.SharedFlow
-import kotlinx.coroutines.flow.asSharedFlow
 import kotlinx.coroutines.flow.receiveAsFlow
 import kotlinx.coroutines.launch
 import kotlinx.serialization.encodeToString
@@ -53,8 +50,8 @@ class TuiWsClient(
     private val _messages = Channel<ServerMessage>(Channel.UNLIMITED)
     val messages: Flow<ServerMessage> = _messages.receiveAsFlow()
 
-    private val _connection = MutableSharedFlow<ConnectionEvent>(extraBufferCapacity = 16)
-    val connection: SharedFlow<ConnectionEvent> = _connection.asSharedFlow()
+    private val _connection = Channel<ConnectionEvent>(Channel.UNLIMITED)
+    val connection: Flow<ConnectionEvent> = _connection.receiveAsFlow()
 
     suspend fun send(message: ClientMessage) {
         runCatching {
@@ -75,7 +72,7 @@ class TuiWsClient(
             runCatching {
                 client.webSocket(method = HttpMethod.Get, host = host, port = port, path = "/stream") {
                     session = this
-                    _connection.tryEmit(ConnectionEvent.Connected)
+                    _connection.send(ConnectionEvent.Connected)
                     delayMs = INITIAL_RETRY_DELAY_MS
                     attempt = 0
                     for (frame in incoming) {
@@ -90,10 +87,10 @@ class TuiWsClient(
                 }
             }.onFailure { }
             session = null
-            _connection.tryEmit(ConnectionEvent.Disconnected)
+            _connection.send(ConnectionEvent.Disconnected)
             attempt++
             val nextRetryAtMs = System.currentTimeMillis() + delayMs
-            _connection.tryEmit(ConnectionEvent.RetryScheduled(attempt = attempt, nextRetryAtMs = nextRetryAtMs))
+            _connection.send(ConnectionEvent.RetryScheduled(attempt = attempt, nextRetryAtMs = nextRetryAtMs))
             delay(delayMs)
             delayMs = minOf(delayMs * 2, MAX_RETRY_DELAY_MS)
         }
diff --git a/apps/tui/src/test/kotlin/com/correx/apps/tui/ws/TuiWsClientConnectionChannelTest.kt b/apps/tui/src/test/kotlin/com/correx/apps/tui/ws/TuiWsClientConnectionChannelTest.kt
new file mode 100644
index 00000000..0449192a
--- /dev/null
+++ b/apps/tui/src/test/kotlin/com/correx/apps/tui/ws/TuiWsClientConnectionChannelTest.kt
@@ -0,0 +1,45 @@
+package com.correx.apps.tui.ws
+
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.runTest
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+
+class TuiWsClientConnectionChannelTest {
+
+    @Test
+    fun `Channel UNLIMITED does not drop connection events under high load`(): Unit = runTest {
+        // Create a test channel to simulate TuiWsClient._connection behavior
+        val channel = Channel<ConnectionEvent>(Channel.UNLIMITED)
+
+        // Collect events from the channel
+        val collected = mutableListOf<ConnectionEvent>()
+        val collectJob = launch {
+            for (event in channel) {
+                collected.add(event)
+            }
+        }
+
+        // Send 10k events rapidly without delay (mix of all event types)
+        val sentCount = 10_000
+        repeat(sentCount) { i ->
+            val event = when (i % 3) {
+                0 -> ConnectionEvent.Connected
+                1 -> ConnectionEvent.Disconnected
+                else -> ConnectionEvent.RetryScheduled(attempt = i, nextRetryAtMs = System.currentTimeMillis() + 1000)
+            }
+            channel.send(event)
+        }
+
+        // Close channel to signal completion
+        channel.close()
+
+        // Wait for collection to complete
+        collectJob.join()
+
+        // Verify all events were received (no silent drops)
+        assertEquals(sentCount, collected.size, "All $sentCount connection events should be received without drops")
+    }
+
+}