fix(chat): render and route group pushes, and cap transcript sender lookups

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

@@ -3,17 +3,24 @@ package com.flipcash.shared.chat.internal
 import com.flipcash.app.persistence.sources.UserProfileDataSource
 import com.flipcash.libs.coroutines.DispatcherProvider
 import com.flipcash.services.controllers.ProfileController
+import com.flipcash.services.models.GetUserProfileError
 import com.flipcash.services.models.UserProfile
 import com.getcode.opencode.model.core.ID
 import com.getcode.utils.TraceType
 import com.getcode.utils.hexEncodedString
 import com.getcode.utils.trace
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.cancelChildren
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.sync.Semaphore
+import kotlinx.coroutines.sync.withPermit
 import javax.inject.Inject
 import javax.inject.Singleton
+import kotlin.time.Duration
+import kotlin.time.Duration.Companion.seconds
 
 /**
  * Resolves the profiles of message senders the roster does not cover.
@@ -29,6 +36,18 @@ import javax.inject.Singleton
  * fills in. That is why there is no suspend `resolve(userId)` here: the transcript is a Paging
  * stream, and a per-item suspend call would stall page composition on the network.
  *
+ * `GetProfile` takes one user id — there is no batch RPC, and no chat RPC that lists members past
+ * the page `GetChat` returns — so one call per unknown sender is the floor. What this class owns is
+ * making sure it is also the ceiling: the caller re-asks on every emission of the transcript's
+ * `combine`, which is once per page load, per pending mutation, and per write to `user_profiles`.
+ * Three things keep that from turning into a call per emission:
+ *
+ * - **A resolved sender stays marked.** Their profile is in `user_profiles`, so the transcript
+ *   reads it from [profiles] and never reaches [request] again.
+ * - **An unresolvable sender stays marked.** See [unresolvable].
+ * - **A sender whose fetch failed for any other reason is held for [RETRY_BACKOFF].** See
+ *   [inFlight].
+ *
  * Owns its own scope because [com.flipcash.shared.chat.internal.delegates.MessagingDelegate] — its
  * only caller — has no `initialize(scope)` and giving it one would churn `RealChatCoordinator` and
  * every construction of it in tests.
@@ -41,51 +60,108 @@ class SenderResolver @Inject constructor(
 ) {
     private val scope = CoroutineScope(dispatchers.IO + SupervisorJob())
 
+    /** Guards [inFlight] and [unresolvable], which are read and written together in [request]. */
+    private val lock = Any()
+
     /**
-     * User-id hexes with a fetch in flight. Marked *before* the launch, so two requests arriving
-     * in the same frame collapse on the calling thread instead of racing to the network.
+     * User-id hexes with a fetch in flight, or cooling down after a failure. Marked *before* the
+     * launch, so two requests arriving in the same frame collapse on the calling thread instead of
+     * racing to the network.
+     *
+     * A transient failure — being offline is the common one — keeps the id marked for
+     * [RETRY_BACKOFF] rather than releasing it immediately. Releasing it immediately makes the
+     * retry cadence "however often the transcript happens to re-emit", which is neither a policy
+     * nor bounded: a scroll while offline is a burst of retries that cannot succeed.
      */
     private val inFlight = mutableSetOf<String>()
 
