feat(chat): wire RemoteMediator for paged message loading and improve chat open speed

apps/flipcash/features/messenger/src/main/kotlin/com/flipcash/app/messenger/internal/ChatViewModel.kt

@@ -70,6 +70,7 @@ import kotlinx.coroutines.launch
 import javax.inject.Inject
 import kotlin.math.min
 import kotlin.time.Duration
+import kotlin.time.Duration.Companion.milliseconds
 import kotlin.time.Duration.Companion.seconds
 import kotlin.time.Instant
 
@@ -277,6 +278,7 @@ internal class ChatViewModel @Inject constructor(
                 if (chatId != null) {
                     dispatchEvent(Event.ChatFound(chatId))
                     chatCoordinator.setActiveChatId(chatId)
+                    chatCoordinator.loadMessages(chatId)
                     chatCoordinator.dismissNotifications(chatId)
                 }
 
@@ -318,15 +320,6 @@ internal class ChatViewModel @Inject constructor(
                 }
             ).launchIn(viewModelScope)
 
-        // trigger message update fetch on open
-        stateFlow.map { it.chatId }
-            .filterNotNull()
-            .distinctUntilChanged()
-            .onEach { chatId ->
-                chatCoordinator.loadMessages(chatId)
-            }
-            .launchIn(viewModelScope)
-
         // Advance read pointer when user scrolls to messages
         eventFlow
             .filterIsInstance<Event.AdvanceReadPointer>()
