Updated on 2026-08-14

data/staking/src/main/java/com/tangem/data/staking/DefaultStakingRepository.kt

@@ -2,7 +2,6 @@ package com.tangem.data.staking
 
 import android.util.Base64
 import arrow.core.getOrElse
-import arrow.core.raise.catch
 import com.squareup.moshi.Moshi
 import com.tangem.blockchain.common.Amount
 import com.tangem.blockchain.common.Blockchain
@@ -30,8 +29,6 @@ import com.tangem.datasource.api.stakekit.models.response.model.YieldBalanceWrap
 import com.tangem.datasource.api.stakekit.models.response.model.transaction.tron.TronStakeKitTransaction
 import com.tangem.datasource.local.token.StakingBalanceStore
 import com.tangem.datasource.local.token.StakingYieldsStore
-import com.tangem.domain.core.lce.LceFlow
-import com.tangem.domain.core.lce.lceFlow
 import com.tangem.domain.staking.model.StakingApproval
 import com.tangem.domain.staking.model.StakingAvailability
 import com.tangem.domain.staking.model.StakingEntryInfo
@@ -57,6 +54,7 @@ import com.tangem.utils.coroutines.CoroutineDispatcherProvider
 import com.tangem.utils.extensions.orZero
 import kotlinx.coroutines.flow.*
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.plus
 import kotlinx.coroutines.withContext
 import timber.log.Timber
 
