feat(chat): group chat RPCs and roster updates

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.RosterUpdate
 import com.flipcash.services.models.chat.TypingNotification
 import com.flipcash.services.models.chat.TypingState
 import com.flipcash.services.models.GetDeltaError
@@ -103,6 +104,13 @@ class EventStreamDelegate @Inject constructor(
     val events: Flow<Event> = _events.receiveAsFlow()
 
     private val sequenceTracker = EventSequenceTracker()
+
+    // In-memory only: Room's chat_metadata has no roster-version column today (rosterSummary is
+    // only ever real when hydrated fresh from the network), so there is nothing durable to key
+    // this off. A cold start always applies the first roster update it sees for a chat, since any
+    // real version is >= 1 > the 0L default here — consistent with ChatMetadata's own
+    // reconstruct-from-Room default.
+    private val rosterVersions = mutableMapOf<ChatId, Long>()
     private var scope: CoroutineScope? = null
     private var eventStreamCollectJob: Job? = null
     private var heartbeatJob: Job? = null
@@ -299,7 +307,7 @@ class EventStreamDelegate @Inject constructor(
 
         trace(
             tag = TAG,
-            message = "applyUpdate: chatId=$chatId, messages=${resolvedMessages.size}, events=${update.events.size}, pointers=${update.pointerUpdates.size}, reactions=${update.reactionUpdates.size}, typing=${update.typingNotifications.size}",
+            message = "applyUpdate: chatId=$chatId, messages=${resolvedMessages.size}, events=${update.events.size}, pointers=${update.pointerUpdates.size}, reactions=${update.reactionUpdates.size}, typing=${update.typingNotifications.size}, roster=${update.rosterUpdates.size}",
             type = TraceType.Process,
         )
 
@@ -376,6 +384,43 @@ class EventStreamDelegate @Inject constructor(
             }
         }
 
+        // --- Process roster updates ---
+        //
+        // Durable (Room), not an overlay: unlike reactions/typing, a roster change mutates the
+        // chat list / membership itself, so it has to land in the same store SyncFeedRequested
+        // and getGroupChatFeed hydrate from. The version guard lives only in memory
+        // (`rosterVersions`) rather than a Room column — see the field's doc comment — so a
+        // process restart re-applies the first roster update it sees per chat, which is safe:
+        // the same upsert/delete calls below are already idempotent.
+
+        for (rosterUpdate in update.rosterUpdates) {
+            val incomingVersion = rosterUpdate.rosterSummary.version
+            val trackedVersion = rosterVersions[chatId] ?: 0L
+            if (incomingVersion <= trackedVersion) {
+                continue
+            }
+            rosterVersions[chatId] = incomingVersion
+
+            when (rosterUpdate) {
+                is RosterUpdate.MemberJoined -> {
+                    memberDataSource.upsert(chatId, listOf(rosterUpdate.member))
+                    val metadata = rosterUpdate.metadata
+                    if (metadata != null) {
+                        metadataDataSource.upsert(metadata)
+                        memberDataSource.upsert(chatId, metadata.members)
+                    }
+                }
+                is RosterUpdate.MemberLeft -> {
+                    if (rosterUpdate.userId == userManager.accountId) {
+                        metadataDataSource.delete(chatId)
+                        memberDataSource.deleteForChat(chatId)
+                    } else {
+                        memberDataSource.removeMember(chatId, rosterUpdate.userId)
+                    }
+                }
+            }
+        }
+
         // --- Eagerly update token balance + count receipts for analytics ---
 
         val selfId = userManager.accountId

apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ChatCoordinatorEventsTest.kt

@@ -16,7 +16,13 @@ import com.flipcash.services.models.chat.ChatMutation
 import com.flipcash.services.models.chat.ChatUpdate
 import com.flipcash.services.models.chat.Emoji
 import com.flipcash.services.models.chat.MessageContent
+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.ReactionUpdate
+import com.flipcash.services.models.chat.RosterSummary
+import com.flipcash.services.models.chat.RosterUpdate
+import com.flipcash.services.models.UserProfile
 import com.flipcash.shared.chat.internal.ChatIdGenerator
 import com.flipcash.shared.chat.internal.ChatStateHolder
 import com.flipcash.shared.chat.internal.RealChatCoordinator
@@ -60,6 +66,7 @@ class ChatCoordinatorEventsTest {
 
     private lateinit var metadataDataSource: ChatMetadataDataSource
     private lateinit var messageDataSource: ChatMessageDataSource
+    private lateinit var memberDataSource: ChatMemberDataSource
     private lateinit var coordinator: RealChatCoordinator
     private lateinit var testDispatchers: TestDispatchers
 
@@ -77,7 +84,7 @@ class ChatCoordinatorEventsTest {
 
         metadataDataSource = mockk(relaxed = true)
         messageDataSource = mockk(relaxed = true)
-        val memberDataSource = mockk<ChatMemberDataSource>(relaxed = true)
+        memberDataSource = mockk(relaxed = true)
         val messagingController = mockk<ChatMessagingController>(relaxed = true)
 
         testDispatchers = TestDispatchers(TestCoroutineScheduler())
@@ -157,6 +164,21 @@ class ChatCoordinatorEventsTest {
         mutations = listOf(ChatMutation.MessageSent(message)),
     )
 
+    private fun chatMember(userId: List<Byte>) = ChatMember(
+        userId = userId,
+        userProfile = UserProfile.Empty,
+        pointers = emptyList(),
+    )
+
+    private fun chatMetadata(rosterVersion: Long) = ChatMetadata(
+        chatId = chatId,
+        type = ChatType.GROUP,
+        members = listOf(chatMember(selfId)),
+        lastMessage = null,
+        lastActivity = Instant.fromEpochSeconds(1000),
+        rosterSummary = RosterSummary(memberCount = 1, version = rosterVersion),
+    )
+
     private suspend fun triggerCollection() {
         coordinator.onUserLoggedIn(mockk(relaxed = true))
     }
@@ -445,6 +467,139 @@ class ChatCoordinatorEventsTest {
 
     // endregion
 
+    // region Roster updates
+
+    @Test
+    fun `roster update with version not greater than tracked is dropped`() = runTest(testDispatchers.dispatcher) {
+        triggerCollection()
+
+        // First update establishes version 2.
+        chatUpdatesChannel.send(ChatUpdate(
+            chatId = chatId,
+            rosterUpdates = listOf(
+                RosterUpdate.MemberJoined(
+                    rosterSummary = RosterSummary(memberCount = 2, version = 2),
+                    member = chatMember(otherId),
+                    metadata = null,
+                ),
+            ),
+        ))
+        advanceTimeBy(500.milliseconds)
+        runCurrent()
+
+        coVerify(exactly = 1) { memberDataSource.upsert(chatId, listOf(chatMember(otherId))) }
+
+        // Stale update at version 1 (<= tracked version 2) must be dropped.
+        chatUpdatesChannel.send(ChatUpdate(
+            chatId = chatId,
+            rosterUpdates = listOf(
+                RosterUpdate.MemberJoined(
+                    rosterSummary = RosterSummary(memberCount = 3, version = 1),
+                    member = chatMember(otherId),
+                    metadata = null,
+                ),
+            ),
+        ))
+        advanceTimeBy(500.milliseconds)
+        runCurrent()
+
+        // Still only the one upsert from the first, accepted update.
+        coVerify(exactly = 1) { memberDataSource.upsert(chatId, listOf(chatMember(otherId))) }
+        coordinator.teardown()
+    }
+
+    @Test
+    fun `recipient join inserts chat metadata and members`() = runTest(testDispatchers.dispatcher) {
+        triggerCollection()
+
+        val metadata = chatMetadata(rosterVersion = 1)
+        chatUpdatesChannel.send(ChatUpdate(
+            chatId = chatId,
+            rosterUpdates = listOf(
+                RosterUpdate.MemberJoined(
+                    rosterSummary = metadata.rosterSummary,
+                    member = chatMember(selfId),
+                    metadata = metadata,
+                ),
+            ),
+        ))
+        advanceTimeBy(500.milliseconds)
+        runCurrent()
+
+        coVerify { memberDataSource.upsert(chatId, listOf(chatMember(selfId))) }
+        coVerify { metadataDataSource.upsert(metadata) }
+        coVerify { memberDataSource.upsert(chatId, metadata.members) }
+        coordinator.teardown()
+    }
+
+    @Test
+    fun `non-recipient join upserts member only`() = runTest(testDispatchers.dispatcher) {
+        triggerCollection()
+
+        chatUpdatesChannel.send(ChatUpdate(
+            chatId = chatId,
+            rosterUpdates = listOf(
+                RosterUpdate.MemberJoined(
+                    rosterSummary = RosterSummary(memberCount = 2, version = 1),
+                    member = chatMember(otherId),
+                    metadata = null,
+                ),
+            ),
+        ))
+        advanceTimeBy(500.milliseconds)
+        runCurrent()
+
+        coVerify { memberDataSource.upsert(chatId, listOf(chatMember(otherId))) }
+        coVerify(exactly = 0) { metadataDataSource.upsert(any<ChatMetadata>()) }
+        coordinator.teardown()
+    }
+
+    @Test
+    fun `recipient leave removes chat metadata and members`() = runTest(testDispatchers.dispatcher) {
+        triggerCollection()
+
+        chatUpdatesChannel.send(ChatUpdate(
+            chatId = chatId,
+            rosterUpdates = listOf(
+                RosterUpdate.MemberLeft(
+                    rosterSummary = RosterSummary(memberCount = 0, version = 1),
+                    userId = selfId,
+                ),
+            ),
+        ))
+        advanceTimeBy(500.milliseconds)
+        runCurrent()
+
+        coVerify { metadataDataSource.delete(chatId) }
+        coVerify { memberDataSource.deleteForChat(chatId) }
+        coVerify(exactly = 0) { memberDataSource.removeMember(chatId, selfId) }
+        coordinator.teardown()
+    }
+
+    @Test
+    fun `non-recipient leave removes member row only`() = runTest(testDispatchers.dispatcher) {
+        triggerCollection()
+
+        chatUpdatesChannel.send(ChatUpdate(
+            chatId = chatId,
+            rosterUpdates = listOf(
+                RosterUpdate.MemberLeft(
+                    rosterSummary = RosterSummary(memberCount = 1, version = 1),
+                    userId = otherId,
+                ),
+            ),
+        ))
+        advanceTimeBy(500.milliseconds)
+        runCurrent()
+
+        coVerify { memberDataSource.removeMember(chatId, otherId) }
+        coVerify(exactly = 0) { metadataDataSource.delete(any()) }
+        coVerify(exactly = 0) { memberDataSource.deleteForChat(any()) }
+        coordinator.teardown()
+    }
+
+    // endregion
+
     // region Session lifecycle
 
     @Test

apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMemberDao.kt

@@ -107,6 +107,9 @@ interface ChatMemberDao {
     @Query("DELETE FROM chat_members WHERE chat_id_hex = :chatIdHex")
     suspend fun deleteForChat(chatIdHex: String)
 
+    @Query("DELETE FROM chat_members WHERE chat_id_hex = :chatIdHex AND user_id_hex = :userIdHex")
+    suspend fun deleteMember(chatIdHex: String, userIdHex: String)
+
     /** Drops the members of [chatIdHex] that are no longer in [keepUserIdHexes]. */
     @Query(
         "DELETE FROM chat_members WHERE chat_id_hex = :chatIdHex " +

apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMetadataDao.kt

@@ -101,6 +101,9 @@ interface ChatMetadataDao {
     @Query("UPDATE chat_metadata SET is_hidden = :hidden WHERE chat_id_hex = :chatIdHex")
     suspend fun updateHidden(chatIdHex: String, hidden: Boolean)
 
+    @Query("DELETE FROM chat_metadata WHERE chat_id_hex = :chatIdHex")
+    suspend fun deleteById(chatIdHex: String)
+
     @Query("DELETE FROM chat_metadata")
     suspend fun deleteAll()
 }

apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/ChatMemberDataSource.kt

@@ -113,6 +113,10 @@ class ChatMemberDataSource @Inject constructor(
         db?.chatMemberDao()?.deleteForChat(mapper.chatIdHex(chatId))
     }
 
+    suspend fun removeMember(chatId: ChatId, userId: ID) {
+        db?.chatMemberDao()?.deleteMember(mapper.chatIdHex(chatId), mapper.userIdHex(userId))
+    }
+
     suspend fun clear() {
         db?.chatMemberDao()?.deleteAll()
     }

apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/ChatMetadataDataSource.kt

@@ -73,6 +73,10 @@ class ChatMetadataDataSource @Inject constructor(
     suspend fun exists(chatId: ChatId): Boolean =
         db?.chatMetadataDao()?.getById(mapper.chatIdHex(chatId)) != null
 
+    suspend fun delete(chatId: ChatId) {
+        db?.chatMetadataDao()?.deleteById(mapper.chatIdHex(chatId))
+    }
+
     fun toMetadata(
         entity: ChatMetadataEntity,
         members: List<ChatMember>,

gradle/libs.versions.toml

@@ -69,7 +69,7 @@ protovalidate-kt = "0.1.2"
 # 0.3.0 is the first release of either package to ship R8 keep rules for its generated
 # messages, which is what lets proguard-rules.pro drop its own.
 ocp-client-protocol = "0.4.0"
-flipcash2-client-protocol = "0.6.0"
+flipcash2-client-protocol = "0.7.0"
 
 # The Android port is the ONLY libphonenumber this app depends on, deliberately. Google's
 # `com.googlecode` artifact used to sit alongside it; the two ship separate copies of the metadata,

services/flipcash/src/main/kotlin/com/flipcash/services/controllers/ChatController.kt

@@ -1,9 +1,11 @@
 package com.flipcash.services.controllers
 
 import com.flipcash.services.models.QueryOptions
+import com.flipcash.services.models.chat.BlobId
 import com.flipcash.services.models.chat.ChatFeedPage
 import com.flipcash.services.models.chat.ChatId
 import com.flipcash.services.models.chat.ChatMetadata
+import com.flipcash.services.models.chat.ChatRules
 import com.flipcash.services.models.chat.ChatType
 import com.flipcash.services.repository.ChatRepository
 import com.flipcash.services.user.UserManager
@@ -31,4 +33,38 @@ class ChatController @Inject constructor(
 
         return repository.getDmChatFeed(owner, queryOptions, chatType)
     }
+
+    suspend fun getGroupChatFeed(
+        queryOptions: QueryOptions = QueryOptions(),
+    ): Result<ChatFeedPage> {
+        val owner = userManager.accountCluster?.authority?.keyPair
+            ?: return Result.failure(Throwable("No account cluster in UserManager"))
+
+        return repository.getGroupChatFeed(owner, queryOptions)
+    }
+
+    suspend fun startChat(
+        title: String,
+        picture: BlobId? = null,
+        rules: ChatRules? = null,
+    ): Result<ChatMetadata> {
+        val owner = userManager.accountCluster?.authority?.keyPair
+            ?: return Result.failure(Throwable("No account cluster in UserManager"))
+
+        return repository.startChat(owner, title, picture, rules)
+    }
+
+    suspend fun joinChat(chatId: ChatId): Result<ChatMetadata> {
+        val owner = userManager.accountCluster?.authority?.keyPair
+            ?: return Result.failure(Throwable("No account cluster in UserManager"))
+
+        return repository.joinChat(owner, chatId)
+    }
+
+    suspend fun leaveChat(chatId: ChatId): Result<Unit> {
+        val owner = userManager.accountCluster?.authority?.keyPair
+            ?: return Result.failure(Throwable("No account cluster in UserManager"))
+
+        return repository.leaveChat(owner, chatId)
+    }
 }