fix(chat): expire stale typing indicators locally

apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/EventStreamDelegate.kt

@@ -21,8 +21,6 @@ import com.flipcash.services.models.chat.ReactionUpdate
 import com.flipcash.services.models.chat.Reactor
 import com.flipcash.services.models.chat.RosterChange
 import com.flipcash.shared.chat.reactions.ReactionState
-import com.flipcash.services.models.chat.TypingNotification
-import com.flipcash.services.models.chat.TypingState
 import com.flipcash.services.models.GetDeltaError
 import com.flipcash.shared.chat.ActiveTypist
 import com.flipcash.shared.chat.EventSequenceTracker
@@ -66,7 +64,8 @@ import kotlin.time.Duration.Companion.seconds
  * - **Reaction overlays** — merges [ReactionUpdate]s into an in-memory
  *   `Map<ChatId, Map<Long, ReactionSummary>>` with last-writer-wins on
  *   `EmojiReaction.version`.
- * - **Typing indicators** — maintains per-chat `Set<ActiveTypist>` in [ChatStateHolder].
+ * - **Typing indicators** — hands typing notifications to [TypingIndicatorTracker], which
+ *   expires typists locally and maintains per-chat `Set<ActiveTypist>` in [ChatStateHolder].
  * - **Eager balance update** — credits incoming cash messages to [TokenCoordinator]
  *   before the server balance refresh arrives.
  * - **Heartbeat** — periodically checks stream liveness and reconnects if dead.
