diff --git a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/RealChatCoordinator.kt b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/RealChatCoordinator.kt index 0286fc1683..5e5678c951 100644 --- a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/RealChatCoordinator.kt +++ b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/RealChatCoordinator.kt @@ -45,7 +45,7 @@ import javax.inject.Singleton import kotlin.time.Duration.Companion.seconds /** - * Thin orchestration shell that implements [ChatCoordinator] by composing three + * Thin orchestration shell that implements [ChatCoordinator] by composing five * focused delegates via Kotlin `by` interface delegation: * * | Delegate | Interface | Responsibility | @@ -54,13 +54,17 @@ import kotlin.time.Duration.Companion.seconds * | [EventStreamDelegate] | [EventStreamOperations] | Event stream, real-time updates, gap-aware sequencing, reactions, typing | * | [DmChatResolverDelegate] | [DmChatResolver] | Resolve a DM's [ChatId] from its participants (derive or look up) | * | [MessagingDelegate] | [MessagingOperations] | Per-chat send/receive, read pointers, paging, notifications | + * | [GroupFeedDelegate] | [GroupOperations] | Group feed sync, join/leave, roster changes | * * **What lives here (and why):** * - **Event routing** — each delegate exposes a `Flow` (backed by a `Channel`); * the `init` block collects both and dispatches cross-delegate calls (e.g. * feed-delegate's `DeltaSyncNeeded` → `eventStreamDelegate.performDeltaSync`, - * event-stream-delegate's `SyncFeedRequested` → `feedDelegate.syncFeed`). + * event-stream-delegate's `SyncFeedRequested` → [syncFeeds]). * All cross-delegate wiring is visible in one place. + * - **Feed composition** — the conversation list is [FeedSyncDelegate]'s DM feeds plus + * [GroupFeedDelegate]'s group feed, so every refresh trigger goes through [syncFeeds] + * rather than either delegate directly. See [refreshFeed]. * - **Lifecycle methods** — [onStart]/[onStop] are inherently cross-cutting * (stream connect/disconnect, heartbeat start/stop, active-chat save/restore). * - **Flow observers** — network reconnect re-syncing the chat feed. @@ -120,12 +124,11 @@ class RealChatCoordinator @Inject constructor( this.cluster.value = cluster feedDelegate.initialize(scope) eventStreamDelegate.initialize(scope) - feedDelegate.observeFeedFromDb() - feedDelegate.syncFeed() groupFeedDelegate.initialize(scope) - groupFeedDelegate.syncGroupFeed() + feedDelegate.observeFeedFromDb() + syncFeeds() eventStreamDelegate.open() - eventStreamDelegate.startHeartbeat { feedDelegate.syncFeed() } + eventStreamDelegate.startHeartbeat { syncFeeds() } } // endregion @@ -166,7 +169,7 @@ class RealChatCoordinator @Inject constructor( .onEach { event -> when (event) { is EventStreamDelegate.Event.SyncFeedRequested -> - feedDelegate.syncFeed() + syncFeeds() is EventStreamDelegate.Event.LoadMessages -> messagingDelegate.loadMessages(event.chatId) is EventStreamDelegate.Event.RosterChanged -> @@ -191,7 +194,7 @@ class RealChatCoordinator @Inject constructor( .debounce(1.seconds) .onEach { trace(tag = TAG, message = "Network connected, re-syncing chat feed", type = TraceType.Process) - feedDelegate.syncFeed() + syncFeeds() eventStreamDelegate.open() } .launchIn(scope) @@ -205,9 +208,9 @@ class RealChatCoordinator @Inject constructor( scope.launch { if (cluster.value != null) { trace(tag = TAG, message = "Lifecycle resumed, syncing chat feed", type = TraceType.Process) - feedDelegate.syncFeed() + syncFeeds() eventStreamDelegate.open() - eventStreamDelegate.startHeartbeat { feedDelegate.syncFeed() } + eventStreamDelegate.startHeartbeat { syncFeeds() } } } } @@ -223,6 +226,33 @@ class RealChatCoordinator @Inject constructor( // region ChatCoordinator + /** + * Overrides the [FeedOperations] delegation, which would otherwise refresh the DM half alone. + * + * The callers — a chat push, a contact payment, a tip payment — are asking for the + * conversation list, and none of them know which half a chat belongs to. + */ + override fun refreshFeed() { + syncFeeds() + } + + /** + * Fetches both halves of the conversation list. + * + * The list is two feeds behind one surface: [FeedSyncDelegate] fetches `CONTACT_DM` and + * `TIP_DM`, and a group's row comes from [GroupFeedDelegate] alone. Every trigger that + * re-syncs means "the list may be stale", which is never true of only one half, so pairing + * them is this class's job rather than something each trigger site remembers. + * + * Both are launch-and-return, and the delegates hold separate jobs, so the two fetches + * overlap rather than queue. A group feed that fails is traced and dropped by + * [GroupFeedDelegate.performGroupFeedSync]; it cannot take the DM list down with it. + */ + private fun syncFeeds() { + feedDelegate.syncFeed() + groupFeedDelegate.syncGroupFeed() + } + override suspend fun teardown() { eventStreamDelegate.stopHeartbeat() eventStreamDelegate.close() diff --git a/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/GroupChatRoutingTest.kt b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/GroupChatRoutingTest.kt index c54d907b31..94678ab472 100644 --- a/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/GroupChatRoutingTest.kt +++ b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/GroupChatRoutingTest.kt @@ -1,5 +1,6 @@ package com.flipcash.shared.chat +import androidx.lifecycle.LifecycleOwner import com.flipcash.app.core.dispatchers.TestDispatchers import com.flipcash.app.persistence.sources.ChatMemberDataSource import com.flipcash.app.persistence.sources.ChatMessageDataSource @@ -12,6 +13,7 @@ import com.flipcash.services.controllers.EventStreamingController import com.flipcash.services.models.UserProfile import com.flipcash.services.models.chat.ChatId import com.flipcash.services.models.chat.ChatMember +import com.flipcash.services.models.chat.ChatType import com.flipcash.services.models.chat.ChatUpdate import com.flipcash.services.models.chat.RosterChange import com.flipcash.services.models.chat.RosterSummary @@ -26,14 +28,17 @@ import com.flipcash.shared.chat.internal.delegates.GroupFeedDelegate import com.flipcash.shared.chat.internal.delegates.MessagingDelegate import com.getcode.utils.network.NetworkConnectivityListener import com.getcode.opencode.model.accounts.AccountCluster +import io.mockk.clearMocks import io.mockk.coEvery import io.mockk.coVerify import io.mockk.every import io.mockk.mockk +import io.mockk.verify import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.flow.receiveAsFlow import kotlinx.coroutines.test.TestCoroutineScheduler +import kotlinx.coroutines.test.TestScope import kotlinx.coroutines.test.runCurrent import kotlinx.coroutines.test.runTest import org.junit.Test @@ -41,23 +46,27 @@ import org.junit.runner.RunWith import org.robolectric.RobolectricTestRunner /** - * The group work reaches the rest of the app through exactly two seams: a roster change on the - * event stream, and the login hook. Both are the coordinator's, so both are checked here rather - * than in the delegates that do the work. + * The group work reaches the rest of the app through the coordinator and nowhere else: a roster + * change on the event stream, and every trigger that re-syncs the conversation list. All of them + * are checked here rather than in the delegates that do the work. + * + * The refresh triggers are the reason this file is not only about routing. A group's row comes + * from `GroupFeedDelegate` alone, so a trigger that reaches `FeedSyncDelegate` by itself leaves + * the group half of the list on whatever the login pass fetched — which is what these assert + * against. */ @OptIn(ExperimentalCoroutinesApi::class) @RunWith(RobolectricTestRunner::class) class GroupChatRoutingTest { - // Every test tears the coordinator down. The heartbeat the login hook starts is a `while - // (true)` on the test scheduler, so a test that leaves it running advances virtual time - // forever instead of finishing. - private val chatId = ChatId("11223344") private val selfId = listOf(1, 2, 3) private val chatUpdatesChannel = Channel(capacity = Channel.UNLIMITED) private val groupFeedDelegate = mockk(relaxed = true) + private val chatController = mockk(relaxed = true).also { + coEvery { it.getDmChatFeed(any(), any()) } returns Result.failure(RuntimeException("not needed")) + } private val testDispatchers = TestDispatchers(TestCoroutineScheduler()) @@ -70,9 +79,6 @@ class GroupChatRoutingTest { every { it.isConnected } returns true every { it.isStreamActive } returns true } - val chatController = mockk(relaxed = true).also { - coEvery { it.getDmChatFeed(any(), any()) } returns Result.failure(RuntimeException("not needed")) - } val metadataDataSource = mockk(relaxed = true) val messageDataSource = mockk(relaxed = true) val memberDataSource = mockk(relaxed = true) @@ -135,40 +141,90 @@ class GroupChatRoutingTest { rosterSummary = RosterSummary(memberCount = 13, version = 5), ) - @Test - fun `a roster change on the stream reaches the group delegate`() = runTest(testDispatchers.dispatcher) { + /** + * Runs [block] against a logged-in coordinator and tears it down afterwards, however it ends. + * + * The `finally` is the point. The heartbeat the login hook starts is a `while (true)` on the + * test scheduler, so a test that fails its assertion before reaching teardown hangs the run + * advancing virtual time instead of reporting the failure. + */ + private suspend fun TestScope.loggedIn(block: suspend (RealChatCoordinator) -> Unit) { val subject = coordinator() subject.onUserLoggedIn(mockk(relaxed = true)) runCurrent() + try { + block(subject) + } finally { + subject.teardown() + } + } - chatUpdatesChannel.send(ChatUpdate(chatId = chatId, rosterUpdates = listOf(change))) - runCurrent() + @Test + fun `a roster change on the stream reaches the group delegate`() = runTest(testDispatchers.dispatcher) { + loggedIn { + chatUpdatesChannel.send(ChatUpdate(chatId = chatId, rosterUpdates = listOf(change))) + runCurrent() - coVerify { groupFeedDelegate.applyRosterChanges(chatId, listOf(change)) } - subject.teardown() + coVerify { groupFeedDelegate.applyRosterChanges(chatId, listOf(change)) } + } } @Test fun `an update with no roster change does not reach the group delegate`() = runTest(testDispatchers.dispatcher) { - val subject = coordinator() - subject.onUserLoggedIn(mockk(relaxed = true)) - runCurrent() + loggedIn { + chatUpdatesChannel.send(ChatUpdate(chatId = chatId)) + runCurrent() - chatUpdatesChannel.send(ChatUpdate(chatId = chatId)) - runCurrent() - - coVerify(exactly = 0) { groupFeedDelegate.applyRosterChanges(any(), any()) } - subject.teardown() + coVerify(exactly = 0) { groupFeedDelegate.applyRosterChanges(any(), any()) } + } } @Test fun `logging in syncs the group feed`() = runTest(testDispatchers.dispatcher) { - val subject = coordinator() - subject.onUserLoggedIn(mockk(relaxed = true)) - runCurrent() + loggedIn { + verify(exactly = 1) { groupFeedDelegate.syncGroupFeed() } + } + } - coVerify { groupFeedDelegate.syncGroupFeed() } - subject.teardown() + @Test + fun `refreshing the feed syncs the group feed`() = runTest(testDispatchers.dispatcher) { + loggedIn { subject -> + clearMocks(groupFeedDelegate, answers = false) + + // What a chat push and both payment delegates call. + subject.refreshFeed() + runCurrent() + + verify(exactly = 1) { groupFeedDelegate.syncGroupFeed() } + } + } + + @Test + fun `refreshing the feed still syncs the DM feeds`() = runTest(testDispatchers.dispatcher) { + loggedIn { subject -> + clearMocks(chatController, answers = false) + + // [RealChatCoordinator.refreshFeed] overrides the FeedOperations delegation, so the + // half that used to be the only one running needs asserting as well as the half that + // did not. + subject.refreshFeed() + runCurrent() + + coVerify(exactly = 1) { chatController.getDmChatFeed(ChatType.CONTACT_DM, any()) } + coVerify(exactly = 1) { chatController.getDmChatFeed(ChatType.TIP_DM, any()) } + } + } + + @Test + fun `resuming from the background syncs the group feed`() = runTest(testDispatchers.dispatcher) { + loggedIn { subject -> + clearMocks(groupFeedDelegate, answers = false) + + subject.onStart(mockk(relaxed = true)) + runCurrent() + + verify(exactly = 1) { groupFeedDelegate.syncGroupFeed() } + } } @Test