From 6c31392de579473cdc21ddc2fbd06201ed4e2a10 Mon Sep 17 00:00:00 2001 From: Brandon McAnsh Date: Tue, 15 Sep 2026 20:32:59 -0400 Subject: [PATCH 1/9] feat(chat): query the feed by page, by type and by membership --- .../app/persistence/dao/ChatMemberDao.kt | 4 + .../app/persistence/dao/ChatMetadataDao.kt | 35 +++++++ .../app/persistence/dao/ChatMemberDaoTest.kt | 16 +++ .../persistence/dao/ChatMetadataDaoTest.kt | 97 +++++++++++++++++++ 4 files changed, 152 insertions(+) diff --git a/apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMemberDao.kt b/apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMemberDao.kt index 45f52955ef..a2ca504711 100644 --- a/apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMemberDao.kt +++ b/apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMemberDao.kt @@ -114,6 +114,10 @@ interface ChatMemberDao { ) suspend fun deleteMembersNotIn(chatIdHex: String, keepUserIdHexes: List) + /** Drops one member of [chatIdHex]. What a `MemberLeft` roster change applies. */ + @Query("DELETE FROM chat_members WHERE chat_id_hex = :chatIdHex AND user_id_hex = :userIdHex") + suspend fun deleteMember(chatIdHex: String, userIdHex: String) + @Query("DELETE FROM chat_members") suspend fun deleteAll() } diff --git a/apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMetadataDao.kt b/apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMetadataDao.kt index f60b544fee..e8e2cc3278 100644 --- a/apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMetadataDao.kt +++ b/apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMetadataDao.kt @@ -1,5 +1,6 @@ package com.flipcash.app.persistence.dao +import androidx.paging.PagingSource import androidx.room.Dao import androidx.room.Insert import androidx.room.OnConflictStrategy @@ -16,6 +17,40 @@ interface ChatMetadataDao { @Query("SELECT * FROM chat_metadata ORDER BY last_activity_epoch_ms DESC") fun observeAll(): Flow> + /** + * The merged conversation list, as a Paging source. + * + * One ordering across every type in [chatTypes] — a group ranks against DMs by activity like + * anything else, so ordering per type and interleaving afterwards would put it in the wrong + * place. `chat_id_hex` breaks a tie: Paging asks for the next page by offset, and an order + * that is not total lets two rows swap between page loads, which duplicates one and loses the + * other. + * + * Chats you are not in are filtered here rather than in the projection so they do not consume + * a page slot. + */ + @Query( + "SELECT * FROM chat_metadata WHERE chat_type IN (:chatTypes) " + + "AND is_hidden = 0 AND is_member = 1 " + + "ORDER BY last_activity_epoch_ms DESC, chat_id_hex DESC" + ) + fun observeFeedPaged(chatTypes: List): PagingSource + + /** The chats of [chatType] this device still considers you a member of. */ + @Query("SELECT chat_id_hex FROM chat_metadata WHERE chat_type = :chatType AND is_member = 1") + suspend fun getChatIdsOfType(chatType: String): List + + /** + * Sets membership without disturbing the rest of the row. Leaving a group keeps its title, + * rules and transcript — the gate reads them to decide what a non-member is allowed to see. + */ + @Query("UPDATE chat_metadata SET is_member = :isMember WHERE chat_id_hex = :chatIdHex") + suspend fun updateMembership(chatIdHex: String, isMember: Boolean) + + /** The roster version already applied to [chatIdHex], or null when the chat is not stored. */ + @Query("SELECT roster_version FROM chat_metadata WHERE chat_id_hex = :chatIdHex") + suspend fun getRosterVersion(chatIdHex: String): Long? + @Query("SELECT * FROM chat_metadata WHERE chat_id_hex = :chatIdHex") suspend fun getById(chatIdHex: String): ChatMetadataEntity? diff --git a/apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMemberDaoTest.kt b/apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMemberDaoTest.kt index a95c33805d..2a2a941951 100644 --- a/apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMemberDaoTest.kt +++ b/apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMemberDaoTest.kt @@ -123,6 +123,22 @@ class ChatMemberDaoTest { assertEquals(9L, dao.getMember(CHAT_HEX, SELF_HEX)?.pointersJson?.single()?.value) } + /** + * What a `MemberLeft` roster change applies. `deleteMembersNotIn` cannot express it: a roster + * change names who left, not who remains, and the stored member list for a large group is only + * a subset — passing it as the keep-list would delete everyone the subset omits. + */ + @Test + fun `deleteMember drops one member and leaves the rest intact`() = runTest { + dao.upsert(member(readPointer(9))) + dao.upsert(member(readPointer(4, OTHER_HEX)).copy(userIdHex = OTHER_HEX)) + + dao.deleteMember(CHAT_HEX, OTHER_HEX) + + assertNull(dao.getMember(CHAT_HEX, OTHER_HEX)) + assertEquals(9L, dao.getMember(CHAT_HEX, SELF_HEX)?.pointersJson?.single()?.value) + } + private companion object { const val CHAT_HEX = "aabb" const val SELF_HEX = "ccdd" diff --git a/apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMetadataDaoTest.kt b/apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMetadataDaoTest.kt index 3c7c4593c7..1545ad1919 100644 --- a/apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMetadataDaoTest.kt +++ b/apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMetadataDaoTest.kt @@ -1,6 +1,7 @@ package com.flipcash.app.persistence.dao import android.content.Context +import androidx.paging.PagingSource import androidx.room.Room import androidx.test.core.app.ApplicationProvider import com.flipcash.app.persistence.FlipcashDatabase @@ -183,6 +184,102 @@ class ChatMetadataDaoTest { assertEquals(4L, stored?.rosterVersion) } + /** + * One ordering across every type is the whole point of the merged feed: a group that belongs + * between two DMs has to come back between them, not after all of them. + */ + @Test + fun `the paged feed orders every type together, newest first`() = runTest { + dao.upsert(entity(chatIdHex = "01", chatType = "CONTACT_DM", lastActivityEpochMs = 3_000)) + dao.upsert(entity(chatIdHex = "02", chatType = "GROUP", lastActivityEpochMs = 2_000)) + dao.upsert(entity(chatIdHex = "03", chatType = "TIP_DM", lastActivityEpochMs = 1_000)) + + val page = dao.observeFeedPaged(listOf("CONTACT_DM", "TIP_DM", "GROUP")).load( + PagingSource.LoadParams.Refresh(null, 10, false) + ) as PagingSource.LoadResult.Page + + assertEquals(listOf("01", "02", "03"), page.data.map { it.chatIdHex }) + } + + @Test + fun `the paged feed leaves out a type the caller did not ask for`() = runTest { + dao.upsert(entity(chatIdHex = "01", chatType = "CONTACT_DM", lastActivityEpochMs = 3_000)) + dao.upsert(entity(chatIdHex = "02", chatType = "GROUP", lastActivityEpochMs = 2_000)) + + val page = dao.observeFeedPaged(listOf("CONTACT_DM")).load( + PagingSource.LoadParams.Refresh(null, 10, false) + ) as PagingSource.LoadResult.Page + + assertEquals(listOf("01"), page.data.map { it.chatIdHex }) + } + + @Test + fun `the paged feed leaves out hidden chats and chats you are not in`() = runTest { + dao.upsert(entity(chatIdHex = "01", chatType = "GROUP", lastActivityEpochMs = 3_000)) + dao.upsert(entity(chatIdHex = "02", chatType = "GROUP", lastActivityEpochMs = 2_000, isHidden = true)) + dao.upsert(entity(chatIdHex = "03", chatType = "GROUP", lastActivityEpochMs = 1_000, isMember = false)) + + val page = dao.observeFeedPaged(listOf("GROUP")).load( + PagingSource.LoadParams.Refresh(null, 10, false) + ) as PagingSource.LoadResult.Page + + assertEquals(listOf("01"), page.data.map { it.chatIdHex }) + } + + /** + * Two chats can share a last-activity timestamp, and `ORDER BY` alone would then leave their + * relative order up to SQLite. Paging asks for the next page by offset, so an unstable order + * duplicates one row and drops the other. + */ + @Test + fun `chats sharing a last activity break the tie on chat id`() = runTest { + dao.upsert(entity(chatIdHex = "01", lastActivityEpochMs = 3_000)) + dao.upsert(entity(chatIdHex = "02", lastActivityEpochMs = 3_000)) + dao.upsert(entity(chatIdHex = "03", lastActivityEpochMs = 3_000)) + + val page = dao.observeFeedPaged(listOf("CONTACT_DM")).load( + PagingSource.LoadParams.Refresh(null, 10, false) + ) as PagingSource.LoadResult.Page + + assertEquals(listOf("03", "02", "01"), page.data.map { it.chatIdHex }) + } + + /** + * What the mediator compares a complete pass against. Only the chats the caller is still in — + * one already cleared is not a removal to rediscover. + */ + @Test + fun `the id listing covers the chats of one type you are still in`() = runTest { + dao.upsert(entity(chatIdHex = "01", chatType = "GROUP")) + dao.upsert(entity(chatIdHex = "02", chatType = "GROUP", isMember = false)) + dao.upsert(entity(chatIdHex = "03", chatType = "CONTACT_DM")) + + assertEquals(listOf("01"), dao.getChatIdsOfType("GROUP")) + } + + @Test + fun `membership can be taken away and given back without touching the row`() = runTest { + dao.upsert(entity(chatType = "GROUP", title = "Flipcash Staff", memberCount = 12, rosterVersion = 4)) + + dao.updateMembership(CHAT_HEX, isMember = false) + + val left = requireNotNull(dao.getById(CHAT_HEX)) + assertEquals(false, left.isMember) + assertEquals("Flipcash Staff", left.title) + assertEquals(12L, left.memberCount) + + dao.updateMembership(CHAT_HEX, isMember = true) + assertEquals(true, dao.getById(CHAT_HEX)?.isMember) + } + + @Test + fun `the roster version reads back, and is null for a chat that is not there`() = runTest { + dao.upsert(entity(chatType = "GROUP", memberCount = 12, rosterVersion = 4)) + + assertEquals(4L, dao.getRosterVersion(CHAT_HEX)) + assertEquals(null, dao.getRosterVersion(OTHER_HEX)) + } + private companion object { const val CHAT_HEX = "aabbccdd" const val OTHER_HEX = "eeff0011" From 4d47d202c386c2a86372527a8a2db1ab9f73fe62 Mon Sep 17 00:00:00 2001 From: Brandon McAnsh Date: Tue, 15 Sep 2026 20:34:18 -0400 Subject: [PATCH 2/9] feat(chat): read the feed by page and write a chat's membership --- .../persistence/sources/build.gradle.kts | 3 ++ .../sources/ChatMemberDataSource.kt | 11 +++++ .../sources/ChatMetadataDataSource.kt | 45 +++++++++++++++++++ .../sources/ChatMetadataDataSourceTest.kt | 42 +++++++++++++++++ 4 files changed, 101 insertions(+) create mode 100644 apps/flipcash/shared/persistence/sources/src/test/kotlin/com/flipcash/app/persistence/sources/ChatMetadataDataSourceTest.kt diff --git a/apps/flipcash/shared/persistence/sources/build.gradle.kts b/apps/flipcash/shared/persistence/sources/build.gradle.kts index bf8a6015ec..7f09ed6ea6 100644 --- a/apps/flipcash/shared/persistence/sources/build.gradle.kts +++ b/apps/flipcash/shared/persistence/sources/build.gradle.kts @@ -14,6 +14,9 @@ android { } dependencies { + testImplementation(kotlin("test")) + testImplementation(libs.bundles.unit.testing) + implementation(libs.bundles.kotlinx.serialization) implementation(libs.bundles.room) diff --git a/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/ChatMemberDataSource.kt b/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/ChatMemberDataSource.kt index 089f609ebf..a3e753e2eb 100644 --- a/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/ChatMemberDataSource.kt +++ b/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/ChatMemberDataSource.kt @@ -109,6 +109,17 @@ class ChatMemberDataSource @Inject constructor( } } + /** + * Drops one member of [chatId]. What a `MemberLeft` roster change applies — [replaceMembers] + * cannot, because a roster change names who left rather than who remains. + */ + suspend fun deleteMember(chatId: ChatId, userId: ID) { + db?.chatMemberDao()?.deleteMember( + chatIdHex = mapper.chatIdHex(chatId), + userIdHex = mapper.userIdHex(userId), + ) + } + suspend fun deleteForChat(chatId: ChatId) { db?.chatMemberDao()?.deleteForChat(mapper.chatIdHex(chatId)) } diff --git a/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/ChatMetadataDataSource.kt b/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/ChatMetadataDataSource.kt index 7c1bec7889..e6b3f57b41 100644 --- a/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/ChatMetadataDataSource.kt +++ b/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/ChatMetadataDataSource.kt @@ -1,5 +1,6 @@ package com.flipcash.app.persistence.sources +import androidx.paging.PagingSource import com.flipcash.app.persistence.FlipcashDatabase import com.flipcash.app.persistence.entities.ChatMetadataEntity import com.flipcash.app.persistence.sources.mapper.chat.ChatEntityMapper @@ -25,6 +26,44 @@ class ChatMetadataDataSource @Inject constructor( fun observeAll(): Flow> = db?.chatMetadataDao()?.observeAll() ?: emptyFlow() + /** + * The merged feed of [chatTypes] as a Paging source. + * + * Unlike every other read here, a closed database cannot answer with an empty result: Paging + * would present that as "no conversations" and stop asking. An error keeps the list in its + * loading state and lets the caller retry once the database is open. + */ + fun observeFeedPaged(chatTypes: List): PagingSource = + db?.chatMetadataDao()?.observeFeedPaged(chatTypes.map { it.name }) + ?: emptyPagingSource() + + /** The chats of [chatType] this device still considers the user a member of, keyed by hex. */ + suspend fun getChatIdsOfType(chatType: ChatType): List = + db?.chatMetadataDao()?.getChatIdsOfType(chatType.name).orEmpty() + + suspend fun setMembership(chatId: ChatId, isMember: Boolean) { + setMembership(mapper.chatIdHex(chatId), isMember) + } + + suspend fun setMembership(chatIdHex: String, isMember: Boolean) { + db?.chatMetadataDao()?.updateMembership(chatIdHex, isMember) + } + + /** The roster version already applied to [chatId]; 0 when the chat is not stored yet. */ + suspend fun getRosterVersion(chatId: ChatId): Long = + db?.chatMetadataDao()?.getRosterVersion(mapper.chatIdHex(chatId)) ?: 0L + + suspend fun updateRoster(chatId: ChatId, memberCount: Long, rosterVersion: Long) { + db?.chatMetadataDao()?.updateRosterIfNewer( + chatIdHex = mapper.chatIdHex(chatId), + memberCount = memberCount, + rosterVersion = rosterVersion, + ) + } + + /** The hex a [chatId] is keyed by. Callers comparing against [getChatIdsOfType] need it. */ + fun chatIdHex(chatId: ChatId): String = mapper.chatIdHex(chatId) + suspend fun upsert(metadata: ChatMetadata) { db?.chatMetadataDao()?.upsert(mapper.toEntity(metadata)) } @@ -82,4 +121,10 @@ class ChatMetadataDataSource @Inject constructor( suspend fun clear() { db?.chatMetadataDao()?.deleteAll() } + + private fun emptyPagingSource() = object : PagingSource() { + override fun getRefreshKey(state: androidx.paging.PagingState): Int? = null + override suspend fun load(params: LoadParams): LoadResult = + LoadResult.Error(Exception("Database not initialized")) + } } diff --git a/apps/flipcash/shared/persistence/sources/src/test/kotlin/com/flipcash/app/persistence/sources/ChatMetadataDataSourceTest.kt b/apps/flipcash/shared/persistence/sources/src/test/kotlin/com/flipcash/app/persistence/sources/ChatMetadataDataSourceTest.kt new file mode 100644 index 0000000000..6c7df7c2e5 --- /dev/null +++ b/apps/flipcash/shared/persistence/sources/src/test/kotlin/com/flipcash/app/persistence/sources/ChatMetadataDataSourceTest.kt @@ -0,0 +1,42 @@ +package com.flipcash.app.persistence.sources + +import androidx.paging.PagingSource +import com.flipcash.app.persistence.entities.ChatMetadataEntity +import com.flipcash.app.persistence.sources.mapper.chat.ChatEntityMapper +import com.flipcash.services.models.chat.ChatId +import com.flipcash.services.models.chat.ChatType +import kotlinx.coroutines.test.runTest +import org.junit.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +/** + * The per-user database is opened at login, so every read here can be attempted before it + * exists. The established answer is a null-safe fallback per method; a paged read is the one + * case where the fallback cannot be an empty result — Paging would render "no conversations" + * and never ask again. + */ +class ChatMetadataDataSourceTest { + + private val dataSource = ChatMetadataDataSource(ChatEntityMapper()) + + @Test + fun `the paged feed reports an error rather than an empty list when the database is closed`() = runTest { + val result = dataSource.observeFeedPaged(listOf(ChatType.GROUP)).load( + PagingSource.LoadParams.Refresh(null, 10, false) + ) + + assertTrue(result is PagingSource.LoadResult.Error) + } + + @Test + fun `reads that have a meaningful empty answer keep returning one`() = runTest { + assertEquals(emptyList(), dataSource.getChatIdsOfType(ChatType.GROUP)) + assertEquals(0L, dataSource.getRosterVersion(ChatId(listOf(0xAB.toByte())))) + } + + @Test + fun `a chat id renders as the hex the tables are keyed by`() { + assertEquals("ab", dataSource.chatIdHex(ChatId(listOf(0xAB.toByte())))) + } +} From d3e0203c3a9849931fcdc1311f6044ead3b991a5 Mon Sep 17 00:00:00 2001 From: Brandon McAnsh Date: Tue, 15 Sep 2026 20:35:24 -0400 Subject: [PATCH 3/9] feat(chat): track how far each source of a merged feed has been paged --- .../sources/mediator/FeedWatermark.kt | 98 +++++++++++++ .../sources/mediator/FeedWatermarkTest.kt | 134 ++++++++++++++++++ 2 files changed, 232 insertions(+) create mode 100644 apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/mediator/FeedWatermark.kt create mode 100644 apps/flipcash/shared/persistence/sources/src/test/kotlin/com/flipcash/app/persistence/sources/mediator/FeedWatermarkTest.kt diff --git a/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/mediator/FeedWatermark.kt b/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/mediator/FeedWatermark.kt new file mode 100644 index 0000000000..f808d117d1 --- /dev/null +++ b/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/mediator/FeedWatermark.kt @@ -0,0 +1,98 @@ +package com.flipcash.app.persistence.sources.mediator + +import com.flipcash.services.models.PagingToken + +/** One source's position in a merged feed. */ +data class FeedSource( + val key: String, + /** What to pass to fetch this source's next page. */ + val token: PagingToken? = null, + /** Oldest last-activity (epoch ms) this source has produced, or null before its first page. */ + val tail: Long? = null, + /** The source reported its last page; there is nothing below [tail]. */ + val exhausted: Boolean = false, + /** The source's last fetch failed. Skipped until it answers again. */ + val failed: Boolean = false, +) + +/** + * Tracks how far each source of a merged feed has been paged, and what that says about the + * merged list as a whole. + * + * Every source orders by most recent activity first and pages with its own opaque token, so + * their pages do not line up: one page of DMs may span a week and one page of groups a year. + * Fetching a page from each and concatenating is only correct down to the point where all of + * them have reported — below the shallowest tail, a source that has not been asked yet may + * still hold an item that belongs higher than one already fetched. + * + * So loading more does not mean "one more page of everything". It means advancing whichever + * source is currently holding the merged list back, which is the non-exhausted source that has + * been paged the least far. + */ +class FeedWatermark(keys: List) { + + private var sources: Map = + keys.associateWithTo(LinkedHashMap()) { FeedSource(key = it) } + + /** Every source's current position, in the order the sources were declared. */ + val all: List get() = sources.values.toList() + + /** The sources whose last fetch failed. */ + val failedKeys: List get() = sources.values.filter { it.failed }.map { it.key } + + /** True once every source has reported its last page. */ + val isComplete: Boolean get() = sources.values.all { it.exhausted } + + /** + * The oldest activity timestamp the merged list is complete down to, or null while a source + * that can still produce items has not produced any yet. + * + * Exhausted sources are excluded: one with nothing left below its tail constrains nothing. + */ + val watermark: Long? + get() { + val live = sources.values.filterNot { it.exhausted } + if (live.any { it.tail == null }) return null + return live.mapNotNull { it.tail }.maxOrNull() + } + + /** + * The source to page next, or null when nothing can be paged — either because every source + * has run out (see [isComplete]) or because the ones left have failed. + */ + fun next(): FeedSource? { + val candidates = sources.values.filterNot { it.exhausted || it.failed } + // A source with no tail caps the watermark at "unknown", so it goes ahead of any source + // that has at least reported where it got to. + return candidates.firstOrNull { it.tail == null } + ?: candidates.maxByOrNull { requireNotNull(it.tail) } + } + + /** + * Records a page from [key]. [tail] is the oldest last-activity in that page, or null when + * the page was empty — in which case the source keeps the tail it already had, because + * forgetting it would restart the source from the top. + */ + fun advance(key: String, tail: Long?, token: PagingToken?, hasMore: Boolean) { + val source = sources[key] ?: return + sources = sources + ( + key to source.copy( + token = token, + tail = tail ?: source.tail, + exhausted = !hasMore, + failed = false, + ) + ) + } + + /** Records that [key]'s fetch failed. It keeps its position and is skipped by [next]. */ + fun fail(key: String) { + val source = sources[key] ?: return + sources = sources + (key to source.copy(failed = true)) + } + + /** Puts every source back to never-paged, for a refresh. */ + fun reset() { + sources = sources.mapValues { (key, _) -> FeedSource(key = key) } + } +} diff --git a/apps/flipcash/shared/persistence/sources/src/test/kotlin/com/flipcash/app/persistence/sources/mediator/FeedWatermarkTest.kt b/apps/flipcash/shared/persistence/sources/src/test/kotlin/com/flipcash/app/persistence/sources/mediator/FeedWatermarkTest.kt new file mode 100644 index 0000000000..d7b78a9668 --- /dev/null +++ b/apps/flipcash/shared/persistence/sources/src/test/kotlin/com/flipcash/app/persistence/sources/mediator/FeedWatermarkTest.kt @@ -0,0 +1,134 @@ +package com.flipcash.app.persistence.sources.mediator + +import org.junit.Test +import kotlin.test.assertEquals +import kotlin.test.assertFalse +import kotlin.test.assertNull +import kotlin.test.assertTrue + +/** + * The merge rule for a feed paged from several sources at once. Two independently-paged feeds + * both ordered by recency are only merged correctly down to the shallower of their two tails — + * below that point one source may still be holding items that belong higher than items already + * fetched from the other. + */ +class FeedWatermarkTest { + + private fun watermark() = FeedWatermark(listOf(DM, GROUP)) + + @Test + fun `a source that has never been paged goes first`() { + val subject = watermark() + + assertEquals(DM, subject.next()?.key) + } + + @Test + fun `the shallowest source is the one to page next`() { + val subject = watermark() + subject.advance(DM, tail = 500, token = null, hasMore = true) + subject.advance(GROUP, tail = 100, token = null, hasMore = true) + + // DM stopped at 500 and GROUP has already reached 100, so everything between them is + // known from GROUP and unknown from DM. DM is what the merged list is waiting on. + assertEquals(DM, subject.next()?.key) + } + + @Test + fun `an exhausted source is never paged again`() { + val subject = watermark() + subject.advance(DM, tail = 500, token = null, hasMore = false) + subject.advance(GROUP, tail = 100, token = null, hasMore = true) + + assertEquals(GROUP, subject.next()?.key) + } + + @Test + fun `the watermark ignores a source that has run out`() { + val subject = watermark() + subject.advance(DM, tail = 500, token = null, hasMore = false) + subject.advance(GROUP, tail = 100, token = null, hasMore = true) + + // DM has nothing left below 500, so it constrains nothing. GROUP is the only source + // that can still produce something, and it is complete down to 100. + assertEquals(100L, subject.watermark) + } + + @Test + fun `the watermark is unknown until every live source has produced a tail`() { + val subject = watermark() + subject.advance(DM, tail = 500, token = null, hasMore = true) + + assertNull(subject.watermark) + } + + @Test + fun `pagination is complete only when every source has run out`() { + val subject = watermark() + subject.advance(DM, tail = 500, token = null, hasMore = false) + assertFalse(subject.isComplete) + + subject.advance(GROUP, tail = 100, token = null, hasMore = false) + assertTrue(subject.isComplete) + } + + /** + * One source failing must not stop the others paging — that is the asymmetry this replaces, + * where a CONTACT_DM failure failed the whole sync. It does block completeness, and so blocks + * the removal reconciliation that needs a whole feed to be sound. + */ + @Test + fun `a failed source is skipped but still holds completeness back`() { + val subject = watermark() + subject.advance(DM, tail = 500, token = null, hasMore = true) + subject.fail(GROUP) + + assertEquals(DM, subject.next()?.key) + + subject.advance(DM, tail = 100, token = null, hasMore = false) + assertNull(subject.next()) + assertFalse(subject.isComplete) + assertEquals(listOf(GROUP), subject.failedKeys) + } + + @Test + fun `a source that answers again is no longer failed`() { + val subject = watermark() + subject.fail(GROUP) + subject.advance(GROUP, tail = 100, token = null, hasMore = true) + + assertEquals(emptyList(), subject.failedKeys) + } + + /** + * An empty page still reports whether the source has more; it just has no tail to contribute. + * Taking its non-existent tail would reset the source to "never paged" and restart it. + */ + @Test + fun `an empty page leaves the tail where it was`() { + val subject = watermark() + subject.advance(DM, tail = 500, token = null, hasMore = true) + subject.advance(DM, tail = null, token = null, hasMore = true) + subject.advance(GROUP, tail = 100, token = null, hasMore = true) + + assertEquals(DM, subject.next()?.key) + } + + @Test + fun `reset puts every source back to never paged`() { + val subject = watermark() + subject.advance(DM, tail = 500, token = null, hasMore = false) + subject.fail(GROUP) + + subject.reset() + + assertEquals(DM, subject.next()?.key) + assertFalse(subject.isComplete) + assertEquals(emptyList(), subject.failedKeys) + } + + private companion object { + const val DM = "CONTACT_DM" + const val GROUP = "GROUP" + } +} From febb8e06224f32468706f77389bd8c2680506a62 Mon Sep 17 00:00:00 2001 From: Brandon McAnsh Date: Tue, 15 Sep 2026 20:39:05 -0400 Subject: [PATCH 4/9] feat(chat): page the merged conversation feed by watermark --- .../mediator/ChatFeedRemoteMediator.kt | 162 ++++++++++++++ .../mediator/ChatFeedRemoteMediatorTest.kt | 201 ++++++++++++++++++ 2 files changed, 363 insertions(+) create mode 100644 apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/mediator/ChatFeedRemoteMediator.kt create mode 100644 apps/flipcash/shared/persistence/sources/src/test/kotlin/com/flipcash/app/persistence/sources/mediator/ChatFeedRemoteMediatorTest.kt diff --git a/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/mediator/ChatFeedRemoteMediator.kt b/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/mediator/ChatFeedRemoteMediator.kt new file mode 100644 index 0000000000..716e5b310c --- /dev/null +++ b/apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/mediator/ChatFeedRemoteMediator.kt @@ -0,0 +1,162 @@ +package com.flipcash.app.persistence.sources.mediator + +import androidx.paging.ExperimentalPagingApi +import androidx.paging.LoadType +import androidx.paging.PagingState +import androidx.paging.RemoteMediator +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.PagingToken +import com.flipcash.services.models.QueryOptions +import com.flipcash.services.models.chat.ChatMetadata +import com.flipcash.services.models.chat.ChatType +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.async +import kotlinx.coroutines.awaitAll +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.withContext + +/** Every remaining source of the merged feed failed, so there is nothing left to page. */ +class FeedSourcesUnavailableException(keys: List) : + Exception("No chat feed source could be paged: ${keys.joinToString()}") + +/** + * Pages the merged conversation feed. + * + * Room is the [androidx.paging.PagingSource]; this only decides what to fetch next and writes + * the result into the same tables the feed delegates write, so the list re-emits off the Room + * change like everything else. + * + * Each load advances exactly one source — the one [FeedWatermark] says the merged list is + * waiting on. That is the difference between this and paging each feed a page at a time: two + * feeds ordered by recency do not have pages that line up, so "one page of each" leaves a + * window where one source has items the other has not been asked for. + * + * **Known trade-off.** The Room query is not filtered to the watermark. Rows below it — fetched + * from the deeper source, not yet matched by the shallower one — are rendered, so a chat can + * move up the list when the other source catches up. Filtering would need an observable + * watermark to invalidate the `PagingSource` on, and would show an empty list on a cold start + * until the first page lands. A briefly over-complete list is the cheaper wrong. + */ +@OptIn(ExperimentalPagingApi::class) +class ChatFeedRemoteMediator( + private val chatTypes: List, + private val controller: ChatController, + private val metadataDataSource: ChatMetadataDataSource, + private val memberDataSource: ChatMemberDataSource, + private val messageDataSource: ChatMessageDataSource, +) : RemoteMediator() { + + private val watermark = FeedWatermark(chatTypes.map { it.name }) + + // Every group id this pass has seen, for the reconciliation at the end of it. + private val seenGroupIds = mutableSetOf() + + /** + * The delegates fetch the first page on login, and Room already holds whatever the last + * session left. Refreshing on every subscribe would re-fetch that first page each time the + * list is opened. + */ + override suspend fun initialize(): InitializeAction = InitializeAction.SKIP_INITIAL_REFRESH + + override suspend fun load( + loadType: LoadType, + state: PagingState, + ): MediatorResult { + return try { + when (loadType) { + LoadType.PREPEND -> return MediatorResult.Success(endOfPaginationReached = true) + + LoadType.REFRESH -> { + watermark.reset() + seenGroupIds.clear() + // Concurrent, so a refresh costs the slower source rather than their sum. + coroutineScope { + chatTypes + .map { type -> async { fetch(type, token = null, limit = state.config.pageSize) } } + .awaitAll() + } + } + + LoadType.APPEND -> { + val source = watermark.next() + ?: return if (watermark.isComplete) { + MediatorResult.Success(endOfPaginationReached = true) + } else { + // Only failed sources remain. Reporting the end of pagination here + // would strand the rest of the feed behind a transient error with + // nothing to retry it; an error lets Paging offer the retry. + MediatorResult.Error(FeedSourcesUnavailableException(watermark.failedKeys)) + } + fetch( + chatType = chatTypes.first { it.name == source.key }, + token = source.token, + limit = state.config.pageSize, + ) + } + } + + if (watermark.isComplete) reconcileGroupRemovals() + + MediatorResult.Success(endOfPaginationReached = watermark.isComplete) + } catch (e: Exception) { + MediatorResult.Error(e) + } + } + + private suspend fun fetch(chatType: ChatType, token: PagingToken?, limit: Int) { + val options = QueryOptions(limit = limit, token = token, descending = true) + val page = when (chatType) { + ChatType.GROUP -> controller.getGroupChatFeed(options) + else -> controller.getDmChatFeed(chatType, options) + }.getOrElse { + // One source erroring stops that source advancing and nothing else. It does keep the + // pass from reporting itself complete, which is what holds back the removal + // reconciliation below — that needs a whole feed to be sound. + watermark.fail(chatType.name) + return + } + + persist(page.chats) + + if (chatType == ChatType.GROUP) { + seenGroupIds += page.chats.map { metadataDataSource.chatIdHex(it.chatId) } + } + + watermark.advance( + key = chatType.name, + tail = page.chats.minOfOrNull { it.lastActivity.toEpochMilliseconds() }, + token = page.pagingToken, + hasMore = page.hasMore, + ) + } + + private suspend fun persist(chats: List) { + withContext(Dispatchers.IO) { + metadataDataSource.upsert(chats) + for (chat in chats) { + memberDataSource.upsert(chat.chatId, chat.members) + chat.lastMessage?.let { messageDataSource.upsert(chat.chatId, listOf(it)) } + } + } + } + + /** + * Clears the membership of any group this device holds that a complete pass did not return. + * + * The event stream is the primary path — `MemberLeft` naming the recipient, and `LeaveChat` + * clearing it optimistically. This is the backstop for a removal that happened while the + * stream was down, and it only runs at the end of pagination: a chat missing from a partial + * pass has not been left, it has not been reached. + */ + private suspend fun reconcileGroupRemovals() { + if (ChatType.GROUP !in chatTypes) return + for (chatIdHex in metadataDataSource.getChatIdsOfType(ChatType.GROUP)) { + if (chatIdHex in seenGroupIds) continue + metadataDataSource.setMembership(chatIdHex, isMember = false) + } + } +} diff --git a/apps/flipcash/shared/persistence/sources/src/test/kotlin/com/flipcash/app/persistence/sources/mediator/ChatFeedRemoteMediatorTest.kt b/apps/flipcash/shared/persistence/sources/src/test/kotlin/com/flipcash/app/persistence/sources/mediator/ChatFeedRemoteMediatorTest.kt new file mode 100644 index 0000000000..9a98c82c85 --- /dev/null +++ b/apps/flipcash/shared/persistence/sources/src/test/kotlin/com/flipcash/app/persistence/sources/mediator/ChatFeedRemoteMediatorTest.kt @@ -0,0 +1,201 @@ +package com.flipcash.app.persistence.sources.mediator + +import androidx.paging.ExperimentalPagingApi +import androidx.paging.LoadType +import androidx.paging.PagingConfig +import androidx.paging.PagingState +import androidx.paging.RemoteMediator +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.chat.ChatFeedPage +import com.flipcash.services.models.chat.ChatId +import com.flipcash.services.models.chat.ChatMetadata +import com.flipcash.services.models.chat.ChatType +import io.mockk.Runs +import io.mockk.coEvery +import io.mockk.coVerify +import io.mockk.just +import io.mockk.mockk +import kotlinx.coroutines.test.runTest +import org.junit.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue +import kotlin.time.Instant + +/** + * The merged feed's paging contract. Its job is to advance one source at a time — the one the + * merged list is waiting on — and to hold the removal reconciliation back until the whole feed + * has been seen, since a chat absent from a partial pass has not been left, only not reached. + */ +@OptIn(ExperimentalPagingApi::class) +class ChatFeedRemoteMediatorTest { + + private val controller = mockk() + private val metadataDataSource = mockk(relaxed = true) + private val memberDataSource = mockk(relaxed = true) + private val messageDataSource = mockk(relaxed = true) + + private fun mediator() = ChatFeedRemoteMediator( + chatTypes = listOf(ChatType.CONTACT_DM, ChatType.GROUP), + controller = controller, + metadataDataSource = metadataDataSource, + memberDataSource = memberDataSource, + messageDataSource = messageDataSource, + ) + + private fun chat(hex: String, type: ChatType, lastActivityMs: Long) = ChatMetadata( + chatId = ChatId(hex), + type = type, + members = emptyList(), + lastMessage = null, + lastActivity = Instant.fromEpochMilliseconds(lastActivityMs), + ) + + private fun page(vararg chats: ChatMetadata, hasMore: Boolean) = + ChatFeedPage(chats = chats.toList(), pagingToken = null, hasMore = hasMore) + + private val state = PagingState( + pages = emptyList(), + anchorPosition = null, + config = PagingConfig(pageSize = 20), + leadingPlaceholderCount = 0, + ) + + private fun dmReturns(vararg pages: ChatFeedPage) { + coEvery { controller.getDmChatFeed(ChatType.CONTACT_DM, any()) } returnsMany + pages.map { Result.success(it) } + } + + private fun groupReturns(vararg pages: ChatFeedPage) { + coEvery { controller.getGroupChatFeed(any()) } returnsMany + pages.map { Result.success(it) } + } + + @Test + fun `a refresh pages every source`() = runTest { + dmReturns(page(chat("01", ChatType.CONTACT_DM, 500), hasMore = true)) + groupReturns(page(chat("02", ChatType.GROUP, 100), hasMore = true)) + + val result = mediator().load(LoadType.REFRESH, state) + + coVerify(exactly = 1) { controller.getDmChatFeed(ChatType.CONTACT_DM, any()) } + coVerify(exactly = 1) { controller.getGroupChatFeed(any()) } + assertEquals(false, (result as RemoteMediator.MediatorResult.Success).endOfPaginationReached) + } + + @Test + fun `a refresh persists what it fetched`() = runTest { + val dm = chat("01", ChatType.CONTACT_DM, 500) + dmReturns(page(dm, hasMore = false)) + groupReturns(page(hasMore = false)) + + mediator().load(LoadType.REFRESH, state) + + coVerify { metadataDataSource.upsert(listOf(dm)) } + coVerify { memberDataSource.upsert(dm.chatId, emptyList()) } + } + + @Test + fun `an append advances only the source the list is waiting on`() = runTest { + dmReturns( + page(chat("01", ChatType.CONTACT_DM, 500), hasMore = true), + page(chat("03", ChatType.CONTACT_DM, 50), hasMore = true), + ) + groupReturns(page(chat("02", ChatType.GROUP, 100), hasMore = true)) + + val subject = mediator() + subject.load(LoadType.REFRESH, state) + subject.load(LoadType.APPEND, state) + + // The group feed has already reached 100 and the DM feed stopped at 500, so the DM feed + // is the one with unknown items in between. + coVerify(exactly = 2) { controller.getDmChatFeed(ChatType.CONTACT_DM, any()) } + coVerify(exactly = 1) { controller.getGroupChatFeed(any()) } + } + + @Test + fun `pagination ends only once every source has run out`() = runTest { + dmReturns(page(chat("01", ChatType.CONTACT_DM, 500), hasMore = false)) + groupReturns( + page(chat("02", ChatType.GROUP, 100), hasMore = true), + page(chat("03", ChatType.GROUP, 50), hasMore = false), + ) + + val subject = mediator() + val refresh = subject.load(LoadType.REFRESH, state) + assertEquals(false, (refresh as RemoteMediator.MediatorResult.Success).endOfPaginationReached) + + val append = subject.load(LoadType.APPEND, state) + assertEquals(true, (append as RemoteMediator.MediatorResult.Success).endOfPaginationReached) + } + + @Test + fun `a prepend has nothing to do`() = runTest { + val result = mediator().load(LoadType.PREPEND, state) + + assertEquals(true, (result as RemoteMediator.MediatorResult.Success).endOfPaginationReached) + } + + /** + * Replaces the old asymmetry, where a CONTACT_DM failure failed the whole sync and a TIP_DM + * failure was swallowed. Every source now fails on its own terms. + */ + @Test + fun `one source failing does not stop the others`() = runTest { + val dm = chat("01", ChatType.CONTACT_DM, 500) + dmReturns(page(dm, hasMore = false)) + coEvery { controller.getGroupChatFeed(any()) } returns Result.failure(Throwable("unreachable")) + + val result = mediator().load(LoadType.REFRESH, state) + + coVerify { metadataDataSource.upsert(listOf(dm)) } + assertEquals(false, (result as RemoteMediator.MediatorResult.Success).endOfPaginationReached) + } + + @Test + fun `an append with only failed sources left surfaces the failure`() = runTest { + dmReturns(page(chat("01", ChatType.CONTACT_DM, 500), hasMore = false)) + coEvery { controller.getGroupChatFeed(any()) } returns Result.failure(Throwable("unreachable")) + + val subject = mediator() + subject.load(LoadType.REFRESH, state) + val result = subject.load(LoadType.APPEND, state) + + assertTrue(result is RemoteMediator.MediatorResult.Error) + } + + @Test + fun `a group missing from a complete pass loses its membership`() = runTest { + val stillThere = chat("01", ChatType.GROUP, 500) + dmReturns(page(hasMore = false)) + groupReturns(page(stillThere, hasMore = false)) + coEvery { metadataDataSource.chatIdHex(stillThere.chatId) } returns "01" + coEvery { metadataDataSource.getChatIdsOfType(ChatType.GROUP) } returns listOf("01", "02") + coEvery { metadataDataSource.setMembership(any(), any()) } just Runs + + mediator().load(LoadType.REFRESH, state) + + coVerify { metadataDataSource.setMembership("02", false) } + coVerify(exactly = 0) { metadataDataSource.setMembership("01", false) } + } + + /** + * A chat below the point the pass reached has not been left — it has not been looked at. + * Reconciling on a partial pass would clear the membership of every conversation past the + * first page. + */ + @Test + fun `an incomplete pass reconciles nothing`() = runTest { + dmReturns(page(hasMore = true)) + groupReturns(page(chat("01", ChatType.GROUP, 500), hasMore = true)) + coEvery { metadataDataSource.chatIdHex(any()) } returns "01" + + mediator().load(LoadType.REFRESH, state) + + coVerify(exactly = 0) { metadataDataSource.getChatIdsOfType(any()) } + coVerify(exactly = 0) { metadataDataSource.setMembership(any(), any()) } + } +} From 348479a88ae63439c88662b89072a3e45c96e41b Mon Sep 17 00:00:00 2001 From: Brandon McAnsh Date: Tue, 15 Sep 2026 20:41:14 -0400 Subject: [PATCH 5/9] feat(chat): project a group into the conversation list --- .../flipcash/shared/chat/ChatCoordinator.kt | 13 +- .../shared/chat/internal/FeedProjection.kt | 54 ++++ .../internal/delegates/FeedSyncDelegate.kt | 47 ++-- .../shared/chat/FeedGroupProjectionTest.kt | 236 ++++++++++++++++++ 4 files changed, 315 insertions(+), 35 deletions(-) create mode 100644 apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/FeedProjection.kt create mode 100644 apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/FeedGroupProjectionTest.kt diff --git a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/ChatCoordinator.kt b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/ChatCoordinator.kt index d09195b2db..d0b4cf076a 100644 --- a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/ChatCoordinator.kt +++ b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/ChatCoordinator.kt @@ -23,11 +23,16 @@ 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> + /** + * 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> - /** Emits the number of [chatType] conversations that have unread messages. */ - fun observeUnreadConversations(chatType: ChatType): Flow + /** Emits the number of conversations of any of [chatTypes] that have unread messages. */ + fun observeUnreadConversations(vararg chatTypes: ChatType): Flow /** Triggers a server-side feed sync. Safe to call redundantly. */ fun refreshFeed() diff --git a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/FeedProjection.kt b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/FeedProjection.kt new file mode 100644 index 0000000000..9f3b6c1c5e --- /dev/null +++ b/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 +} diff --git a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/FeedSyncDelegate.kt b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/FeedSyncDelegate.kt index addfab9e8a..a5a92bce8f 100644 --- a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/FeedSyncDelegate.kt +++ b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/FeedSyncDelegate.kt @@ -10,12 +10,13 @@ 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.flipcash.shared.chat.ChatHydrationState import com.flipcash.shared.chat.ChatSummary import com.flipcash.shared.chat.FeedOperations import com.flipcash.shared.chat.FeedSyncState import com.flipcash.shared.chat.internal.ChatStateHolder +import com.flipcash.shared.chat.internal.isRenderable +import com.flipcash.shared.chat.internal.unreadCount import com.flipcash.services.user.UserManager import com.getcode.opencode.model.core.ID import com.getcode.utils.TraceType @@ -101,41 +102,22 @@ class FeedSyncDelegate @Inject constructor( // region FeedOperations - override fun feed(chatType: ChatType): Flow> = - stateHolder.state.map { state -> + override fun feed(vararg chatTypes: ChatType): Flow> { + val requested = chatTypes.toSet() + return stateHolder.state.map { state -> val selfId = userManager.accountId val selfPhone = userManager.profile?.verifiedPhoneNumber - val isSelf = { member: ChatMember -> - member.userId == selfId || (selfPhone != null && member.userProfile.verifiedPhoneNumber == selfPhone) - } state.feed - .filter { it.type == chatType } - // Hidden chats (e.g. a DM the user blocked) must not surface in the feed. - .filter { !it.isHidden } - .mapNotNull { metadata -> - val otherMember = metadata.members.firstOrNull { !isSelf(it) } - ?: return@mapNotNull null - - // 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. - if (!isDmAddressable(chatType, otherMember.userProfile)) return@mapNotNull null - - val readPointer = metadata.members - .firstOrNull { it.userId == selfId } - ?.pointers - ?.firstOrNull { it.type == PointerType.READ } - ?.value ?: 0L - - val unreadCount = metadata.lastMessage?.let { lastMsg -> - if (lastMsg.messageId > readPointer && lastMsg.senderId != selfId) 1 else 0 - } ?: 0 - - ChatSummary(metadata = metadata, unreadCount = unreadCount) + .filter { it.type in requested } + .filter { isRenderable(it, selfId, selfPhone) } + .map { metadata -> + ChatSummary(metadata = metadata, unreadCount = unreadCount(metadata, selfId)) } } + } - override fun observeUnreadConversations(chatType: ChatType): Flow { - return feed(chatType).map { summaries -> summaries.count { it.unreadCount > 0 } } + override fun observeUnreadConversations(vararg chatTypes: ChatType): Flow { + return feed(*chatTypes).map { summaries -> summaries.count { it.unreadCount > 0 } } } override fun refreshFeed() { @@ -221,7 +203,10 @@ class FeedSyncDelegate @Inject constructor( metadataEntities: List, membersByChat: Map>, ): List { - return metadataEntities.map { entity -> + // `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. + return metadataEntities.filter { it.isMember }.map { entity -> val members = membersByChat[entity.chatIdHex] ?: emptyList() // 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 diff --git a/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/FeedGroupProjectionTest.kt b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/FeedGroupProjectionTest.kt new file mode 100644 index 0000000000..d013df1607 --- /dev/null +++ b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/FeedGroupProjectionTest.kt @@ -0,0 +1,236 @@ +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.models.chat.RosterSummary +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.every +import io.mockk.mockk +import kotlinx.coroutines.ExperimentalCoroutinesApi +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 + +/** + * A group is not a DM with more people in it. It has no counterparty to resolve a title from, and + * its roster arrives a page at a time, so the signed-in user's own member row — the one the unread + * count is read off — may not be among the members the device holds. + */ +@OptIn(ExperimentalCoroutinesApi::class) +class FeedGroupProjectionTest { + + private val selfId = listOf(1, 2, 3) + private val otherId = listOf(4, 5, 6) + + private val dmHex = "aabbccdd" + private val groupHex = "11223344" + + private fun entity(chatIdHex: String, type: ChatType, isMember: Boolean = true) = + ChatMetadataEntity( + chatIdHex = chatIdHex, + chatType = type.name, + lastActivityEpochMs = 2_000, + lastMessageId = 2, + isMember = isMember, + ) + + private fun message(messageId: Long) = ChatMessage( + messageId = messageId, + senderId = otherId, + content = listOf(MessageContent.Text("hi")), + timestamp = Instant.fromEpochSeconds(messageId), + unreadSeq = messageId, + ) + + private fun selfMember(readPointer: Long) = ChatMember( + userId = selfId, + userProfile = UserProfile.Empty, + pointers = listOf( + MessagePointer( + type = PointerType.READ, + userId = selfId, + value = readPointer, + timestamp = Instant.fromEpochSeconds(1_000), + ) + ), + ) + + private val addressableOther = ChatMember( + userId = otherId, + userProfile = UserProfile.Empty.copy( + displayName = "Ada", + phoneNumber = VerifiableContactMethod("+15551234567", verified = true), + ), + pointers = emptyList(), + ) + + /** Wires a [FeedSyncDelegate] over a fixed set of rows and rosters. */ + private inner class Harness( + entities: List, + members: Map>, + ) { + private val messageDataSource = mockk(relaxed = true).also { + coEvery { it.getLatestVisible(any()) } returns message(2) + } + + private val metadataDataSource = mockk(relaxed = true).also { source -> + every { source.observeAll() } returns flowOf(entities) + every { source.toMetadata(any(), any(), any()) } answers { + val row = firstArg() + ChatMetadata( + chatId = ChatId(row.chatIdHex), + type = ChatType.valueOf(row.chatType), + members = secondArg(), + lastMessage = thirdArg(), + lastActivity = Instant.fromEpochMilliseconds(row.lastActivityEpochMs), + title = row.title, + rosterSummary = RosterSummary(memberCount = row.memberCount, version = row.rosterVersion), + ) + } + } + + private val memberDataSource = mockk(relaxed = true).also { source -> + every { source.observeAll() } returns flowOf(members) + } + + val delegate = FeedSyncDelegate( + chatController = mockk(relaxed = true), + metadataDataSource = metadataDataSource, + messageDataSource = messageDataSource, + memberDataSource = memberDataSource, + stateHolder = ChatStateHolder(), + userManager = mockk(relaxed = true).also { + every { it.accountId } returns selfId + every { it.profile } returns null + }, + ) + + suspend fun feed(scope: TestScope, vararg types: ChatType): List { + delegate.initialize(scope.backgroundScope) + delegate.observeFeedFromDb() + scope.runCurrent() + return delegate.feed(*types).first() + } + } + + @Test + fun `a group with no counterparty still reaches the feed`() = runTest { + val harness = Harness( + entities = listOf(entity(groupHex, ChatType.GROUP)), + members = mapOf(groupHex to listOf(selfMember(readPointer = 2))), + ) + + val feed = harness.feed(this, ChatType.GROUP) + + assertEquals(1, feed.size) + assertEquals(ChatId(groupHex), feed.single().metadata.chatId) + } + + /** + * The roster is paged, so the members the device holds are a slice of a larger list. Treating + * a missing self row as a read pointer of zero would put an unread splat on every group until + * the slice happened to include the signed-in user. + */ + @Test + fun `a group whose roster does not include you reports no unread`() = runTest { + val harness = Harness( + entities = listOf(entity(groupHex, ChatType.GROUP)), + members = mapOf(groupHex to listOf(addressableOther)), + ) + + val feed = harness.feed(this, ChatType.GROUP) + + assertEquals(0, feed.single().unreadCount) + } + + @Test + fun `a group counts unread against your own read pointer`() = runTest { + val harness = Harness( + entities = listOf(entity(groupHex, ChatType.GROUP)), + members = mapOf(groupHex to listOf(selfMember(readPointer = 1), addressableOther)), + ) + + val feed = harness.feed(this, ChatType.GROUP) + + assertEquals(1, feed.single().unreadCount) + } + + /** DM behaviour is unchanged: an absent self row still reads as a pointer of zero. */ + @Test + fun `a DM whose roster does not include you still counts unread`() = runTest { + val harness = Harness( + entities = listOf(entity(dmHex, ChatType.CONTACT_DM)), + members = mapOf(dmHex to listOf(addressableOther)), + ) + + val feed = harness.feed(this, ChatType.CONTACT_DM) + + assertEquals(1, feed.single().unreadCount) + } + + @Test + fun `a group you have left is not in the feed`() = runTest { + val harness = Harness( + entities = listOf(entity(groupHex, ChatType.GROUP, isMember = false)), + members = mapOf(groupHex to listOf(selfMember(readPointer = 2))), + ) + + val feed = harness.feed(this, ChatType.GROUP) + + assertEquals(emptyList(), feed) + } + + @Test + fun `asking for several types returns all of them`() = runTest { + val harness = Harness( + entities = listOf(entity(dmHex, ChatType.CONTACT_DM), entity(groupHex, ChatType.GROUP)), + members = mapOf( + dmHex to listOf(selfMember(readPointer = 2), addressableOther), + groupHex to listOf(selfMember(readPointer = 2)), + ), + ) + + val feed = harness.feed(this, ChatType.CONTACT_DM, ChatType.GROUP) + + assertEquals( + listOf(ChatId(dmHex), ChatId(groupHex)), + feed.map { it.metadata.chatId }, + ) + } + + @Test + fun `asking for one type leaves the others out`() = runTest { + val harness = Harness( + entities = listOf(entity(dmHex, ChatType.CONTACT_DM), entity(groupHex, ChatType.GROUP)), + members = mapOf( + dmHex to listOf(selfMember(readPointer = 2), addressableOther), + groupHex to listOf(selfMember(readPointer = 2)), + ), + ) + + val feed = harness.feed(this, ChatType.CONTACT_DM) + + assertEquals(listOf(ChatId(dmHex)), feed.map { it.metadata.chatId }) + } +} From d95bc86eb56d904749528183adf60d77e164dda0 Mon Sep 17 00:00:00 2001 From: Brandon McAnsh Date: Tue, 15 Sep 2026 20:45:32 -0400 Subject: [PATCH 6/9] feat(chat): page the conversation list off the database --- apps/flipcash/shared/chat/build.gradle.kts | 1 + .../flipcash/shared/chat/ChatCoordinator.kt | 10 ++ .../internal/delegates/FeedSyncDelegate.kt | 49 ++++++ .../com/flipcash/shared/chat/FeedPagedTest.kt | 148 ++++++++++++++++++ gradle/libs.versions.toml | 1 + 5 files changed, 209 insertions(+) create mode 100644 apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/FeedPagedTest.kt diff --git a/apps/flipcash/shared/chat/build.gradle.kts b/apps/flipcash/shared/chat/build.gradle.kts index 89adabc474..6004b3e160 100644 --- a/apps/flipcash/shared/chat/build.gradle.kts +++ b/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) diff --git a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/ChatCoordinator.kt b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/ChatCoordinator.kt index d0b4cf076a..4f9cf6947c 100644 --- a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/ChatCoordinator.kt +++ b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/ChatCoordinator.kt @@ -34,6 +34,16 @@ interface FeedOperations { /** Emits the number of conversations of any of [chatTypes] that have unread messages. */ fun observeUnreadConversations(vararg chatTypes: ChatType): Flow + /** + * 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> + /** Triggers a server-side feed sync. Safe to call redundantly. */ fun refreshFeed() diff --git a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/FeedSyncDelegate.kt b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/FeedSyncDelegate.kt index a5a92bce8f..1c8d89df9a 100644 --- a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/FeedSyncDelegate.kt +++ b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/FeedSyncDelegate.kt @@ -1,9 +1,18 @@ +@file:OptIn(ExperimentalPagingApi::class) + package com.flipcash.shared.chat.internal.delegates +import androidx.paging.ExperimentalPagingApi +import androidx.paging.Pager +import androidx.paging.PagingConfig +import androidx.paging.PagingData +import androidx.paging.filter +import androidx.paging.map 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.app.persistence.sources.mediator.ChatFeedRemoteMediator import com.flipcash.services.controllers.ChatController import com.flipcash.services.models.chat.ChatId import com.flipcash.services.models.chat.ChatMember @@ -62,6 +71,10 @@ class FeedSyncDelegate @Inject constructor( companion object { private const val TAG = "FeedSyncDelegate" + + // The feed's rows are cheap and the merge in the mediator costs a round trip per source + // per page, so this is larger than it would be for a transcript. + private const val FEED_PAGE_SIZE = 20 } sealed interface Event { @@ -120,6 +133,42 @@ class FeedSyncDelegate @Inject constructor( return feed(*chatTypes).map { summaries -> summaries.count { it.unreadCount > 0 } } } + override fun feedPaged(vararg chatTypes: ChatType): Flow> { + val types = chatTypes.toList() + return Pager( + config = PagingConfig(pageSize = FEED_PAGE_SIZE), + remoteMediator = ChatFeedRemoteMediator( + chatTypes = types, + controller = chatController, + metadataDataSource = metadataDataSource, + memberDataSource = memberDataSource, + messageDataSource = messageDataSource, + ), + ) { + metadataDataSource.observeFeedPaged(types) + }.flow.map { page -> + val selfId = userManager.accountId + val selfPhone = userManager.profile?.verifiedPhoneNumber + page + .map { entity -> toSummary(entity, selfId) } + // After the map, not before: the rules read a ChatMetadata, and PagingData carries + // entities until this point. + .filter { isRenderable(it.metadata, selfId, selfPhone) } + } + } + + /** + * Rebuilds one row into a [ChatSummary], the same way [buildFeedFromDb] rebuilds the whole + * list — the members and the newest visible message are separate reads because the row holds + * neither. + */ + private suspend fun toSummary(entity: ChatMetadataEntity, selfId: ID?): ChatSummary { + val members = memberDataSource.getMembersForChat(entity.chatIdHex) + val lastMessage = entity.lastMessageId?.let { messageDataSource.getLatestVisible(entity.chatIdHex) } + val metadata = metadataDataSource.toMetadata(entity, members, lastMessage) + return ChatSummary(metadata = metadata, unreadCount = unreadCount(metadata, selfId)) + } + override fun refreshFeed() { syncFeed() } diff --git a/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/FeedPagedTest.kt b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/FeedPagedTest.kt new file mode 100644 index 0000000000..22bcb2ee45 --- /dev/null +++ b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/FeedPagedTest.kt @@ -0,0 +1,148 @@ +package com.flipcash.shared.chat + +import androidx.paging.PagingSource +import androidx.paging.PagingState +import androidx.paging.testing.asSnapshot +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.chat.ChatFeedPage +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.every +import io.mockk.mockk +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.test.runTest +import org.junit.Test +import kotlin.test.assertEquals +import kotlin.time.Instant + +/** + * The paged feed reads the same rows from the same table as [FeedSyncDelegate.feed], through + * Room's `PagingSource` instead of the whole list in state. The projection is shared, so what is + * checked here is the wiring: rows come out as [ChatSummary], and a row the projection rejects + * does not. + */ +@OptIn(ExperimentalCoroutinesApi::class) +class FeedPagedTest { + + private val selfId = listOf(1, 2, 3) + private val otherId = listOf(4, 5, 6) + private val groupHex = "11223344" + private val hiddenHex = "55667788" + + private fun entity(chatIdHex: String, isHidden: Boolean = false) = ChatMetadataEntity( + chatIdHex = chatIdHex, + chatType = ChatType.GROUP.name, + lastActivityEpochMs = 2_000, + lastMessageId = 2, + isHidden = isHidden, + ) + + private val selfMember = ChatMember( + userId = selfId, + userProfile = UserProfile.Empty, + pointers = listOf( + MessagePointer( + type = PointerType.READ, + userId = selfId, + value = 1, + timestamp = Instant.fromEpochSeconds(1_000), + ) + ), + ) + + private val lastMessage = ChatMessage( + messageId = 2, + senderId = otherId, + content = listOf(MessageContent.Text("hi")), + timestamp = Instant.fromEpochSeconds(2), + unreadSeq = 2, + ) + + /** A `PagingSource` over a fixed list, standing in for the Room-generated one. */ + private class FixedPagingSource( + private val rows: List, + ) : PagingSource() { + override fun getRefreshKey(state: PagingState): Int? = null + override suspend fun load(params: LoadParams): LoadResult = + LoadResult.Page(data = rows, prevKey = null, nextKey = null) + } + + private fun delegate(rows: List): FeedSyncDelegate { + val metadataDataSource = mockk(relaxed = true).also { source -> + every { source.observeFeedPaged(any()) } returns FixedPagingSource(rows) + every { source.toMetadata(any(), any(), any()) } answers { + val row = firstArg() + ChatMetadata( + chatId = ChatId(row.chatIdHex), + type = ChatType.valueOf(row.chatType), + members = secondArg(), + lastMessage = thirdArg(), + lastActivity = Instant.fromEpochMilliseconds(row.lastActivityEpochMs), + isHidden = row.isHidden, + ) + } + } + + val memberDataSource = mockk(relaxed = true).also { + coEvery { it.getMembersForChat(any()) } returns listOf(selfMember) + } + + val messageDataSource = mockk(relaxed = true).also { + coEvery { it.getLatestVisible(any()) } returns lastMessage + } + + // Explicit rather than relaxed: `Result` is a value class, so a relaxed mock hands the + // mediator an Object it cannot cast, and the paging boundary call blows up. + val chatController = mockk(relaxed = true).also { + coEvery { it.getGroupChatFeed(any()) } returns + Result.success(ChatFeedPage(chats = emptyList(), pagingToken = null, hasMore = false)) + } + + return FeedSyncDelegate( + chatController = chatController, + metadataDataSource = metadataDataSource, + messageDataSource = messageDataSource, + memberDataSource = memberDataSource, + stateHolder = ChatStateHolder(), + userManager = mockk(relaxed = true).also { + every { it.accountId } returns selfId + every { it.profile } returns null + }, + ) + } + + @Test + fun `a paged row comes out as a summary with its unread count`() = runTest { + val page = delegate(listOf(entity(groupHex))) + .feedPaged(ChatType.GROUP) + .asSnapshot() + + assertEquals(1, page.size) + assertEquals(ChatId(groupHex), page.single().metadata.chatId) + assertEquals(1, page.single().unreadCount) + } + + @Test + fun `a hidden row is left out of the paged feed`() = runTest { + val page = delegate(listOf(entity(hiddenHex, isHidden = true))) + .feedPaged(ChatType.GROUP) + .asSnapshot() + + assertEquals(emptyList(), page) + } +} diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 4a14781fd2..e6776837c1 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -148,6 +148,7 @@ navigation3-ui = { module = "androidx.navigation3:navigation3-ui", version.ref = lifecycle-viewmodel-navigation3 = { module = "androidx.lifecycle:lifecycle-viewmodel-navigation3", version.ref = "navigation3-lifecycle" } androidx-browser = { module = "androidx.browser:browser", version.ref = "androidx-browser" } androidx-paging-runtime = { module = "androidx.paging:paging-runtime-ktx", version.ref = "androidx-paging" } +androidx-paging-testing = { module = "androidx.paging:paging-testing", version.ref = "androidx-paging" } androidx-room-runtime = { module = "androidx.room:room-runtime", version.ref = "androidx-room" } androidx-room-compiler = { module = "androidx.room:room-compiler", version.ref = "androidx-room" } androidx-room-ktx = { module = "androidx.room:room-ktx", version.ref = "androidx-room" } From 13bbba0da98a335529f91e5662781e3d7e525524 Mon Sep 17 00:00:00 2001 From: Brandon McAnsh Date: Tue, 15 Sep 2026 20:46:35 -0400 Subject: [PATCH 7/9] feat(chat): apply roster changes by version and refetch past a gap --- .../shared/chat/internal/RosterStateHolder.kt | 98 +++++++++++ .../shared/chat/RosterStateHolderTest.kt | 152 ++++++++++++++++++ 2 files changed, 250 insertions(+) create mode 100644 apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/RosterStateHolder.kt create mode 100644 apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/RosterStateHolderTest.kt diff --git a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/RosterStateHolder.kt b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/RosterStateHolder.kt new file mode 100644 index 0000000000..8cd27dbf6e --- /dev/null +++ b/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) { + 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" + } +} diff --git a/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/RosterStateHolderTest.kt b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/RosterStateHolderTest.kt new file mode 100644 index 0000000000..3cbc3cfd9c --- /dev/null +++ b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/RosterStateHolderTest.kt @@ -0,0 +1,152 @@ +package com.flipcash.shared.chat + +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.UserProfile +import com.flipcash.services.models.chat.ChatId +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.RosterChange +import com.flipcash.services.models.chat.RosterSummary +import com.flipcash.shared.chat.internal.RosterStateHolder +import io.mockk.coEvery +import io.mockk.coVerify +import io.mockk.mockk +import kotlinx.coroutines.test.runTest +import org.junit.Test +import kotlin.time.Instant + +/** + * A roster change applies only when its version is newer than the stored one, and a version that + * skips ahead means the device missed a change it cannot reconstruct — the repair is a refetch, + * not an apply. + */ +class RosterStateHolderTest { + + private val chatId = ChatId("11223344") + private val joinerId = listOf(7, 7, 7) + private val leaverId = listOf(8, 8, 8) + + private val controller = mockk(relaxed = true) + private val metadataDataSource = mockk(relaxed = true) + private val memberDataSource = mockk(relaxed = true) + + private val subject = RosterStateHolder( + chatController = controller, + metadataDataSource = metadataDataSource, + memberDataSource = memberDataSource, + ) + + private fun joiner(userId: List = joinerId) = ChatMember( + userId = userId, + userProfile = UserProfile.Empty.copy(displayName = "Ada"), + pointers = emptyList(), + ) + + private fun joined(version: Long, memberCount: Long = 13) = RosterChange.MemberJoined( + member = joiner(), + metadata = null, + rosterSummary = RosterSummary(memberCount = memberCount, version = version), + ) + + private fun left(version: Long, memberCount: Long = 11) = RosterChange.MemberLeft( + userId = leaverId, + rosterSummary = RosterSummary(memberCount = memberCount, version = version), + ) + + private fun storedVersion(version: Long) { + coEvery { metadataDataSource.getRosterVersion(chatId) } returns version + } + + @Test + fun `a join one version ahead is applied`() = runTest { + storedVersion(4) + + subject.apply(chatId, listOf(joined(version = 5))) + + coVerify { memberDataSource.upsert(chatId, listOf(joiner())) } + coVerify { metadataDataSource.updateRoster(chatId, memberCount = 13, rosterVersion = 5) } + } + + @Test + fun `a leave one version ahead removes the member`() = runTest { + storedVersion(4) + + subject.apply(chatId, listOf(left(version = 5))) + + coVerify { memberDataSource.deleteMember(chatId, leaverId) } + coVerify { metadataDataSource.updateRoster(chatId, memberCount = 11, rosterVersion = 5) } + } + + @Test + fun `a change older than the stored version is dropped`() = runTest { + storedVersion(9) + + subject.apply(chatId, listOf(joined(version = 4))) + + coVerify(exactly = 0) { memberDataSource.upsert(any(), any()) } + coVerify(exactly = 0) { metadataDataSource.updateRoster(any(), any(), any()) } + } + + /** Re-delivery is normal on a stream that guarantees at-least-once. */ + @Test + fun `a change at the stored version is dropped`() = runTest { + storedVersion(5) + + subject.apply(chatId, listOf(joined(version = 5))) + + coVerify(exactly = 0) { memberDataSource.upsert(any(), any()) } + } + + @Test + fun `a version that skips ahead refetches the chat instead of applying`() = runTest { + storedVersion(4) + val refetched = ChatMetadata( + chatId = chatId, + type = ChatType.GROUP, + members = listOf(joiner()), + lastMessage = null, + lastActivity = Instant.fromEpochSeconds(1_000), + rosterSummary = RosterSummary(memberCount = 20, version = 9), + ) + coEvery { controller.getChat(chatId) } returns Result.success(refetched) + + subject.apply(chatId, listOf(joined(version = 7))) + + coVerify { controller.getChat(chatId) } + coVerify { metadataDataSource.upsert(refetched) } + coVerify { memberDataSource.upsert(chatId, refetched.members) } + // The skipped change is not applied on top of what the refetch returned. + coVerify(exactly = 0) { metadataDataSource.updateRoster(chatId, 13, 7) } + } + + @Test + fun `a failed refetch leaves the stored roster alone`() = runTest { + storedVersion(4) + coEvery { controller.getChat(chatId) } returns Result.failure(Throwable("unreachable")) + + subject.apply(chatId, listOf(joined(version = 7))) + + coVerify(exactly = 0) { metadataDataSource.upsert(any()) } + coVerify(exactly = 0) { metadataDataSource.updateRoster(any(), any(), any()) } + } + + /** + * The stream carries no ordering guarantee for roster changes, so a batch can arrive newest + * first. Applied in arrival order, the older change would look like a gap and trigger a + * refetch for something the device already had. + */ + @Test + fun `a batch is applied in version order`() = runTest { + val versions = mutableListOf(4L, 5L, 6L) + coEvery { metadataDataSource.getRosterVersion(chatId) } answers { versions.removeAt(0) } + + subject.apply(chatId, listOf(joined(version = 6), joined(version = 5))) + + coVerify(exactly = 0) { controller.getChat(any()) } + coVerify { metadataDataSource.updateRoster(chatId, memberCount = 13, rosterVersion = 5) } + coVerify { metadataDataSource.updateRoster(chatId, memberCount = 13, rosterVersion = 6) } + } +} From c5468d272a08e3fe8699731d07f7bf3ea2b6cce4 Mon Sep 17 00:00:00 2001 From: Brandon McAnsh Date: Tue, 15 Sep 2026 20:50:45 -0400 Subject: [PATCH 8/9] feat(chat): sync, join and leave a group through its own delegate --- .../internal/delegates/GroupFeedDelegate.kt | 165 ++++++++++++++ .../shared/chat/GroupFeedDelegateTest.kt | 213 ++++++++++++++++++ 2 files changed, 378 insertions(+) create mode 100644 apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/GroupFeedDelegate.kt create mode 100644 apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/GroupFeedDelegateTest.kt diff --git a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/GroupFeedDelegate.kt b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/GroupFeedDelegate.kt new file mode 100644 index 0000000000..574d13b10f --- /dev/null +++ b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/GroupFeedDelegate.kt @@ -0,0 +1,165 @@ +package com.flipcash.shared.chat.internal.delegates + +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.chat.ChatId +import com.flipcash.services.models.chat.ChatMetadata +import com.flipcash.services.models.chat.RosterChange +import com.flipcash.services.user.UserManager +import com.flipcash.shared.chat.internal.RosterStateHolder +import com.getcode.utils.TraceType +import com.getcode.utils.trace +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Job +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.receiveAsFlow +import kotlinx.coroutines.launch +import javax.inject.Inject +import javax.inject.Singleton + +/** + * Owns the group side of the conversation list: fetching the group feed, joining and leaving, and + * the two roster changes that mean something to the list rather than to the roster. + * + * Beside [FeedSyncDelegate] rather than inside it. The DM sync treats a contact-feed failure as a + * failure of the whole sync; a group-feed failure must not take the DM list down with it, and the + * group feed pages where the DM feed does not. Folding the two together would put a + * `when (chatType)` in every method and make one list's failure mode reachable from the other. + * + * **Cross-delegate communication:** as in [FeedSyncDelegate], work that belongs to another + * delegate is emitted on [events] and routed by + * [RealChatCoordinator][com.flipcash.shared.chat.internal.RealChatCoordinator] rather than called + * directly. + * + * Requires [initialize] with a [CoroutineScope] before [syncGroupFeed] can be launched. + */ +@Singleton +class GroupFeedDelegate @Inject constructor( + private val chatController: ChatController, + private val metadataDataSource: ChatMetadataDataSource, + private val memberDataSource: ChatMemberDataSource, + private val messageDataSource: ChatMessageDataSource, + private val rosterStateHolder: RosterStateHolder, + private val userManager: UserManager, +) { + + sealed interface Event { + /** A group whose transcript the device does not have yet. */ + data class LoadMessages(val chatId: ChatId) : Event + } + + private val _events = Channel(Channel.UNLIMITED) + val events: Flow = _events.receiveAsFlow() + + private var scope: CoroutineScope? = null + private var syncJob: Job? = null + + internal fun initialize(scope: CoroutineScope) { + this.scope = scope + } + + internal fun syncGroupFeed() { + val scope = scope ?: return + syncJob?.cancel() + syncJob = scope.launch { performGroupFeedSync() } + } + + internal fun cancelJobs() { + syncJob?.cancel() + syncJob = null + } + + /** + * Fetches the first page of the group feed and caches it. + * + * Deliberately does not reconcile removals. This is one page, and a group below it has not + * been left — it has not been reached. That comparison needs a whole pass, which is what + * [com.flipcash.app.persistence.sources.mediator.ChatFeedRemoteMediator] knows about and this + * does not. + */ + internal suspend fun performGroupFeedSync() { + val page = chatController.getGroupChatFeed().getOrElse { + trace(tag = TAG, message = "Group feed sync failed", type = TraceType.Error) + return + } + persist(page.chats) + } + + /** + * Joins [chatId], caching the chat the server returns so the list has it before any sync runs. + */ + suspend fun join(chatId: ChatId): Result { + return chatController.joinChat(chatId) + .onSuccess { metadata -> + persist(listOf(metadata)) + metadataDataSource.setMembership(metadata.chatId, isMember = true) + _events.send(Event.LoadMessages(metadata.chatId)) + } + .map { } + } + + /** + * Leaves [chatId]. + * + * The row is cleared first so the chat drops out of the list on the tap rather than a round + * trip later, and put back if the call fails — the same optimistic shape as + * [FeedSyncDelegate.setChatHidden], with the restore that a remote call needs. + */ + suspend fun leave(chatId: ChatId): Result { + metadataDataSource.setMembership(chatId, isMember = false) + return chatController.leaveChat(chatId) + .onFailure { + trace(tag = TAG, message = "Leave failed for $chatId", type = TraceType.Error) + metadataDataSource.setMembership(chatId, isMember = true) + } + } + + /** + * Handles the list-level meaning of [changes], then hands them to [RosterStateHolder] for the + * roster itself. + * + * Two of the changes say something about whether the chat belongs in your list at all: a join + * carrying a full snapshot is your own join, and a departure naming you is your own removal. + * Everything else is a roster edit. + * + * The list arm runs first so a self-join costs no extra round trip: persisting the snapshot + * writes its roster version, which is exactly the version the change carries, so the holder + * sees an already-applied change instead of a gap to refetch. + */ + suspend fun applyRosterChanges(chatId: ChatId, changes: List) { + val selfId = userManager.accountId + for (change in changes) { + when (change) { + is RosterChange.MemberJoined -> { + // Set only when you are the one who joined. Assigned to a local first: a smart + // cast will not hold across a module boundary. + val metadata = change.metadata ?: continue + persist(listOf(metadata)) + metadataDataSource.setMembership(metadata.chatId, isMember = true) + _events.send(Event.LoadMessages(metadata.chatId)) + } + + is RosterChange.MemberLeft -> { + if (change.userId != selfId) continue + metadataDataSource.setMembership(chatId, isMember = false) + } + } + } + rosterStateHolder.apply(chatId, changes) + } + + private suspend fun persist(chats: List) { + metadataDataSource.upsert(chats) + for (chat in chats) { + memberDataSource.upsert(chat.chatId, chat.members) + chat.lastMessage?.let { messageDataSource.upsert(chat.chatId, listOf(it)) } + } + } + + private companion object { + const val TAG = "GroupFeedDelegate" + } +} diff --git a/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/GroupFeedDelegateTest.kt b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/GroupFeedDelegateTest.kt new file mode 100644 index 0000000000..7fe77fe449 --- /dev/null +++ b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/GroupFeedDelegateTest.kt @@ -0,0 +1,213 @@ +package com.flipcash.shared.chat + +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.chat.ChatFeedPage +import com.flipcash.services.models.chat.ChatId +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.RosterChange +import com.flipcash.services.models.chat.RosterSummary +import com.flipcash.services.user.UserManager +import com.flipcash.shared.chat.internal.RosterStateHolder +import com.flipcash.shared.chat.internal.delegates.GroupFeedDelegate +import io.mockk.coEvery +import io.mockk.coVerify +import io.mockk.every +import io.mockk.mockk +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.test.runTest +import org.junit.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue +import kotlin.time.Instant + +/** + * Group membership is a state the device has to hold locally, because the two things that change + * it — your own join and your own removal — reach the client as a roster change on a stream that + * can be down, and as a call whose result the list should not wait for. + */ +@OptIn(ExperimentalCoroutinesApi::class) +class GroupFeedDelegateTest { + + private val chatId = ChatId("11223344") + private val selfId = listOf(1, 2, 3) + private val otherId = listOf(4, 5, 6) + + private val controller = mockk(relaxed = true) + private val metadataDataSource = mockk(relaxed = true) + private val memberDataSource = mockk(relaxed = true) + private val messageDataSource = mockk(relaxed = true) + private val rosterStateHolder = mockk(relaxed = true) + + private val subject = GroupFeedDelegate( + chatController = controller, + metadataDataSource = metadataDataSource, + memberDataSource = memberDataSource, + messageDataSource = messageDataSource, + rosterStateHolder = rosterStateHolder, + userManager = mockk(relaxed = true).also { + every { it.accountId } returns selfId + }, + ) + + private fun member(userId: List) = ChatMember( + userId = userId, + userProfile = UserProfile.Empty.copy(displayName = "Ada"), + pointers = emptyList(), + ) + + private fun group(version: Long = 4) = ChatMetadata( + chatId = chatId, + type = ChatType.GROUP, + members = listOf(member(selfId)), + lastMessage = null, + lastActivity = Instant.fromEpochSeconds(1_000), + title = "Flipcash Staff", + rosterSummary = RosterSummary(memberCount = 12, version = version), + ) + + @Test + fun `a sync persists the groups it fetched`() = runTest { + coEvery { controller.getGroupChatFeed(any()) } returns Result.success( + ChatFeedPage(chats = listOf(group()), pagingToken = null, hasMore = false) + ) + + subject.performGroupFeedSync() + + coVerify { metadataDataSource.upsert(listOf(group())) } + coVerify { memberDataSource.upsert(chatId, group().members) } + } + + /** + * A page is not the whole feed, so a group absent from it has not been left. Reconciliation + * belongs to the paged mediator, which knows when a pass finished. + */ + @Test + fun `a sync does not clear the membership of a group it did not see`() = runTest { + coEvery { controller.getGroupChatFeed(any()) } returns Result.success( + ChatFeedPage(chats = emptyList(), pagingToken = null, hasMore = true) + ) + + subject.performGroupFeedSync() + + coVerify(exactly = 0) { metadataDataSource.setMembership(any(), any()) } + } + + @Test + fun `a failed sync writes nothing`() = runTest { + coEvery { controller.getGroupChatFeed(any()) } returns Result.failure(Throwable("unreachable")) + + subject.performGroupFeedSync() + + coVerify(exactly = 0) { metadataDataSource.upsert(any>()) } + } + + @Test + fun `joining persists the chat and asks for its messages`() = runTest { + coEvery { controller.joinChat(chatId) } returns Result.success(group()) + + val result = subject.join(chatId) + + assertTrue(result.isSuccess) + coVerify { metadataDataSource.upsert(listOf(group())) } + coVerify { memberDataSource.upsert(chatId, group().members) } + coVerify { metadataDataSource.setMembership(chatId, isMember = true) } + assertEquals(GroupFeedDelegate.Event.LoadMessages(chatId), subject.events.first()) + } + + @Test + fun `a failed join writes nothing`() = runTest { + coEvery { controller.joinChat(chatId) } returns Result.failure(Throwable("rules not met")) + + val result = subject.join(chatId) + + assertTrue(result.isFailure) + coVerify(exactly = 0) { metadataDataSource.setMembership(any(), any()) } + } + + /** + * The row is cleared before the call, so the chat leaves the list on the tap rather than a + * round trip later. + */ + @Test + fun `leaving clears the membership before the call returns`() = runTest { + coEvery { controller.leaveChat(chatId) } returns Result.success(Unit) + + subject.leave(chatId) + + coVerify(exactly = 1) { metadataDataSource.setMembership(chatId, isMember = false) } + coVerify(exactly = 0) { metadataDataSource.setMembership(chatId, isMember = true) } + } + + @Test + fun `a failed leave puts the chat back`() = runTest { + coEvery { controller.leaveChat(chatId) } returns Result.failure(Throwable("unreachable")) + + val result = subject.leave(chatId) + + assertTrue(result.isFailure) + coVerify { metadataDataSource.setMembership(chatId, isMember = false) } + coVerify { metadataDataSource.setMembership(chatId, isMember = true) } + } + + /** `metadata` is set only on your own join, and is the whole chat to put in the list. */ + @Test + fun `your own join adds the chat to the list`() = runTest { + val change = RosterChange.MemberJoined( + member = member(selfId), + metadata = group(version = 5), + rosterSummary = RosterSummary(memberCount = 12, version = 5), + ) + + subject.applyRosterChanges(chatId, listOf(change)) + + coVerify { metadataDataSource.upsert(listOf(group(version = 5))) } + coVerify { metadataDataSource.setMembership(chatId, isMember = true) } + assertEquals(GroupFeedDelegate.Event.LoadMessages(chatId), subject.events.first()) + } + + @Test + fun `someone else joining is left to the roster holder`() = runTest { + val change = RosterChange.MemberJoined( + member = member(otherId), + metadata = null, + rosterSummary = RosterSummary(memberCount = 13, version = 5), + ) + + subject.applyRosterChanges(chatId, listOf(change)) + + coVerify(exactly = 0) { metadataDataSource.setMembership(any(), any()) } + coVerify { rosterStateHolder.apply(chatId, listOf(change)) } + } + + @Test + fun `being removed drops the chat out of the list`() = runTest { + val change = RosterChange.MemberLeft( + userId = selfId, + rosterSummary = RosterSummary(memberCount = 11, version = 5), + ) + + subject.applyRosterChanges(chatId, listOf(change)) + + coVerify { metadataDataSource.setMembership(chatId, isMember = false) } + } + + @Test + fun `someone else being removed leaves your membership alone`() = runTest { + val change = RosterChange.MemberLeft( + userId = otherId, + rosterSummary = RosterSummary(memberCount = 11, version = 5), + ) + + subject.applyRosterChanges(chatId, listOf(change)) + + coVerify(exactly = 0) { metadataDataSource.setMembership(any(), any()) } + coVerify { rosterStateHolder.apply(chatId, listOf(change)) } + } +} From f8f2ec57f893832c1a5a693a3660fad828224481 Mon Sep 17 00:00:00 2001 From: Brandon McAnsh Date: Tue, 15 Sep 2026 21:12:27 -0400 Subject: [PATCH 9/9] feat(chat): route roster changes and the group sync through the coordinator --- .../flipcash/shared/chat/ChatCoordinator.kt | 23 +- .../chat/internal/RealChatCoordinator.kt | 26 ++- .../internal/delegates/EventStreamDelegate.kt | 16 ++ .../internal/delegates/GroupFeedDelegate.kt | 7 +- .../chat/ChatCoordinatorEagerBalanceTest.kt | 2 + .../shared/chat/ChatCoordinatorEventsTest.kt | 2 + .../chat/ChatCoordinatorTeardownTest.kt | 2 + .../shared/chat/ChatHistoryHydrationTest.kt | 2 + .../shared/chat/GroupChatRoutingTest.kt | 204 ++++++++++++++++++ .../shared/chat/ReceivedCounterTest.kt | 2 + .../flipcash/shared/chat/ReceivedEventTest.kt | 2 + 11 files changed, 283 insertions(+), 5 deletions(-) create mode 100644 apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/GroupChatRoutingTest.kt diff --git a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/ChatCoordinator.kt b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/ChatCoordinator.kt index 4f9cf6947c..11b6d5abcd 100644 --- a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/ChatCoordinator.kt +++ b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/ChatCoordinator.kt @@ -244,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 + + /** + * Leaves [chatId]. The chat drops out of the list immediately and comes back if the call + * fails. + */ + suspend fun leave(chatId: ChatId): Result +} + +interface ChatCoordinator : + FeedOperations, + EventStreamOperations, + DmChatResolver, + MessagingOperations, + GroupOperations { /** Full observable snapshot of chat state (feed, typing, reactions, active chat). */ val state: StateFlow diff --git a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/RealChatCoordinator.kt b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/RealChatCoordinator.kt index 1028a3518b..8a1f3a4544 100644 --- a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/RealChatCoordinator.kt +++ b/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,9 +76,11 @@ 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, @@ -82,7 +88,8 @@ class RealChatCoordinator @Inject constructor( 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() diff --git a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/EventStreamDelegate.kt b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/EventStreamDelegate.kt index 387aaf37be..2032051dfe 100644 --- a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/EventStreamDelegate.kt +++ b/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) : Event } private val _events = Channel(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()) { diff --git a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/GroupFeedDelegate.kt b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/GroupFeedDelegate.kt index 574d13b10f..21e6289784 100644 --- a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/GroupFeedDelegate.kt +++ b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/GroupFeedDelegate.kt @@ -8,6 +8,7 @@ import com.flipcash.services.models.chat.ChatId import com.flipcash.services.models.chat.ChatMetadata import com.flipcash.services.models.chat.RosterChange import com.flipcash.services.user.UserManager +import com.flipcash.shared.chat.GroupOperations import com.flipcash.shared.chat.internal.RosterStateHolder import com.getcode.utils.TraceType import com.getcode.utils.trace @@ -44,7 +45,7 @@ class GroupFeedDelegate @Inject constructor( private val messageDataSource: ChatMessageDataSource, private val rosterStateHolder: RosterStateHolder, private val userManager: UserManager, -) { +) : GroupOperations { sealed interface Event { /** A group whose transcript the device does not have yet. */ @@ -91,7 +92,7 @@ class GroupFeedDelegate @Inject constructor( /** * Joins [chatId], caching the chat the server returns so the list has it before any sync runs. */ - suspend fun join(chatId: ChatId): Result { + override suspend fun join(chatId: ChatId): Result { return chatController.joinChat(chatId) .onSuccess { metadata -> persist(listOf(metadata)) @@ -108,7 +109,7 @@ class GroupFeedDelegate @Inject constructor( * trip later, and put back if the call fails — the same optimistic shape as * [FeedSyncDelegate.setChatHidden], with the restore that a remote call needs. */ - suspend fun leave(chatId: ChatId): Result { + override suspend fun leave(chatId: ChatId): Result { metadataDataSource.setMembership(chatId, isMember = false) return chatController.leaveChat(chatId) .onFailure { diff --git a/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ChatCoordinatorEagerBalanceTest.kt b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ChatCoordinatorEagerBalanceTest.kt index af581eae0d..c029b1621c 100644 --- a/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ChatCoordinatorEagerBalanceTest.kt +++ b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ChatCoordinatorEagerBalanceTest.kt @@ -131,6 +131,8 @@ class ChatCoordinatorEagerBalanceTest { userManager = userManager, networkObserver = mockk(relaxed = true), dispatchers = testDispatchers, + groupFeedDelegate = mockk(relaxed = true), + featureFlags = mockk(relaxed = true), ) } diff --git a/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ChatCoordinatorEventsTest.kt b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ChatCoordinatorEventsTest.kt index c9757aec71..82846e894b 100644 --- a/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ChatCoordinatorEventsTest.kt +++ b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ChatCoordinatorEventsTest.kt @@ -134,6 +134,8 @@ class ChatCoordinatorEventsTest { userManager = userManager, networkObserver = mockk(relaxed = true), dispatchers = testDispatchers, + groupFeedDelegate = mockk(relaxed = true), + featureFlags = mockk(relaxed = true), ) } diff --git a/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ChatCoordinatorTeardownTest.kt b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ChatCoordinatorTeardownTest.kt index b8c8fc21c9..61d109e430 100644 --- a/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ChatCoordinatorTeardownTest.kt +++ b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ChatCoordinatorTeardownTest.kt @@ -112,6 +112,8 @@ class ChatCoordinatorTeardownTest { userManager = userManager, networkObserver = mockk(relaxed = true), dispatchers = TestDispatchers(TestCoroutineScheduler()), + groupFeedDelegate = mockk(relaxed = true), + featureFlags = mockk(relaxed = true), ) } diff --git a/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ChatHistoryHydrationTest.kt b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ChatHistoryHydrationTest.kt index d8255c5c91..f5c469108a 100644 --- a/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ChatHistoryHydrationTest.kt +++ b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ChatHistoryHydrationTest.kt @@ -144,6 +144,8 @@ class ChatHistoryHydrationTest { userManager = userManager, networkObserver = mockk(relaxed = true), dispatchers = TestDispatchers(testScheduler), + groupFeedDelegate = mockk(relaxed = true), + featureFlags = mockk(relaxed = true), ) } diff --git a/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/GroupChatRoutingTest.kt b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/GroupChatRoutingTest.kt new file mode 100644 index 0000000000..4fa156c2c7 --- /dev/null +++ b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/GroupChatRoutingTest.kt @@ -0,0 +1,204 @@ +package com.flipcash.shared.chat + +import com.flipcash.app.core.dispatchers.TestDispatchers +import com.flipcash.app.featureflags.FeatureFlag +import com.flipcash.app.featureflags.FeatureFlagController +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.app.tokens.TokenCoordinator +import com.flipcash.services.controllers.ChatController +import com.flipcash.services.controllers.ChatMessagingController +import com.flipcash.services.controllers.EventStreamingController +import com.flipcash.services.models.UserProfile +import com.flipcash.services.models.chat.ChatId +import com.flipcash.services.models.chat.ChatMember +import com.flipcash.services.models.chat.ChatUpdate +import com.flipcash.services.models.chat.RosterChange +import com.flipcash.services.models.chat.RosterSummary +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.GroupFeedDelegate +import com.flipcash.shared.chat.internal.delegates.MessagingDelegate +import com.getcode.utils.network.NetworkConnectivityListener +import com.getcode.opencode.model.accounts.AccountCluster +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.runCurrent +import kotlinx.coroutines.test.runTest +import org.junit.Test +import org.junit.runner.RunWith +import org.robolectric.RobolectricTestRunner + +/** + * The group work reaches the rest of the app through exactly two seams: a roster change on the + * event stream, and the login hook. Both are the coordinator's, so both are checked here rather + * than in the delegates that do the work. + */ +@OptIn(ExperimentalCoroutinesApi::class) +@RunWith(RobolectricTestRunner::class) +class GroupChatRoutingTest { + + // Every test tears the coordinator down. The heartbeat the login hook starts is a `while + // (true)` on the test scheduler, so a test that leaves it running advances virtual time + // forever instead of finishing. + + private val chatId = ChatId("11223344") + private val selfId = listOf(1, 2, 3) + private val chatUpdatesChannel = Channel(capacity = Channel.UNLIMITED) + + private val groupFeedDelegate = mockk(relaxed = true) + private val featureFlags = mockk(relaxed = true) + + private val testDispatchers = TestDispatchers(TestCoroutineScheduler()) + + private fun coordinator(): RealChatCoordinator { + val userManager = mockk(relaxed = true).also { + every { it.accountId } returns selfId + } + val eventStreamingController = mockk(relaxed = true).also { + every { it.chatUpdates } returns chatUpdatesChannel.receiveAsFlow() + every { it.isConnected } returns true + every { it.isStreamActive } returns true + } + val chatController = mockk(relaxed = true).also { + coEvery { it.getDmChatFeed(any(), any()) } returns Result.failure(RuntimeException("not needed")) + } + val metadataDataSource = mockk(relaxed = true) + val messageDataSource = mockk(relaxed = true) + val memberDataSource = mockk(relaxed = true) + val stateHolder = ChatStateHolder() + + return RealChatCoordinator( + feedDelegate = FeedSyncDelegate( + chatController = chatController, + metadataDataSource = metadataDataSource, + messageDataSource = messageDataSource, + memberDataSource = memberDataSource, + stateHolder = stateHolder, + userManager = userManager, + ), + eventStreamDelegate = EventStreamDelegate( + eventStreamingController = eventStreamingController, + messagingController = mockk(relaxed = true), + metadataDataSource = metadataDataSource, + messageDataSource = messageDataSource, + memberDataSource = memberDataSource, + tokenCoordinator = mockk(relaxed = true), + userManager = userManager, + stateHolder = stateHolder, + analytics = mockk(relaxed = true), + exchange = mockk(relaxed = true), + ), + dmChatResolverDelegate = DmChatResolverDelegate( + chatIdGenerator = ChatIdGenerator(), + userManager = userManager, + contactDataSource = mockk(relaxed = true), + memberDataSource = memberDataSource, + ), + messagingDelegate = MessagingDelegate( + chatController = chatController, + messagingController = mockk(relaxed = true), + metadataDataSource = metadataDataSource, + messageDataSource = messageDataSource, + memberDataSource = memberDataSource, + notificationManager = mockk(relaxed = true), + userManager = userManager, + stateHolder = stateHolder, + analytics = mockk(relaxed = true), + ), + groupFeedDelegate = groupFeedDelegate, + stateHolder = stateHolder, + userManager = userManager, + networkObserver = mockk(relaxed = true), + featureFlags = featureFlags, + dispatchers = testDispatchers, + ) + } + + private val change = RosterChange.MemberJoined( + member = ChatMember( + userId = listOf(9, 9, 9), + userProfile = UserProfile.Empty.copy(displayName = "Ada"), + pointers = emptyList(), + ), + metadata = null, + rosterSummary = RosterSummary(memberCount = 13, version = 5), + ) + + @Test + fun `a roster change on the stream reaches the group delegate`() = runTest(testDispatchers.dispatcher) { + coEvery { featureFlags.get(FeatureFlag.GroupChats) } returns true + val subject = coordinator() + subject.onUserLoggedIn(mockk(relaxed = true)) + runCurrent() + + chatUpdatesChannel.send(ChatUpdate(chatId = chatId, rosterUpdates = listOf(change))) + runCurrent() + + coVerify { groupFeedDelegate.applyRosterChanges(chatId, listOf(change)) } + subject.teardown() + } + + @Test + fun `an update with no roster change does not reach the group delegate`() = runTest(testDispatchers.dispatcher) { + coEvery { featureFlags.get(FeatureFlag.GroupChats) } returns true + val subject = coordinator() + subject.onUserLoggedIn(mockk(relaxed = true)) + runCurrent() + + chatUpdatesChannel.send(ChatUpdate(chatId = chatId)) + runCurrent() + + coVerify(exactly = 0) { groupFeedDelegate.applyRosterChanges(any(), any()) } + subject.teardown() + } + + @Test + fun `logging in syncs the group feed when the flag is on`() = runTest(testDispatchers.dispatcher) { + coEvery { featureFlags.get(FeatureFlag.GroupChats) } returns true + + val subject = coordinator() + subject.onUserLoggedIn(mockk(relaxed = true)) + runCurrent() + + coVerify { groupFeedDelegate.syncGroupFeed() } + subject.teardown() + } + + @Test + fun `logging in does not sync the group feed when the flag is off`() = runTest(testDispatchers.dispatcher) { + coEvery { featureFlags.get(FeatureFlag.GroupChats) } returns false + + val subject = coordinator() + subject.onUserLoggedIn(mockk(relaxed = true)) + runCurrent() + + coVerify(exactly = 0) { groupFeedDelegate.syncGroupFeed() } + subject.teardown() + } + + @Test + fun `teardown cancels the group delegate's jobs`() = runTest(testDispatchers.dispatcher) { + coEvery { featureFlags.get(FeatureFlag.GroupChats) } returns true + val subject = coordinator() + subject.onUserLoggedIn(mockk(relaxed = true)) + runCurrent() + + subject.teardown() + + coVerify { groupFeedDelegate.cancelJobs() } + } +} diff --git a/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ReceivedCounterTest.kt b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ReceivedCounterTest.kt index 6bf34ce80d..a349a99aad 100644 --- a/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ReceivedCounterTest.kt +++ b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ReceivedCounterTest.kt @@ -152,6 +152,8 @@ class ReceivedCounterTest { userManager = userManager, networkObserver = mockk(relaxed = true), dispatchers = testDispatchers, + groupFeedDelegate = mockk(relaxed = true), + featureFlags = mockk(relaxed = true), ) } diff --git a/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ReceivedEventTest.kt b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ReceivedEventTest.kt index 8ffa9ce541..1b75fda32f 100644 --- a/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ReceivedEventTest.kt +++ b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ReceivedEventTest.kt @@ -155,6 +155,8 @@ class ReceivedEventTest { userManager = userManager, networkObserver = mockk(relaxed = true), dispatchers = testDispatchers, + groupFeedDelegate = mockk(relaxed = true), + featureFlags = mockk(relaxed = true), ) }