@@ -385,15 +383,22 @@ internal class DefaultStakingRepository(
         refresh: Boolean,
     ) = withContext(dispatchers.io) {
         if (!stakingFeatureToggle.isStakingEnabled) return@withContext
+
         try {
-            isYieldBalanceFetching.update {
-                it + (userWalletId to true)
-            }
             cacheRegistry.invokeOnExpire(
                 key = getYieldBalancesKey(userWalletId),
                 skipCache = refresh,
                 block = {
-                    val yields = getEnabledYields()
+                    isYieldBalanceFetching.update {
+                        it + (userWalletId to true)
+                    }
+
+                    val yields = getEnabledYields().ifEmpty {
+                        Timber.i("No enabled yields for $userWalletId")
+                        stakingBalanceStore.store(userWalletId, emptySet())
+
+                        return@invokeOnExpire
+                    }
                     val availableCurrencies = cryptoCurrencies
                         .mapNotNull { currency ->
                             val addresses = walletManagersFacade.getAddresses(userWalletId, currency.network)
@@ -410,10 +415,15 @@ internal class DefaultStakingRepository(
                         }
                         .map { getBalanceRequestData(it.first.value, it.second) }
                         .ifEmpty {
-                            cacheRegistry.invalidate(getYieldBalancesKey(userWalletId))
-                            error("No addresses found")
+                            Timber.i("No yield balances available for $userWalletId")
+                            stakingBalanceStore.store(userWalletId, emptySet())
+
+                            return@invokeOnExpire
                         }
-                    val result = stakeKitApi.getMultipleYieldBalances(availableCurrencies).getOrThrow()
+
+                    val result = stakeKitApi
+                        .getMultipleYieldBalances(availableCurrencies)
+                        .getOrThrow()
 
                     stakingBalanceStore.store(userWalletId, result)
                 },
@@ -446,27 +456,22 @@ internal class DefaultStakingRepository(
         }
     }.cancellable()
 
-    override fun getMultiYieldBalanceLce(
+    override fun getMultiYieldBalance(
         userWalletId: UserWalletId,
         cryptoCurrencies: List<CryptoCurrency>,
-    ): LceFlow<Throwable, YieldBalanceList> = lceFlow {
+    ): Flow<YieldBalanceList> = channelFlow {
         if (!stakingFeatureToggle.isStakingEnabled) {
             send(YieldBalanceList.Empty)
         } else {
-            launch(dispatchers.io) {
-                combine(
-                    stakingBalanceStore.get(userWalletId),
-                    isYieldBalanceFetching.map { it.getOrElse(userWalletId) { false } },
-                ) { result, isFetching ->
-                    val balances = yieldBalanceListConverter.convert(result)
-                    send(balances, isStillLoading = isFetching)
-                }.collect()
-            }
+            stakingBalanceStore.get(userWalletId)
+                .onEach {
+                    val balances = yieldBalanceListConverter.convert(it)
+                    send(balances)
+                }
+                .launchIn(scope = this + dispatchers.io)
+
             withContext(dispatchers.io) {
-                catch(
-                    block = { fetchMultiYieldBalance(userWalletId, cryptoCurrencies, refresh = false) },
-                    catch = { raise(it) },
-                )
+                fetchMultiYieldBalance(userWalletId, cryptoCurrencies, refresh = false)
             }
         }
     }

data/tokens/src/main/kotlin/com/tangem/data/tokens/repository/DefaultCurrenciesRepository.kt

@@ -42,6 +42,7 @@ import com.tangem.utils.coroutines.CoroutineDispatcherProvider
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.flow.*
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.plus
 import kotlinx.coroutines.withContext
 import timber.log.Timber
 import com.tangem.blockchain.common.FeePaidCurrency as FeePaidSdkCurrency
@@ -206,18 +207,14 @@ internal class DefaultCurrenciesRepository(
         }
     }
 
-    override fun getWalletCurrenciesUpdates(userWalletId: UserWalletId): LceFlow<Throwable, List<CryptoCurrency>> {
-        return lceFlow {
-            val userWallet = catch({ getUserWallet(userWalletId) }) {
-                raise(it)
-            }
+    override fun getWalletCurrenciesUpdates(userWalletId: UserWalletId): Flow<List<CryptoCurrency>> {
+        return channelFlow {
+            val userWallet = getUserWallet(userWalletId)
 
             if (userWallet.isMultiCurrency) {
-                getMultiCurrencyWalletCurrenciesUpdatesLce(userWalletId).collect(::send)
+                getMultiCurrencyWalletCurrenciesUpdates(userWalletId).collect(::send)
             } else {
-                val currency = catch({ getSingleCurrencyWalletPrimaryCurrency(userWalletId) }) {
-                    raise(it)
-                }
+                val currency = getSingleCurrencyWalletPrimaryCurrency(userWalletId)
                 send(listOf(currency))
             }
         }
@@ -260,16 +257,14 @@ internal class DefaultCurrenciesRepository(
             val userWallet = getUserWallet(userWalletId)
             ensureIsCorrectUserWallet(userWallet, isMultiCurrencyWalletExpected = true)
 
-            launch(dispatchers.io) {
-                getMultiCurrencyWalletCurrencies(userWallet)
-                    .collectLatest(::send)
-            }
+            getMultiCurrencyWalletCurrencies(userWallet)
+                .onEach { send(it) }
+                .launchIn(scope = this + dispatchers.io)
 
             withContext(dispatchers.io) {
                 fetchTokensIfCacheExpired(userWallet, refresh = false)
             }
         }
-            .cancellable()
     }
 
     override fun getMultiCurrencyWalletCurrenciesUpdatesLce(

data/tokens/src/main/kotlin/com/tangem/data/tokens/repository/DefaultNetworksRepository.kt

@@ -54,29 +54,25 @@ internal class DefaultNetworksRepository(
         userWalletId: UserWalletId,
         networks: Set<Network>,
     ): Flow<Set<NetworkStatus>> = channelFlow {
-        launch(dispatchers.io) {
-            networksStatusesStore.get(userWalletId)
-                .collectLatest(::send)
-        }
+        networksStatusesStore.get(userWalletId)
+            .onEach(::send)
+            .launchIn(scope = this + dispatchers.io)
 
         withContext(dispatchers.io) {
-            fetchNetworksStatusesIfCacheExpired(userWalletId, networks, false)
+            fetchNetworksStatusesIfCacheExpired(userWalletId, networks, refresh = false)
         }
     }
-        .cancellable()
 
     override fun getNetworkStatusesUpdatesLce(
         userWalletId: UserWalletId,
         networks: Set<Network>,
     ): LceFlow<Throwable, Set<NetworkStatus>> = lceFlow {
-        launch(dispatchers.io) {
-            combine(
-                networksStatusesStore.get(userWalletId),
-                isNetworkStatusesFetching.map { it.getOrElse(userWalletId) { false } },
-            ) { statuses, isFetching ->
-                send(statuses, isStillLoading = isFetching)
-            }.collect()
-        }
+        combine(
+            networksStatusesStore.get(userWalletId),
+            isNetworkStatusesFetching.map { it.getOrElse(userWalletId) { false } },
+        ) { statuses, isFetching ->
+            send(statuses, isStillLoading = isFetching || networks.size != statuses.size)
+        }.launchIn(scope = this + dispatchers.io)
 
         withContext(dispatchers.io) {
             catch({ fetchNetworksStatusesIfCacheExpired(userWalletId, networks, refresh = false) }) {
@@ -185,25 +181,32 @@ internal class DefaultNetworksRepository(
         userWalletId: UserWalletId,
         networks: Set<Network>,
         refresh: Boolean,
-    ) {
-        val currencies = getCurrencies(userWalletId, networks)
-        val networksDeferred = networks.mapNotNull { network ->
-            fetchNetworkStatusIfCacheExpired(userWalletId, network, currencies, refresh)
+    ) = coroutineScope {
+        if (refresh) {
+            val statusesToRefresh = networks.map { NetworkStatus(it, NetworkStatus.Loading) }
+            networksStatusesStore.storeAll(userWalletId, statusesToRefresh)
         }
 
-        if (networksDeferred.isNotEmpty()) {
-            try {
-                isNetworkStatusesFetching.update {
-                    it + (userWalletId to true)
-                }
+        val currencies = getCurrencies(userWalletId, networks)
+        val networksDeferred = networks.mapNotNull { network ->
+            coroutineScope {
+                val key = getNetworksStatusesCacheKey(userWalletId, network)
 
-                networksDeferred.awaitAll()
-            } finally {
-                isNetworkStatusesFetching.update {
-                    it - userWalletId
+                if (refresh || cacheRegistry.isExpired(key)) {
+                    async {
+                        cacheRegistry.invokeOnExpire(
+                            key = key,
+                            skipCache = refresh,
+                            block = { fetchNetworkStatus(userWalletId, network, currencies) },
+                        )
+                    }
+                } else {
+                    null
                 }
             }
         }
+
+        networksDeferred.awaitAll()
     }
 
     private suspend fun fetchNetworksPendingTransactions(
@@ -222,31 +225,13 @@ internal class DefaultNetworksRepository(
         }
     }
 
-    private suspend fun fetchNetworkStatusIfCacheExpired(
-        userWalletId: UserWalletId,
-        network: Network,
-        currencies: Sequence<CryptoCurrency>,
-        refresh: Boolean,
-    ): Deferred<Unit>? = coroutineScope {
-        val key = getNetworksStatusesCacheKey(userWalletId, network)
-        if (refresh || cacheRegistry.isExpired(key)) {
-            async {
-                cacheRegistry.invokeOnExpire(
-                    key = key,
-                    skipCache = refresh,
-                    block = { fetchNetworkStatus(userWalletId, network, currencies) },
-                )
-            }
-        } else {
-            null
-        }
-    }
-
     private suspend fun fetchNetworkStatus(
         userWalletId: UserWalletId,
         network: Network,
         currencies: Sequence<CryptoCurrency>,
     ) {
+        networksStatusesStore.store(userWalletId, NetworkStatus(network, NetworkStatus.Loading))
+
         val result = walletManagersFacade.update(
             userWalletId = userWalletId,
             network = network,