Skip to content

Commit 3d0d096

Browse files
committed
fix(chat): back chats up by their applied event cursor, not by message presence
A feed sync never backfilled a transcript. It asked `hasMessages` to decide whether a chat had ever been fetched, but the same sync writes each chat's last-message preview as a message row moments earlier — so the answer was always yes, and the newest page was never pulled. Chats opened to a single preview message until the user scrolled. Underneath that sat a worse problem: `latest_event_sequence` was written by two parties meaning two different things. The event stream advanced it to the sequence the client had actually applied; the feed sync overwrote it with the server's reported head via `@Insert(REPLACE)`. Once a sync landed, the next `GetDelta(after:)` resumed from head and silently skipped every event in between. The same whole-row replace also zeroed `analytics_counted_through`, re-counting received messages already counted. The column now has one meaning — the cursor the client has applied: - `ChatMetadataDao.upsert` inserts, or updates only the server-owned columns in place. The two client-owned watermarks are never touched by a sync. - `ChatEntityMapper` stops round-tripping the server head. A row starts at 0, meaning "this transcript has never been fetched"; a chat rebuilt from the database reports 0 for the head — unknown, not "no events". - `MessagingDelegate.loadMessages` seats the cursor at the newest page's frontier, and only ever forward. - The sync's decision reads the cursor: unseated means load the newest page (never a delta, which would re-pull the whole history from sequence 0), behind the server head means delta-sync the missed window. The pre-sync snapshot #1331 added comes out with it. That workaround read the cursor and `hasMessages` *before* the sync wrote, and carried the old cursor on `DeltaSyncNeeded`, because the write clobbered the row it was about to read. Nothing clobbers it now: the sync reads the cursor after its own write, and `performDeltaSync` reads it for itself again. This matches the iOS fix in code-ios-app#628.
1 parent 0b8218c commit 3d0d096

11 files changed

Lines changed: 539 additions & 96 deletions

File tree

