diff --git a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/EventStreamDelegate.kt b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/EventStreamDelegate.kt index 86b3487e78..b02ed6dd41 100644 --- a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/EventStreamDelegate.kt +++ b/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>` with last-writer-wins on * `EmojiReaction.version`. - * - **Typing indicators** — maintains per-chat `Set` in [ChatStateHolder]. + * - **Typing indicators** — hands typing notifications to [TypingIndicatorTracker], which + * expires typists locally and maintains per-chat `Set` 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, - 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 } diff --git a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/TypingIndicatorTracker.kt b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/TypingIndicatorTracker.kt new file mode 100644 index 0000000000..bcdb270ff0 --- /dev/null +++ b/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>() + 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, 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) } + } +} diff --git a/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/EventStreamTypingTest.kt b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/EventStreamTypingTest.kt new file mode 100644 index 0000000000..81d4560498 --- /dev/null +++ b/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(1) + private val alice = listOf(2) + private val chat = ChatId("aabbccdd") + + private val updates = Channel(Channel.UNLIMITED) + private val stateHolder = ChatStateHolder() + + private val delegate = EventStreamDelegate( + eventStreamingController = mockk(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(relaxed = true).also { every { it.accountId } returns self }, + stateHolder = stateHolder, + analytics = mockk(relaxed = true), + exchange = mockk(relaxed = true), + ) + + private fun typing(vararg notifications: Pair, 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()) + } +} diff --git a/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/TypingIndicatorTrackerTest.kt b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/TypingIndicatorTrackerTest.kt new file mode 100644 index 0000000000..e8e8b3f1e4 --- /dev/null +++ b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/TypingIndicatorTrackerTest.kt @@ -0,0 +1,169 @@ +package com.flipcash.shared.chat + +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.internal.ChatStateHolder +import com.flipcash.shared.chat.internal.delegates.TypingIndicatorTracker +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.test.TestScope +import kotlinx.coroutines.test.advanceTimeBy +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 +import kotlin.time.Duration +import kotlin.time.Duration.Companion.milliseconds +import kotlin.time.Duration.Companion.seconds +import kotlin.time.Instant + +/** + * The server relays typing best-effort and never times a typist out itself, so a STOPPED that + * never arrives must not leave the indicator up. Mirrors the iOS `ConversationController` typing + * cases. + */ +@OptIn(ExperimentalCoroutinesApi::class) +@RunWith(RobolectricTestRunner::class) +class TypingIndicatorTrackerTest { + + private val self = listOf(1) + private val alice = listOf(2) + private val bob = listOf(3) + private val chat = ChatId("aabbccdd") + private val otherChat = ChatId("eeff0011") + + private class Harness(scope: TestScope) { + val stateHolder = ChatStateHolder() + val tracker = TypingIndicatorTracker( + stateHolder = stateHolder, + timeSource = scope.testScheduler.timeSource, + now = { Instant.fromEpochMilliseconds(scope.testScheduler.currentTime) }, + ).also { it.initialize(scope.backgroundScope) } + + fun typists(chatId: ChatId): List = + stateHolder.current.typingIndicators[chatId].orEmpty().sortedBy { it.since } + } + + private fun TestScope.harness() = Harness(this) + + private fun Harness.send(chatId: ChatId, userId: List, state: TypingState, selfId: List? = self) = + tracker.apply(chatId, listOf(TypingNotification(userId, state)), selfId) + + private fun TestScope.advance(duration: Duration) { + advanceTimeBy(duration) + runCurrent() + } + + @Test + fun `a typist with no further notification expires after 10 seconds`() = runTest { + val h = harness() + h.send(chat, alice, TypingState.STARTED_TYPING) + runCurrent() + assertEquals(listOf(alice), h.typists(chat).map { it.userId }) + + advance(9.seconds + 999.milliseconds) + assertEquals(listOf(alice), h.typists(chat).map { it.userId }) + + advance(1.milliseconds) + assertTrue(h.typists(chat).isEmpty()) + assertTrue(chat !in h.stateHolder.current.typingIndicators) + } + + @Test + fun `a STILL heartbeat extends the deadline without moving the typist`() = runTest { + val h = harness() + h.send(chat, alice, TypingState.STARTED_TYPING) + advance(1.seconds) + h.send(chat, bob, TypingState.STARTED_TYPING) + advance(7.seconds) + + h.send(chat, alice, TypingState.STILL_TYPING) + runCurrent() + val startedAt = h.typists(chat).first { it.userId == alice }.since + + // Past alice's original deadline (10s), inside her refreshed one (18s). + advance(4.seconds) + assertEquals(listOf(alice), h.typists(chat).map { it.userId }) + assertEquals(startedAt, h.typists(chat).single().since) + assertEquals(Instant.fromEpochMilliseconds(0), startedAt) + + advance(6.seconds) + assertTrue(h.typists(chat).isEmpty()) + } + + @Test + fun `a heartbeat keeps a typist ahead of a later starter`() = runTest { + val h = harness() + h.send(chat, alice, TypingState.STARTED_TYPING) + advance(1.seconds) + h.send(chat, bob, TypingState.STARTED_TYPING) + advance(1.seconds) + h.send(chat, alice, TypingState.STILL_TYPING) + runCurrent() + + assertEquals(listOf(alice, bob), h.typists(chat).map { it.userId }) + } + + @Test + fun `STOPPED and TIMED_OUT remove the typist`() = runTest { + val h = harness() + h.send(chat, alice, TypingState.STARTED_TYPING) + h.send(chat, bob, TypingState.STARTED_TYPING) + runCurrent() + + h.send(chat, alice, TypingState.STOPPED_TYPING) + assertEquals(listOf(bob), h.typists(chat).map { it.userId }) + + h.send(chat, bob, TypingState.TYPING_TIMED_OUT) + assertTrue(chat !in h.stateHolder.current.typingIndicators) + } + + @Test + fun `the signed-in user's own typing is ignored`() = runTest { + val h = harness() + h.send(chat, self, TypingState.STARTED_TYPING) + runCurrent() + + assertTrue(h.typists(chat).isEmpty()) + } + + @Test + fun `typists expire independently, across chats`() = runTest { + val h = harness() + h.send(chat, alice, TypingState.STARTED_TYPING) + advance(3.seconds) + h.send(otherChat, bob, TypingState.STARTED_TYPING) + advance(3.seconds) + h.send(chat, bob, TypingState.STARTED_TYPING) + + advance(4.seconds) // t=10: alice in chat lapses + assertEquals(listOf(bob), h.typists(chat).map { it.userId }) + assertEquals(listOf(bob), h.typists(otherChat).map { it.userId }) + + advance(3.seconds) // t=13: bob in otherChat lapses + assertEquals(listOf(bob), h.typists(chat).map { it.userId }) + assertTrue(h.typists(otherChat).isEmpty()) + + advance(3.seconds) // t=16: bob in chat lapses + assertTrue(h.stateHolder.current.typingIndicators.isEmpty()) + } + + @Test + fun `clear drops every typist and nothing re-arms`() = runTest { + val h = harness() + h.send(chat, alice, TypingState.STARTED_TYPING) + h.send(otherChat, bob, TypingState.STARTED_TYPING) + runCurrent() + + h.tracker.clear() + assertTrue(h.stateHolder.current.typingIndicators.isEmpty()) + + // A typist arriving after the clear still expires on its own deadline. + h.send(chat, bob, TypingState.STARTED_TYPING) + advance(10.seconds) + assertTrue(h.stateHolder.current.typingIndicators.isEmpty()) + } +}