fix(chat): create the member row when recording a read pointer

`ChatMemberDao.updatePointers` was a bare `UPDATE ... WHERE chat_id_hex = ?
AND user_id_hex = ?`, which matches nothing when the member row is absent, so
the pointer was dropped without a trace. Both callers can reach it in that
state: a read is written the moment a message is on screen, which can beat the
feed sync that writes the membership, and the event stream applies pointer
updates for members the feed has not written yet.

It is now `advancePointer`, which inserts the row when it is missing and
otherwise carries the member's other pointers over. The read-merge-write moved
into the DAO with it, so a feed sync landing between the read and the write can
no longer be lost.

Author: bmc08gt

apps/flipcash/shared/persistence/db/src/main/kotlin/com/flipcash/app/persistence/dao/ChatMemberDao.kt

@@ -72,8 +72,31 @@ 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)
+    /**
+     * Records [pointer] for [userIdHex] in [chatIdHex], creating the member row if it is not
+     * there yet.
+     *
+     * 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. Only the member's pointer of the same type is displaced; the rest carry
+     * over.
+     */
+    @Transaction
+    suspend fun advancePointer(
+        chatIdHex: String,
+        userIdHex: String,
+        pointer: MessagePointerSerialized,
+    ) {
+        val existing = getMember(chatIdHex, userIdHex)?.pointersJson.orEmpty()
+        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)

apps/flipcash/shared/persistence/db/src/test/kotlin/com/flipcash/app/persistence/dao/ChatMemberDaoTest.kt

@@ -53,11 +53,7 @@ class ChatMemberDaoTest {
     @Test
     fun `upsert does not rewind a locally advanced read pointer`() = runTest {
         dao.upsert(member(readPointer(1)))
-        dao.updatePointers(
-            CHAT_HEX,
-            SELF_HEX,
-            """[{"type":"READ","userIdHex":"$SELF_HEX","value":9,"timestampEpochSeconds":0}]""",
-        )
+        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)))
@@ -87,6 +83,25 @@ class ChatMemberDaoTest {
         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 `deleteMembersNotIn drops departed members and leaves the rest intact`() = runTest {
         dao.upsert(member(readPointer(9)))

apps/flipcash/shared/persistence/sources/src/main/kotlin/com/flipcash/app/persistence/sources/ChatMemberDataSource.kt

@@ -69,19 +69,19 @@ class ChatMemberDataSource @Inject constructor(
         }
     }
 
+    /**
+     * Records [pointer] for its member in [chatId], whether or not that member has synced yet.
+     *
+     * 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()
-
-        val merged = existingPointers
-            .filter { it.type != pointer.type.name }
-            .plus(mapper.pointerSerialized(pointer))
-
-        dao.updatePointers(chatIdHex, userIdHex, mapper.pointersToJson(merged))
+        dao.advancePointer(
+            chatIdHex = mapper.chatIdHex(chatId),
+            userIdHex = mapper.userIdHex(pointer.userId),
+            pointer = mapper.pointerSerialized(pointer),
+        )
     }
 
     /**