feat(chat): scaffold GetRoster and EditChat

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

@@ -365,6 +365,16 @@ class EventStreamDelegate @Inject constructor(
                     // and a mute and the unmute after it can land either way round.
                     metadataDataSource.updateViewerState(chatId, metaUpdate.viewerState)
                 }
+                is MetadataUpdate.TitleChanged -> {
+                    // Best-effort, applied as received (unversioned, unlike the roster and
+                    // viewer-state overlays above): a client that suspects a miss refetches via
+                    // GetChat rather than waiting on a later update to correct it.
+                    metadataDataSource.updateTitle(chatId, metaUpdate.newTitle)
+                }
+                is MetadataUpdate.PictureChanged -> {
+                    // Same best-effort contract as TitleChanged above.
+                    metadataDataSource.updatePicture(chatId, metaUpdate.newPicture)
+                }
             }
         }
 

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

@@ -0,0 +1,217 @@
+package com.flipcash.shared.chat
+
+import com.flipcash.app.core.dispatchers.TestDispatchers
+import com.flipcash.app.persistence.sources.ChatMemberDataSource
+import com.flipcash.app.persistence.sources.ChatMessageDataSource
+import com.flipcash.app.persistence.sources.ChatMetadataDataSource
+import com.flipcash.app.persistence.sources.ContactDataSource
+import com.flipcash.services.controllers.ChatController
+import com.flipcash.services.controllers.ChatMessagingController
+import com.flipcash.services.controllers.EventStreamingController
+import com.flipcash.services.models.chat.ChatId
+import com.flipcash.services.models.chat.ChatUpdate
+import com.flipcash.services.models.chat.MediaItem
+import com.flipcash.services.models.chat.MetadataUpdate
+import com.flipcash.services.user.UserManager
+import com.flipcash.shared.chat.internal.ChatIdGenerator
+import com.flipcash.shared.chat.internal.ChatStateHolder
+import com.flipcash.shared.chat.internal.RealChatCoordinator
+import com.flipcash.shared.chat.internal.delegates.DmChatResolverDelegate
+import com.flipcash.shared.chat.internal.delegates.EventStreamDelegate
+import com.flipcash.shared.chat.internal.delegates.FeedSyncDelegate
+import com.flipcash.shared.chat.internal.delegates.MessagingDelegate
+import com.getcode.utils.network.NetworkConnectivityListener
+import io.mockk.coEvery
+import io.mockk.coVerify
+import io.mockk.every
+import io.mockk.mockk
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.flow.receiveAsFlow
+import kotlinx.coroutines.test.TestCoroutineScheduler
+import kotlinx.coroutines.test.advanceTimeBy
+import kotlinx.coroutines.test.runCurrent
+import kotlinx.coroutines.test.runTest
+import org.junit.Before
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.robolectric.RobolectricTestRunner
+import kotlin.time.Duration.Companion.milliseconds
+
+/**
+ * The title/picture arm of the event stream: `TitleChanged` and `PictureChanged` are applied
+ * unconditionally, unlike the versioned [MetadataUpdate.ViewerStateChanged] handling covered in
+ * [ViewerStateStreamTest] — neither carries a version, so there is nothing for the delegate to
+ * gate the write on. Delivery is best-effort, by contract; a client that suspects a miss is
+ * expected to refetch via `GetChat` rather than relying on the stream to correct itself.
+ */
+@OptIn(ExperimentalCoroutinesApi::class)
+@RunWith(RobolectricTestRunner::class)
+class ChatMetadataEditStreamTest {
+
+    private val selfId = listOf<Byte>(1, 2, 3)
+    private val chatId = ChatId("aabbccdd")
+
+    private val chatUpdatesChannel = Channel<ChatUpdate>(capacity = Channel.UNLIMITED)
+
+    private lateinit var metadataDataSource: ChatMetadataDataSource
+    private lateinit var coordinator: RealChatCoordinator
+    private lateinit var testDispatchers: TestDispatchers
+
+    @Before
+    fun setUp() {
+        val userManager = mockk<UserManager>(relaxed = true)
+        every { userManager.accountId } returns selfId
+        val eventStreamingController = mockk<EventStreamingController>(relaxed = true)
+        every { eventStreamingController.chatUpdates } returns chatUpdatesChannel.receiveAsFlow()
+        every { eventStreamingController.isConnected } returns true
+        every { eventStreamingController.isStreamActive } returns true
+
+        val chatController = mockk<ChatController>(relaxed = true)
+        coEvery { chatController.getDmChatFeed(any(), any()) } returns
+            Result.failure(RuntimeException("not needed"))
+
+        testDispatchers = TestDispatchers(TestCoroutineScheduler())
+
+        val stateHolder = ChatStateHolder()
+        val memberDataSource = mockk<ChatMemberDataSource>(relaxed = true)
+        val messagingController = mockk<ChatMessagingController>(relaxed = true)
+        metadataDataSource = mockk(relaxed = true)
+        val messageDataSource = mockk<ChatMessageDataSource>(relaxed = true)
+
+        val feedDelegate = FeedSyncDelegate(
+            chatController = chatController,
+            metadataDataSource = metadataDataSource,
+            messageDataSource = messageDataSource,
+            memberDataSource = memberDataSource,
+            stateHolder = stateHolder,
+            userManager = userManager,
+        )
+
+        val eventStreamDelegate = EventStreamDelegate(
+            eventStreamingController = eventStreamingController,
+            messagingController = messagingController,
+            metadataDataSource = metadataDataSource,
+            messageDataSource = messageDataSource,
+            memberDataSource = memberDataSource,
+            tokenCoordinator = mockk(relaxed = true),
+            userManager = userManager,
+            stateHolder = stateHolder,
+            analytics = mockk(relaxed = true),
+            exchange = mockk(relaxed = true),
+        )
+
+        val messagingDelegate = MessagingDelegate(
+            chatController = chatController,
+            messagingController = messagingController,
+            metadataDataSource = metadataDataSource,
+            messageDataSource = messageDataSource,
+            memberDataSource = memberDataSource,
+            notificationManager = mockk(relaxed = true),
+            userManager = userManager,
+            stateHolder = stateHolder,
+            analytics = mockk(relaxed = true),
+            senderResolver = mockk(relaxed = true),
+        )
+
+        val dmChatResolverDelegate = DmChatResolverDelegate(
+            chatIdGenerator = ChatIdGenerator(),
+            userManager = userManager,
+            contactDataSource = mockk<ContactDataSource>(relaxed = true),
+            memberDataSource = memberDataSource,
+        )
+
+        coordinator = RealChatCoordinator(
+            feedDelegate = feedDelegate,
+            eventStreamDelegate = eventStreamDelegate,
+            dmChatResolverDelegate = dmChatResolverDelegate,
+            messagingDelegate = messagingDelegate,
+            stateHolder = stateHolder,
+            draftStore = mockk<ChatDraftStore>(relaxed = true),
+            userManager = userManager,
+            networkObserver = mockk<NetworkConnectivityListener>(relaxed = true),
+            dispatchers = testDispatchers,
+            groupFeedDelegate = mockk(relaxed = true),
+        )
+    }
+
+    private suspend fun triggerCollection() {
+        coordinator.onUserLoggedIn(mockk(relaxed = true))
+    }
+
+    @Test
+    fun `a title change from the stream is stored`() = runTest(testDispatchers.dispatcher) {
+        tornDown {
+            triggerCollection()
+
+            chatUpdatesChannel.send(
+                ChatUpdate(
+                    chatId = chatId,
+                    metadataUpdates = listOf(MetadataUpdate.TitleChanged("New title")),
+                ),
+            )
+            advanceTimeBy(1_000.milliseconds)
+            runCurrent()
+
+            coVerify(exactly = 1) { metadataDataSource.updateTitle(chatId, "New title") }
+        }
+    }
+
+    @Test
+    fun `a picture change from the stream is stored`() = runTest(testDispatchers.dispatcher) {
+        tornDown {
+            triggerCollection()
+            val picture = MediaItem(renditions = emptyList())
+
+            chatUpdatesChannel.send(
+                ChatUpdate(
+                    chatId = chatId,
+                    metadataUpdates = listOf(MetadataUpdate.PictureChanged(picture)),
+                ),
+            )
+            advanceTimeBy(1_000.milliseconds)
+            runCurrent()
+
+            coVerify(exactly = 1) { metadataDataSource.updatePicture(chatId, picture) }
+        }
+    }
+
+    @Test
+    fun `an update carrying no title or picture change leaves them alone`() =
+        runTest(testDispatchers.dispatcher) {
+            tornDown {
+                triggerCollection()
+
+                chatUpdatesChannel.send(
+                    ChatUpdate(
+                        chatId = chatId,
+                        metadataUpdates = listOf(
+                            MetadataUpdate.LastActivityChanged(kotlin.time.Instant.fromEpochSeconds(2_000)),
+                        ),
+                    ),
+                )
+                advanceTimeBy(1_000.milliseconds)
+                runCurrent()
+
+                coVerify(exactly = 0) { metadataDataSource.updateTitle(any(), any()) }
+                coVerify(exactly = 0) { metadataDataSource.updatePicture(any(), any()) }
+            }
+        }
+
+    /**
+     * Runs [block], then tears the coordinator down however it ends.
+     *
+     * The `finally` is the point. Logging in starts a heartbeat that is a `while (true)` of delays
+     * on the coordinator's own scope rather than the test's `backgroundScope`, and `runTest` drains
+     * the scheduler once the body returns. A failing assertion that skipped the teardown would
+     * leave that loop advancing virtual time with nothing to stop it, hanging the run instead of
+     * reporting the failure.
+     */
+    private suspend fun tornDown(block: suspend () -> Unit) {
+        try {
+            block()
+        } finally {
+            coordinator.teardown()
+        }
+    }
+}

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

