feat(chat): group chats in the conversation feed

apps/flipcash/shared/chat/build.gradle.kts

@@ -10,6 +10,7 @@ dependencies {
     testImplementation(kotlin("test"))
     testImplementation(libs.bundles.unit.testing)
     testImplementation(libs.robolectric)
+    testImplementation(libs.androidx.paging.testing)
 
     implementation(libs.bundles.kotlinx.serialization)
 

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

@@ -23,11 +23,26 @@ import kotlinx.coroutines.flow.StateFlow
  * Implemented by [com.flipcash.shared.chat.internal.delegates.FeedSyncDelegate].
  */
 interface FeedOperations {
-    /** Reactive list of [chatType] conversations, sorted by last activity. */
-    fun feed(chatType: ChatType): Flow<List<ChatSummary>>
+    /**
+     * Reactive list of conversations of any of [chatTypes], sorted by last activity.
+     *
+     * Several types at once because a group and a DM share one list on screen; asking for one
+     * type is still the common case and reads the same as it did.
+     */
+    fun feed(vararg chatTypes: ChatType): Flow<List<ChatSummary>>
+
+    /** Emits the number of conversations of any of [chatTypes] that have unread messages. */
+    fun observeUnreadConversations(vararg chatTypes: ChatType): Flow<Int>
 
-    /** Emits the number of [chatType] conversations that have unread messages. */
-    fun observeUnreadConversations(chatType: ChatType): Flow<Int>
+    /**
+     * The same conversations as [feed], paged.
+     *
+     * Room is the page source and [com.flipcash.app.persistence.sources.mediator.ChatFeedRemoteMediator]
+     * fetches more when the local rows run out, so the list is not bounded by what one sync put in
+     * memory. Unlike [feed] it does not read [ChatState.feed], so the two can be collected at once
+     * without one starving the other.
+     */
+    fun feedPaged(vararg chatTypes: ChatType): Flow<PagingData<ChatSummary>>
 
     /** Triggers a server-side feed sync. Safe to call redundantly. */
     fun refreshFeed()
@@ -229,7 +244,28 @@ interface MessagingOperations {
  *
  * @see com.flipcash.shared.chat.internal.RealChatCoordinator
  */
-interface ChatCoordinator : FeedOperations, EventStreamOperations, DmChatResolver, MessagingOperations {
+/**
+ * Membership of a group chat.
+ *
+ * Implemented by [com.flipcash.shared.chat.internal.delegates.GroupFeedDelegate].
+ */
+interface GroupOperations {
+    /** Joins [chatId], caching the chat so it is in the list before the next sync. */
+    suspend fun join(chatId: ChatId): Result<Unit>
+
+    /**
+     * Leaves [chatId]. The chat drops out of the list immediately and comes back if the call
+     * fails.
+     */
+    suspend fun leave(chatId: ChatId): Result<Unit>
+}
+
+interface ChatCoordinator :
+    FeedOperations,
+    EventStreamOperations,
+    DmChatResolver,
+    MessagingOperations,
+    GroupOperations {
     /** Full observable snapshot of chat state (feed, typing, reactions, active chat). */
     val state: StateFlow<ChatState>
 

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

@@ -0,0 +1,54 @@
+package com.flipcash.shared.chat.internal
+
+import com.flipcash.services.models.chat.ChatMember
+import com.flipcash.services.models.chat.ChatMetadata
+import com.flipcash.services.models.chat.ChatType
+import com.flipcash.services.models.chat.PointerType
+import com.flipcash.services.models.chat.isDmAddressable
+import com.getcode.opencode.model.core.ID
+
+/**
+ * The two rules a conversation row has to pass to be shown, kept apart from the delegate so the
+ * list and its paged form apply exactly the same ones.
+ */
+
+private fun isSelf(member: ChatMember, selfId: ID?, selfPhone: String?): Boolean =
+    member.userId == selfId ||
+        (selfPhone != null && member.userProfile.verifiedPhoneNumber == selfPhone)
+
+/**
+ * Whether [metadata] belongs in the conversation list.
+ *
+ * A DM is its counterparty — with no addressable other member there is no name to draw, so it is
+ * dropped. A group carries its own title and picture and is renderable on its own, however much of
+ * its roster the device happens to hold.
+ */
+internal fun isRenderable(metadata: ChatMetadata, selfId: ID?, selfPhone: String?): Boolean {
+    // Hidden chats (e.g. a DM the user blocked) must not surface in the feed.
+    if (metadata.isHidden) return false
+    if (metadata.type == ChatType.GROUP) return true
+
+    val otherMember = metadata.members.firstOrNull { !isSelf(it, selfId, selfPhone) } ?: return false
+    // Contact DMs require a resolvable identity (phone / display name). Tip DMs are identified by
+    // user id and have no phone by design, so they are never dropped.
+    return isDmAddressable(metadata.type, otherMember.userProfile)
+}
+
+/**
+ * 1 when the newest visible message is past the signed-in user's READ pointer and someone else
+ * sent it, 0 otherwise. The feed shows a splat rather than a number, so counting past one buys
+ * nothing.
+ *
+ * A group's roster is paged: the members the device holds are a slice, and the signed-in user's
+ * own row may not be in it. Falling back to a pointer of zero there would splat every group until
+ * the slice caught up, so an absent self row means "not known yet" for a group. On a DM both
+ * members always arrive together, so absence keeps its old meaning and the count is unchanged.
+ */
+internal fun unreadCount(metadata: ChatMetadata, selfId: ID?): Int {
+    val self = metadata.members.firstOrNull { it.userId == selfId }
+    if (self == null && metadata.type == ChatType.GROUP) return 0
+
+    val readPointer = self?.pointers?.firstOrNull { it.type == PointerType.READ }?.value ?: 0L
+    val lastMessage = metadata.lastMessage ?: return 0
+    return if (lastMessage.messageId > readPointer && lastMessage.senderId != selfId) 1 else 0
+}

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

@@ -5,6 +5,8 @@ package com.flipcash.shared.chat.internal
 import androidx.lifecycle.DefaultLifecycleObserver
 import androidx.lifecycle.LifecycleOwner
 import androidx.lifecycle.ProcessLifecycleOwner
+import com.flipcash.app.featureflags.FeatureFlag
+import com.flipcash.app.featureflags.FeatureFlagController
 import com.flipcash.libs.coroutines.DispatcherProvider
 import com.flipcash.services.models.chat.ChatId
 import com.flipcash.services.user.UserManager
@@ -13,9 +15,11 @@ import com.flipcash.shared.chat.ChatState
 import com.flipcash.shared.chat.DmChatResolver
 import com.flipcash.shared.chat.EventStreamOperations
 import com.flipcash.shared.chat.FeedOperations
+import com.flipcash.shared.chat.GroupOperations
 import com.flipcash.shared.chat.MessagingOperations
 import com.flipcash.shared.chat.internal.delegates.EventStreamDelegate
 import com.flipcash.shared.chat.internal.delegates.FeedSyncDelegate
+import com.flipcash.shared.chat.internal.delegates.GroupFeedDelegate
 import com.flipcash.shared.chat.internal.delegates.DmChatResolverDelegate
 import com.flipcash.shared.chat.internal.delegates.MessagingDelegate
 import com.getcode.opencode.model.accounts.AccountCluster
@@ -72,17 +76,20 @@ class RealChatCoordinator @Inject constructor(
     private val eventStreamDelegate: EventStreamDelegate,
     private val dmChatResolverDelegate: DmChatResolverDelegate,
     private val messagingDelegate: MessagingDelegate,
+    private val groupFeedDelegate: GroupFeedDelegate,
     private val stateHolder: ChatStateHolder,
     private val userManager: UserManager,
     private val networkObserver: NetworkConnectivityListener,
+    private val featureFlags: FeatureFlagController,
     private val dispatchers: DispatcherProvider,
 ) : ChatCoordinator,
     SessionListener,
     DefaultLifecycleObserver,
     FeedOperations by feedDelegate,
     EventStreamOperations by eventStreamDelegate,
     DmChatResolver by dmChatResolverDelegate,
-    MessagingOperations by messagingDelegate {
+    MessagingOperations by messagingDelegate,
+    GroupOperations by groupFeedDelegate {
 
     companion object {
         private const val TAG = "ChatCoordinator"
@@ -118,6 +125,12 @@ class RealChatCoordinator @Inject constructor(
         eventStreamDelegate.initialize(scope)
         feedDelegate.observeFeedFromDb()
         feedDelegate.syncFeed()
+        groupFeedDelegate.initialize(scope)
+        // The only gate on the branch. Everything behind it is built and tested; what it decides
+        // is whether the device asks the server for groups at all.
+        if (featureFlags.get(FeatureFlag.GroupChats)) {
+            groupFeedDelegate.syncGroupFeed()
+        }
         eventStreamDelegate.open()
         eventStreamDelegate.startHeartbeat { feedDelegate.syncFeed() }
     }
@@ -163,6 +176,16 @@ class RealChatCoordinator @Inject constructor(
                         feedDelegate.syncFeed()
                     is EventStreamDelegate.Event.LoadMessages ->
                         messagingDelegate.loadMessages(event.chatId)
+                    is EventStreamDelegate.Event.RosterChanged ->
+                        groupFeedDelegate.applyRosterChanges(event.chatId, event.changes)
+                }
+            }.launchIn(scope)
+
+        groupFeedDelegate.events
+            .onEach { event ->
+                when (event) {
+                    is GroupFeedDelegate.Event.LoadMessages ->
+                        messagingDelegate.loadMessages(event.chatId)
                 }
             }.launchIn(scope)
 
@@ -209,6 +232,7 @@ class RealChatCoordinator @Inject constructor(
         eventStreamDelegate.stopHeartbeat()
         eventStreamDelegate.close()
         feedDelegate.cancelJobs()
+        groupFeedDelegate.cancelJobs()
         networkObserverJob?.cancel()
         stateHolder.reset()
         eventStreamDelegate.clearAll()

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

@@ -0,0 +1,98 @@
+package com.flipcash.shared.chat.internal
+
+import com.flipcash.app.persistence.sources.ChatMemberDataSource
+import com.flipcash.app.persistence.sources.ChatMetadataDataSource
+import com.flipcash.services.controllers.ChatController
+import com.flipcash.services.models.chat.ChatId
+import com.flipcash.services.models.chat.RosterChange
+import com.getcode.utils.TraceType
+import com.getcode.utils.trace
+import javax.inject.Inject
+import javax.inject.Singleton
+
+/**
+ * Decides which roster changes apply to a chat, and repairs the roster when one cannot.
+ *
+ * The state it arbitrates is `chat_metadata.roster_version`, not a field of its own — a version
+ * held in memory would diverge from the row the moment a feed sync wrote a newer one. What this
+ * class owns is the rule, which is deliberately not the rule the event log follows:
+ *
+ * - **Newer wins, equal loses.** Every change carries the roster's summary *after* it, so a change
+ *   at or below the stored version has nothing to add. Re-delivery is normal on an at-least-once
+ *   stream.
+ * - **A skipped version is a refetch, not an apply.** There is no delta to fill the hole with, and
+ *   applying the change anyway would write a member count that does not describe the roster the
+ *   device holds. [ChatController.getChat] returns the current one instead.
+ *
+ * Kept out of [com.flipcash.shared.chat.EventSequenceTracker], whose contract is that
+ * sequence N+1 follows N and a hole is fetched. Folding a rule with the opposite repair into it
+ * would make that invariant conditional on the kind of update being handled.
+ */
+@Singleton
+class RosterStateHolder @Inject constructor(
+    private val chatController: ChatController,
+    private val metadataDataSource: ChatMetadataDataSource,
+    private val memberDataSource: ChatMemberDataSource,
+) {
+
+    /**
+     * Applies [changes] to [chatId]'s cached roster, oldest version first.
+     *
+     * Sorting matters: the changes arrive in a batch with no ordering guarantee, and out of order
+     * the older of two consecutive versions reads as a gap and costs a refetch for something the
+     * device is about to be told anyway.
+     */
+    suspend fun apply(chatId: ChatId, changes: List<RosterChange>) {
+        for (change in changes.sortedBy { it.rosterSummary.version }) {
+            val stored = metadataDataSource.getRosterVersion(chatId)
+            val incoming = change.rosterSummary.version
+
+            when {
+                incoming <= stored -> continue
+                incoming > stored + 1 -> refetch(chatId, stored, incoming)
+                else -> applyChange(chatId, change)
+            }
+        }
+    }
+
+    private suspend fun applyChange(chatId: ChatId, change: RosterChange) {
+        when (change) {
+            is RosterChange.MemberJoined -> memberDataSource.upsert(chatId, listOf(change.member))
+            is RosterChange.MemberLeft -> memberDataSource.deleteMember(chatId, change.userId)
+        }
+        metadataDataSource.updateRoster(
+            chatId = chatId,
+            memberCount = change.rosterSummary.memberCount,
+            rosterVersion = change.rosterSummary.version,
+        )
+    }
+
+    /**
+     * Replaces the cached roster with the server's current one.
+     *
+     * Members are merged rather than replaced. A group's member list is a page of a roster whose
+     * true size is `member_count`, so deleting everything absent from one response would drop
+     * members the device legitimately holds — including the senders a transcript needs to name.
+     * Departures come through [RosterChange.MemberLeft]; this is only here to get the count and
+     * the version back in step with the server.
+     */
+    private suspend fun refetch(chatId: ChatId, stored: Long, incoming: Long) {
+        trace(
+            tag = TAG,
+            message = "Roster version gap on $chatId: stored $stored, incoming $incoming",
+            type = TraceType.Silent,
+        )
+        val metadata = chatController.getChat(chatId).getOrElse {
+            // Leaving the stored version alone is what makes this retryable: the next change on
+            // this chat still reads as a gap, and tries again.
+            trace(tag = TAG, message = "Roster refetch failed for $chatId", type = TraceType.Error)
+            return
+        }
+        metadataDataSource.upsert(metadata)
+        memberDataSource.upsert(chatId, metadata.members)
+    }
+
+    private companion object {
+        const val TAG = "RosterStateHolder"
+    }
+}

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

@@ -16,6 +16,7 @@ import com.flipcash.services.models.chat.MessageContent
 import com.flipcash.services.models.chat.MetadataUpdate
 import com.flipcash.services.models.chat.ReactionSummary
 import com.flipcash.services.models.chat.ReactionUpdate
+import com.flipcash.services.models.chat.RosterChange
 import com.flipcash.services.models.chat.TypingNotification
 import com.flipcash.services.models.chat.TypingState
 import com.flipcash.services.models.GetDeltaError
@@ -97,6 +98,12 @@ class EventStreamDelegate @Inject constructor(
     sealed interface Event {
         data object SyncFeedRequested : Event
         data class LoadMessages(val chatId: ChatId) : Event
+
+        /**
+         * [chatId]'s roster changed. Emitted rather than applied here: the roster is the group
+         * delegate's, and this delegate has no business knowing what a membership means.
+         */
+        data class RosterChanged(val chatId: ChatId, val changes: List<RosterChange>) : Event
     }
 
     private val _events = Channel<Event>(Channel.UNLIMITED)
@@ -362,6 +369,15 @@ class EventStreamDelegate @Inject constructor(
             metadataDataSource.updateLastActivity(chatId, msg.timestamp.toEpochMilliseconds())
         }
 
+        // --- Roster changes ---
+
+        // Convergent like the reaction overlay below, and unlike the event log above: applied by
+        // version rather than gap-filled by sequence. Routed out because the version arbitration
+        // and the membership writes both live on the group side.
+        if (update.rosterUpdates.isNotEmpty()) {
+            _events.send(Event.RosterChanged(chatId, update.rosterUpdates))
+        }
+
         // --- Process reaction updates ---
 
         if (update.reactionUpdates.isNotEmpty()) {