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 535b70adc3..1028a3518b 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 @@ -145,6 +145,8 @@ class RealChatCoordinator @Inject constructor( messagingDelegate.loadMessages(event.chatId) is FeedSyncDelegate.Event.DeltaSyncNeeded -> eventStreamDelegate.performDeltaSync(event.chatId) + is FeedSyncDelegate.Event.ReadPointerUnreported -> + messagingDelegate.reportReadPointer(event.chatId, event.messageId) // Arrives after every catch-up item above it, because this is one sequential // collector over a FIFO channel. Anything reading chat history as evidence — // the wallet's "send a tip" milestone — waits for this rather than for the 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 13109f4812..f2123b6bf7 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 @@ -29,17 +29,21 @@ import com.getcode.utils.TraceType import com.getcode.utils.trace import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job +import kotlinx.coroutines.awaitCancellation import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.delay import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.distinctUntilChanged import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.firstOrNull import kotlinx.coroutines.flow.map import kotlinx.coroutines.flow.receiveAsFlow import kotlinx.coroutines.launch +import kotlinx.coroutines.withTimeoutOrNull import javax.inject.Inject import javax.inject.Singleton import kotlin.time.Clock +import kotlin.time.Duration import kotlin.time.Duration.Companion.seconds /** @@ -86,6 +90,8 @@ class EventStreamDelegate @Inject constructor( companion object { private const val TAG = "EventStreamDelegate" private val GAP_FILL_DELAY = 2.seconds + private val HEARTBEAT_INTERVAL = 30.seconds + private val REOPEN_BACKOFF_BASE = 1.seconds } sealed interface Event { @@ -164,22 +170,57 @@ class EventStreamDelegate @Inject constructor( eventStreamingController.close() } + /** + * Supervises the stream: whenever it is found down, syncs the feed and reopens it. + * + * Each pass waits for whichever comes first, a [HEARTBEAT_INTERVAL] tick or + * [EventStreamingController.streamFailures] reporting a stream that ended with nothing + * retrying it. The failure signal is what keeps such a stream from costing a whole interval: + * a status outside the reconnect loop's retryable set, or a ping timeout, is never retried by + * the loop, and the tick was the only thing that noticed. + * + * Reopening backs off from the second consecutive attempt on, so a stream that fails as soon + * as it opens cannot spin the loop. The backoff tops out at [HEARTBEAT_INTERVAL], which is + * where every failure used to wait. + */ internal fun startHeartbeat(onReconnect: () -> Unit) { val scope = scope ?: return stopHeartbeat() heartbeatJob = scope.launch { + var consecutiveReopens = 0 + while (true) { - delay(30.seconds) - if (!eventStreamingController.isStreamActive) { - trace(tag = TAG, message = "Heartbeat: event stream dead, syncing feed and reconnecting", type = TraceType.Process) - onReconnect() - eventStreamingController.close() - open() + // A failures flow that completes means no signal is coming, not that one just + // arrived — waiting out the tick keeps this a supervisor rather than a spin. + val failed = withTimeoutOrNull(HEARTBEAT_INTERVAL) { + eventStreamingController.streamFailures.firstOrNull() ?: awaitCancellation() + } != null + + // Only the failure-driven wake can arrive back to back; the tick already spaces + // itself out. Liveness is read after the wait, so a stream another trigger + // reopened in the meantime is left alone. + if (failed) delay(reopenBackoff(consecutiveReopens)) + + if (eventStreamingController.isStreamActive) { + consecutiveReopens = 0 + continue } + consecutiveReopens++ + + trace(tag = TAG, message = "Event stream down, syncing feed and reconnecting", type = TraceType.Process) + onReconnect() + eventStreamingController.close() + open() } } } + private fun reopenBackoff(consecutiveReopens: Int): Duration = when { + consecutiveReopens <= 0 -> Duration.ZERO + else -> (REOPEN_BACKOFF_BASE * (1 shl (consecutiveReopens - 1).coerceAtMost(5))) + .coerceAtMost(HEARTBEAT_INTERVAL) + } + internal fun stopHeartbeat() { heartbeatJob?.cancel() heartbeatJob = null @@ -301,8 +342,10 @@ class EventStreamDelegate @Inject constructor( when (metaUpdate) { is MetadataUpdate.FullRefresh -> { metadataDataSource.upsert(metaUpdate.metadata) - memberDataSource.deleteForChat(metaUpdate.metadata.chatId) - memberDataSource.upsert(metaUpdate.metadata.chatId, metaUpdate.metadata.members) + memberDataSource.replaceMembers( + metaUpdate.metadata.chatId, + metaUpdate.metadata.members, + ) metaUpdate.metadata.lastMessage?.let { msg -> messageDataSource.upsert(metaUpdate.metadata.chatId, listOf(msg)) } 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 60fe42e866..cf5cd9f282 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 @@ -17,6 +17,7 @@ import com.flipcash.shared.chat.FeedOperations import com.flipcash.shared.chat.FeedSyncState import com.flipcash.shared.chat.internal.ChatStateHolder import com.flipcash.services.user.UserManager +import com.getcode.opencode.model.core.ID import com.getcode.utils.TraceType import com.getcode.utils.trace import kotlinx.coroutines.CoroutineScope @@ -71,6 +72,12 @@ class FeedSyncDelegate @Inject constructor( */ data class DeltaSyncNeeded(val chatId: ChatId) : Event + /** + * The client's own READ pointer for [chatId] is ahead of the copy the feed just returned, + * so the server never took the advance. The consumer re-reports [messageId]. + */ + data class ReadPointerUnreported(val chatId: ChatId, val messageId: Long) : Event + /** * Emitted last by every successful sync, after any catch-up above it. * @@ -184,6 +191,32 @@ class FeedSyncDelegate @Inject constructor( feedObserverJob = null } + /** + * Emits [Event.ReadPointerUnreported] when the stored READ pointer for [chat] is ahead of the + * one the feed just returned. + * + * A read is written locally the moment the message is on screen and reported to the server + * afterwards, and nothing retries a report that fails. The local pointer survives — the member + * row keeps whichever value is further ahead — so the two copies can disagree indefinitely, + * leaving every other device and the pushes this one receives treating the chat as unread. A + * feed payload is the server's own copy, so the sync is where they can be compared. + */ + private suspend fun reportUnreportedRead(chat: ChatMetadata, selfId: ID) { + val server = chat.members + .firstOrNull { it.userId == selfId } + ?.pointers + ?.firstOrNull { it.type == PointerType.READ } + ?.value + ?: 0L + + // Read after the merge above, so this is max(local, server): ahead of `server` only when + // a local advance never reached it. + val local = memberDataSource.getSelfReadPointer(chat.chatId, selfId) + if (local > server) { + _events.send(Event.ReadPointerUnreported(chat.chatId, local)) + } + } + private suspend fun buildFeedFromDb( metadataEntities: List, membersByChat: Map>, @@ -229,7 +262,11 @@ class FeedSyncDelegate @Inject constructor( stateHolder.update { it.copy(feedSyncState = FeedSyncState.Synced) } trace(tag = TAG, message = "Feed synced: ${chats.size} chats", type = TraceType.Process) + val selfId = userManager.accountId + for (chat in chats) { + if (selfId != null) reportUnreportedRead(chat, selfId) + // The applied cursor, not the presence of messages, is what says whether a // transcript was ever pulled: the loop above persists each chat's last-message // preview, so "has messages" is true for nearly every chat in a feed the client diff --git a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/MessagingDelegate.kt b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/MessagingDelegate.kt index 4babb09c8c..01d2a08fab 100644 --- a/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/MessagingDelegate.kt +++ b/apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/delegates/MessagingDelegate.kt @@ -28,6 +28,8 @@ import com.flipcash.shared.chat.MessagingOperations import com.flipcash.shared.chat.internal.ChatStateHolder import com.flipcash.services.user.UserManager import com.getcode.opencode.model.core.ID +import com.getcode.utils.TraceType +import com.getcode.utils.trace import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.distinctUntilChanged @@ -235,6 +237,22 @@ class MessagingDelegate @Inject constructor( return messagingController.advancePointer(chatId, PointerType.READ, messageId) } + /** + * Re-sends a READ pointer the server never took, without touching the local copy or the + * analytics that go with a genuine read — both already happened when the message was seen. + */ + internal suspend fun reportReadPointer(chatId: ChatId, messageId: Long) { + messagingController.advancePointer(chatId, PointerType.READ, messageId) + .onFailure { + trace( + tag = TAG, + message = "Re-reporting read pointer $messageId failed", + type = TraceType.Error, + error = it, + ) + } + } + override suspend fun markAsRead(chatId: ChatId): Result { val messageId = stateHolder.current.feed .firstOrNull { it.chatId == chatId } @@ -292,4 +310,8 @@ class MessagingDelegate @Inject constructor( } // endregion + + private companion object { + const val TAG = "MessagingDelegate" + } } diff --git a/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/EventStreamHeartbeatTest.kt b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/EventStreamHeartbeatTest.kt new file mode 100644 index 0000000000..f0d7207c40 --- /dev/null +++ b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/EventStreamHeartbeatTest.kt @@ -0,0 +1,106 @@ +package com.flipcash.shared.chat + +import com.flipcash.services.controllers.EventStreamingController +import com.flipcash.shared.chat.internal.ChatStateHolder +import com.flipcash.shared.chat.internal.delegates.EventStreamDelegate +import io.mockk.every +import io.mockk.mockk +import io.mockk.verify +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.flow.emptyFlow +import kotlinx.coroutines.flow.receiveAsFlow +import kotlinx.coroutines.test.TestScope +import kotlinx.coroutines.test.advanceTimeBy +import kotlinx.coroutines.test.runCurrent +import kotlinx.coroutines.test.runTest +import org.junit.Test +import org.junit.runner.RunWith +import org.robolectric.RobolectricTestRunner +import kotlin.time.Duration.Companion.milliseconds +import kotlin.time.Duration.Companion.seconds + +/** + * The heartbeat is the only thing that reopens a stream the reconnect loop has stopped retrying — + * a status outside its retryable set, a ping timeout, or its attempts running out. Polling for + * that on a fixed interval costs up to the whole interval in missed messages. + */ +@OptIn(ExperimentalCoroutinesApi::class) +@RunWith(RobolectricTestRunner::class) +class EventStreamHeartbeatTest { + + private class Harness { + val failures = Channel(capacity = Channel.CONFLATED) + + val controller = mockk(relaxed = true).also { controller -> + every { controller.streamFailures } returns failures.receiveAsFlow() + every { controller.chatUpdates } returns emptyFlow() + every { controller.blobUpdates } returns emptyFlow() + // Nothing is connected for the whole test: every wake finds the stream down. + every { controller.isStreamActive } returns false + every { controller.isConnected } returns false + } + + val delegate = EventStreamDelegate( + eventStreamingController = controller, + messagingController = mockk(relaxed = true), + metadataDataSource = mockk(relaxed = true), + messageDataSource = mockk(relaxed = true), + memberDataSource = mockk(relaxed = true), + tokenCoordinator = mockk(relaxed = true), + userManager = mockk(relaxed = true), + stateHolder = ChatStateHolder(), + analytics = mockk(relaxed = true), + exchange = mockk(relaxed = true), + ) + + fun start(scope: TestScope) { + delegate.initialize(scope.backgroundScope) + delegate.startHeartbeat { } + scope.runCurrent() + } + } + + @Test + fun `a stream that gives up is reopened without waiting out the tick`() = runTest { + val harness = Harness() + harness.start(this) + + harness.failures.trySend(Unit) + advanceTimeBy(100.milliseconds) + runCurrent() + + verify(exactly = 1) { harness.controller.open(any()) } + } + + @Test + fun `a stream failing as soon as it opens is not reopened in a hot loop`() = runTest { + val harness = Harness() + harness.start(this) + + harness.failures.trySend(Unit) + advanceTimeBy(100.milliseconds) + runCurrent() + + // The reopened stream fails immediately too. + harness.failures.trySend(Unit) + advanceTimeBy(500.milliseconds) + runCurrent() + verify(exactly = 1) { harness.controller.open(any()) } + + advanceTimeBy(1.seconds) + runCurrent() + verify(exactly = 2) { harness.controller.open(any()) } + } + + @Test + fun `a dead stream is still reopened at the tick`() = runTest { + val harness = Harness() + harness.start(this) + + advanceTimeBy(31.seconds) + runCurrent() + + verify(exactly = 1) { harness.controller.open(any()) } + } +} diff --git a/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/FeedSyncReadPointerReportTest.kt b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/FeedSyncReadPointerReportTest.kt new file mode 100644 index 0000000000..7a87628675 --- /dev/null +++ b/apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/FeedSyncReadPointerReportTest.kt @@ -0,0 +1,137 @@ +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.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.MessagePointer +import com.flipcash.services.models.chat.PointerType +import com.flipcash.services.models.UserProfile +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.launchIn +import kotlinx.coroutines.flow.onEach +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.test.assertTrue +import kotlin.time.Instant + +/** + * The READ pointer is written locally the moment a message is seen and reported to the server + * afterwards, and nothing retries a report that fails. The local copy therefore survives — the + * member row keeps whichever pointer is further ahead — while the server's stays behind, and every + * other device, plus the pushes this one receives, goes on treating the chat as unread. + * + * A feed payload is the server's own copy, so the sync is where the two can be compared. + */ +@OptIn(ExperimentalCoroutinesApi::class) +class FeedSyncReadPointerReportTest { + + private val selfId = listOf(1, 2, 3) + private val chatId = ChatId("aabbccdd") + + private fun member(pointer: Long) = ChatMember( + userId = selfId, + userProfile = UserProfile.Empty, + pointers = listOf( + MessagePointer( + type = PointerType.READ, + userId = selfId, + value = pointer, + timestamp = Instant.fromEpochSeconds(1_000), + ) + ), + ) + + private class Harness(serverPointer: Long, localPointer: Long, selfId: List, chatId: ChatId) { + val chats = listOf( + ChatMetadata( + chatId = chatId, + type = ChatType.TIP_DM, + members = listOf( + ChatMember( + userId = selfId, + userProfile = UserProfile.Empty, + pointers = listOf( + MessagePointer( + type = PointerType.READ, + userId = selfId, + value = serverPointer, + timestamp = Instant.fromEpochSeconds(1_000), + ) + ), + ) + ), + lastMessage = null, + lastActivity = Instant.fromEpochSeconds(1_000), + ) + ) + + val memberDataSource = mockk(relaxed = true).also { source -> + coEvery { source.getSelfReadPointer(any(), any()) } returns localPointer + } + + val chatController = mockk(relaxed = true).also { controller -> + coEvery { controller.getDmChatFeed(ChatType.CONTACT_DM, any()) } returns + Result.success(ChatFeedPage(emptyList(), null, false)) + coEvery { controller.getDmChatFeed(ChatType.TIP_DM, any()) } returns + Result.success(ChatFeedPage(chats, null, false)) + } + + val delegate = FeedSyncDelegate( + chatController = chatController, + metadataDataSource = mockk(relaxed = true), + messageDataSource = mockk(relaxed = true), + memberDataSource = memberDataSource, + stateHolder = ChatStateHolder(), + userManager = mockk(relaxed = true).also { + every { it.accountId } returns selfId + }, + ) + + fun sync(scope: TestScope): List { + val events = mutableListOf() + delegate.events.onEach { events += it }.launchIn(scope.backgroundScope) + delegate.initialize(scope.backgroundScope) + delegate.syncFeed() + scope.runCurrent() + return events + } + } + + @Test + fun `a read the server never took is re-reported`() = runTest { + val events = Harness(serverPointer = 3, localPointer = 9, selfId = selfId, chatId = chatId) + .sync(this) + + assertEquals( + listOf(FeedSyncDelegate.Event.ReadPointerUnreported(chatId, 9)), + events.filterIsInstance(), + "the local pointer is ahead of the server's, so the advance never landed", + ) + } + + @Test + fun `a read the server already has is not re-reported`() = runTest { + val events = Harness(serverPointer = 9, localPointer = 9, selfId = selfId, chatId = chatId) + .sync(this) + + assertTrue( + events.none { it is FeedSyncDelegate.Event.ReadPointerUnreported }, + "the server is level with the client; there is nothing to report", + ) + } +} 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 7f90333dd7..45f52955ef 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 @@ -5,6 +5,7 @@ import androidx.room.Insert import androidx.room.OnConflictStrategy import androidx.room.Query import androidx.room.Transaction +import com.flipcash.app.persistence.converters.MessagePointerSerialized import com.flipcash.app.persistence.entities.ChatMemberEntity import com.flipcash.app.persistence.entities.ChatMemberWithProfile import kotlinx.coroutines.flow.Flow @@ -25,10 +26,32 @@ interface ChatMemberDao { fun observeAll(): Flow> @Insert(onConflict = OnConflictStrategy.REPLACE) - suspend fun upsert(entity: ChatMemberEntity) + suspend fun insertOrReplace(entity: ChatMemberEntity) - @Insert(onConflict = OnConflictStrategy.REPLACE) - suspend fun upsert(entities: List) + /** + * Writes server truth for a member, keeping whichever copy of each pointer is further ahead. + * + * Deliberately not a whole-row REPLACE: `pointers_json` carries the READ pointer, which the + * client advances locally the instant a message is seen and only reports afterwards. A feed + * payload is the server's view as of the fetch, so replacing the row rewinds every advance + * the server has not acknowledged yet — and the chat goes back to reporting unread. + */ + @Transaction + suspend fun upsert(entity: ChatMemberEntity) { + val existing = getMember(entity.chatIdHex, entity.userIdHex) + ?: return insertOrReplace(entity) + + insertOrReplace( + entity.copy( + pointersJson = mergePointers(existing.pointersJson, entity.pointersJson), + ) + ) + } + + @Transaction + suspend fun upsert(entities: List) { + for (entity in entities) upsert(entity) + } @Query("SELECT * FROM chat_members WHERE chat_id_hex = :chatIdHex AND user_id_hex = :userIdHex LIMIT 1") suspend fun getMember(chatIdHex: String, userIdHex: String): ChatMemberEntity? @@ -49,12 +72,74 @@ interface ChatMemberDao { ) suspend fun getChatIdForMember(userIdHex: String, chatType: String): String? - @Query("UPDATE chat_members SET pointers_json = :pointersJson WHERE chat_id_hex = :chatIdHex AND user_id_hex = :userIdHex") - suspend fun updatePointers(chatIdHex: String, userIdHex: String, pointersJson: String) + /** + * Moves [userIdHex]'s pointer of [pointer]'s type forward in [chatIdHex], creating the member + * row if it is not there yet. The member's other pointers carry over. + * + * The row is not a given at this point. A read is written the moment a message is on screen, + * and a pointer update can arrive off the event stream for a member the feed has not written + * yet — the bare `UPDATE` this replaces matched nothing in either case and dropped the + * pointer silently. + * + * Forward only, on the same reasoning as [mergePointers]: the stream echoes a member's + * pointer as the server last saw it, which can be behind a local advance that has not been + * reported yet, and applying it would put the chat back to unread. + */ + @Transaction + suspend fun advancePointer( + chatIdHex: String, + userIdHex: String, + pointer: MessagePointerSerialized, + ) { + val existing = getMember(chatIdHex, userIdHex)?.pointersJson.orEmpty() + val current = existing.firstOrNull { it.type == pointer.type } + if (current != null && current.value >= pointer.value) return + + insertOrReplace( + ChatMemberEntity( + chatIdHex = chatIdHex, + userIdHex = userIdHex, + pointersJson = existing.filterNot { it.type == pointer.type } + pointer, + ) + ) + } @Query("DELETE FROM chat_members WHERE chat_id_hex = :chatIdHex") suspend fun deleteForChat(chatIdHex: String) + /** Drops the members of [chatIdHex] that are no longer in [keepUserIdHexes]. */ + @Query( + "DELETE FROM chat_members WHERE chat_id_hex = :chatIdHex " + + "AND user_id_hex NOT IN (:keepUserIdHexes)" + ) + suspend fun deleteMembersNotIn(chatIdHex: String, keepUserIdHexes: List) + @Query("DELETE FROM chat_members") suspend fun deleteAll() } + +/** + * The pointers a member row should end up holding. + * + * A pointer only ever moves forward on either side — the client bumps its own READ pointer as + * messages are seen, and the server only ever raises the copy it hands back — so taking the + * greater value per (type, member) lets a refresh carry a pointer forward without rewinding a + * local advance it has not been told about yet. + */ +private fun mergePointers( + existing: List?, + incoming: List?, +): List? { + if (existing.isNullOrEmpty()) return incoming + if (incoming.isNullOrEmpty()) return existing + + val merged = existing.associateByTo(LinkedHashMap()) { it.type to it.userIdHex } + for (pointer in incoming) { + val key = pointer.type to pointer.userIdHex + val current = merged[key] + if (current == null || pointer.value > current.value) { + merged[key] = pointer + } + } + return merged.values.toList() +} 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 new file mode 100644 index 0000000000..a95c33805d --- /dev/null +++ b/apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMemberDaoTest.kt @@ -0,0 +1,131 @@ +package com.flipcash.app.persistence.dao + +import android.content.Context +import androidx.room.Room +import androidx.test.core.app.ApplicationProvider +import com.flipcash.app.persistence.FlipcashDatabase +import com.flipcash.app.persistence.converters.MessagePointerSerialized +import com.flipcash.app.persistence.entities.ChatMemberEntity +import kotlinx.coroutines.test.runTest +import org.junit.After +import org.junit.Before +import org.junit.Test +import org.junit.runner.RunWith +import org.robolectric.RobolectricTestRunner +import kotlin.test.assertEquals +import kotlin.test.assertNull + +/** + * Covers what a feed sync is allowed to overwrite on a member row. `pointers_json` holds the + * READ pointer, which the client advances locally the moment a message is seen and only then + * reports to the server. A feed payload is server truth as of the fetch, so a whole-row replace + * rewinds an advance the server has not acknowledged yet and the chat re-reports as unread. + */ +@RunWith(RobolectricTestRunner::class) +class ChatMemberDaoTest { + + private lateinit var db: FlipcashDatabase + private lateinit var dao: ChatMemberDao + + @Before + fun setUp() { + val context = ApplicationProvider.getApplicationContext() + db = Room.inMemoryDatabaseBuilder(context, FlipcashDatabase::class.java) + .allowMainThreadQueries() + .build() + dao = db.chatMemberDao() + } + + @After + fun tearDown() { + db.close() + } + + private fun readPointer(value: Long, userIdHex: String = SELF_HEX) = + MessagePointerSerialized(type = "READ", userIdHex = userIdHex, value = value) + + private fun member(vararg pointers: MessagePointerSerialized) = ChatMemberEntity( + chatIdHex = CHAT_HEX, + userIdHex = SELF_HEX, + pointersJson = pointers.toList(), + ) + + @Test + fun `upsert does not rewind a locally advanced read pointer`() = runTest { + dao.upsert(member(readPointer(1))) + dao.advancePointer(CHAT_HEX, SELF_HEX, readPointer(9)) + + // A feed sync issued before the advance reached the server carries the old pointer. + dao.upsert(member(readPointer(1))) + + assertEquals(9L, dao.getMember(CHAT_HEX, SELF_HEX)?.pointersJson?.single()?.value) + } + + @Test + fun `upsert applies a server pointer that is ahead of the local one`() = runTest { + dao.upsert(member(readPointer(1))) + + // Read on another device: the server is ahead and must win. + dao.upsert(member(readPointer(12))) + + assertEquals(12L, dao.getMember(CHAT_HEX, SELF_HEX)?.pointersJson?.single()?.value) + } + + @Test + fun `upsert keeps a local pointer while applying the other member's`() = runTest { + dao.upsert(member(readPointer(9), readPointer(3, OTHER_HEX))) + + // The server has seen the peer read further, but not yet our own advance to 9. + dao.upsert(member(readPointer(1), readPointer(7, OTHER_HEX))) + + val pointers = dao.getMember(CHAT_HEX, SELF_HEX)?.pointersJson.orEmpty() + assertEquals(9L, pointers.single { it.userIdHex == SELF_HEX }.value) + assertEquals(7L, pointers.single { it.userIdHex == OTHER_HEX }.value) + } + + @Test + fun `advancePointer records a read for a member that has not synced yet`() = runTest { + // The chat is open and a message is on screen before the feed writes the membership. + dao.advancePointer(CHAT_HEX, SELF_HEX, readPointer(4)) + + assertEquals(4L, dao.getMember(CHAT_HEX, SELF_HEX)?.pointersJson?.single()?.value) + } + + @Test + fun `advancePointer leaves the member's other pointers alone`() = runTest { + dao.upsert(member(readPointer(2), MessagePointerSerialized("DELIVERED", SELF_HEX, 5))) + + dao.advancePointer(CHAT_HEX, SELF_HEX, readPointer(6)) + + val pointers = dao.getMember(CHAT_HEX, SELF_HEX)?.pointersJson.orEmpty() + assertEquals(6L, pointers.single { it.type == "READ" }.value) + assertEquals(5L, pointers.single { it.type == "DELIVERED" }.value) + } + + @Test + fun `advancePointer does not lower a pointer that is already ahead`() = runTest { + dao.advancePointer(CHAT_HEX, SELF_HEX, readPointer(9)) + + // The stream echoes our own READ pointer as the server last saw it, behind the local one. + dao.advancePointer(CHAT_HEX, SELF_HEX, readPointer(4)) + + assertEquals(9L, dao.getMember(CHAT_HEX, SELF_HEX)?.pointersJson?.single()?.value) + } + + @Test + fun `deleteMembersNotIn drops departed members and leaves the rest intact`() = runTest { + dao.upsert(member(readPointer(9))) + dao.upsert(member(readPointer(4, OTHER_HEX)).copy(userIdHex = OTHER_HEX)) + + dao.deleteMembersNotIn(CHAT_HEX, listOf(SELF_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" + const val OTHER_HEX = "eeff" + } +} 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 deb9b9470f..089f609ebf 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 @@ -69,19 +69,44 @@ class ChatMemberDataSource @Inject constructor( } } + /** + * Moves [pointer]'s member forward in [chatId], whether or not that member has synced yet. + * A pointer already ahead of [pointer] stays where it is. + * + * The read-merge-write runs inside the DAO so a feed sync landing between the two halves + * cannot be lost. + */ suspend fun updatePointers(chatId: ChatId, pointer: MessagePointer) { val dao = db?.chatMemberDao() ?: return - val chatIdHex = mapper.chatIdHex(chatId) - val userIdHex = mapper.userIdHex(pointer.userId) - - val existing = dao.getMember(chatIdHex, userIdHex) - val existingPointers = existing?.pointersJson ?: emptyList() + dao.advancePointer( + chatIdHex = mapper.chatIdHex(chatId), + userIdHex = mapper.userIdHex(pointer.userId), + pointer = mapper.pointerSerialized(pointer), + ) + } - val merged = existingPointers - .filter { it.type != pointer.type.name } - .plus(mapper.pointerSerialized(pointer)) + /** + * Replaces a chat's membership with [members], keeping the rows that survive instead of + * clearing the table first. A wipe takes each member's pointers with it, and the refresh + * that follows can only restore what the server knew at fetch time — so a read the client + * has advanced locally but not yet reported would be lost. + */ + suspend fun replaceMembers(chatId: ChatId, members: List) { + if (members.isEmpty()) { + deleteForChat(chatId) + return + } - dao.updatePointers(chatIdHex, userIdHex, mapper.pointersToJson(merged)) + val database = db ?: return + val hex = mapper.chatIdHex(chatId) + database.withTransaction { + database.userProfileDao().upsertFull(members.map { mapper.toProfileEntity(it) }) + database.chatMemberDao().upsert(members.map { mapper.toEntity(hex, it) }) + database.chatMemberDao().deleteMembersNotIn( + chatIdHex = hex, + keepUserIdHexes = members.map { mapper.userIdHex(it.userId) }, + ) + } } suspend fun deleteForChat(chatId: ChatId) { diff --git a/services/flipcash/src/main/kotlin/com/flipcash/services/controllers/EventStreamingController.kt b/services/flipcash/src/main/kotlin/com/flipcash/services/controllers/EventStreamingController.kt index e8f48019b3..4a0269142e 100644 --- a/services/flipcash/src/main/kotlin/com/flipcash/services/controllers/EventStreamingController.kt +++ b/services/flipcash/src/main/kotlin/com/flipcash/services/controllers/EventStreamingController.kt @@ -24,6 +24,17 @@ class EventStreamingController @Inject constructor( private val _blobUpdates = Channel(capacity = Channel.UNLIMITED) val blobUpdates: Flow = _blobUpdates.receiveAsFlow() + // Conflated: the signal is "there is no stream any more", which two failures state no better + // than one, and it has to survive being emitted while the consumer is between waits. + private val _streamFailures = Channel(capacity = Channel.CONFLATED) + + /** + * Emits when a stream ends with nothing retrying it — a status outside the reconnect loop's + * retryable set, or the loop exhausting its attempts. Reopening is the consumer's call; this + * only says the connection is gone, so a supervisor need not poll to find out. + */ + val streamFailures: Flow = _streamFailures.receiveAsFlow() + // Guards all reads/writes of [streamRef]. open()/close() are invoked // concurrently from multiple triggers (login, lifecycle onStart, network // reconnect, feature-flag, heartbeat) on the multi-threaded IO scope, so @@ -43,6 +54,10 @@ class EventStreamingController @Inject constructor( return true } + // A pending failure describes the stream being replaced here. Left queued, it would wake + // the supervisor against the new stream before it has had a chance to connect. + while (_streamFailures.tryReceive().isSuccess) Unit + val owner = userManager.accountCluster?.authority?.keyPair ?: run { trace("EventStreamingController: No account cluster, cannot open stream") return false @@ -66,7 +81,10 @@ class EventStreamingController @Inject constructor( // event creates a fresh stream. Only clear if it is still THIS // stream — never null out a newer ref opened after us. synchronized(lock) { - if (streamRef === openedRef) streamRef = null + if (streamRef === openedRef) { + streamRef = null + _streamFailures.trySend(Unit) + } } }, ) diff --git a/services/opencode/src/main/kotlin/com/getcode/opencode/internal/bidi/BidirectionalStreamReference.kt b/services/opencode/src/main/kotlin/com/getcode/opencode/internal/bidi/BidirectionalStreamReference.kt index 309cbdf74d..3a22c086b4 100644 --- a/services/opencode/src/main/kotlin/com/getcode/opencode/internal/bidi/BidirectionalStreamReference.kt +++ b/services/opencode/src/main/kotlin/com/getcode/opencode/internal/bidi/BidirectionalStreamReference.kt @@ -176,4 +176,17 @@ class BidirectionalStreamReference( fun activateStream() { isStreamActive = true } + + /** + * Reports the stream down without tearing the reference down. + * + * [cancel] and [destroy] both cancel [coroutineScope], which on this reference is where + * the reconnect loop itself runs, so neither can express "no live stream right now, still + * retrying". Callers poll [isActive] to decide whether to step in and reopen; without this, + * liveness stays stuck at whatever the last successful activation set and they sit out the + * whole retry window. + */ + fun deactivateStream() { + isStreamActive = false + } } \ No newline at end of file diff --git a/services/opencode/src/main/kotlin/com/getcode/opencode/internal/bidi/OpenStream.kt b/services/opencode/src/main/kotlin/com/getcode/opencode/internal/bidi/OpenStream.kt index fcd2beca4b..7db44b65e7 100644 --- a/services/opencode/src/main/kotlin/com/getcode/opencode/internal/bidi/OpenStream.kt +++ b/services/opencode/src/main/kotlin/com/getcode/opencode/internal/bidi/OpenStream.kt @@ -65,6 +65,12 @@ fun openBidirectionalStream( var attempt = 0 while (attempt++ <= maxReconnectAttempts) { + // Nothing is connected until this attempt's first response activates it. The flag + // has to say so for the whole gap — backoff included — because the callers that + // poll it (the chat heartbeat, lifecycle resume) use it to decide whether the + // stream needs reopening, and a stale `true` makes every one of them a no-op. + streamRef.deactivateStream() + val backoffMs = computeBackoffMs(attempt, reconnectDelayMs, maxReconnectDelayMs) if (backoffMs > 0) { delay(backoffMs) diff --git a/services/opencode/src/test/kotlin/com/getcode/opencode/internal/bidi/OpenBidirectionalStreamTest.kt b/services/opencode/src/test/kotlin/com/getcode/opencode/internal/bidi/OpenBidirectionalStreamTest.kt index 592ceab1b3..572c13a54d 100644 --- a/services/opencode/src/test/kotlin/com/getcode/opencode/internal/bidi/OpenBidirectionalStreamTest.kt +++ b/services/opencode/src/test/kotlin/com/getcode/opencode/internal/bidi/OpenBidirectionalStreamTest.kt @@ -4,6 +4,7 @@ package com.getcode.opencode.internal.bidi import io.grpc.Status import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.awaitCancellation import kotlinx.coroutines.channels.ClosedSendChannelException import kotlinx.coroutines.flow.first import kotlinx.coroutines.flow.flow @@ -11,6 +12,7 @@ import kotlinx.coroutines.test.advanceUntilIdle import kotlinx.coroutines.test.runTest import org.junit.Test import kotlin.test.assertEquals +import kotlin.test.assertFalse import kotlin.test.assertTrue import kotlin.time.Duration.Companion.seconds import kotlin.time.TestTimeSource @@ -257,4 +259,53 @@ class OpenBidirectionalStreamTest { streamRef.destroy() } + + /** + * Liveness must describe the stream that exists right now, not the last one that ever + * connected. The loop retries internally, so between a failure and the next activation + * there is no stream — and every external recovery trigger (the chat heartbeat, lifecycle + * resume, network reconnect) reads this flag to decide whether to step in. + * + * Here the stream activates, fails with UNAVAILABLE, and the retry never emits. The + * reference must report inactive while the loop is parked waiting on that attempt. + */ + @Test + fun `stream reports inactive while the reconnect loop is between attempts`() = runTest { + var attemptCount = 0 + + val streamRef = BidirectionalStreamReference(this, "test-stream") + streamRef.retain() + + openBidirectionalStream>( + streamRef = streamRef, + apiCall = { requestFlow -> + attemptCount++ + flow { + requestFlow.first() + if (attemptCount == 1) { + emit("activation") + throw Status.UNAVAILABLE.asRuntimeException() + } + // The reconnect never gets a response — the stream is down. + awaitCancellation() + } + }, + initialRequest = { "req" }, + responseHandler = { _: String, _: (String) -> Unit -> }, + onError = { }, + reconnectOnUnavailable = true, + maxReconnectAttempts = 3, + reconnectDelayMs = 0, + ) + + advanceUntilIdle() + + assertEquals(2, attemptCount, "Should have reconnected once after UNAVAILABLE") + assertFalse( + streamRef.isActive, + "Reference reports a live stream while the loop is still waiting to reconnect" + ) + + streamRef.destroy() + } }