@@ -241,6 +241,19 @@ interface ChatMetadataDao {
     @Query("UPDATE chat_metadata SET is_hidden = :hidden WHERE chat_id_hex = :chatIdHex")
     suspend fun updateHidden(chatIdHex: String, hidden: Boolean)
 
+    /**
+     * Applied unconditionally, unlike [updateRosterIfNewer] and [updateViewerStateIfNewer]:
+     * `MetadataUpdate.TitleChanged` carries no version, so there is nothing to gate on. Delivery
+     * is best-effort; a client that suspects a miss refetches via `GetChat` rather than relying
+     * on this to converge on its own.
+     */
+    @Query("UPDATE chat_metadata SET title = :title WHERE chat_id_hex = :chatIdHex")
+    suspend fun updateTitle(chatIdHex: String, title: String)
+
+    /** Unconditional, for the same reason as [updateTitle]: `MetadataUpdate.PictureChanged` carries no version. */
+    @Query("UPDATE chat_metadata SET picture_json = :pictureJson WHERE chat_id_hex = :chatIdHex")
+    suspend fun updatePicture(chatIdHex: String, pictureJson: MediaItem)
+
     @Query("DELETE FROM chat_metadata")
     suspend fun deleteAll()
 }

apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMetadataDaoTest.kt