+    /**
+     * User-id hexes the server has no profile for, which are never asked for again.
+     *
+     * `NOT_FOUND` is an answer, not a failure to get one — the account is gone, or was never
+     * visible to this viewer. Treating it as transient is what made a single deleted sender in a
+     * group cost a `GetProfile` on every re-emission for the life of the process.
+     */
+    private val unresolvable = mutableSetOf<String>()
+
+    /**
+     * Caps how many profile fetches are on the network at once.
+     *
+     * Opening a group whose loaded pages carry senders past the roster page asks for all of them in
+     * one frame. [inFlight] collapses duplicates but says nothing about the fan-out across distinct
+     * ids, and none of these are urgent — a bubble renders without its name either way. The permit
+     * is held for the call alone, so a backing-off id occupies nothing.
+     */
+    private val fetchLimit = Semaphore(MAX_CONCURRENT_FETCHES)
+
     /** Every profile this device holds, keyed by user-id hex. The transcript indexes into it. */
     val profiles: Flow<Map<String, UserProfile>> = userProfileDataSource.observeProfiles()
 
     /**
      * Asks for [userId]'s profile if nothing has asked already. Returns immediately.
-     *
-     * A failure releases the id rather than caching the miss: the common cause is being offline,
-     * and a permanently poisoned id would leave the sender nameless for the life of the process.
      */
     fun request(userId: ID) {
         val hex = userId.hexEncodedString()
-        synchronized(inFlight) {
+        synchronized(lock) {
+            if (hex in unresolvable) return
             if (!inFlight.add(hex)) return
         }
         scope.launch {
-            profileController.getProfileForUser(userId)
+            fetchLimit.withPermit { profileController.getProfileForUser(userId) }
                 .onSuccess { profile ->
                     userProfileDataSource.store(userId, profile)
                     // Kept marked on success: the profile is in `user_profiles` now, so the
                     // transcript reads it from [profiles] and never asks again.
                 }
                 .onFailure { error ->
-                    synchronized(inFlight) { inFlight.remove(hex) }
+                    if (error is GetUserProfileError.NotFound) {
+                        synchronized(lock) { unresolvable.add(hex) }
+                        trace(
+                            tag = TAG,
+                            message = "no profile exists for sender $hex; will not ask again",
+                            type = TraceType.Silent,
+                        )
+                        return@onFailure
+                    }
+
                     trace(
                         tag = TAG,
                         message = "failed to resolve sender $hex",
                         error = error,
                         type = TraceType.Error,
                     )
+                    // Held, not released: the next emission of the transcript is not a signal that
+                    // the network came back, and asking again on it would just fail again.
+                    delay(RETRY_BACKOFF)
+                    synchronized(lock) { inFlight.remove(hex) }
                 }
         }
     }
 
-    /** Drops what this process has asked for, so a re-login re-resolves against the new account. */
+    /**
+     * Drops what this process has asked for, so a re-login re-resolves against the new account.
+     *
+     * Outstanding work is cancelled rather than left to finish. A fetch in flight was authorized by
+     * the account that just went away, and a backoff still sleeping would wake to release an id the
+     * next account may already be fetching.
+     */
     fun clear() {
-        synchronized(inFlight) { inFlight.clear() }
+        synchronized(lock) {
+            inFlight.clear()
+            unresolvable.clear()
+        }
+        scope.coroutineContext.cancelChildren()
     }
 
     private companion object {
         const val TAG = "SenderResolver"
+
+        /**
+         * Enough to fill a screen of unfamiliar senders without opening a connection per bubble.
+         */
+        const val MAX_CONCURRENT_FETCHES = 6
+
+        /** How long a sender whose fetch failed is left alone before anything asks again. */
+        val RETRY_BACKOFF: Duration = 30.seconds
     }
 }

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

@@ -4,29 +4,37 @@ import app.cash.turbine.test
 import com.flipcash.app.core.dispatchers.TestDispatchers
 import com.flipcash.app.persistence.sources.UserProfileDataSource
 import com.flipcash.services.controllers.ProfileController
+import com.flipcash.services.models.GetUserProfileError
 import com.flipcash.services.models.UserProfile
 import com.flipcash.shared.chat.internal.SenderResolver
 import io.mockk.coEvery
 import io.mockk.coVerify
 import io.mockk.every
 import io.mockk.mockk
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.test.TestCoroutineScheduler
+import kotlinx.coroutines.test.advanceTimeBy
 import kotlinx.coroutines.test.advanceUntilIdle
+import kotlinx.coroutines.test.runCurrent
 import kotlinx.coroutines.test.runTest
 import org.junit.Test
 import kotlin.test.assertEquals
+import kotlin.time.Duration.Companion.minutes
+import kotlin.time.Duration.Companion.seconds
 
 /**
  * `GetProfile` is single-user, so every distinct unknown sender in a group costs one call. The
- * guarantee this test exists for is that it costs no more than one: the transcript asks for a
- * sender on every recomposition that sees a bubble from them, and an unguarded resolver would
- * fan that out into a call per frame.
+ * guarantee these tests exist for is that it costs no more than one: the transcript asks for a
+ * sender on every emission that sees a bubble from them — a page load, a pending mutation, any
+ * write to `user_profiles` — and an unguarded resolver would fan that out into a call per emission.
  *
  * The dedupe marks a user id as in-flight *before* launching the fetch, so the second request
  * is dropped on the calling thread. That makes the behaviour deterministic under
  * `StandardTestDispatcher` rather than dependent on when the coroutine happens to be scheduled.
  */
