Skip to content

Commit ab25ef7

Browse files
bmc08gtbmc08gt
committed
chore(chat): decompose ChatCoordinator into focused delegates
Split the 735-line monolithic ChatCoordinator into three delegates using Kotlin interface delegation, following the SessionController pattern: - FeedSyncDelegate: feed sync, DB observation, ChatSummary projection - EventStreamDelegate: event stream, applyUpdate, gap tracking, reactions, typing indicators - MessagingDelegate: per-chat messaging, read pointers, paging, notifications ChatCoordinator becomes an interface with sub-interfaces (FeedOperations, EventStreamOperations, MessagingOperations). RealChatCoordinator is the thin shell that wires delegates via Channel<Event> flows, handles lifecycle, and observes feature flags. Signed-off-by: Brandon McAnsh <git@bmcreations.dev>
1 parent 43dbecd commit ab25ef7

9 files changed

Lines changed: 1241 additions & 737 deletions

File tree

‎apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/ChatCoordinator.kt‎

Lines changed: 81 additions & 697 deletions
Large diffs are not rendered by default.

‎apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/inject/ChatModule.kt‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,20 +1,28 @@
11
package com.flipcash.shared.chat.inject
22

33
import com.flipcash.shared.chat.ChatCoordinator
4+
import com.flipcash.shared.chat.internal.RealChatCoordinator
45
import com.getcode.opencode.providers.SessionListener
56
import dagger.Binds
67
import dagger.Module
78
import dagger.hilt.InstallIn
89
import dagger.hilt.components.SingletonComponent
910
import dagger.multibindings.IntoSet
11+
import javax.inject.Singleton
1012

1113
@Module
1214
@InstallIn(SingletonComponent::class)
1315
abstract class ChatModule {
1416

17+
@Binds
18+
@Singleton
19+
abstract fun bindChatCoordinator(
20+
impl: RealChatCoordinator
21+
): ChatCoordinator
22+
1523
@Binds
1624
@IntoSet
1725
abstract fun bindSessionListener(
18-
coordinator: ChatCoordinator
26+
impl: RealChatCoordinator
1927
): SessionListener
2028
}

‎apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/ChatStateHolder.kt‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
package com.flipcash.shared.chat.internal
2+
3+
import com.flipcash.shared.chat.ChatState
4+
import kotlinx.coroutines.flow.MutableStateFlow
5+
import kotlinx.coroutines.flow.StateFlow
6+
import kotlinx.coroutines.flow.asStateFlow
7+
import kotlinx.coroutines.flow.update
8+
import javax.inject.Inject
9+
import javax.inject.Singleton
10+
11+
@Singleton
12+
class ChatStateHolder @Inject constructor() {
13+
private val _state = MutableStateFlow(ChatState())
14+
val state: StateFlow<ChatState> = _state.asStateFlow()
15+
val current: ChatState get() = _state.value
16+
17+
fun update(transform: (ChatState) -> ChatState) {
18+
_state.update(transform)
19+
}
20+
21+
fun reset() {
22+
_state.value = ChatState()
23+
}
24+
}

‎apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/RealChatCoordinator.kt‎

