Merge pull request #522 from code-payments/fix/network-state-balance-update

feat: improve network resiliency when starting in a no network state

Author: bmc08gt

api/src/main/java/com/getcode/db/ConversationDao.kt

@@ -1,5 +1,7 @@
 package com.getcode.db
 
+import androidx.paging.PagingData
+import androidx.paging.PagingSource
 import androidx.room.Dao
 import androidx.room.Delete
 import androidx.room.Insert
@@ -19,6 +21,10 @@ interface ConversationDao {
     @Insert(onConflict = OnConflictStrategy.REPLACE)
     suspend fun upsertConversations(vararg conversation: Conversation)
 
+    @RewriteQueriesToDropUnusedColumns
+    @Query("SELECT * FROM conversations")
+    fun observeConversations(): PagingSource<Int, Conversation>
+
     @RewriteQueriesToDropUnusedColumns
     @Query("SELECT * FROM conversations LEFT JOIN conversation_pointers ON conversations.idBase58 = conversation_pointers.conversationIdBase58 WHERE conversations.idBase58 = :id")
     fun observeConversation(id: String): Flow<ConversationWithLastPointers?>

api/src/main/java/com/getcode/mapper/ConversationMapper.kt

@@ -4,7 +4,6 @@ import com.getcode.model.Conversation
 import com.getcode.model.chat.Chat
 import com.getcode.model.chat.ChatType
 import com.getcode.model.chat.self
-import com.getcode.network.TipController
 import com.getcode.network.localized
 import com.getcode.network.repository.base58
 import com.getcode.util.resources.ResourceHelper

api/src/main/java/com/getcode/model/Feature.kt

@@ -22,13 +22,13 @@ data class TipCardOnHomeScreenFeature(
     override val available: Boolean = true, // always available
 ): Feature
 
-data class TipChatFeature(
-    override val enabled: Boolean = BetaOptions.Defaults.tipsChatEnabled,
+data class ConversationsFeature(
+    override val enabled: Boolean = BetaOptions.Defaults.conversationsEnabled,
     override val available: Boolean = true, // always available
 ): Feature
 
-data class TipChatCashFeature(
-    override val enabled: Boolean = BetaOptions.Defaults.tipsChatCashEnabled,
+data class ConversationCashFeature(
+    override val enabled: Boolean = BetaOptions.Defaults.conversationCashEnabled,
     override val available: Boolean = true, // always available
 ): Feature
 

api/src/main/java/com/getcode/model/PrefBool.kt

@@ -46,8 +46,8 @@ sealed class PrefsBool(val value: String) {
     data object BUY_MODULE_ENABLED : PrefsBool("buy_kin_enabled"), BetaFlag
     data object CHAT_UNSUB_ENABLED: PrefsBool("chat_unsub_enabled"), BetaFlag
     data object TIPS_ENABLED : PrefsBool("tips_enabled"), BetaFlag
-    data object TIPS_CHAT_ENABLED: PrefsBool("tips_chat_enabled"), BetaFlag
-    data object TIPS_CHAT_CASH_ENABLED: PrefsBool("tips_chat_cash_enabled"), BetaFlag
+    data object CONVERSATIONS_ENABLED: PrefsBool("conversations_enabled"), BetaFlag
+    data object CONVERSATION_CASH_ENABLED: PrefsBool("convo_cash_enabled"), BetaFlag
     data object BALANCE_CURRENCY_SELECTION_ENABLED: PrefsBool("balance_currency_enabled"), BetaFlag
     data object KADO_WEBVIEW_ENABLED : PrefsBool("kado_inapp_enabled"), BetaFlag
     data object SHARE_TWEET_TO_TIP : PrefsBool("share_tweet_to_tip"), BetaFlag

api/src/main/java/com/getcode/model/chat/Chat.kt

@@ -2,6 +2,7 @@ package com.getcode.model.chat
 
 import com.getcode.model.Cursor
 import com.getcode.model.ID
+import kotlinx.serialization.Serializable
 import java.util.UUID
 
 /**
@@ -18,6 +19,7 @@ import java.util.UUID
  * @param cursor [Cursor] value for this chat for reference in subsequent GetChatsRequest
  * @param messages List of messages within this chat
  */
+@Serializable
 data class Chat(
     val id: ID,
     val type: ChatType,
@@ -135,6 +137,9 @@ data class Chat(
 val Chat.isV2: Boolean
     get() = members.isNotEmpty()
 
+val Chat.isNotification: Boolean
+    get() = type == ChatType.Notification
+
 val Chat.isConversation: Boolean
     get() = type == ChatType.TwoWay
 

api/src/main/java/com/getcode/model/chat/ChatMessage.kt

@@ -4,6 +4,8 @@ import com.getcode.ed25519.Ed25519.KeyPair
 import com.getcode.model.Cursor
 import com.getcode.model.ID
 import com.getcode.model.MessageStatus
+import com.getcode.utils.serializer.UUIDSerializer
+import kotlinx.serialization.Serializable
 import java.util.UUID
 
 /**
@@ -18,8 +20,10 @@ import java.util.UUID
  * @param contents Ordered message content. A message may have more than one piece of content.
  * @param status Derived [MessageStatus] from [Pointer]'s in [ChatMember].
  */
+@Serializable
 data class ChatMessage(
     val id: ID, // time based UUID in v2
+    @Serializable(with = UUIDSerializer::class)
     val senderId: UUID?,
     val isFromSelf: Boolean,
     val cursor: Cursor,

api/src/main/java/com/getcode/network/BalanceController.kt

@@ -13,6 +13,7 @@ import com.getcode.solana.organizer.Organizer
 import com.getcode.solana.organizer.Tray
 import com.getcode.utils.FormatUtils
 import com.getcode.utils.network.NetworkConnectivityListener
+import com.getcode.utils.network.retryable
 import com.getcode.utils.trace
 import io.reactivex.rxjava3.core.Completable
 import kotlinx.coroutines.CoroutineScope
@@ -23,6 +24,7 @@ import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.combine
 import kotlinx.coroutines.flow.distinctUntilChanged
+import kotlinx.coroutines.flow.flatMapLatest
 import kotlinx.coroutines.flow.flowOn
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.map
@@ -38,8 +40,7 @@ data class BalanceDisplay(
     val marketValue: Double = 0.0,
     val formattedValue: String = "",
     val currency: Currency? = null,
-
-    )
+)
 
 open class BalanceController @Inject constructor(
     exchange: Exchange,
@@ -65,25 +66,34 @@ open class BalanceController @Inject constructor(
             .stateIn(scope, SharingStarted.Eagerly, BalanceDisplay())
 
     init {
-        combine(
-            exchange.observeLocalRate()
-                .flowOn(Dispatchers.IO)
-                .onEach {
-                    val display = _balanceDisplay.value ?: BalanceDisplay()
-                    _balanceDisplay.value = display.copy(currency = getCurrencyFromCode(it.currency))
+        networkObserver.state
+            .map { it.connected }
+            .onEach { connected ->
+                if (connected) {
+                    retryable({ fetchBalanceSuspend() })
+                }
+            }
+            .flatMapLatest {
+                combine(
+                    exchange.observeLocalRate()
+                        .flowOn(Dispatchers.IO)
+                        .onEach {
+                            val display = _balanceDisplay.value ?: BalanceDisplay()
+                            _balanceDisplay.value =
+                                display.copy(currency = getCurrencyFromCode(it.currency))
+                        }
+                        .onEach { exchange.fetchRatesIfNeeded() },
+                    balanceRepository.balanceFlow,
+                ) { rate, balance ->
+                    rate to balance.coerceAtLeast(0.0)
+                }.map { (rate, balance) ->
+                    refreshBalance(balance, rate)
                 }
-                .onEach { exchange.fetchRatesIfNeeded() },
-            balanceRepository.balanceFlow,
-            networkObserver.state
-        ) { rate, balance, _ ->
-            rate to balance.coerceAtLeast(0.0)
-        }.map { (rate, balance) ->
-            refreshBalance(balance, rate)
-        }.distinctUntilChanged().onEach { (marketValue, amountText) ->
-            val display = _balanceDisplay.value ?: BalanceDisplay()
-            _balanceDisplay.value =
-                display.copy(marketValue = marketValue, formattedValue = amountText)
-        }.launchIn(scope)
+            }.distinctUntilChanged().onEach { (marketValue, amountText) ->
+                val display = _balanceDisplay.value ?: BalanceDisplay()
+                _balanceDisplay.value =
+                    display.copy(marketValue = marketValue, formattedValue = amountText)
+            }.launchIn(scope)
     }
 
     fun setTray(organizer: Organizer, tray: Tray) {
@@ -155,6 +165,7 @@ open class BalanceController @Inject constructor(
 
 
     suspend fun fetchBalanceSuspend() {
+        Timber.d("fetching balance")
         if (SessionManager.isAuthenticated() != true) {
             Timber.d("FetchBalance - Not authenticated")
             return

api/src/main/java/com/getcode/network/HistoryController.kt renamed to api/src/main/java/com/getcode/network/ChatHistoryController.kt

@@ -17,13 +17,12 @@ import com.getcode.model.Cursor
 import com.getcode.model.ID
 import com.getcode.model.MessageStatus
 import com.getcode.model.chat.ChatMember
-import com.getcode.model.chat.ChatType
 import com.getcode.model.chat.Identity
 import com.getcode.model.chat.Platform
 import com.getcode.model.chat.Title
 import com.getcode.model.chat.isConversation
+import com.getcode.model.chat.isNotification
 import com.getcode.model.chat.selfId
-import com.getcode.model.description
 import com.getcode.network.client.Client
 import com.getcode.network.client.advancePointer
 import com.getcode.network.client.fetchChats
@@ -34,39 +33,42 @@ import com.getcode.network.repository.encodeBase64
 import com.getcode.network.source.ChatMessagePagingSource
 import com.getcode.util.resources.ResourceHelper
 import com.getcode.util.resources.ResourceType
-import com.getcode.utils.ErrorUtils
 import com.getcode.utils.TraceType
 import com.getcode.utils.trace
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.GlobalScope
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.StateFlow
-import kotlinx.coroutines.flow.asStateFlow
 import kotlinx.coroutines.flow.filterNotNull
 import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.stateIn
 import kotlinx.coroutines.flow.update
-import kotlinx.coroutines.launch
-import okhttp3.internal.toImmutableList
 import timber.log.Timber
 import java.util.Locale
 import javax.inject.Inject
 import javax.inject.Singleton
 
-// TODO: See if we can merge this into [ChatController]
 @Singleton
-class HistoryController @Inject constructor(
+class ChatHistoryController @Inject constructor(
     private val client: Client,
-    private val resources: ResourceHelper,
     private val tipController: TipController,
     private val conversationMapper: ConversationMapper,
     private val conversationMessageMapper: ConversationMessageMapper,
 ) : CoroutineScope by CoroutineScope(Dispatchers.IO) {
 
-    private val _chats = MutableStateFlow<List<Chat>?>(null)
+    private val chatEntries = MutableStateFlow<List<Chat>?>(null)
+    val notifications: StateFlow<List<Chat>?>
+        get() = chatEntries
+            .map { it?.filter { entry -> entry.isNotification } }
+            .stateIn(this, SharingStarted.Eagerly, emptyList())
+
     val chats: StateFlow<List<Chat>?>
-        get() = _chats.asStateFlow()
+        get() = chatEntries
+            .map { it?.filter { entry -> entry.isConversation } }
+            .stateIn(this, SharingStarted.Eagerly, emptyList())
 
     var loadingMessages: Boolean = false
 
@@ -86,9 +88,9 @@ class HistoryController @Inject constructor(
         pagerMap[chatId] ?: ChatMessagePagingSource(
             client = client,
             owner = owner()!!,
-            chat = _chats.value?.find { it.id == chatId },
+            chat = chatEntries.value?.find { it.id == chatId },
             onMessagesFetched = { messages ->
-                val chat = _chats.value?.find { it.id == chatId } ?: return@ChatMessagePagingSource
+                val chat = chatEntries.value?.find { it.id == chatId } ?: return@ChatMessagePagingSource
                 updateChatWithMessages(chat, messages)
             }
         ).also {
@@ -99,14 +101,14 @@ class HistoryController @Inject constructor(
     fun updateChatWithMessages(chat: Chat, messages: List<ChatMessage>) {
         val updatedMessages = (chat.messages + messages).distinctBy { it.id }
         val updatedChat = chat.copy(messages = updatedMessages)
-        val chats = _chats.value?.map {
+        val chats = chatEntries.value?.map {
             if (it.id == updatedChat.id) {
                 updatedChat
             } else {
                 it
             }
         }?.sortedByDescending { it.lastMessageMillis }
-        _chats.update { chats }
+        chatEntries.update { chats }
     }
 
     fun chatFlow(chatId: ID) =
@@ -132,7 +134,7 @@ class HistoryController @Inject constructor(
         if (!update) {
             pagerMap.clear()
             chatFlows.clear()
-            _chats.value = containers
+            chatEntries.value = containers
 
             loadingMessages = true
         }
@@ -151,13 +153,13 @@ class HistoryController @Inject constructor(
         }
 
         loadingMessages = false
-        _chats.value = updatedWithMessages.sortedByDescending { it.lastMessageMillis }
+        chatEntries.value = updatedWithMessages.sortedByDescending { it.lastMessageMillis }
     }
 
     suspend fun advanceReadPointer(chatId: ID) {
         val owner = owner() ?: return
 
-        _chats.update {
+        chatEntries.update {
             it?.toMutableList()?.apply chats@{
                 indexOfFirst { chat -> chat.id == chatId }
                     .takeIf { index -> index >= 0 }
@@ -180,7 +182,7 @@ class HistoryController @Inject constructor(
     }
 
     fun advanceReadPointerUpTo(chatId: ID, timestamp: Long) {
-        _chats.update {
+        chatEntries.update {
             it?.toMutableList()?.apply chats@{
                 indexOfFirst { chat -> chat.id == chatId }
                     .takeIf { index -> index >= 0 }
@@ -198,7 +200,7 @@ class HistoryController @Inject constructor(
     suspend fun setMuted(chat: Chat, muted: Boolean): Result<Boolean> {
         val owner = owner() ?: return Result.failure(Throwable("No owner detected"))
 
-        _chats.update {
+        chatEntries.update {
             it?.toMutableList()?.apply chats@{
                 indexOfFirst { item -> item.id == chat.id }
                     .takeIf { index -> index >= 0 }
@@ -216,7 +218,7 @@ class HistoryController @Inject constructor(
     suspend fun setSubscribed(chat: Chat, subscribed: Boolean): Result<Boolean> {
         val owner = owner() ?: return Result.failure(Throwable("No owner detected"))
 
-        _chats.update {
+        chatEntries.update {
             it?.toMutableList()?.apply chats@{
                 indexOfFirst { item -> item.id == chat.id }
                     .takeIf { index -> index >= 0 }
@@ -250,7 +252,6 @@ class HistoryController @Inject constructor(
                         MessageStatus.Delivered
                     )
                 }
-
             }
             .onFailure {
                 Timber.e(t = it, "Failed to fetch messages for $encodedId.")

api/src/main/java/com/getcode/network/ConversationController.kt

@@ -47,7 +47,7 @@ interface ConversationController {
 }
 
 class ConversationStreamController @Inject constructor(
-    private val historyController: HistoryController,
+    private val historyController: ChatHistoryController,
     private val exchange: Exchange,
     private val chatService: ChatServiceV2,
     private val conversationMapper: ConversationMapper,

api/src/main/java/com/getcode/network/api/CurrencyApi.kt

@@ -7,16 +7,18 @@ import io.grpc.ManagedChannel
 import io.reactivex.rxjava3.core.Scheduler
 import io.reactivex.rxjava3.core.Single
 import io.reactivex.rxjava3.schedulers.Schedulers
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.flowOn
 import javax.inject.Inject
 
 class CurrencyApi @Inject constructor(
     managedChannel: ManagedChannel,
-    private val scheduler: Scheduler = Schedulers.io()
 ) : GrpcApi(managedChannel) {
     private val api = CurrencyGrpc.newStub(managedChannel)
 
-    fun getRates(request: CurrencyService.GetAllRatesRequest = CurrencyService.GetAllRatesRequest.getDefaultInstance()): Single<CurrencyService.GetAllRatesResponse> =
+    fun getRates(request: CurrencyService.GetAllRatesRequest = CurrencyService.GetAllRatesRequest.getDefaultInstance()): Flow<CurrencyService.GetAllRatesResponse> =
         api::getAllRates
-            .callAsSingle(request)
-            .subscribeOn(scheduler)
+            .callAsCancellableFlow(request)
+            .flowOn(Dispatchers.IO)
 }