@@ -120,6 +119,7 @@ class EventStreamDelegate @Inject constructor(
     private var scope: CoroutineScope? = null
     private var eventStreamCollectJob: Job? = null
     private var heartbeatJob: Job? = null
+    private val typingTracker = TypingIndicatorTracker(stateHolder)
 
     /**
      * One [ReactionState] per message touched by a stream update, keyed by chat then message id.
@@ -198,6 +198,7 @@ class EventStreamDelegate @Inject constructor(
 
     internal fun initialize(scope: CoroutineScope) {
         this.scope = scope
+        typingTracker.initialize(scope)
     }
 
     internal fun open() {
@@ -327,6 +328,7 @@ class EventStreamDelegate @Inject constructor(
     internal fun clearAll() {
         sequenceTracker.clearAll()
         reactionStates.clear()
+        typingTracker.clear()
     }
 
     private fun ensureCollector(scope: CoroutineScope) {
@@ -518,15 +520,7 @@ class EventStreamDelegate @Inject constructor(
         // --- Typing indicators ---
 
         if (update.typingNotifications.isNotEmpty()) {
-            stateHolder.update { state ->
-                val currentTypists = state.typingIndicators[chatId]?.toMutableSet() ?: mutableSetOf()
-                for (notification in update.typingNotifications) {
-                    applyTypingNotification(currentTypists, notification)
-                }
-                state.copy(
-                    typingIndicators = state.typingIndicators + (chatId to currentTypists.toSet())
-                )
-            }
+            typingTracker.apply(chatId, update.typingNotifications, selfId = userManager.accountId)
         }
     }
 
@@ -565,21 +559,5 @@ class EventStreamDelegate @Inject constructor(
         sequenceTracker.setGapFillJob(chatId, job)
     }
 
-    private fun applyTypingNotification(
-        typists: MutableSet<ActiveTypist>,
-        notification: TypingNotification,
-    ) {
-        when (notification.state) {
-            TypingState.STARTED_TYPING, TypingState.STILL_TYPING -> {
-                typists.removeAll { it.userId == notification.userId }
-                typists.add(ActiveTypist(userId = notification.userId, since = Clock.System.now()))
-            }
-            TypingState.STOPPED_TYPING, TypingState.TYPING_TIMED_OUT -> {
-                typists.removeAll { it.userId == notification.userId }
-            }
-            TypingState.UNKNOWN -> Unit
-        }
-    }
-
     // endregion
 }

apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/TypingIndicatorTracker.kt

@@ -0,0 +1,126 @@
+package com.flipcash.shared.chat.internal.delegates
+
+import com.flipcash.services.models.chat.ChatId
+import com.flipcash.services.models.chat.TypingNotification
+import com.flipcash.services.models.chat.TypingState
+import com.flipcash.shared.chat.ActiveTypist
+import com.flipcash.shared.chat.ChatState
+import com.flipcash.shared.chat.internal.ChatStateHolder
+import com.getcode.opencode.model.core.ID
+import com.getcode.utils.TraceType
+import com.getcode.utils.trace
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
+import kotlin.time.Clock
+import kotlin.time.ComparableTimeMark
+import kotlin.time.Duration
+import kotlin.time.Duration.Companion.seconds
+import kotlin.time.Instant
+import kotlin.time.TimeSource
+
+/**
+ * Other members' typing state per chat, and the only writer of [ChatState.typingIndicators].
+ *
+ * The server relays typing best-effort with no timeout of its own, so a lost STOPPED (the sender
+ * backgrounded or killed, a dropped or reopened stream) would leave a typist on screen for good.
+ * Every STARTED/STILL gives the typist a deadline [expiry] out; one job, armed for the earliest
+ * deadline across all chats, drops whoever has passed theirs and re-arms for the next. Matches
+ * iOS's `ConversationTyping`.
+ *
+ * Mutations happen under a lock: notifications arrive on the stream collector and expiry runs on
+ * its own job, both on the coordinator's multi-threaded scope.
+ */
+internal class TypingIndicatorTracker(
+    private val stateHolder: ChatStateHolder,
+    private val timeSource: TimeSource.WithComparableMarks = TimeSource.Monotonic,
+    private val expiry: Duration = INCOMING_EXPIRY,
+    private val now: () -> Instant = { Clock.System.now() },
+) {
+
+    companion object {
+        private const val TAG = "TypingIndicatorTracker"
+
+        // Both clients send STILL every 3s while typing, so this rides out a couple of lost ones.
+        // Same value as iOS's `incomingExpiry`.
+        val INCOMING_EXPIRY = 10.seconds
+    }
+
+    private class Typist(val since: Instant, val deadline: ComparableTimeMark)
+
+    private val lock = Any()
+    private val typists = mutableMapOf<ChatId, MutableMap<ID, Typist>>()
+    private var scope: CoroutineScope? = null
+    private var expiryJob: Job? = null
+
+    fun initialize(scope: CoroutineScope) = synchronized(lock) {
+        this.scope = scope
+    }
+
+    /** Applies [notifications] for [chatId], ignoring the signed-in user's own ([selfId]). */
+    fun apply(chatId: ChatId, notifications: List<TypingNotification>, selfId: ID?) = synchronized(lock) {
+        for (notification in notifications) {
+            if (notification.userId == selfId) continue
+            when (notification.state) {
+                TypingState.STARTED_TYPING, TypingState.STILL_TYPING -> {
+                    val chat = typists.getOrPut(chatId) { mutableMapOf() }
+                    // A STILL heartbeat extends the deadline but keeps the typist's place in line.
+                    val since = chat[notification.userId]?.since ?: now()
+                    chat[notification.userId] = Typist(since, timeSource.markNow() + expiry)
+                }
+                TypingState.STOPPED_TYPING, TypingState.TYPING_TIMED_OUT -> remove(chatId, notification.userId)
+                TypingState.UNKNOWN -> Unit
+            }
+        }
+        publish()
+        scheduleExpiry()
+    }
+
+    /** Drops every typist and stops the expiry job. */
+    fun clear() = synchronized(lock) {
+        typists.clear()
+        expiryJob?.cancel()
+        expiryJob = null
+        publish()
+    }
+
+    private fun remove(chatId: ChatId, userId: ID) {
+        val chat = typists[chatId] ?: return
+        chat.remove(userId)
+        if (chat.isEmpty()) typists.remove(chatId)
+    }
+
+    private fun scheduleExpiry() {
+        expiryJob?.cancel()
+        expiryJob = null
+        val earliest = typists.values.flatMap { it.values }.minOfOrNull { it.deadline } ?: return
+        val scope = scope ?: return
+        expiryJob = scope.launch {
+            // Negative while the deadline is ahead of us.
+            delay(-earliest.elapsedNow())
+            sweep()
+        }
+    }
+
+    private fun sweep() = synchronized(lock) {
+        val current = timeSource.markNow()
+        for ((chatId, chat) in typists.entries.toList()) {
+            for ((userId, typist) in chat.entries.toList()) {
+                if (typist.deadline <= current) {
+                    trace(tag = TAG, message = "Expiring stale typist in $chatId", type = TraceType.Process)
+                    remove(chatId, userId)
+                }
+            }
+        }
+        publish()
+        scheduleExpiry()
+    }
+
+    private fun publish() {
+        val snapshot = typists.mapValues { (_, chat) ->
+            chat.map { (userId, typist) -> ActiveTypist(userId = userId, since = typist.since) }.toSet()
+        }
+        stateHolder.update { it.copy(typingIndicators = snapshot) }
+    }
+}

apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/EventStreamTypingTest.kt

@@ -0,0 +1,81 @@
+package com.flipcash.shared.chat
+
+import com.flipcash.services.controllers.EventStreamingController
+import com.flipcash.services.models.chat.ChatId
+import com.flipcash.services.models.chat.ChatUpdate
+import com.flipcash.services.models.chat.TypingNotification
+import com.flipcash.services.models.chat.TypingState
+import com.flipcash.services.user.UserManager
+import com.flipcash.shared.chat.internal.ChatStateHolder
+import com.flipcash.shared.chat.internal.delegates.EventStreamDelegate
+import io.mockk.every
+import io.mockk.mockk
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.flow.emptyFlow
+import kotlinx.coroutines.flow.receiveAsFlow
+import kotlinx.coroutines.test.runCurrent
+import kotlinx.coroutines.test.runTest
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.robolectric.RobolectricTestRunner
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+
+/** Typing notifications from the stream reach [ChatState.typingIndicators], minus the user's own. */
+@OptIn(ExperimentalCoroutinesApi::class)
+@RunWith(RobolectricTestRunner::class)
+class EventStreamTypingTest {
+
+    private val self = listOf<Byte>(1)
+    private val alice = listOf<Byte>(2)
+    private val chat = ChatId("aabbccdd")
+
+    private val updates = Channel<ChatUpdate>(Channel.UNLIMITED)
+    private val stateHolder = ChatStateHolder()
+
+    private val delegate = EventStreamDelegate(
+        eventStreamingController = mockk<EventStreamingController>(relaxed = true).also {
+            every { it.chatUpdates } returns updates.receiveAsFlow()
+            every { it.streamFailures } returns emptyFlow()
+            every { it.isConnected } returns false
+        },
+        messagingController = mockk(relaxed = true),
+        metadataDataSource = mockk(relaxed = true),
+        messageDataSource = mockk(relaxed = true),
+        memberDataSource = mockk(relaxed = true),
+        tokenCoordinator = mockk(relaxed = true),
+        userManager = mockk<UserManager>(relaxed = true).also { every { it.accountId } returns self },
+        stateHolder = stateHolder,
+        analytics = mockk(relaxed = true),
+        exchange = mockk(relaxed = true),
+    )
+
+    private fun typing(vararg notifications: Pair<List<Byte>, TypingState>) = ChatUpdate(
+        chatId = chat,
+        typingNotifications = notifications.map { (userId, state) -> TypingNotification(userId, state) },
+    )
+
+    @Test
+    fun `a counterpart's typing shows and the user's own does not`() = runTest {
+        delegate.initialize(backgroundScope)
+        delegate.open()
+
+        updates.send(typing(self to TypingState.STARTED_TYPING, alice to TypingState.STARTED_TYPING))
+        runCurrent()
+
+        assertEquals(listOf(alice), stateHolder.current.typingIndicators[chat]?.map { it.userId })
+    }
+
+    @Test
+    fun `clearAll drops typists`() = runTest {
+        delegate.initialize(backgroundScope)
+        delegate.open()
+        updates.send(typing(alice to TypingState.STARTED_TYPING))
+        runCurrent()
+
+        delegate.clearAll()
+
+        assertTrue(stateHolder.current.typingIndicators.isEmpty())
+    }
+}