@@ -541,7 +534,7 @@ internal class ChatViewModel @Inject constructor(
                     ).onSuccess { amount ->
                         dispatchEvent(Event.SendStateUpdated(success = true))
                         stateFlow.value.chatId?.let { chatCoordinator.loadMessages(it) }
-                        delay(400)
+                        delay(400.milliseconds)
                         dispatchEvent(
                             Dispatchers.Main,
                             Event.SendComplete(amount.localFiat.nativeAmount)

apps/flipcash/features/messenger/src/main/kotlin/com/flipcash/app/messenger/internal/screens/components/MessageList.kt

@@ -155,14 +155,16 @@ internal fun MessageList(
             }
         }
 
-        // Chat start shows contact info container
-        item {
-            Box(modifier = Modifier.fillParentMaxWidth(), contentAlignment = Alignment.Center) {
-                ContactInfoContainer(
-                    contact = state.chattingWith,
-                    modifier = Modifier
-                        .padding(horizontal = CodeTheme.dimens.grid.x12)
-                )
+        // Chat start shows contact info container (only after messages have loaded)
+        if (messages.itemCount > 0) {
+            item {
+                Box(modifier = Modifier.fillParentMaxWidth(), contentAlignment = Alignment.Center) {
+                    ContactInfoContainer(
+                        contact = state.chattingWith,
+                        modifier = Modifier
+                            .padding(horizontal = CodeTheme.dimens.grid.x12)
+                    )
+                }
             }
         }
     }

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

@@ -1,17 +1,19 @@
-@file:OptIn(ExperimentalCoroutinesApi::class)
+@file:OptIn(ExperimentalCoroutinesApi::class, ExperimentalPagingApi::class)
 
 package com.flipcash.shared.chat
 
 import androidx.core.app.NotificationManagerCompat
 import androidx.lifecycle.DefaultLifecycleObserver
 import androidx.lifecycle.LifecycleOwner
 import androidx.lifecycle.ProcessLifecycleOwner
+import androidx.paging.ExperimentalPagingApi
 import androidx.paging.Pager
 import androidx.paging.PagingConfig
 import androidx.paging.PagingData
 import androidx.paging.map
 import com.flipcash.app.contacts.device.DeviceContact
 import com.flipcash.app.persistence.sources.ChatMemberDataSource
+import com.flipcash.app.persistence.sources.mediator.ChatMessageRemoteMediator
 import com.flipcash.app.persistence.sources.ChatMessageDataSource
 import com.flipcash.app.persistence.sources.ChatMetadataDataSource
 import com.flipcash.app.persistence.sources.ContactDataSource
@@ -173,6 +175,7 @@ class ChatCoordinator @Inject constructor(
     fun observeMessagesPaged(chatId: ChatId): Flow<PagingData<ChatMessage>> {
         return Pager(
             config = PagingConfig(pageSize = 50),
+            remoteMediator = ChatMessageRemoteMediator(chatId, messagingController, messageDataSource),
         ) {
             messageDataSource.observeForChat(chatId)
         }.flow.map { page ->
@@ -195,7 +198,7 @@ class ChatCoordinator @Inject constructor(
             .distinctUntilChanged()
     }
 
-    suspend fun loadMessages(chatId: ChatId, limit: Int = 100) {
+    suspend fun loadMessages(chatId: ChatId) {
         messagingController.getMessages(chatId)
             .onSuccess { messages ->
                 messageDataSource.upsert(chatId, messages)
@@ -335,6 +338,11 @@ class ChatCoordinator @Inject constructor(
 
                 _state.update { it.copy(feed = page.chats, feedSyncState = FeedSyncState.Synced) }
                 trace(tag = TAG, message = "Feed synced: ${page.chats.size} chats", type = TraceType.Process)
+
+                // Prefetch first page of messages for chats with no cached messages
+                page.chats
+                    .filterNot { messageDataSource.hasMessages(it.chatId) }
+                    .forEach { chat -> loadMessages(chat.chatId) }
             }
             .onFailure { error ->
                 _state.update { it.copy(feedSyncState = FeedSyncState.Error) }

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

@@ -92,6 +92,12 @@ class ChatMessageDataSource @Inject constructor(
     suspend fun getLatest(chatIdHex: String): ChatMessage? =
         db?.chatMessageDao()?.getLatest(chatIdHex)?.let { toChatMessage(it) }
 
+    suspend fun hasMessages(chatId: ChatId): Boolean =
+        db?.chatMessageDao()?.getLatest(mapper.chatIdHex(chatId)) != null
+
+    suspend fun getLatestMessageId(chatId: ChatId): Long? =
+        db?.chatMessageDao()?.getLatest(mapper.chatIdHex(chatId))?.messageId
+
     suspend fun upsert(chatId: ChatId, messages: List<ChatMessage>) {
         val hex = mapper.chatIdHex(chatId)
         val entities = messages.map { mapper.toEntity(hex, it) }

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

@@ -11,6 +11,7 @@ import com.flipcash.services.models.QueryOptions
 import com.flipcash.services.models.chat.ChatId
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.withContext
+import java.nio.ByteBuffer
 
 @OptIn(ExperimentalPagingApi::class)
 class ChatMessageRemoteMediator(
@@ -28,14 +29,19 @@ class ChatMessageRemoteMediator(
         state: PagingState<Int, ChatMessageEntity>,
     ): MediatorResult {
         return try {
-            when (loadType) {
-                LoadType.REFRESH -> Unit
+            val token = when (loadType) {
+                LoadType.REFRESH -> null
                 LoadType.PREPEND -> return MediatorResult.Success(endOfPaginationReached = true)
-                LoadType.APPEND -> Unit
+                LoadType.APPEND -> {
+                    val lastItem = state.lastItemOrNull()
+                        ?: return MediatorResult.Success(endOfPaginationReached = true)
+                    lastItem.messageId.toPagingToken()
+                }
             }
 
             val queryOptions = QueryOptions(
                 limit = state.config.pageSize,
+                token = token,
                 descending = true,
             )
 
@@ -52,3 +58,7 @@ class ChatMessageRemoteMediator(
         }
     }
 }
+
+private fun Long.toPagingToken(): List<Byte> {
+    return ByteBuffer.allocate(Long.SIZE_BYTES).putLong(this).array().toList()
+}