Lines changed: 227 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,227 @@
1+
@file:OptIn(ExperimentalCoroutinesApi::class, FlowPreview::class)
2+
3+
package com.flipcash.shared.chat.internal
4+
5+
import androidx.lifecycle.DefaultLifecycleObserver
6+
import androidx.lifecycle.LifecycleOwner
7+
import androidx.lifecycle.ProcessLifecycleOwner
8+
import com.flipcash.app.featureflags.FeatureFlag
9+
import com.flipcash.app.featureflags.FeatureFlagController
10+
import com.flipcash.libs.coroutines.DispatcherProvider
11+
import com.flipcash.services.models.chat.ChatId
12+
import com.flipcash.services.user.UserManager
13+
import com.flipcash.shared.chat.ChatCoordinator
14+
import com.flipcash.shared.chat.ChatState
15+
import com.flipcash.shared.chat.EventStreamOperations
16+
import com.flipcash.shared.chat.FeedOperations
17+
import com.flipcash.shared.chat.MessagingOperations
18+
import com.flipcash.shared.chat.internal.delegates.EventStreamDelegate
19+
import com.flipcash.shared.chat.internal.delegates.FeedSyncDelegate
20+
import com.flipcash.shared.chat.internal.delegates.MessagingDelegate
21+
import com.getcode.opencode.model.accounts.AccountCluster
22+
import com.getcode.opencode.providers.SessionListener
23+
import com.getcode.utils.TraceType
24+
import com.getcode.utils.network.NetworkConnectivityListener
25+
import com.getcode.utils.trace
26+
import kotlinx.coroutines.CoroutineScope
27+
import kotlinx.coroutines.ExperimentalCoroutinesApi
28+
import kotlinx.coroutines.FlowPreview
29+
import kotlinx.coroutines.Job
30+
import kotlinx.coroutines.SupervisorJob
31+
import kotlinx.coroutines.flow.MutableStateFlow
32+
import kotlinx.coroutines.flow.StateFlow
33+
import kotlinx.coroutines.flow.combine
34+
import kotlinx.coroutines.flow.debounce
35+
import kotlinx.coroutines.flow.distinctUntilChanged
36+
import kotlinx.coroutines.flow.filter
37+
import kotlinx.coroutines.flow.filterNotNull
38+
import kotlinx.coroutines.flow.flatMapLatest
39+
import kotlinx.coroutines.flow.launchIn
40+
import kotlinx.coroutines.flow.map
41+
import kotlinx.coroutines.flow.onEach
42+
import kotlinx.coroutines.launch
43+
import javax.inject.Inject
44+
import javax.inject.Singleton
45+
import kotlin.time.Duration.Companion.seconds
46+
47+
/**
48+
* Thin orchestration shell that implements [ChatCoordinator] by composing three
49+
* focused delegates via Kotlin `by` interface delegation:
50+
*
51+
* | Delegate | Interface | Responsibility |
52+
* |----------|-----------|----------------|
53+
* | [FeedSyncDelegate] | [FeedOperations] | Feed sync, DB observation, unread counts |
54+
* | [EventStreamDelegate] | [EventStreamOperations] | Event stream, real-time updates, gap-aware sequencing, reactions, typing |
55+
* | [MessagingDelegate] | [MessagingOperations] | Per-chat send/receive, read pointers, paging, notifications |
56+
*
57+
* **What lives here (and why):**
58+
* - **Event routing** — each delegate exposes a `Flow<Event>` (backed by a `Channel`);
59+
* the `init` block collects both and dispatches cross-delegate calls (e.g.
60+
* feed-delegate's `DeltaSyncNeeded` → `eventStreamDelegate.performDeltaSync`,
61+
* event-stream-delegate's `SyncFeedRequested` → `feedDelegate.syncFeed`).
62+
* All cross-delegate wiring is visible in one place.
63+
* - **Lifecycle methods** — [onStart]/[onStop] are inherently cross-cutting
64+
* (stream connect/disconnect, heartbeat start/stop, active-chat save/restore).
65+
* - **Flow observers** — network reconnect and feature-flag transitions that
66+
* gate whether the chat subsystem should be active.
67+
*
68+
* Delegates require [initialize] with a shared [CoroutineScope] before use;
69+
* this happens in [onUserLoggedIn].
70+
*/
71+
@Singleton
72+
class RealChatCoordinator @Inject constructor(
73+
private val feedDelegate: FeedSyncDelegate,
74+
private val eventStreamDelegate: EventStreamDelegate,
75+
private val messagingDelegate: MessagingDelegate,
76+
private val stateHolder: ChatStateHolder,
77+
private val userManager: UserManager,
78+
private val featureFlags: FeatureFlagController,
79+
networkObserver: NetworkConnectivityListener,
80+
dispatchers: DispatcherProvider,
81+
) : ChatCoordinator,
82+
SessionListener,
83+
DefaultLifecycleObserver,
84+
FeedOperations by feedDelegate,
85+
EventStreamOperations by eventStreamDelegate,
86+
MessagingOperations by messagingDelegate {
87+
88+
companion object {
89+
private const val TAG = "ChatCoordinator"
90+
}
91+
92+
private val supervisorJob = SupervisorJob()
93+
private val scope = CoroutineScope(dispatchers.IO + supervisorJob)
94+
private val cluster = MutableStateFlow<AccountCluster?>(null)
95+
private var flagObserverJob: Job? = null
96+
private var networkObserverJob: Job? = null
97+
private var backgroundedActiveChat: ChatId? = null
98+
99+
override val state: StateFlow<ChatState>
100+
get() = stateHolder.state
101+
102+
// region SessionListener
103+
104+
override suspend fun onUserLoggedIn(cluster: AccountCluster) {
105+
trace(tag = TAG, message = "User logged in, hydrating chat", type = TraceType.User)
106+
this.cluster.value = cluster
107+
feedDelegate.initialize(scope)
108+
eventStreamDelegate.initialize(scope)
109+
feedDelegate.observeFeedFromDb()
110+
feedDelegate.syncFeed()
111+
eventStreamDelegate.open()
112+
eventStreamDelegate.startHeartbeat { feedDelegate.syncFeed() }
113+
observeFeatureFlag()
114+
}
115+
116+
// endregion
117+
118+
// region Lifecycle
119+
120+
init {
121+
ProcessLifecycleOwner.get().lifecycle.addObserver(this)
122+
123+
feedDelegate.events
124+
.onEach { event ->
125+
when (event) {
126+
is FeedSyncDelegate.Event.LoadMessages ->
127+
messagingDelegate.loadMessages(event.chatId)
128+
is FeedSyncDelegate.Event.DeltaSyncNeeded ->
129+
eventStreamDelegate.performDeltaSync(event.chatId)
130+
}
131+
}.launchIn(scope)
132+
133+
eventStreamDelegate.events
134+
.onEach { event ->
135+
when (event) {
136+
is EventStreamDelegate.Event.SyncFeedRequested ->
137+
feedDelegate.syncFeed()
138+
is EventStreamDelegate.Event.LoadMessages ->
139+
messagingDelegate.loadMessages(event.chatId)
140+
}
141+
}.launchIn(scope)
142+
143+
networkObserverJob = cluster.filterNotNull()
144+
.flatMapLatest { networkObserver.state }
145+
.distinctUntilChanged()
146+
.filter { it.connected }
147+
.debounce(1.seconds)
148+
.onEach {
149+
if (!isChatEnabled()) return@onEach
150+
trace(tag = TAG, message = "Network connected, re-syncing chat feed", type = TraceType.Process)
151+
feedDelegate.syncFeed()
152+
eventStreamDelegate.open()
153+
}
154+
.launchIn(scope)
155+
}
156+
157+
override fun onStart(owner: LifecycleOwner) {
158+
backgroundedActiveChat?.let {
159+
messagingDelegate.setActiveChatId(it)
160+
backgroundedActiveChat = null
161+
}
162+
scope.launch {
163+
if (cluster.value != null && isChatEnabled()) {
164+
trace(tag = TAG, message = "Lifecycle resumed, syncing chat feed", type = TraceType.Process)
165+
feedDelegate.syncFeed()
166+
eventStreamDelegate.open()
167+
eventStreamDelegate.startHeartbeat { feedDelegate.syncFeed() }
168+
}
169+
}
170+
}
171+
172+
override fun onStop(owner: LifecycleOwner) {
173+
backgroundedActiveChat = stateHolder.current.activeChat
174+
messagingDelegate.setActiveChatId(null)
175+
eventStreamDelegate.stopHeartbeat()
176+
eventStreamDelegate.close()
177+
}
178+
179+
// endregion
180+
181+
// region ChatCoordinator
182+
183+
override suspend fun reset() {
184+
eventStreamDelegate.stopHeartbeat()
185+
eventStreamDelegate.close()
186+
feedDelegate.cancelJobs()
187+
flagObserverJob?.cancel()
188+
networkObserverJob?.cancel()
189+
stateHolder.reset()
190+
eventStreamDelegate.clearAll()
191+
cluster.value = null
192+
messagingDelegate.clear()
193+
supervisorJob.cancel()
194+
trace(tag = TAG, message = "reset complete", type = TraceType.Process)
195+
}
196+
197+
// endregion
198+
199+
// region Internal
200+
201+
private suspend fun isChatEnabled(): Boolean {
202+
val featureFlag = featureFlags.get(FeatureFlag.PhoneNumberSend)
203+
val serverFlag = userManager.state.value.flags?.enablePhoneNumberSend == true
204+
return featureFlag || serverFlag
205+
}
206+
207+
private fun observeFeatureFlag() {
208+
flagObserverJob?.cancel()
209+
flagObserverJob = combine(
210+
featureFlags.observe(FeatureFlag.PhoneNumberSend),
211+
userManager.state.map { it.flags?.enablePhoneNumberSend == true },
212+
) { featureFlag, serverFlag -> featureFlag || serverFlag }
213+
.distinctUntilChanged()
214+
.filter { it }
215+
.onEach {
216+
if (cluster.value != null) {
217+
trace(tag = TAG, message = "Chat feature enabled, syncing feed", type = TraceType.Process)
218+
feedDelegate.syncFeed()
219+
eventStreamDelegate.open()
220+
eventStreamDelegate.startHeartbeat { feedDelegate.syncFeed() }
221+
}
222+
}
223+
.launchIn(scope)
224+
}
225+
226+
// endregion
227+
}

0 commit comments

Comments
 (0)