perf(chat): rebuild the conversation list with one preview read and one stamp read

apps/flipcash/shared/chat/src/commonMain/kotlin/com/flipcash/shared/chat/internal/delegates/FeedSyncDelegate.kt

@@ -19,6 +19,7 @@ import com.flipcash.services.controllers.ChatMessagingController
 import com.flipcash.services.models.GetMessageError
 import com.flipcash.services.models.chat.ChatId
 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.PointerType
@@ -290,15 +291,17 @@ class FeedSyncDelegate @Inject constructor(
             // a row whose newest message changed without its metadata row changing (an edit, a
             // delete, a preview written a beat after the row) kept its old preview until some
             // unrelated write. Distinct, so only a change to some chat's newest visible message
-            // rebuilds, not every message write; the initial value keeps a slow first read from
-            // holding the list up.
-            messageDataSource.observeLatestVisibleChanges()
-                .onStart { emit(emptyList()) }
+            // rebuilds, not every message write. The build reads its previews from here rather
+            // than querying them again. The initial null keeps a slow first read from holding the
+            // list up; a build that sees it reads the previews itself.
+            messageDataSource.observeLatestVisibleByChat()
+                .map<Map<String, ChatMessage>, Map<String, ChatMessage>?> { it }
+                .onStart { emit(null) }
                 .distinctUntilChanged(),
             archiveStore.observeArchived(),
             markedUnreadStore.observeMarked(),
-        ) { metadataEntities, membersByChat, _, archived, marked ->
-            Triple(buildFeedFromDb(metadataEntities, membersByChat), archived, marked)
+        ) { metadataEntities, membersByChat, latestVisible, archived, marked ->
+            Triple(buildFeedFromDb(metadataEntities, membersByChat, latestVisible), archived, marked)
         }.onEach { (built, archived, marked) ->
             val (feed, readStamps) = built
             val staged = drainStagedStamps()
@@ -464,27 +467,31 @@ class FeedSyncDelegate @Inject constructor(
     private suspend fun buildFeedFromDb(
         metadataEntities: List<ChatMetadataEntity>,
         membersByChat: Map<String, List<ChatMember>>,
+        observedLatestVisible: Map<String, ChatMessage>?,
     ): Pair<List<ChatMetadata>, Map<Pair<ChatId, Long>, Long>> {
         // `is_member` is a column rather than a field on ChatMetadata: a chat you have left is
         // still a chat you can be shown (Plan C's gate reads the same row), so the flag is dropped
         // here, at the edge of the list, rather than carried through the domain model.
-        val latestVisible = messageDataSource.getLatestVisibleByChat()
+        val latestVisible = observedLatestVisible ?: messageDataSource.getLatestVisibleByChat()
         val selfId = userManager.accountId
-        val readStamps = mutableMapOf<Pair<ChatId, Long>, Long>()
-        val feed = metadataEntities.filter { it.isMember }.map { entity ->
-            val members = membersByChat[entity.chatIdHex] ?: emptyList()
+        val rows = metadataEntities.filter { it.isMember }
+        val feed = rows.map { entity ->
             // Deliberately the newest *visible* message, not the newest row: deleting the newest
             // message drops the feed back to the one before it, so the preview reads that message
             // instead of "Message deleted" and its unread splat clears with it (the fallback sits
             // at or below the read pointer whenever the deleted message was the only unread one).
             val lastMessage = entity.lastMessageId?.let { latestVisible[entity.chatIdHex] }
-            metadataDataSource.toMetadata(entity, members, lastMessage).also { metadata ->
-                // Only the stamp a row can ask for: the one on the message the READ pointer names.
-                val readPointer = selfReadPointer(metadata, selfId) ?: return@also
-                messageDataSource.getUnreadSeq(entity.chatIdHex, readPointer)
-                    ?.let { readStamps[metadata.chatId to readPointer] = it }
-            }
+            metadataDataSource.toMetadata(entity, membersByChat[entity.chatIdHex] ?: emptyList(), lastMessage)
+        }
+        // Only the stamp a row can ask for: the one on the message the READ pointer names. Read
+        // for every row at once; one query per chat was most of a rebuild.
+        val pointers = feed.zip(rows).mapNotNull { (metadata, entity) ->
+            selfReadPointer(metadata, selfId)?.let { Triple(metadata.chatId, entity.chatIdHex, it) }
         }
+        val stamps = messageDataSource.getUnreadSeqs(pointers.map { (_, hex, pointer) -> hex to pointer })
+        val readStamps = pointers.mapNotNull { (chatId, hex, pointer) ->
+            stamps[hex to pointer]?.let { (chatId to pointer) to it }
+        }.toMap()
         return feed to readStamps
     }
 

apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/ChatArchiveFeedTest.kt

@@ -90,6 +90,7 @@ class ChatArchiveFeedTest {
         val messageDataSource = mockk<ChatMessageDataSource>(relaxed = true).also {
             coEvery { it.getLatestVisibleByChat() } returns entities.associate { e -> e.chatIdHex to message(2) }
             coEvery { it.getUnreadSeq(any(), any()) } answers { secondArg() }
+            coEvery { it.getUnreadSeqs(any()) } answers { firstArg<Collection<Pair<String, Long>>>().associateWith { key -> key.second } }
         }
         val metadataDataSource = mockk<ChatMetadataDataSource>(relaxed = true).also { source ->
             every { source.observeAll() } returns flowOf(entities)

apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/ChatListUnreadCountTest.kt

@@ -109,6 +109,7 @@ class ChatListUnreadCountTest {
         val messageDataSource = mockk<ChatMessageDataSource>(relaxed = true).also {
             coEvery { it.getLatestVisibleByChat() } returns entities.associate { e -> e.chatIdHex to message(2) }
             coEvery { it.getUnreadSeq(any(), any()) } answers { secondArg() }
+            coEvery { it.getUnreadSeqs(any()) } answers { firstArg<Collection<Pair<String, Long>>>().associateWith { key -> key.second } }
         }
         val metadataDataSource = mockk<ChatMetadataDataSource>(relaxed = true).also { source ->
             every { source.observeAll() } returns flowOf(entities)

apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/ChatMarkedUnreadFeedTest.kt

@@ -101,6 +101,7 @@ class ChatMarkedUnreadFeedTest {
         val messageDataSource = mockk<ChatMessageDataSource>(relaxed = true).also {
             coEvery { it.getLatestVisibleByChat() } returns entities.associate { e -> e.chatIdHex to message(2) }
             coEvery { it.getUnreadSeq(any(), any()) } answers { secondArg() }
+            coEvery { it.getUnreadSeqs(any()) } answers { firstArg<Collection<Pair<String, Long>>>().associateWith { key -> key.second } }
         }
         val metadataDataSource = mockk<ChatMetadataDataSource>(relaxed = true).also { source ->
             every { source.observeAll() } returns flowOf(entities)

apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/FeedDeletedMessageTest.kt

@@ -85,7 +85,12 @@ class FeedDeletedMessageTest {
             ),
         )
 
-        val messageDataSource = mockk<ChatMessageDataSource>(relaxed = true)
+        // The READ pointer's message carries a stamp equal to its id, as `message()` builds them.
+        val messageDataSource = mockk<ChatMessageDataSource>(relaxed = true).also {
+            coEvery { it.getUnreadSeqs(any()) } answers {
+                firstArg<Collection<Pair<String, Long>>>().associateWith { key -> key.second }
+            }
+        }
 
         /** Stubs the chat's newest visible message, as the batched read returns it. */
         fun latestVisible(message: ChatMessage?) {

apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/FeedGroupProjectionTest.kt

@@ -94,6 +94,7 @@ class FeedGroupProjectionTest {
             coEvery { it.getLatestVisibleByChat() } returns entities.associate { e -> e.chatIdHex to message(2) }
             // Every stored message is stamped with its own id, as [message] builds them.
             coEvery { it.getUnreadSeq(any(), any()) } answers { secondArg() }
+            coEvery { it.getUnreadSeqs(any()) } answers { firstArg<Collection<Pair<String, Long>>>().associateWith { key -> key.second } }
         }
 
         private val metadataDataSource = mockk<ChatMetadataDataSource>(relaxed = true).also { source ->

apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/FeedObservedPreviewTest.kt

@@ -0,0 +1,172 @@
+package com.flipcash.shared.chat
+
+import com.flipcash.app.persistence.entities.ChatMetadataEntity
+import com.flipcash.app.persistence.sources.ChatMemberDataSource
+import com.flipcash.app.persistence.sources.ChatMessageDataSource
+import com.flipcash.app.persistence.sources.ChatMetadataDataSource
+import com.flipcash.services.controllers.ChatController
+import com.flipcash.services.models.UserProfile
+import com.flipcash.services.models.VerifiableContactMethod
+import com.flipcash.services.models.chat.ChatId
+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.MessageContent
+import com.flipcash.services.models.chat.MessagePointer
+import com.flipcash.services.models.chat.PointerType
+import com.flipcash.services.user.UserManager
+import com.flipcash.shared.chat.internal.ChatStateHolder
+import com.flipcash.shared.chat.internal.delegates.FeedSyncDelegate
+import io.mockk.coEvery
+import io.mockk.coVerify
+import io.mockk.every
+import io.mockk.mockk
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.flow.flowOf
+import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.runCurrent
+import kotlinx.coroutines.test.runTest
+import org.junit.Test
+import kotlin.test.assertEquals
+import kotlin.time.Instant
+
+/**
+ * The conversation list takes its previews from the stream that tells it they changed, and reads
+ * every row's unread stamp in one call, rather than querying the previews a second time and the
+ * stamps once per chat on every rebuild. The list it builds is the same either way.
+ */
+@OptIn(ExperimentalCoroutinesApi::class)
+class FeedObservedPreviewTest {
+
+    private val selfId = listOf<Byte>(1, 2, 3)
+    private val otherId = listOf<Byte>(4, 5, 6)
+    private val chatHex = "aabbccdd"
+    private val readPointer = 2L
+
+    private fun message(messageId: Long, text: String) = ChatMessage(
+        messageId = messageId,
+        senderId = otherId,
+        content = listOf(MessageContent.Text(text)),
+        timestamp = Instant.fromEpochSeconds(messageId),
+        unreadSeq = messageId,
+    )
+
+    private inner class Harness(observed: Map<String, ChatMessage>?, direct: Map<String, ChatMessage>) {
+        private val entity = ChatMetadataEntity(
+            chatIdHex = chatHex,
+            chatType = ChatType.CONTACT_DM.name,
+            lastActivityEpochMs = 5_000,
+            lastMessageId = 5,
+        )
+
+        private val members = listOf(
+            ChatMember(
+                userId = selfId,
+                userProfile = UserProfile.Empty,
+                pointers = listOf(
+                    MessagePointer(
+                        type = PointerType.READ,
+                        userId = selfId,
+                        value = readPointer,
+                        timestamp = Instant.fromEpochSeconds(1_000),
+                    ),
+                ),
+            ),
+            ChatMember(
+                userId = otherId,
+                userProfile = UserProfile.Empty.copy(
+                    displayName = "Ada",
+                    phoneNumber = VerifiableContactMethod("+15551234567", verified = true),
+                ),
+                pointers = emptyList(),
+            ),
+        )
+
+        /** What the preview stream emits; null leaves it silent, as a database not yet open. */
+        val previews = observed?.let { MutableStateFlow(it) }
+
+        val messageDataSource = mockk<ChatMessageDataSource>(relaxed = true).also {
+            if (previews != null) every { it.observeLatestVisibleByChat() } returns previews
+            coEvery { it.getLatestVisibleByChat() } returns direct
+            coEvery { it.getUnreadSeqs(any()) } answers {
+                firstArg<Collection<Pair<String, Long>>>().associateWith { key -> key.second }
+            }
+        }
+
+        val delegate = FeedSyncDelegate(
+            messagingController = mockk(relaxed = true),
+            chatController = mockk<ChatController>(relaxed = true),
+            metadataDataSource = mockk<ChatMetadataDataSource>(relaxed = true).also { source ->
+                every { source.observeAll() } returns flowOf(listOf(entity))
+                every { source.toMetadata(any(), any(), any()) } answers {
+                    ChatMetadata(
+                        chatId = ChatId(chatHex),
+                        type = ChatType.CONTACT_DM,
+                        members = secondArg(),
+                        lastMessage = thirdArg(),
+                        lastActivity = Instant.fromEpochSeconds(5),
+                    )
+                }
+            },
+            messageDataSource = messageDataSource,
+            memberDataSource = mockk<ChatMemberDataSource>(relaxed = true).also { source ->
+                every { source.observeAll() } returns flowOf(mapOf(chatHex to members))
+            },
+            stateHolder = ChatStateHolder(),
+            userManager = mockk<UserManager>(relaxed = true).also {
+                every { it.accountId } returns selfId
+                every { it.profile } returns null
+            },
+        )
+
+        fun start(scope: TestScope) {
+            delegate.initialize(scope.backgroundScope)
+            delegate.observeFeedFromDb()
+            scope.runCurrent()
+        }
+
+        suspend fun summary(): ChatSummary? = delegate.feed(ChatType.CONTACT_DM).first().firstOrNull()
+    }
+
+    @Test
+    fun `a preview change rebuilds from the streamed previews without querying them again`() = runTest {
+        val harness = Harness(
+            observed = mapOf(chatHex to message(4, "four")),
+            direct = mapOf(chatHex to message(1, "stale direct read")),
+        )
+        harness.start(this)
+        coVerify(atMost = 1) { harness.messageDataSource.getLatestVisibleByChat() }
+
+        harness.previews!!.value = mapOf(chatHex to message(5, "five"))
+        runCurrent()
+
+        assertEquals(5L, harness.summary()?.metadata?.lastMessage?.messageId)
+        assertEquals(3, harness.summary()?.unreadCount)
+        coVerify(atMost = 1) { harness.messageDataSource.getLatestVisibleByChat() }
+    }
+
+    @Test
+    fun `every row's stamp is read in one call`() = runTest {
+        val harness = Harness(observed = mapOf(chatHex to message(5, "five")), direct = emptyMap())
+        harness.start(this)
+
+        assertEquals(3, harness.summary()?.unreadCount)
+        coVerify { harness.messageDataSource.getUnreadSeqs(listOf(chatHex to readPointer)) }
+        coVerify(exactly = 0) { harness.messageDataSource.getUnreadSeq(any(), any()) }
+    }
+
+    @Test
+    fun `the list built from streamed previews matches the one built from a direct read`() = runTest {
+        val previews = mapOf(chatHex to message(5, "five"))
+        val streamed = Harness(observed = previews, direct = emptyMap())
+        val direct = Harness(observed = null, direct = previews)
+        streamed.start(this)
+        direct.start(this)
+
+        assertEquals(direct.summary(), streamed.summary())
+        assertEquals(5L, streamed.summary()?.metadata?.lastMessage?.messageId)
+    }
+}

apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/FeedReadStampFetchTest.kt

@@ -107,6 +107,10 @@ class FeedReadStampFetchTest {
             messageDataSource = mockk<ChatMessageDataSource>(relaxed = true).also {
                 coEvery { it.getLatestVisibleByChat() } returns mapOf(chatHex to message(9))
                 coEvery { it.getUnreadSeq(chatHex, readPointer) } returns storedStamp
+                coEvery { it.getUnreadSeqs(any()) } answers {
+                    val key = chatHex to readPointer
+                    if (key in firstArg<Collection<Pair<String, Long>>>()) listOfNotNull(storedStamp?.let { key to it }).toMap() else emptyMap()
+                }
             },
             memberDataSource = mockk<ChatMemberDataSource>(relaxed = true).also {
                 every { it.observeAll() } returns flowOf(mapOf(chatHex to members))