diff --git a/apps/flipcash/shared/chat/src/commonMain/kotlin/com/flipcash/shared/chat/internal/delegates/FeedSyncDelegate.kt b/apps/flipcash/shared/chat/src/commonMain/kotlin/com/flipcash/shared/chat/internal/delegates/FeedSyncDelegate.kt index d9aa337b6e..d2ed768bd5 100644 --- a/apps/flipcash/shared/chat/src/commonMain/kotlin/com/flipcash/shared/chat/internal/delegates/FeedSyncDelegate.kt +++ b/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?> { 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, membersByChat: Map>, + observedLatestVisible: Map?, ): Pair, Map, 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, 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 } diff --git a/apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/ChatArchiveFeedTest.kt b/apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/ChatArchiveFeedTest.kt index 79c8d7b0f9..797c2fcf06 100644 --- a/apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/ChatArchiveFeedTest.kt +++ b/apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/ChatArchiveFeedTest.kt @@ -90,6 +90,7 @@ class ChatArchiveFeedTest { val messageDataSource = mockk(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>>().associateWith { key -> key.second } } } val metadataDataSource = mockk(relaxed = true).also { source -> every { source.observeAll() } returns flowOf(entities) diff --git a/apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/ChatListUnreadCountTest.kt b/apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/ChatListUnreadCountTest.kt index a153aaadc0..2b343f564a 100644 --- a/apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/ChatListUnreadCountTest.kt +++ b/apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/ChatListUnreadCountTest.kt @@ -109,6 +109,7 @@ class ChatListUnreadCountTest { val messageDataSource = mockk(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>>().associateWith { key -> key.second } } } val metadataDataSource = mockk(relaxed = true).also { source -> every { source.observeAll() } returns flowOf(entities) diff --git a/apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/ChatMarkedUnreadFeedTest.kt b/apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/ChatMarkedUnreadFeedTest.kt index c1213b6c99..b4605437e4 100644 --- a/apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/ChatMarkedUnreadFeedTest.kt +++ b/apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/ChatMarkedUnreadFeedTest.kt @@ -101,6 +101,7 @@ class ChatMarkedUnreadFeedTest { val messageDataSource = mockk(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>>().associateWith { key -> key.second } } } val metadataDataSource = mockk(relaxed = true).also { source -> every { source.observeAll() } returns flowOf(entities) diff --git a/apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/FeedDeletedMessageTest.kt b/apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/FeedDeletedMessageTest.kt index 83dc72149a..8541ae5c22 100644 --- a/apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/FeedDeletedMessageTest.kt +++ b/apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/FeedDeletedMessageTest.kt @@ -85,7 +85,12 @@ class FeedDeletedMessageTest { ), ) - val messageDataSource = mockk(relaxed = true) + // The READ pointer's message carries a stamp equal to its id, as `message()` builds them. + val messageDataSource = mockk(relaxed = true).also { + coEvery { it.getUnreadSeqs(any()) } answers { + firstArg>>().associateWith { key -> key.second } + } + } /** Stubs the chat's newest visible message, as the batched read returns it. */ fun latestVisible(message: ChatMessage?) { diff --git a/apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/FeedGroupProjectionTest.kt b/apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/FeedGroupProjectionTest.kt index 04b12b9d44..adbc6d8c0d 100644 --- a/apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/FeedGroupProjectionTest.kt +++ b/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>>().associateWith { key -> key.second } } } private val metadataDataSource = mockk(relaxed = true).also { source -> diff --git a/apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/FeedObservedPreviewTest.kt b/apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/FeedObservedPreviewTest.kt new file mode 100644 index 0000000000..8afa3780c7 --- /dev/null +++ b/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(1, 2, 3) + private val otherId = listOf(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?, direct: Map) { + 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(relaxed = true).also { + if (previews != null) every { it.observeLatestVisibleByChat() } returns previews + coEvery { it.getLatestVisibleByChat() } returns direct + coEvery { it.getUnreadSeqs(any()) } answers { + firstArg>>().associateWith { key -> key.second } + } + } + + val delegate = FeedSyncDelegate( + messagingController = mockk(relaxed = true), + chatController = mockk(relaxed = true), + metadataDataSource = mockk(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(relaxed = true).also { source -> + every { source.observeAll() } returns flowOf(mapOf(chatHex to members)) + }, + stateHolder = ChatStateHolder(), + userManager = mockk(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) + } +} diff --git a/apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/FeedReadStampFetchTest.kt b/apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/FeedReadStampFetchTest.kt index cf8888cb2b..ab5cb3e75a 100644 --- a/apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/FeedReadStampFetchTest.kt +++ b/apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/FeedReadStampFetchTest.kt @@ -107,6 +107,10 @@ class FeedReadStampFetchTest { messageDataSource = mockk(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>>()) listOfNotNull(storedStamp?.let { key to it }).toMap() else emptyMap() + } }, memberDataSource = mockk(relaxed = true).also { every { it.observeAll() } returns flowOf(mapOf(chatHex to members)) diff --git a/apps/flipcash/shared/persistence/db/src/androidHostTest/kotlin/com/flipcash/app/persistence/dao/ChatMessageDaoTest.kt b/apps/flipcash/shared/persistence/db/src/androidHostTest/kotlin/com/flipcash/app/persistence/dao/ChatMessageDaoTest.kt index 4cfa6c5d3d..732fc0d32d 100644 --- a/apps/flipcash/shared/persistence/db/src/androidHostTest/kotlin/com/flipcash/app/persistence/dao/ChatMessageDaoTest.kt +++ b/apps/flipcash/shared/persistence/db/src/androidHostTest/kotlin/com/flipcash/app/persistence/dao/ChatMessageDaoTest.kt @@ -9,9 +9,11 @@ import com.flipcash.app.persistence.converters.EmojiReactionSerialized import com.flipcash.app.persistence.converters.MessageContentSerialized import com.flipcash.app.persistence.converters.ReactionSummarySerialized import com.flipcash.app.persistence.entities.ChatMessageEntity +import com.flipcash.app.persistence.entities.ChatMetadataEntity import com.flipcash.app.persistence.entities.EncryptionState import com.flipcash.app.persistence.entities.MessageStatus import kotlinx.coroutines.flow.first +import kotlinx.coroutines.runBlocking import kotlinx.coroutines.test.runTest import kotlinx.serialization.json.Json import org.junit.After @@ -22,6 +24,7 @@ import org.robolectric.RobolectricTestRunner import kotlin.test.assertEquals import kotlin.test.assertNotEquals import kotlin.test.assertNull +import kotlin.test.assertTrue /** * Covers the split between the newest stored row and the newest row that still has content. The @@ -43,8 +46,17 @@ class ChatMessageDaoTest { .allowMainThreadQueries() .build() dao = db.chatMessageDao() + // The chat-list preview reads only chats with a metadata row. + runBlocking { listOf(CHAT_HEX, OTHER_HEX).forEach { db.chatMetadataDao().insertIfAbsent(metadata(it)) } } } + private fun metadata(chatIdHex: String) = ChatMetadataEntity( + chatIdHex = chatIdHex, + chatType = "CONTACT_DM", + lastActivityEpochMs = 0, + lastMessageId = null, + ) + @After fun tearDown() { db.close() @@ -143,6 +155,52 @@ class ChatMessageDaoTest { assertEquals(listOf(OTHER_HEX), dao.getLatestVisibleForAllChats().map { it.chatIdHex }) } + @Test + fun `getLatestVisibleForAllChats leaves out a chat with no metadata row`() = runTest { + dao.upsert(text(1, "one")) + dao.upsert(text(2, "two").copy(chatIdHex = NO_METADATA_HEX)) + + assertEquals(listOf(CHAT_HEX), dao.getLatestVisibleForAllChats().map { it.chatIdHex }) + assertEquals(listOf(CHAT_HEX), dao.observeLatestVisibleForAllChats().first().map { it.chatIdHex }) + } + + @Test + fun `getUnreadSeqs returns the stamp of each stored key and leaves out the rest`() = runTest { + dao.upsert(listOf(text(1, "one"), text(2, "two").copy(unreadSeq = 20))) + dao.upsert(text(5, "five").copy(chatIdHex = OTHER_HEX, unreadSeq = 50)) + + val stamps = dao.getUnreadSeqs( + listOf(CHAT_HEX to 1L, CHAT_HEX to 2L, OTHER_HEX to 5L, CHAT_HEX to 5L, OTHER_HEX to 9L, CHAT_HEX to 2L), + ) + + assertEquals(mapOf((CHAT_HEX to 1L) to 1L, (CHAT_HEX to 2L) to 20L, (OTHER_HEX to 5L) to 50L), stamps) + assertEquals(emptyMap(), dao.getUnreadSeqs(emptyList())) + } + + @Test + fun `getUnreadSeqs agrees with getUnreadSeq across more keys than one statement takes`() = runTest { + val ids = 1L..(UNREAD_SEQ_CHUNK * 2 + 7L) + dao.upsert(ids.filter { it % 3 != 0L }.map { text(it, "m$it").copy(unreadSeq = it * 10) }) + + val stamps = dao.getUnreadSeqs(ids.map { CHAT_HEX to it }) + + val expected = ids.mapNotNull { id -> dao.getUnreadSeq(CHAT_HEX, id)?.let { (CHAT_HEX to id) to it } }.toMap() + assertEquals(expected, stamps) + assertEquals(ids.count { it % 3 != 0L }, stamps.size) + } + + @Test + fun `getUnreadSeqs looks each key up by primary key`() { + val plan = db.openHelper.writableDatabase.query( + "EXPLAIN QUERY PLAN SELECT m.chat_id_hex, m.message_id, m.unread_seq FROM (VALUES (?,?),(?,?)) k " + + "JOIN chat_messages m ON m.chat_id_hex = k.column1 AND m.message_id = k.column2", + arrayOf(CHAT_HEX, 1L, OTHER_HEX, 2L), + ).use { c -> buildList { while (c.moveToNext()) add(c.getString(c.getColumnIndexOrThrow("detail"))) } } + + assertTrue(plan.any { "SEARCH m USING" in it && "PRIMARY KEY" in it || "sqlite_autoindex_chat_messages_1" in it }, plan.toString()) + assertTrue(plan.none { it.startsWith("SCAN m") }, plan.toString()) + } + @Test fun `on a shared timestamp both latest-visible reads pick the highest message id`() = runTest { val sameMs = 5_000L @@ -914,6 +972,7 @@ class ChatMessageDaoTest { private companion object { const val CHAT_HEX = "aabb" const val OTHER_HEX = "ccdd" + const val NO_METADATA_HEX = "eeff" const val SENDER_HEX = "1122" const val SELF_HEX = "3344" const val CLIENT_HEX = "eeff" diff --git a/apps/flipcash/shared/persistence/db/src/commonMain/kotlin/com/flipcash/app/persistence/dao/ChatMessageDao.kt b/apps/flipcash/shared/persistence/db/src/commonMain/kotlin/com/flipcash/app/persistence/dao/ChatMessageDao.kt index 5a5f201f0b..332c57ace2 100644 --- a/apps/flipcash/shared/persistence/db/src/commonMain/kotlin/com/flipcash/app/persistence/dao/ChatMessageDao.kt +++ b/apps/flipcash/shared/persistence/db/src/commonMain/kotlin/com/flipcash/app/persistence/dao/ChatMessageDao.kt @@ -5,6 +5,9 @@ import androidx.room.Dao import androidx.room.Insert import androidx.room.OnConflictStrategy import androidx.room.Query +import androidx.room.RawQuery +import androidx.room.RoomRawQuery +import androidx.room.ColumnInfo import androidx.room.Transaction import com.flipcash.app.persistence.converters.mergeReactionsJson import com.flipcash.app.persistence.entities.ChatMessageEntity @@ -24,6 +27,16 @@ private const val NOT_REQUEST_SHARE = "AND NOT (COALESCE(content_json, '') LIKE '%\"type\":\"cash\"%' " + "AND COALESCE(content_json, '') LIKE '%\"paymentRequestId\":\"%') " +/** Keys per statement in [ChatMessageDao.getUnreadSeqs]: two bind variables each, under 999. */ +internal const val UNREAD_SEQ_CHUNK = 400 + +/** One row of [ChatMessageDao.getUnreadSeqs]. */ +data class UnreadSeqRow( + @ColumnInfo(name = "chat_id_hex") val chatIdHex: String, + @ColumnInfo(name = "message_id") val messageId: Long, + @ColumnInfo(name = "unread_seq") val unreadSeq: Long?, +) + /** [NOT_REQUEST_SHARE] for the correlated subqueries that alias the table as `m`. */ private const val NOT_REQUEST_SHARE_M = "AND NOT (COALESCE(m.content_json, '') LIKE '%\"type\":\"cash\"%' " + @@ -90,7 +103,12 @@ interface ChatMessageDao { suspend fun getLatestVisible(chatIdHex: String): ChatMessageEntity? /** - * [getLatestVisible] for every chat at once: one row per chat that has a message with content. + * [getLatestVisible] for every chat at once: one row per chat that has a `chat_metadata` row and + * a message with content. Messages of a chat with no metadata row are left out; the list builds + * only from metadata rows, so it never reads them. + * + * Driven from `chat_metadata` rather than `SELECT DISTINCT chat_id_hex FROM chat_messages`, + * which walks an index over every stored message to find the chats. * * The conversation list needs this for every chat it builds, and one query per chat costs a * round trip each. A correlated subquery rather than a window function, which needs SQLite 3.25 @@ -104,7 +122,7 @@ interface ChatMessageDao { "AND (m.encryption_state IS NULL OR m.encryption_state != 'KEY_PENDING') " + NOT_REQUEST_SHARE_M + "ORDER BY m.timestamp_epoch_ms DESC, m.message_id DESC LIMIT 1) " + - "FROM (SELECT DISTINCT chat_id_hex FROM chat_messages) c)" + "FROM chat_metadata c)" ) suspend fun getLatestVisibleForAllChats(): List @@ -121,7 +139,7 @@ interface ChatMessageDao { "AND (m.encryption_state IS NULL OR m.encryption_state != 'KEY_PENDING') " + NOT_REQUEST_SHARE_M + "ORDER BY m.timestamp_epoch_ms DESC, m.message_id DESC LIMIT 1) " + - "FROM (SELECT DISTINCT chat_id_hex FROM chat_messages) c)" + "FROM chat_metadata c)" ) fun observeLatestVisibleForAllChats(): Flow> @@ -317,6 +335,39 @@ interface ChatMessageDao { @Query("SELECT unread_seq FROM chat_messages WHERE chat_id_hex = :chatIdHex AND message_id = :messageId") suspend fun getUnreadSeq(chatIdHex: String, messageId: Long): Long? + @RawQuery + suspend fun queryUnreadSeqs(query: RoomRawQuery): List + + /** + * [getUnreadSeq] for many (chat id hex, message id) keys. A key whose message isn't stored, or + * whose stamp is null, is absent from the result. + * + * One statement per [UNREAD_SEQ_CHUNK] keys rather than one per key: the conversation list + * reads a stamp for every chat on every rebuild. The chunk keeps a statement under the 999 + * bind variables of the SQLite that minSdk ships. + */ + @Transaction + suspend fun getUnreadSeqs(keys: Collection>): Map, Long> { + if (keys.isEmpty()) return emptyMap() + val stamps = HashMap, Long>(keys.size) + for (chunk in keys.distinct().chunked(UNREAD_SEQ_CHUNK)) { + val values = chunk.joinToString(",") { "(?,?)" } + val query = RoomRawQuery( + "SELECT m.chat_id_hex, m.message_id, m.unread_seq FROM (VALUES $values) k " + + "JOIN chat_messages m ON m.chat_id_hex = k.column1 AND m.message_id = k.column2", + ) { statement -> + chunk.forEachIndexed { i, (chatIdHex, messageId) -> + statement.bindText(2 * i + 1, chatIdHex) + statement.bindLong(2 * i + 2, messageId) + } + } + for (row in queryUnreadSeqs(query)) { + row.unreadSeq?.let { stamps[row.chatIdHex to row.messageId] = it } + } + } + return stamps + } + @Transaction suspend fun upsert(entity: ChatMessageEntity) { // Event-sequence guard: skip if the stored sequence is strictly newer (last-writer-wins). diff --git a/apps/flipcash/shared/persistence/sources/src/commonMain/kotlin/com/flipcash/app/persistence/sources/ChatMessageDataSource.kt b/apps/flipcash/shared/persistence/sources/src/commonMain/kotlin/com/flipcash/app/persistence/sources/ChatMessageDataSource.kt index da0bdb8b1a..50f7506852 100644 --- a/apps/flipcash/shared/persistence/sources/src/commonMain/kotlin/com/flipcash/app/persistence/sources/ChatMessageDataSource.kt +++ b/apps/flipcash/shared/persistence/sources/src/commonMain/kotlin/com/flipcash/app/persistence/sources/ChatMessageDataSource.kt @@ -141,12 +141,14 @@ class ChatMessageDataSource @Inject constructor( .orEmpty() /** - * Emits whenever the set of newest-visible messages changes (one per chat). Carries the rows - * only so callers can de-duplicate on them; the list reads previews with - * [getLatestVisibleByChat]. + * [getLatestVisibleByChat] as a stream: emits when some chat's newest visible message changes, + * not on every message write. */ - fun observeLatestVisibleChanges(): Flow> = - db?.chatMessageDao()?.observeLatestVisibleForAllChats()?.distinctUntilChanged() ?: emptyFlow() + fun observeLatestVisibleByChat(): Flow> = + db?.chatMessageDao()?.observeLatestVisibleForAllChats() + ?.distinctUntilChanged() + ?.map { rows -> rows.associate { it.chatIdHex to toChatMessage(it) } } + ?: emptyFlow() /** * The server's unread stamp on one stored message, or null when that message isn't stored — @@ -155,6 +157,10 @@ class ChatMessageDataSource @Inject constructor( suspend fun getUnreadSeq(chatIdHex: String, messageId: Long): Long? = db?.chatMessageDao()?.getUnreadSeq(chatIdHex, messageId) + /** [getUnreadSeq] for many (chat id hex, message id) keys; keys with no stamp are absent. */ + suspend fun getUnreadSeqs(keys: Collection>): Map, Long> = + db?.chatMessageDao()?.getUnreadSeqs(keys).orEmpty() + suspend fun hasMessages(chatId: ChatId): Boolean = db?.chatMessageDao()?.getLatest(mapper.chatIdHex(chatId)) != null