Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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 |
Expand All @@ -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<Event>` (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.
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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 ->
Expand All @@ -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)
Expand All @@ -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() }
}
}
}
Expand All @@ -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()
Expand Down
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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
Expand All @@ -26,38 +28,45 @@ 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
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<Byte>(1, 2, 3)
private val chatUpdatesChannel = Channel<ChatUpdate>(capacity = Channel.UNLIMITED)

private val groupFeedDelegate = mockk<GroupFeedDelegate>(relaxed = true)
private val chatController = mockk<ChatController>(relaxed = true).also {
coEvery { it.getDmChatFeed(any(), any()) } returns Result.failure(RuntimeException("not needed"))
}

private val testDispatchers = TestDispatchers(TestCoroutineScheduler())

Expand All @@ -70,9 +79,6 @@ class GroupChatRoutingTest {
every { it.isConnected } returns true
every { it.isStreamActive } returns true
}
val chatController = mockk<ChatController>(relaxed = true).also {
coEvery { it.getDmChatFeed(any(), any()) } returns Result.failure(RuntimeException("not needed"))
}
val metadataDataSource = mockk<ChatMetadataDataSource>(relaxed = true)
val messageDataSource = mockk<ChatMessageDataSource>(relaxed = true)
val memberDataSource = mockk<ChatMemberDataSource>(relaxed = true)
Expand Down Expand Up @@ -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<AccountCluster>(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<AccountCluster>(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<AccountCluster>(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<LifecycleOwner>(relaxed = true))
runCurrent()

verify(exactly = 1) { groupFeedDelegate.syncGroupFeed() }
}
}

@Test
Expand Down
Loading