fix(chat): prevent duplicate event streams from concurrent open

open()/close() did a non-atomic check-then-act on streamRef while being
invoked from multiple triggers (login, lifecycle onStart, network
reconnect, feature-flag, heartbeat) on the multi-threaded IO scope. Two
threads could both observe a null ref and each open a gRPC stream, leaving
a second orphaned stream whose reconnect loop kept running. The server
rejects the duplicate with ABORTED 'stream already exists' and the two
streams knock each other offline in a persistent duel.

Serialize open/close under a lock so only one stream can exist, and guard
onError to clear only its own ref. Adds a 32-thread concurrency test.

Author: bmc08gt

services/flipcash/src/main/kotlin/com/flipcash/services/controllers/EventStreamingController.kt

@@ -20,13 +20,20 @@ class EventStreamingController @Inject constructor(
     private val _chatUpdates = Channel<ChatUpdate>(capacity = Channel.UNLIMITED)
     val chatUpdates: Flow<ChatUpdate> = _chatUpdates.receiveAsFlow()
 
+    // Guards all reads/writes of [streamRef]. open()/close() are invoked
+    // concurrently from multiple triggers (login, lifecycle onStart, network
+    // reconnect, feature-flag, heartbeat) on the multi-threaded IO scope, so
+    // the check-then-act on streamRef must be atomic — otherwise two threads
+    // both observe a null ref and each open a stream, producing a second
+    // orphaned gRPC stream and the server's "ABORTED: stream already exists".
+    private val lock = Any()
     private var streamRef: EventStreamReference? = null
 
-    val isConnected: Boolean get() = streamRef != null
+    val isConnected: Boolean get() = synchronized(lock) { streamRef != null }
 
-    val isStreamActive: Boolean get() = streamRef?.isActive == true
+    val isStreamActive: Boolean get() = synchronized(lock) { streamRef?.isActive == true }
 
-    fun open(scope: CoroutineScope): Boolean {
+    fun open(scope: CoroutineScope): Boolean = synchronized(lock) {
         if (streamRef != null) {
             trace("EventStreamingController: Stream already open, skipping")
             return true
@@ -37,7 +44,8 @@ class EventStreamingController @Inject constructor(
             return false
         }
 
-        streamRef = repository.openEventStream(
+        lateinit var openedRef: EventStreamReference
+        openedRef = repository.openEventStream(
             scope = scope,
             owner = owner,
             onEvent = { update ->
@@ -47,15 +55,18 @@ class EventStreamingController @Inject constructor(
             onError = { error ->
                 trace("EventStreamingController: Stream error: ${error.message}")
                 // Clear the ref so the next heartbeat, lifecycle, or network
-                // event creates a fresh stream. The framework guarantees this
-                // fires only once per stream, so no risk of clearing a newer ref.
-                streamRef = null
+                // event creates a fresh stream. Only clear if it is still THIS
+                // stream — never null out a newer ref opened after us.
+                synchronized(lock) {
+                    if (streamRef === openedRef) streamRef = null
+                }
             },
         )
+        streamRef = openedRef
         return true
     }
 
-    fun close() {
+    fun close() = synchronized(lock) {
         streamRef?.destroy()
         streamRef = null
     }

services/flipcash/src/test/kotlin/com/flipcash/services/controllers/EventStreamingControllerTest.kt

@@ -11,10 +11,15 @@ import io.mockk.every
 import io.mockk.mockk
 import io.mockk.verify
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.test.UnconfinedTestDispatcher
 import kotlinx.coroutines.test.runTest
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+import kotlin.concurrent.thread
 import kotlin.test.Test
+import kotlin.test.assertEquals
 import kotlin.test.assertNotNull
 
 @OptIn(ExperimentalCoroutinesApi::class)
@@ -69,21 +74,57 @@ class EventStreamingControllerTest {
     fun `chatUpdates flow is accessible`() {
         assertNotNull(controller.chatUpdates)
     }
+
+    @Test
+    fun `concurrent open calls create only one stream`() {
+        stubOwner()
+        val scope = CoroutineScope(Dispatchers.Default)
+        val threadCount = 32
+        val startLatch = CountDownLatch(1)
+        val doneLatch = CountDownLatch(threadCount)
+
+        repeat(threadCount) {
+            thread {
+                startLatch.await()
+                controller.open(scope)
+                doneLatch.countDown()
+            }
+        }
+
+        // Release all threads at once to maximize the check-then-act race.
+        startLatch.countDown()
+        doneLatch.await(5, TimeUnit.SECONDS)
+
+        // Exactly one gRPC stream must be opened; opening two produces the
+        // server-side "ABORTED: stream already exists" duel.
+        assertEquals(1, repository.openCount)
+    }
 }
 
 private class FakeEventStreamingRepository : EventStreamingRepository {
-    var opened = false
+    private val lock = Any()
+    var openCount = 0
+        private set
+    val opened: Boolean get() = openCount > 0
     var lastStreamRef: EventStreamReference? = null
+        private set
 
     override fun openEventStream(
         scope: CoroutineScope,
         owner: Ed25519.KeyPair,
         onEvent: (ChatUpdate) -> Unit,
         onError: (Throwable) -> Unit,
     ): EventStreamReference {
-        opened = true
-        val ref = mockk<EventStreamReference>(relaxed = true)
-        lastStreamRef = ref
-        return ref
+        // Widen the controller's check-then-act window so an unsynchronized
+        // open() lets multiple threads reach here concurrently. Mock creation
+        // and counting are serialized here so the assertion measures the
+        // controller's concurrency, not mockk's.
+        Thread.sleep(20)
+        return synchronized(lock) {
+            openCount++
+            val ref = mockk<EventStreamReference>(relaxed = true)
+            lastStreamRef = ref
+            ref
+        }
     }
 }