+@OptIn(ExperimentalCoroutinesApi::class)
 class SenderResolverTest {
 
     private val scheduler = TestCoroutineScheduler()
@@ -74,21 +82,80 @@ class SenderResolverTest {
         resolver.clear()
     }
 
+    // region Failures
+
     @Test
-    fun `a failed fetch releases the id so a later request retries`() = runTest(dispatchers.dispatcher) {
-        coEvery { profileController.getProfileForUser(userId) } returns
-            Result.failure(RuntimeException("offline"))
-        val resolver = subject()
+    fun `a transient failure is held for the backoff window, then retried`() =
+        runTest(dispatchers.dispatcher) {
+            coEvery { profileController.getProfileForUser(userId) } returns
+                Result.failure(RuntimeException("offline"))
+            val resolver = subject()
 
-        resolver.request(userId)
-        advanceUntilIdle()
-        resolver.request(userId)
-        advanceUntilIdle()
+            resolver.request(userId)
+            runCurrent()
 
-        coVerify(exactly = 2) { profileController.getProfileForUser(userId) }
-        coVerify(exactly = 0) { userProfileDataSource.store(any(), any()) }
-        resolver.clear()
-    }
+            // A re-emission inside the window is not evidence the network came back, so it does
+            // not buy another attempt.
+            resolver.request(userId)
+            runCurrent()
+            coVerify(exactly = 1) { profileController.getProfileForUser(userId) }
+
+            advanceTimeBy(31.seconds)
+            runCurrent()
+            resolver.request(userId)
+            runCurrent()
+
+            coVerify(exactly = 2) { profileController.getProfileForUser(userId) }
+            coVerify(exactly = 0) { userProfileDataSource.store(any(), any()) }
+            resolver.clear()
+        }
+
+    @Test
+    fun `a sender the server has no profile for is never asked about again`() =
+        runTest(dispatchers.dispatcher) {
+            coEvery { profileController.getProfileForUser(userId) } returns
+                Result.failure(GetUserProfileError.NotFound())
+            val resolver = subject()
+
+            resolver.request(userId)
+            advanceUntilIdle()
+
+            // NOT_FOUND is an answer, not a failure to get one: waiting out a backoff and asking
+            // again would get the same answer for as long as the transcript keeps showing the
+            // bubble.
+            advanceTimeBy(5.minutes)
+            resolver.request(userId)
+            advanceUntilIdle()
+
+            coVerify(exactly = 1) { profileController.getProfileForUser(userId) }
+            resolver.clear()
+        }
+
+    // endregion
+
+    @Test
+    fun `fetches are capped so opening a group does not fan out per bubble`() =
+        runTest(dispatchers.dispatcher) {
+            val gate = CompletableDeferred<Unit>()
+            var started = 0
+            coEvery { profileController.getProfileForUser(any()) } coAnswers {
+                started++
+                gate.await()
+                Result.success(profile)
+            }
+            val resolver = subject()
+
+            repeat(20) { index -> resolver.request(ByteArray(16) { index.toByte() }.toList()) }
+            advanceUntilIdle()
+
+            assertEquals(6, started)
+
+            gate.complete(Unit)
+            advanceUntilIdle()
+
+            assertEquals(20, started)
+            resolver.clear()
+        }
 
     @Test
     fun `resolved profiles are observed off the shared user_profiles table`() =

apps/flipcash/shared/notifications/src/main/kotlin/com/flipcash/app/notifications/ChatTapTargetPlanner.kt

@@ -0,0 +1,36 @@
+package com.flipcash.app.notifications
+
+import com.flipcash.services.models.chat.ChatType
+
+/** Where tapping a chat notification lands. */
+enum class ChatTapTarget {
+    /** The conversation itself, through the app's chat-by-id link. */
+    Conversation,
+
+    /**
+     * A plain launch, which opens the app on the camera.
+     *
+     * The fallback for a chat with no screen to open, rather than firing a link nothing routes:
+     * an unrouted link still leaves the app on whatever it was showing, having taken the tap.
+     */
+    AppLauncher,
+}
+
+/**
+ * Whether a chat push can open its conversation.
+ *
+ * The Chats tab lists tip DMs and groups (`TipFlowViewModel`: `feed(TIP_DM, GROUP)`), and both
+ * open by chat id from there, so both are reachable through the `/tip/chat/{id}` link the router
+ * resolves to that tab. A `CONTACT_DM` is not: the Send tab and direct-send flow that used to
+ * open one were removed, and `AppRouter` deliberately routes nothing for them.
+ *
+ * [chatType] is the type resolved by [planConversationStyling], not what the payload declared.
+ * `chat_metadata` is optional on the wire, and a group push that omits it would otherwise land on
+ * the camera. Null and [ChatType.UNKNOWN] read the same: neither the push nor the device could
+ * say what the chat is, and a chat screen opened for a type the app has no surface for would
+ * dead-end.
+ */
+fun planChatTapTarget(chatType: ChatType?): ChatTapTarget = when (chatType) {
+    ChatType.TIP_DM, ChatType.GROUP -> ChatTapTarget.Conversation
+    ChatType.CONTACT_DM, ChatType.UNKNOWN, null -> ChatTapTarget.AppLauncher
+}

apps/flipcash/shared/notifications/src/main/kotlin/com/flipcash/app/notifications/ConversationStylingPlanner.kt

@@ -0,0 +1,58 @@
+package com.flipcash.app.notifications
+
+import com.flipcash.services.models.chat.ChatType
+
+/**
+ * How a chat push should render as a conversation.
+ *
+ * Pure for the same reason [planSenderLookup] is: the rule is what's worth testing, and stating
+ * it needs neither a database nor a live service.
+ */
+data class ConversationStyling(
+    /** The chat's type, whichever source could answer. Also what the sender lookup keys off. */
+    val chatType: ChatType,
+    /**
+     * Whether `MessagingStyle` should treat this as a group.
+     *
+     * Without it the platform renders every message as a one-to-one: no conversation title, and
+     * the sender's name dropped from the line because in a DM the notification's own title
+     * already says who it is.
+     */
+    val isGroupConversation: Boolean,
+    /**
+     * What to call the group, or null to leave whatever is already on screen.
+     *
+     * Null rather than a placeholder because a re-posted notification carries the previous
+     * title forward, and a group whose row hasn't synced yet is better untitled than renamed.
+     */
+    val conversationTitle: String?,
+)
+
+/**
+ * Decides whether a chat push is a group conversation and what to call it.
+ *
+ * The type is taken from [payloadChatType] first and [storedChatType] second. `chat_metadata` is
+ * optional on the wire, and a push that omits it would otherwise pass for a DM — which is how a
+ * group ends up rendered as one, with an arbitrary participant's name on it. The device already
+ * knows the type of any chat it has synced, so it only falls through to [ChatType.UNKNOWN] for a
+ * group this device has never seen.
+ *
+ * @param payloadChatType the type the push declares, null when it carries no chat metadata
+ * @param storedChatType the type in `chat_metadata`, [ChatType.UNKNOWN] when the chat isn't stored
+ * @param storedTitle the group's name in `chat_metadata`, null for a DM or an unstored chat
+ */
+fun planConversationStyling(
+    payloadChatType: ChatType?,
+    storedChatType: ChatType,
+    storedTitle: String?,
+): ConversationStyling {
+    val chatType = payloadChatType?.takeUnless { it == ChatType.UNKNOWN } ?: storedChatType
+    val isGroup = chatType == ChatType.GROUP
+    return ConversationStyling(
+        chatType = chatType,
+        isGroupConversation = isGroup,
+        // A DM is named by its counterparty, so a title on one would be the wrong name in a
+        // place the user can't correct. Blank is the same as absent.
+        conversationTitle = storedTitle?.takeIf { isGroup && it.isNotBlank() },
+    )
+}