@@ -170,6 +170,30 @@ class ChatMetadataDaoTest {
         assertEquals(false, stored?.isMember)
     }
 
+    /**
+     * `MetadataUpdate.TitleChanged` carries no version, unlike the roster and viewer-state
+     * overlays above, so there is nothing to gate the write on.
+     */
+    @Test
+    fun `updateTitle overwrites the title unconditionally`() = runTest {
+        dao.upsert(entity(chatType = "GROUP", title = "Old title"))
+
+        dao.updateTitle(CHAT_HEX, "New title")
+
+        assertEquals("New title", dao.getById(CHAT_HEX)?.title)
+    }
+
+    /** Same unconditional contract as `updateTitle`, for `MetadataUpdate.PictureChanged`. */
+    @Test
+    fun `updatePicture overwrites the picture unconditionally`() = runTest {
+        dao.upsert(entity(chatType = "GROUP"))
+        val picture = MediaItem(renditions = emptyList())
+
+        dao.updatePicture(CHAT_HEX, picture)
+
+        assertEquals(picture, dao.getById(CHAT_HEX)?.pictureJson)
+    }
+
     @Test
     fun `a newer roster version replaces the roster columns`() = runTest {
         dao.upsert(entity(chatType = "GROUP", memberCount = 12, rosterVersion = 4))

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

@@ -9,6 +9,7 @@ import com.flipcash.services.models.chat.ChatMember
 import com.flipcash.services.models.chat.ChatMessage
 import com.flipcash.services.models.chat.ChatMetadata
 import com.flipcash.services.models.chat.ChatType
+import com.flipcash.services.models.chat.MediaItem
 import com.flipcash.services.models.chat.ViewerState
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.emptyFlow
@@ -130,6 +131,16 @@ class ChatMetadataDataSource @Inject constructor(
     suspend fun getTitle(chatId: ChatId): String? =
         db?.chatMetadataDao()?.getTitle(mapper.chatIdHex(chatId))
 
+    /** Applies a `MetadataUpdate.TitleChanged` for [chatId]. See ChatMetadataDao.updateTitle. */
+    suspend fun updateTitle(chatId: ChatId, title: String) {
+        db?.chatMetadataDao()?.updateTitle(mapper.chatIdHex(chatId), title)
+    }
+
+    /** Applies a `MetadataUpdate.PictureChanged` for [chatId]. See ChatMetadataDao.updatePicture. */
+    suspend fun updatePicture(chatId: ChatId, picture: MediaItem) {
+        db?.chatMetadataDao()?.updatePicture(mapper.chatIdHex(chatId), picture)
+    }
+
     suspend fun getAnalyticsCountedThrough(chatId: ChatId): Long =
         db?.chatMetadataDao()?.getAnalyticsCountedThrough(mapper.chatIdHex(chatId)) ?: 0L
 

gradle/libs.versions.toml

@@ -69,7 +69,7 @@ protovalidate-kt = "0.1.3"
 # 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.5.0"
-flipcash2-client-protocol = "0.10.0"
+flipcash2-client-protocol = "0.11.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

@@ -5,8 +5,10 @@ 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.ChatType
+import com.flipcash.services.models.chat.EditChatParameters
 import com.flipcash.services.models.chat.IdempotencyKey
 import com.flipcash.services.models.chat.MuteState
+import com.flipcash.services.models.chat.RosterPage
 import com.flipcash.services.models.chat.StartChatParameters
 import com.flipcash.services.models.chat.ViewerState
 import com.flipcash.services.models.chat.ViewMode
@@ -68,6 +70,31 @@ class ChatController @Inject constructor(
         return repository.startChat(owner, parameters, idempotencyKey)
     }
 
+    /**
+     * One page of [chatId]'s roster. See [ChatRepository.getRoster] for paging and merge
+     * semantics.
+     */
+    suspend fun getRoster(
+        chatId: ChatId,
+        queryOptions: QueryOptions = QueryOptions(),
+    ): Result<RosterPage> {
+        val owner = userManager.accountCluster?.authority?.keyPair
+            ?: return Result.failure(Throwable("No account cluster in UserManager"))
+
+        return repository.getRoster(owner, chatId, queryOptions)
+    }
+
+    /** Edits [chatId] per [parameters]. See [ChatRepository.editChat] for no-op semantics. */
+    suspend fun editChat(
+        chatId: ChatId,
+        parameters: EditChatParameters,
+    ): Result<ChatMetadata> {
+        val owner = userManager.accountCluster?.authority?.keyPair
+            ?: return Result.failure(Throwable("No account cluster in UserManager"))
+
+        return repository.editChat(owner, chatId, parameters)
+    }
+
     suspend fun joinChat(chatId: ChatId): Result<ChatMetadata> {
         val owner = userManager.accountCluster?.authority?.keyPair
             ?: return Result.failure(Throwable("No account cluster in UserManager"))

services/flipcash/src/main/kotlin/com/flipcash/services/internal/domain/ChatMetadataMapper.kt

@@ -33,6 +33,12 @@ class ChatMetadataMapper @Inject constructor(
                     userProfile = userProfileMapper.map(member.userProfile)
                         .let { if (it.userId == null) it.copy(userId = member.userId.toId()) else it },
                     pointers = member.pointersList.map { it.toPointer() },
+                    joinedAt = if (member.hasJoinedAt()) {
+                        Instant.fromEpochSeconds(member.joinedAt.seconds, member.joinedAt.nanos)
+                    } else {
+                        null
+                    },
+                    version = member.version,
                 )
             },
             lastMessage = if (from.hasLastMessage()) from.lastMessage.toChatMessage() else null,