feat(notifications): plan push handling as data

.gitignore

@@ -43,3 +43,7 @@ maestro/screenshots/results/
 maestro/screenshots/diffs/
 
 docs/superpowers/
+
+# Raw spike capture: full device logcat, so it can carry unrelated personal
+# content. Keep the summarised results, not the source logs.
+docs/spikes/raw/

apps/flipcash/shared/chat/src/main/kotlin/com/flipcash/shared/chat/ChatCoordinator.kt

@@ -160,6 +160,15 @@ interface MessagingOperations {
     /** Fetches the full message history for [chatId] from the server and persists locally. */
     suspend fun loadMessages(chatId: ChatId)
 
+    /**
+     * Persists a message the server delivered inside a push payload, without an RPC.
+     *
+     * Idempotent on `eventSequence`: a copy that is not newer than the stored row is
+     * dropped, so re-delivery of the same push and a following [loadMessages] converge
+     * on the same transcript.
+     */
+    suspend fun applyPushedMessage(chatId: ChatId, message: ChatMessage)
+
     /**
      * Sends a text message to [chatId]. Returns the server-confirmed [ChatMessage].
      *

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

@@ -51,8 +51,8 @@ import kotlin.time.Duration.Companion.seconds
  * local persistence and in-memory state.
  *
  * Responsibilities:
- * - **Message persistence** — resolves messages from `ChatUpdate.events` (preferred)
- *   or the deprecated `newMessages` field, then upserts to Room.
+ * - **Message persistence** — resolves messages from `ChatUpdate.events`, then
+ *   upserts to Room.
  * - **Gap-aware event sequencing** — uses [EventSequenceTracker] to maintain a
  *   contiguous frontier. Only the highest contiguous sequence is persisted; if a
  *   gap is detected, a timed [getDelta][performDeltaSync] backfill is scheduled.
@@ -290,17 +290,12 @@ class EventStreamDelegate @Inject constructor(
     private suspend fun applyUpdate(update: ChatUpdate) {
         val chatId = update.chatId
 
-        // --- Resolve messages: prefer events, fall back to deprecated newMessages ---
+        // --- Resolve messages from the event log ---
 
-        val resolvedMessages = if (update.events.isNotEmpty()) {
-            update.events
-                .flatMap { event -> event.mutations.map { it.message } }
-                .sortedBy { it.eventSequence }
-                .distinctBy { it.messageId }
-        } else {
-            @Suppress("DEPRECATION")
-            update.newMessages
-        }
+        val resolvedMessages = update.events
+            .flatMap { event -> event.mutations.map { it.message } }
+            .sortedBy { it.eventSequence }
+            .distinctBy { it.messageId }
 
         trace(
             tag = TAG,

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

@@ -193,6 +193,20 @@ class MessagingDelegate @Inject constructor(
             }
     }
 
+    override suspend fun applyPushedMessage(chatId: ChatId, message: ChatMessage) {
+        // The DAO's upsert already drops a copy older than the stored row, so the only
+        // ordering this has to protect is the metadata below it.
+        messageDataSource.upsert(chatId, listOf(message))
+
+        // Deliberately not advancing the event-log cursor. A push carries one message, not
+        // a page, so seating the cursor at its sequence would let a later catch-up resume
+        // from a frontier it never actually fetched and skip whatever it missed in between.
+        if (message.messageId > (metadataDataSource.getLastMessageId(chatId) ?: 0L)) {
+            metadataDataSource.updateLastMessageId(chatId, message.messageId)
+            metadataDataSource.updateLastActivity(chatId, message.timestamp.toEpochMilliseconds())
+        }
+    }
+
     override suspend fun sendMessage(
         chatId: ChatId,
         content: String,

apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ChatCoordinatorEagerBalanceTest.kt

@@ -9,8 +9,10 @@ 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.chat.ChatEvent
 import com.flipcash.services.models.chat.ChatId
 import com.flipcash.services.models.chat.ChatMessage
+import com.flipcash.services.models.chat.ChatMutation
 import com.flipcash.services.models.chat.ChatUpdate
 import com.flipcash.services.models.chat.MessageContent
 import com.flipcash.shared.chat.internal.ChatIdGenerator
@@ -156,12 +158,24 @@ class ChatCoordinatorEagerBalanceTest {
         unreadSeq = 0,
     )
 
+    // Each update carries its messages as event-log events. Sequences run
+    // contiguously from 1 across the whole test so the gap detector stays quiet
+    // and these tests exercise only the behaviour they name.
+    private var nextEventSequence = 1L
+
     private fun chatUpdate(vararg messages: ChatMessage) = ChatUpdate(
         chatId = chatId,
-        newMessages = messages.toList(),
         pointerUpdates = emptyList(),
         typingNotifications = emptyList(),
         metadataUpdates = emptyList(),
+        events = messages.map { message ->
+            ChatEvent(
+                sequence = nextEventSequence++,
+                count = 1,
+                ts = message.timestamp,
+                mutations = listOf(ChatMutation.MessageSent(message)),
+            )
+        },
     )
 
     private suspend fun triggerCollection() {

apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ChatCoordinatorEventsTest.kt

@@ -161,56 +161,7 @@ class ChatCoordinatorEventsTest {
         coordinator.onUserLoggedIn(mockk(relaxed = true))
     }
 
-    // region Events vs newMessages
-
-    @Test
-    fun `events are preferred over deprecated newMessages`() = runTest(testDispatchers.dispatcher) {
-        triggerCollection()
-
-        val eventMsg = textMessage(id = 1, eventSequence = 1)
-        val deprecatedMsg = textMessage(id = 99)
-
-        @Suppress("DEPRECATION")
-        val update = ChatUpdate(
-            chatId = chatId,
-            newMessages = listOf(deprecatedMsg),
-            events = listOf(chatEvent(1, eventMsg)),
-        )
-        chatUpdatesChannel.send(update)
-        advanceTimeBy(1_000.milliseconds)
-        runCurrent()
-
-        // Should upsert the event message, not the deprecated one
-        coVerify {
-            messageDataSource.upsert(chatId, match { messages ->
-                messages.size == 1 && messages[0].messageId == 1L
-            })
-        }
-        coordinator.teardown()
-    }
-
-    @Test
-    fun `falls back to newMessages when events is empty`() = runTest(testDispatchers.dispatcher) {
-        triggerCollection()
-
-        val msg = textMessage(id = 42)
-        @Suppress("DEPRECATION")
-        val update = ChatUpdate(
-            chatId = chatId,
-            newMessages = listOf(msg),
-            events = emptyList(),
-        )
-        chatUpdatesChannel.send(update)
-        advanceTimeBy(1_000.milliseconds)
-        runCurrent()
-
-        coVerify {
-            messageDataSource.upsert(chatId, match { messages ->
-                messages.size == 1 && messages[0].messageId == 42L
-            })
-        }
-        coordinator.teardown()
-    }
+    // region Event message resolution
 
     @Test
     fun `multiple events are flattened and deduped by messageId`() = runTest(testDispatchers.dispatcher) {

apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/MessagingPushedMessageTest.kt

@@ -0,0 +1,135 @@
+package com.flipcash.shared.chat
+
+import com.flipcash.app.persistence.sources.ChatMessageDataSource
+import com.flipcash.app.persistence.sources.ChatMetadataDataSource
+import com.flipcash.services.controllers.ChatMessagingController
+import com.flipcash.services.models.chat.ChatId
+import com.flipcash.services.models.chat.ChatMessage
+import com.flipcash.services.models.chat.MessageContent
+import com.flipcash.shared.chat.internal.delegates.MessagingDelegate
+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
+
+/**
+ * Covers the write a push-carried message performs in place of a fetch.
+ *
+ * Three properties matter here, and none of them are visible from the planner: the write
+ * reaches the same data source a fetched page would, the event cursor stays where it was,
+ * and the feed row only moves forward.
+ */
+class MessagingPushedMessageTest {
+
+    private val chatId = ChatId("aabbccdd")
+
+    private fun message(id: Long, eventSequence: Long, epochSeconds: Long = 1_757_000_000) =
+        ChatMessage(
+            messageId = id,
+            senderId = listOf<Byte>(4, 5, 6),
+            content = listOf(MessageContent.Text("msg-$id")),
+            timestamp = Instant.fromEpochSeconds(epochSeconds),
+            unreadSeq = 0,
+            eventSequence = eventSequence,
+        )
+
+    private fun delegateWith(
+        metadataDataSource: ChatMetadataDataSource,
+        messageDataSource: ChatMessageDataSource = mockk(relaxed = true),
+        messagingController: ChatMessagingController = mockk(relaxed = true),
+    ) = MessagingDelegate(
+        chatController = mockk(relaxed = true),
+        messagingController = messagingController,
+        metadataDataSource = metadataDataSource,
+        messageDataSource = messageDataSource,
+        memberDataSource = mockk(relaxed = true),
+        notificationManager = mockk(relaxed = true),
+        userManager = mockk(relaxed = true),
+        stateHolder = mockk(relaxed = true),
+        analytics = mockk(relaxed = true),
+    )
+
+    @Test
+    fun `a pushed message is written through the same source a fetched page uses`() = runTest {
+        val pushed = message(id = 12, eventSequence = 9)
+        val messageDataSource = mockk<ChatMessageDataSource>(relaxed = true)
+        val metadataDataSource = mockk<ChatMetadataDataSource>(relaxed = true)
+        coEvery { metadataDataSource.getLastMessageId(chatId) } returns 11
+
+        delegateWith(metadataDataSource, messageDataSource).applyPushedMessage(chatId, pushed)
+
+        coVerify(exactly = 1) { messageDataSource.upsert(chatId, listOf(pushed)) }
+    }
+
+    @Test
+    fun `applying a pushed message makes no network call`() = runTest {
+        val messagingController = mockk<ChatMessagingController>(relaxed = true)
+        val metadataDataSource = mockk<ChatMetadataDataSource>(relaxed = true)
+        coEvery { metadataDataSource.getLastMessageId(chatId) } returns 0
+
+        delegateWith(metadataDataSource, messagingController = messagingController)
+            .applyPushedMessage(chatId, message(id = 1, eventSequence = 3))
+
+        coVerify(exactly = 0) { messagingController.getMessages(any(), any()) }
+    }
+
+    @Test
+    fun `a pushed message does not seat the event cursor`() = runTest {
+        val metadataDataSource = mockk<ChatMetadataDataSource>(relaxed = true)
+        coEvery { metadataDataSource.getLastMessageId(chatId) } returns 0
+
+        delegateWith(metadataDataSource).applyPushedMessage(chatId, message(id = 12, eventSequence = 99))
+
+        // A push carries one message, not a page. Seating the cursor at its sequence would let a
+        // later catch-up resume from a frontier it never fetched.
+        coVerify(exactly = 0) { metadataDataSource.updateLatestEventSequence(chatId, any()) }
+    }
+
+    @Test
+    fun `a newer pushed message moves the feed row forward`() = runTest {
+        val metadataDataSource = mockk<ChatMetadataDataSource>(relaxed = true)
+        coEvery { metadataDataSource.getLastMessageId(chatId) } returns 11
+
+        delegateWith(metadataDataSource)
+            .applyPushedMessage(chatId, message(id = 12, eventSequence = 9, epochSeconds = 1_757_000_042))
+
+        coVerify(exactly = 1) { metadataDataSource.updateLastMessageId(chatId, 12) }
+        coVerify(exactly = 1) { metadataDataSource.updateLastActivity(chatId, 1_757_000_042_000) }
+    }
+
+    @Test
+    fun `a re-delivered push does not rewind the feed row`() = runTest {
+        val metadataDataSource = mockk<ChatMetadataDataSource>(relaxed = true)
+        coEvery { metadataDataSource.getLastMessageId(chatId) } returns 20
+
+        delegateWith(metadataDataSource).applyPushedMessage(chatId, message(id = 12, eventSequence = 9))
+
+        coVerify(exactly = 0) { metadataDataSource.updateLastMessageId(chatId, any()) }
+        coVerify(exactly = 0) { metadataDataSource.updateLastActivity(chatId, any()) }
+    }
+
+    @Test
+    fun `the same push applied twice moves the feed row once`() = runTest {
+        val metadataDataSource = mockk<ChatMetadataDataSource>(relaxed = true)
+        coEvery { metadataDataSource.getLastMessageId(chatId) } returns 11 andThen 12
+        val pushed = message(id = 12, eventSequence = 9)
+        val delegate = delegateWith(metadataDataSource)
+
+        delegate.applyPushedMessage(chatId, pushed)
+        delegate.applyPushedMessage(chatId, pushed)
+
+        coVerify(exactly = 1) { metadataDataSource.updateLastMessageId(chatId, 12) }
+    }
+
+    @Test
+    fun `a chat with no stored message id accepts the first push`() = runTest {
+        val metadataDataSource = mockk<ChatMetadataDataSource>(relaxed = true)
+        coEvery { metadataDataSource.getLastMessageId(chatId) } returns null
+
+        delegateWith(metadataDataSource).applyPushedMessage(chatId, message(id = 1, eventSequence = 3))
+
+        coVerify(exactly = 1) { metadataDataSource.updateLastMessageId(chatId, 1) }
+    }
+}

apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ReceivedCounterTest.kt

@@ -11,8 +11,10 @@ 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.chat.ChatEvent
 import com.flipcash.services.models.chat.ChatId
 import com.flipcash.services.models.chat.ChatMessage
+import com.flipcash.services.models.chat.ChatMutation
 import com.flipcash.services.models.chat.ChatUpdate
 import com.flipcash.services.models.chat.MessageContent
 import com.flipcash.services.user.UserManager
@@ -180,15 +182,24 @@ class ReceivedCounterTest {
         unreadSeq = 0,
     )
 
-    // newMessages is deprecated in favour of `events`, but applyUpdate still falls
-    // back to it and every existing chat test builds updates this way. Matching the
-    // existing harness keeps these tests readable next to their neighbours.
+    // Each update carries its messages as event-log events. Sequences run
+    // contiguously from 1 across the whole test so the gap detector stays quiet
+    // and these tests exercise only the behaviour they name.
+    private var nextEventSequence = 1L
+
     private fun chatUpdate(vararg messages: ChatMessage) = ChatUpdate(
         chatId = chatId,
-        newMessages = messages.toList(),
         pointerUpdates = emptyList(),
         typingNotifications = emptyList(),
         metadataUpdates = emptyList(),
+        events = messages.map { message ->
+            ChatEvent(
+                sequence = nextEventSequence++,
+                count = 1,
+                ts = message.timestamp,
+                mutations = listOf(ChatMutation.MessageSent(message)),
+            )
+        },
     )
 
     private suspend fun TestScope.deliver(vararg messages: ChatMessage) {

apps/flipcash/shared/chat/src/test/kotlin/com/flipcash/shared/chat/ReceivedEventTest.kt

@@ -10,8 +10,10 @@ 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.chat.ChatEvent
 import com.flipcash.services.models.chat.ChatId
 import com.flipcash.services.models.chat.ChatMessage
+import com.flipcash.services.models.chat.ChatMutation
 import com.flipcash.services.models.chat.ChatType
 import com.flipcash.services.models.chat.ChatUpdate
 import com.flipcash.services.models.chat.MessageContent
@@ -183,15 +185,24 @@ class ReceivedEventTest {
         unreadSeq = 0,
     )
 
-    // newMessages is deprecated in favour of `events`, but applyUpdate still falls
-    // back to it and every existing chat test builds updates this way. Matching the
-    // existing harness keeps these tests readable next to their neighbours.
+    // Each update carries its messages as event-log events. Sequences run
+    // contiguously from 1 across the whole test so the gap detector stays quiet
+    // and these tests exercise only the behaviour they name.
+    private var nextEventSequence = 1L
+
     private fun chatUpdate(vararg messages: ChatMessage) = ChatUpdate(
         chatId = chatId,
-        newMessages = messages.toList(),
         pointerUpdates = emptyList(),
         typingNotifications = emptyList(),
         metadataUpdates = emptyList(),
+        events = messages.map { message ->
+            ChatEvent(
+                sequence = nextEventSequence++,
+                count = 1,
+                ts = message.timestamp,
+                mutations = listOf(ChatMutation.MessageSent(message)),
+            )
+        },
     )
 
     private suspend fun TestScope.deliver(vararg messages: ChatMessage) {

apps/flipcash/shared/notifications/build.gradle.kts

@@ -11,6 +11,7 @@ dependencies {
     implementation(project(":apps:flipcash:shared:authentication"))
     implementation(project(":apps:flipcash:shared:chat"))
     implementation(project(":apps:flipcash:shared:contacts"))
+    implementation(project(":apps:flipcash:shared:featureflags"))
     implementation(project(":apps:flipcash:shared:persistence:sources"))
     implementation(project(":apps:flipcash:shared:phone"))
     implementation(project(":apps:flipcash:shared:push"))