‎apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/internal/RealChatCoordinator.kt‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -144,7 +144,7 @@ class RealChatCoordinator @Inject constructor(
144144
is FeedSyncDelegate.Event.LoadMessages ->
145145
messagingDelegate.loadMessages(event.chatId)
146146
is FeedSyncDelegate.Event.DeltaSyncNeeded ->
147-
eventStreamDelegate.performDeltaSync(event.chatId, event.afterSequence)
147+
eventStreamDelegate.performDeltaSync(event.chatId)
148148
// Arrives after every catch-up item above it, because this is one sequential
149149
// collector over a FIFO channel. Anything reading chat history as evidence —
150150
// the wallet's "send a tip" milestone — waits for this rather than for the

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

Lines changed: 6 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -186,16 +186,14 @@ class EventStreamDelegate @Inject constructor(
186186
}
187187

188188
/**
189-
* Backfills a chat from [afterSequence] onward.
189+
* Backfills a chat from its applied cursor onward.
190190
*
191-
* Callers that have already written the server's sequence to `chat_metadata` before asking for
192-
* a delta must pass the sequence the cache held *beforehand* — reading it here would return
193-
* the value they just wrote and ask the server for everything after its own latest event,
194-
* which is always nothing. The live gap-fill path has no such write in front of it and omits
195-
* the argument.
191+
* The cursor is read from `chat_metadata`, which holds what the client has actually applied:
192+
* nothing writes the server's head there — a feed sync refreshes only the server-owned columns
193+
* (`ChatMetadataDao.upsert`), and the mapper drops the head on the way in.
196194
*/
197-
internal suspend fun performDeltaSync(chatId: ChatId, afterSequence: Long? = null) {
198-
val afterSequence = afterSequence ?: metadataDataSource.getLatestEventSequence(chatId)
195+
internal suspend fun performDeltaSync(chatId: ChatId) {
196+
val afterSequence = metadataDataSource.getLatestEventSequence(chatId)
199197
trace(tag = TAG, message = "Delta sync for $chatId from sequence $afterSequence", type = TraceType.Process)
200198

201199
try {

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

Lines changed: 18 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -64,10 +64,11 @@ class FeedSyncDelegate @Inject constructor(
6464
sealed interface Event {
6565
data class LoadMessages(val chatId: ChatId) : Event
6666
/**
67-
* @param afterSequence the sequence the local cache held *before* this sync overwrote it;
68-
* the delta must be requested from there, not from the row we just wrote.
67+
* The chat's applied cursor is behind the feed's head. The consumer reads that cursor from
68+
* `chat_metadata` itself: a feed sync never writes it (see `ChatMetadataDao.upsert`), so
69+
* the row still holds what the client has actually applied.
6970
*/
70-
data class DeltaSyncNeeded(val chatId: ChatId, val afterSequence: Long) : Event
71+
data class DeltaSyncNeeded(val chatId: ChatId) : Event
7172

7273
/**
7374
* Emitted last by every successful sync, after any catch-up above it.
@@ -216,35 +217,10 @@ class FeedSyncDelegate @Inject constructor(
216217
Result.success(contact.chats + (tip?.chats ?: emptyList()))
217218
}
218219

219-
/** What the local cache held for a chat before a sync wrote to it. */
220-
private data class CachedChatState(
221-
val hasMessages: Boolean,
222-
val latestEventSequence: Long,
223-
)
224-
225-
private suspend fun performFeedSync() {
220+
internal suspend fun performFeedSync() {
226221
stateHolder.update { it.copy(feedSyncState = FeedSyncState.Syncing) }
227222
fetchCombinedFeed()
228223
.onSuccess { chats ->
229-
// Snapshot the cache before writing to it. The writes below stamp the server's
230-
// latestEventSequence onto every metadata row and give every chat at least one
231-
// message, so the catch-up checks at the end of this sync — which are reads of
232-
// exactly those two things — would otherwise always find the chat current and
233-
// never fire.
234-
//
235-
// Both branches being inert had the same consequence: after a re-login, where
236-
// logout has cleared the chat cache (ChatCoordinator.reset), a chat is left
237-
// holding only the one message the feed carried. Anything derived from chat
238-
// history then reads as if the rest never happened — the wallet's "send a tip"
239-
// milestone looks for an outgoing TIPPED message and re-shows the new-user
240-
// tutorial to an account that has already tipped.
241-
val cached = chats.associate { chat ->
242-
chat.chatId to CachedChatState(
243-
hasMessages = messageDataSource.hasMessages(chat.chatId),
244-
latestEventSequence = metadataDataSource.getLatestEventSequence(chat.chatId),
245-
)
246-
}
247-
248224
metadataDataSource.upsert(chats)
249225

250226
for (chat in chats) {
@@ -258,19 +234,19 @@ class FeedSyncDelegate @Inject constructor(
258234
trace(tag = TAG, message = "Feed synced: ${chats.size} chats", type = TraceType.Process)
259235

260236
for (chat in chats) {
261-
val before = cached[chat.chatId] ?: continue
262-
if (chat.latestEventSequence > 0) {
263-
if (before.latestEventSequence > 0 &&
264-
before.latestEventSequence < chat.latestEventSequence
265-
) {
266-
_events.send(
267-
Event.DeltaSyncNeeded(chat.chatId, before.latestEventSequence)
268-
)
269-
continue
270-
}
271-
}
272-
if (!before.hasMessages) {
273-
_events.send(Event.LoadMessages(chat.chatId))
237+
// The applied cursor, not the presence of messages, is what says whether a
238+
// transcript was ever pulled: the loop above persists each chat's last-message
239+
// preview, so "has messages" is true for nearly every chat in a feed the client
240+
// has otherwise never fetched. Only a message load or an applied delta seats a
241+
// cursor.
242+
val cursor = metadataDataSource.getLatestEventSequence(chat.chatId)
243+
when {
244+
// Never fetched: take the newest page. Resuming a delta from 0 would instead
245+
// re-pull the entire history as a "gap".
246+
cursor <= 0L -> _events.send(Event.LoadMessages(chat.chatId))
247+
// Fetched, but the server has moved on: stream the missed window.
248+
chat.latestEventSequence > cursor ->
249+
_events.send(Event.DeltaSyncNeeded(chat.chatId))
274250
}
275251
}
276252

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

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -155,6 +155,15 @@ class MessagingDelegate @Inject constructor(
155155
.onSuccess { messages ->
156156
messageDataSource.upsert(chatId, messages)
157157

158+
// Seat the event-log cursor at the newest page's frontier. It is what marks this
159+
// transcript as fetched, and it lets a following catch-up resume from head and
160+
// append genuinely newer messages instead of re-pulling the whole history from
161+
// sequence 0. Only ever advanced — a page older than the cursor must not rewind it.
162+
val head = messages.maxOfOrNull { it.eventSequence } ?: 0L
163+
if (head > metadataDataSource.getLatestEventSequence(chatId)) {
164+
metadataDataSource.updateLatestEventSequence(chatId, head)
165+
}
166+
158167
val latest = messages.maxByOrNull { it.messageId } ?: return@onSuccess
159168
metadataDataSource.updateLastMessageId(chatId, latest.messageId)
160169
metadataDataSource.updateLastActivity(chatId, latest.timestamp.toEpochMilliseconds())
Lines changed: 145 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,145 @@
1+
package com.flipcash.shared.chat
2+
3+
import com.flipcash.app.persistence.sources.ChatMessageDataSource
4+
import com.flipcash.app.persistence.sources.ChatMetadataDataSource
5+
import com.flipcash.services.controllers.ChatController
6+
import com.flipcash.services.models.chat.ChatFeedPage
7+
import com.flipcash.services.models.chat.ChatId
8+
import com.flipcash.services.models.chat.ChatMetadata
9+
import com.flipcash.services.models.chat.ChatType
10+
import com.flipcash.shared.chat.internal.delegates.FeedSyncDelegate
11+
import io.mockk.coEvery
12+
import io.mockk.mockk
13+
import kotlinx.coroutines.ExperimentalCoroutinesApi
14+
import kotlinx.coroutines.launch
15+
import kotlinx.coroutines.test.TestScope
16+
import kotlinx.coroutines.test.UnconfinedTestDispatcher
17+
import kotlinx.coroutines.test.runCurrent
18+
import kotlinx.coroutines.test.runTest
19+
import kotlin.time.Instant
20+
import org.junit.Assert.assertEquals
21+
import org.junit.Test
22+
23+
/**
24+
* Covers which chats a feed sync decides to backfill. The decision is keyed on the locally
25+
* applied event cursor — never on whether the chat holds any message rows, because the sync
26+
* itself persists every chat's last-message preview as a row.
27+
*/
28+
@OptIn(ExperimentalCoroutinesApi::class)
29+
class FeedSyncBackfillTest {
30+
31+
private fun metadata(hex: String, latestEventSequence: Long) = ChatMetadata(
32+
chatId = ChatId(hex),
33+
type = ChatType.CONTACT_DM,
34+
members = emptyList(),
35+
lastMessage = null,
36+
lastActivity = Instant.fromEpochSeconds(1000),
37+
latestEventSequence = latestEventSequence,
38+
)
39+
40+
/**
41+
* Runs one feed sync over a single chat whose server head is [serverHead] and whose locally
42+
* applied cursor is [localCursor], and returns the events it emitted. Every successful sync
43+
* ends with [FeedSyncDelegate.Event.CatchUpComplete], so that marker closes each expectation
44+
* below; what varies is the backfill decision in front of it.
45+
*/
46+
private suspend fun TestScope.backfillEvents(
47+
serverHead: Long,
48+
localCursor: Long,
49+
): List<FeedSyncDelegate.Event> {
50+
val chat = metadata(CHAT_HEX, serverHead)
51+
52+
val chatController = mockk<ChatController>(relaxed = true)
53+
coEvery { chatController.getDmChatFeed(ChatType.CONTACT_DM, any()) } returns
54+
Result.success(ChatFeedPage(listOf(chat), null, false))
55+
coEvery { chatController.getDmChatFeed(ChatType.TIP_DM, any()) } returns
56+
Result.success(ChatFeedPage(emptyList(), null, false))
57+
58+
val metadataDataSource = mockk<ChatMetadataDataSource>(relaxed = true)
59+
coEvery { metadataDataSource.getLatestEventSequence(chat.chatId) } returns localCursor
60+
61+
// Nothing here stubs `hasMessages`: the decision must not consult it. The sync persists
62+
// every chat's last-message preview, so it would answer true even for a chat whose
63+
// transcript was never pulled.
64+
val messageDataSource = mockk<ChatMessageDataSource>(relaxed = true)
65+
66+
val delegate = FeedSyncDelegate(
67+
chatController = chatController,
68+
metadataDataSource = metadataDataSource,
69+
messageDataSource = messageDataSource,
70+
memberDataSource = mockk(relaxed = true),
71+
stateHolder = mockk(relaxed = true),
72+
userManager = mockk(relaxed = true),
73+
)
74+
75+
val received = mutableListOf<FeedSyncDelegate.Event>()
76+
val collector = launch(UnconfinedTestDispatcher(testScheduler)) {
77+
delegate.events.collect { received += it }
78+
}
79+
delegate.performFeedSync()
80+
runCurrent()
81+
collector.cancel()
82+
return received
83+
}
84+
85+
@Test
86+
fun `chat that was never fetched is backfilled with its newest page`() = runTest {
87+
// The sync has just written this chat's last-message preview, so it holds a message row —
88+
// only the unseated cursor reveals that its transcript was never pulled.
89+
val events = backfillEvents(serverHead = 12, localCursor = 0)
90+
91+
assertEquals(
92+
listOf(
93+
FeedSyncDelegate.Event.LoadMessages(ChatId(CHAT_HEX)),
94+
FeedSyncDelegate.Event.CatchUpComplete,
95+
),
96+
events,
97+
)
98+
}
99+
100+
@Test
101+
fun `chat that was never fetched is never delta synced from zero`() = runTest {
102+
// A delta after sequence 0 re-pulls the entire history instead of the newest page.
103+
val events = backfillEvents(serverHead = 0, localCursor = 0)
104+
105+
assertEquals(
106+
listOf(
107+
FeedSyncDelegate.Event.LoadMessages(ChatId(CHAT_HEX)),
108+
FeedSyncDelegate.Event.CatchUpComplete,
109+
),
110+
events,
111+
)
112+
}
113+
114+
@Test
115+
fun `chat behind the server head catches up with a delta sync`() = runTest {
116+
val events = backfillEvents(serverHead = 12, localCursor = 5)
117+
118+
assertEquals(
119+
listOf(
120+
FeedSyncDelegate.Event.DeltaSyncNeeded(ChatId(CHAT_HEX)),
121+
FeedSyncDelegate.Event.CatchUpComplete,
122+
),
123+
events,
124+
)
125+
}
126+
127+
@Test
128+
fun `chat already at the server head needs no backfill`() = runTest {
129+
val events = backfillEvents(serverHead = 12, localCursor = 12)
130+
131+
assertEquals(listOf(FeedSyncDelegate.Event.CatchUpComplete), events)
132+
}
133+
134+
@Test
135+
fun `chat ahead of a stale feed head needs no backfill`() = runTest {
136+
// The stream can apply events the feed page predates; that is not a gap.
137+
val events = backfillEvents(serverHead = 12, localCursor = 20)
138+
139+
assertEquals(listOf(FeedSyncDelegate.Event.CatchUpComplete), events)
140+
}
141+
142+
private companion object {
143+
const val CHAT_HEX = "aabbccdd"
144+
}
145+
}

0 commit comments

Comments
 (0)