Updated on 2026-08-14
This commit is contained in:
Tangem
parent
2887693055
commit
990b06c3ed
16 changed files with 725 additions and 198 deletions
|
|
@ -17,6 +17,7 @@ import com.tangem.domain.quotes.multi.MultiQuoteStatusFetcher
|
|||
import com.tangem.domain.staking.StakingIdFactory
|
||||
import com.tangem.domain.staking.multi.MultiStakingBalanceFetcher
|
||||
import com.tangem.domain.staking.utils.StakingCleaner
|
||||
import com.tangem.domain.tokens.BalanceFetchingOperations
|
||||
import com.tangem.domain.tokens.GetCryptoCurrencyActionsUseCase
|
||||
import com.tangem.domain.tokens.repository.CurrenciesRepository
|
||||
import com.tangem.domain.walletmanager.WalletManagersFacade
|
||||
|
|
@ -121,18 +122,28 @@ internal object AccountStatusUseCaseModule {
|
|||
|
||||
@Provides
|
||||
@Singleton
|
||||
fun provideCryptoCurrencyBalanceFetcher(
|
||||
fun provideBalanceFetchingOperations(
|
||||
multiNetworkStatusFetcher: MultiNetworkStatusFetcher,
|
||||
multiQuoteStatusFetcher: MultiQuoteStatusFetcher,
|
||||
multiStakingBalanceFetcher: MultiStakingBalanceFetcher,
|
||||
stakingIdFactory: StakingIdFactory,
|
||||
appScope: AppCoroutineScope,
|
||||
): CryptoCurrencyBalanceFetcher {
|
||||
return CryptoCurrencyBalanceFetcher(
|
||||
): BalanceFetchingOperations {
|
||||
return BalanceFetchingOperations(
|
||||
multiNetworkStatusFetcher = multiNetworkStatusFetcher,
|
||||
multiQuoteStatusFetcher = multiQuoteStatusFetcher,
|
||||
multiStakingBalanceFetcher = multiStakingBalanceFetcher,
|
||||
stakingIdFactory = stakingIdFactory,
|
||||
)
|
||||
}
|
||||
|
||||
@Provides
|
||||
@Singleton
|
||||
fun provideCryptoCurrencyBalanceFetcher(
|
||||
balanceFetchingOperations: BalanceFetchingOperations,
|
||||
appScope: AppCoroutineScope,
|
||||
): CryptoCurrencyBalanceFetcher {
|
||||
return CryptoCurrencyBalanceFetcher(
|
||||
balanceFetchingOperations = balanceFetchingOperations,
|
||||
parallelUpdatingScope = appScope,
|
||||
)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,128 +1,101 @@
|
|||
package com.tangem.domain.account.status.utils
|
||||
|
||||
import arrow.core.Either
|
||||
import com.tangem.domain.models.currency.CryptoCurrency
|
||||
import com.tangem.domain.models.wallet.UserWalletId
|
||||
import com.tangem.domain.networks.multi.MultiNetworkStatusFetcher
|
||||
import com.tangem.domain.quotes.multi.MultiQuoteStatusFetcher
|
||||
import com.tangem.domain.staking.StakingIdFactory
|
||||
import com.tangem.domain.staking.multi.MultiStakingBalanceFetcher
|
||||
import com.tangem.domain.tokens.wallet.FetchingSource
|
||||
import kotlinx.coroutines.*
|
||||
import com.tangem.domain.tokens.BalanceFetchingOperations
|
||||
import com.tangem.domain.tokens.FetchErrorFormatter
|
||||
import com.tangem.domain.tokens.FetchingSource
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.sync.Mutex
|
||||
import kotlinx.coroutines.sync.withLock
|
||||
import timber.log.Timber
|
||||
import java.util.concurrent.ConcurrentHashMap
|
||||
|
||||
/**
|
||||
* Utility class responsible for fetching and refreshing the balances of various crypto currencies
|
||||
* associated with a user's wallet.
|
||||
*
|
||||
* @property multiNetworkStatusFetcher Fetcher for updating network statuses.
|
||||
* @property multiQuoteStatusFetcher Fetcher for updating quote statuses.
|
||||
* @property multiStakingBalanceFetcher Fetcher for updating staking balances.
|
||||
* @property stakingIdFactory Factory for creating staking IDs.
|
||||
* @property parallelUpdatingScope Coroutine scope for parallel balance updates.
|
||||
* Uses [BalanceFetchingOperations] for the actual fetching logic.
|
||||
* Uses per-wallet mutex to allow concurrent refreshes for different wallets while preventing
|
||||
* concurrent refreshes for the same wallet.
|
||||
*
|
||||
* @property balanceFetchingOperations shared operations for fetching balance data
|
||||
* @property parallelUpdatingScope coroutine scope for parallel balance updates
|
||||
*
|
||||
[REDACTED_AUTHOR]
|
||||
*/
|
||||
class CryptoCurrencyBalanceFetcher(
|
||||
private val multiNetworkStatusFetcher: MultiNetworkStatusFetcher,
|
||||
private val multiQuoteStatusFetcher: MultiQuoteStatusFetcher,
|
||||
private val multiStakingBalanceFetcher: MultiStakingBalanceFetcher,
|
||||
private val stakingIdFactory: StakingIdFactory,
|
||||
private val balanceFetchingOperations: BalanceFetchingOperations,
|
||||
private val parallelUpdatingScope: CoroutineScope,
|
||||
) {
|
||||
|
||||
private val mutex = Mutex()
|
||||
private val mutexMap = ConcurrentHashMap<UserWalletId, Mutex>()
|
||||
|
||||
/**
|
||||
* Fire-and-forget balance refresh for a single currency.
|
||||
*/
|
||||
operator fun invoke(userWalletId: UserWalletId, currency: CryptoCurrency) {
|
||||
invoke(userWalletId = userWalletId, currencies = listOf(currency))
|
||||
}
|
||||
|
||||
/**
|
||||
* Fire-and-forget balance refresh for multiple currencies.
|
||||
* Launches in [parallelUpdatingScope] and uses per-wallet mutex to prevent concurrent refreshes
|
||||
* for the same wallet while allowing parallel refreshes for different wallets.
|
||||
*/
|
||||
operator fun invoke(userWalletId: UserWalletId, currencies: List<CryptoCurrency>) {
|
||||
if (currencies.isEmpty()) return
|
||||
|
||||
parallelUpdatingScope.launch {
|
||||
mutex.withLock {
|
||||
getMutex(userWalletId).withLock {
|
||||
refreshBalances(userWalletId, currencies)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Suspending balance refresh for a single currency.
|
||||
* Awaits completion before returning.
|
||||
*/
|
||||
suspend fun invokeAndAwait(userWalletId: UserWalletId, currency: CryptoCurrency) {
|
||||
invokeAndAwait(userWalletId = userWalletId, currencies = listOf(currency))
|
||||
}
|
||||
|
||||
/**
|
||||
* Suspending balance refresh for multiple currencies.
|
||||
* Awaits completion before returning, uses per-wallet mutex to prevent concurrent refreshes
|
||||
* for the same wallet while allowing parallel refreshes for different wallets.
|
||||
*/
|
||||
suspend fun invokeAndAwait(userWalletId: UserWalletId, currencies: List<CryptoCurrency>) {
|
||||
if (currencies.isEmpty()) return
|
||||
|
||||
mutex.withLock {
|
||||
getMutex(userWalletId).withLock {
|
||||
refreshBalances(userWalletId, currencies)
|
||||
}
|
||||
}
|
||||
|
||||
private fun getMutex(userWalletId: UserWalletId): Mutex {
|
||||
return mutexMap.computeIfAbsent(userWalletId) { Mutex() }
|
||||
}
|
||||
|
||||
private suspend fun refreshBalances(userWalletId: UserWalletId, currencies: List<CryptoCurrency>) {
|
||||
coroutineScope {
|
||||
val results = listOf(
|
||||
async {
|
||||
FetchingSource.NETWORK to refreshNetworks(userWalletId = userWalletId, currencies = currencies)
|
||||
},
|
||||
async {
|
||||
FetchingSource.STAKING to refreshStakingBalances(
|
||||
userWalletId = userWalletId,
|
||||
currencies = currencies,
|
||||
)
|
||||
},
|
||||
async { FetchingSource.QUOTE to refreshQuotes(currencies = currencies) },
|
||||
)
|
||||
.awaitAll()
|
||||
val errors = balanceFetchingOperations.fetchAll(
|
||||
userWalletId = userWalletId,
|
||||
currencies = currencies,
|
||||
sources = FETCHING_SOURCES,
|
||||
)
|
||||
|
||||
val errors = results.mapNotNull { (source, maybeResult) ->
|
||||
val error = maybeResult.leftOrNull() ?: return@mapNotNull null
|
||||
|
||||
source to error
|
||||
}
|
||||
|
||||
if (errors.isNotEmpty()) {
|
||||
val message = "Failed to fetch next sources for $userWalletId:\n" +
|
||||
errors.joinToString(separator = "\n") { "${it.first.name} – ${it.second}" }
|
||||
|
||||
Timber.e(message)
|
||||
}
|
||||
if (errors.isNotEmpty()) {
|
||||
Timber.e(FetchErrorFormatter.format(userWalletId, errors))
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun refreshNetworks(
|
||||
userWalletId: UserWalletId,
|
||||
currencies: List<CryptoCurrency>,
|
||||
): Either<Throwable, Unit> {
|
||||
return multiNetworkStatusFetcher(
|
||||
params = MultiNetworkStatusFetcher.Params(
|
||||
userWalletId = userWalletId,
|
||||
networks = currencies.mapTo(hashSetOf(), CryptoCurrency::network),
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
private suspend fun refreshStakingBalances(
|
||||
userWalletId: UserWalletId,
|
||||
currencies: List<CryptoCurrency>,
|
||||
): Either<Throwable, Unit> {
|
||||
val stakingIds = currencies.mapNotNullTo(hashSetOf()) {
|
||||
stakingIdFactory.create(userWalletId = userWalletId, cryptoCurrency = it).getOrNull()
|
||||
}
|
||||
|
||||
return multiStakingBalanceFetcher(
|
||||
params = MultiStakingBalanceFetcher.Params(userWalletId = userWalletId, stakingIds = stakingIds),
|
||||
)
|
||||
}
|
||||
|
||||
private suspend fun refreshQuotes(currencies: List<CryptoCurrency>): Either<Throwable, Unit> {
|
||||
return multiQuoteStatusFetcher(
|
||||
params = MultiQuoteStatusFetcher.Params(
|
||||
currenciesIds = currencies.mapNotNullTo(hashSetOf()) { it.id.rawCurrencyId },
|
||||
appCurrencyId = null,
|
||||
),
|
||||
private companion object {
|
||||
val FETCHING_SOURCES = setOf(
|
||||
FetchingSource.NETWORK,
|
||||
FetchingSource.QUOTE,
|
||||
FetchingSource.STAKING,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,316 @@
|
|||
package com.tangem.domain.account.status.utils
|
||||
|
||||
import com.google.common.truth.Truth.assertThat
|
||||
import com.tangem.common.test.domain.token.MockCryptoCurrencyFactory
|
||||
import com.tangem.domain.models.currency.CryptoCurrency
|
||||
import com.tangem.domain.models.wallet.UserWalletId
|
||||
import com.tangem.domain.tokens.BalanceFetchingOperations
|
||||
import com.tangem.domain.tokens.FetchingSource
|
||||
import com.tangem.test.mock.MockAccounts
|
||||
import io.mockk.*
|
||||
import kotlinx.coroutines.ExperimentalCoroutinesApi
|
||||
import kotlinx.coroutines.delay
|
||||
import kotlinx.coroutines.test.TestScope
|
||||
import kotlinx.coroutines.test.advanceUntilIdle
|
||||
import kotlinx.coroutines.test.runTest
|
||||
import org.junit.jupiter.api.*
|
||||
import java.util.concurrent.atomic.AtomicInteger
|
||||
|
||||
/**
|
||||
* Tests for [CryptoCurrencyBalanceFetcher]
|
||||
*
|
||||
[REDACTED_AUTHOR]
|
||||
*/
|
||||
@OptIn(ExperimentalCoroutinesApi::class)
|
||||
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
|
||||
internal class CryptoCurrencyBalanceFetcherTest {
|
||||
|
||||
private val cryptoCurrencyFactory = MockCryptoCurrencyFactory()
|
||||
private val balanceFetchingOperations: BalanceFetchingOperations = mockk()
|
||||
|
||||
private val userWalletId = MockAccounts.userWalletId
|
||||
private val userWalletId2 = UserWalletId("012")
|
||||
|
||||
@AfterEach
|
||||
fun tearDown() {
|
||||
clearMocks(balanceFetchingOperations)
|
||||
}
|
||||
|
||||
private fun createFetcher(testScope: TestScope): CryptoCurrencyBalanceFetcher {
|
||||
return CryptoCurrencyBalanceFetcher(
|
||||
balanceFetchingOperations = balanceFetchingOperations,
|
||||
parallelUpdatingScope = testScope,
|
||||
)
|
||||
}
|
||||
|
||||
@Nested
|
||||
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
|
||||
inner class InvokeWithEmptyCurrencies {
|
||||
|
||||
@Test
|
||||
fun `invoke with empty currencies does not fetch`() = runTest {
|
||||
// Arrange
|
||||
val fetcher = createFetcher(this)
|
||||
val currencies = emptyList<CryptoCurrency>()
|
||||
|
||||
// Act
|
||||
fetcher(userWalletId = userWalletId, currencies = currencies)
|
||||
advanceUntilIdle()
|
||||
|
||||
// Assert
|
||||
coVerify(exactly = 0) { balanceFetchingOperations.fetchAll(any(), any(), any()) }
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `invokeAndAwait with empty currencies does not fetch`() = runTest {
|
||||
// Arrange
|
||||
val fetcher = createFetcher(this)
|
||||
val currencies = emptyList<CryptoCurrency>()
|
||||
|
||||
// Act
|
||||
fetcher.invokeAndAwait(userWalletId = userWalletId, currencies = currencies)
|
||||
|
||||
// Assert
|
||||
coVerify(exactly = 0) { balanceFetchingOperations.fetchAll(any(), any(), any()) }
|
||||
}
|
||||
}
|
||||
|
||||
@Nested
|
||||
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
|
||||
inner class InvokeWithSingleCurrency {
|
||||
|
||||
@Test
|
||||
fun `invoke with single currency fetches all sources`() = runTest {
|
||||
// Arrange
|
||||
val fetcher = createFetcher(this)
|
||||
val currency = cryptoCurrencyFactory.ethereum
|
||||
coEvery {
|
||||
balanceFetchingOperations.fetchAll(
|
||||
userWalletId = userWalletId,
|
||||
currencies = listOf(currency),
|
||||
sources = any(),
|
||||
)
|
||||
} returns emptyMap()
|
||||
|
||||
// Act
|
||||
fetcher(userWalletId = userWalletId, currency = currency)
|
||||
advanceUntilIdle()
|
||||
|
||||
// Assert
|
||||
coVerify(exactly = 1) {
|
||||
balanceFetchingOperations.fetchAll(
|
||||
userWalletId = userWalletId,
|
||||
currencies = listOf(currency),
|
||||
sources = setOf(FetchingSource.NETWORK, FetchingSource.QUOTE, FetchingSource.STAKING),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `invokeAndAwait with single currency fetches all sources`() = runTest {
|
||||
// Arrange
|
||||
val fetcher = createFetcher(this)
|
||||
val currency = cryptoCurrencyFactory.ethereum
|
||||
coEvery {
|
||||
balanceFetchingOperations.fetchAll(
|
||||
userWalletId = userWalletId,
|
||||
currencies = listOf(currency),
|
||||
sources = any(),
|
||||
)
|
||||
} returns emptyMap()
|
||||
|
||||
// Act
|
||||
fetcher.invokeAndAwait(userWalletId = userWalletId, currency = currency)
|
||||
|
||||
// Assert
|
||||
coVerify(exactly = 1) {
|
||||
balanceFetchingOperations.fetchAll(
|
||||
userWalletId = userWalletId,
|
||||
currencies = listOf(currency),
|
||||
sources = setOf(FetchingSource.NETWORK, FetchingSource.QUOTE, FetchingSource.STAKING),
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Nested
|
||||
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
|
||||
inner class InvokeWithMultipleCurrencies {
|
||||
|
||||
@Test
|
||||
fun `invoke with multiple currencies passes correct parameters`() = runTest {
|
||||
// Arrange
|
||||
val fetcher = createFetcher(this)
|
||||
val currencies = cryptoCurrencyFactory.ethereumAndStellar
|
||||
coEvery {
|
||||
balanceFetchingOperations.fetchAll(
|
||||
userWalletId = userWalletId,
|
||||
currencies = currencies,
|
||||
sources = any(),
|
||||
)
|
||||
} returns emptyMap()
|
||||
|
||||
// Act
|
||||
fetcher(userWalletId = userWalletId, currencies = currencies)
|
||||
advanceUntilIdle()
|
||||
|
||||
// Assert
|
||||
coVerify(exactly = 1) {
|
||||
balanceFetchingOperations.fetchAll(
|
||||
userWalletId = userWalletId,
|
||||
currencies = currencies,
|
||||
sources = setOf(FetchingSource.NETWORK, FetchingSource.QUOTE, FetchingSource.STAKING),
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Nested
|
||||
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
|
||||
inner class ErrorHandling {
|
||||
|
||||
@Test
|
||||
fun `invoke logs errors when fetch fails`() = runTest {
|
||||
// Arrange
|
||||
val fetcher = createFetcher(this)
|
||||
val currency = cryptoCurrencyFactory.ethereum
|
||||
val error = RuntimeException("Network error")
|
||||
val errors = mapOf(FetchingSource.NETWORK to error)
|
||||
|
||||
coEvery {
|
||||
balanceFetchingOperations.fetchAll(any(), any(), any())
|
||||
} returns errors
|
||||
|
||||
// Act
|
||||
fetcher(userWalletId = userWalletId, currency = currency)
|
||||
advanceUntilIdle()
|
||||
|
||||
// Assert - fetchAll was called, FetchErrorFormatter.format is used internally (object, no mock needed)
|
||||
coVerify(exactly = 1) {
|
||||
balanceFetchingOperations.fetchAll(any(), any(), any())
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `invoke continues even when some sources fail`() = runTest {
|
||||
// Arrange
|
||||
val fetcher = createFetcher(this)
|
||||
val currency = cryptoCurrencyFactory.ethereum
|
||||
val networkError = RuntimeException("Network error")
|
||||
val errors = mapOf(FetchingSource.NETWORK to networkError)
|
||||
|
||||
coEvery {
|
||||
balanceFetchingOperations.fetchAll(any(), any(), any())
|
||||
} returns errors
|
||||
|
||||
// Act & Assert - should not throw
|
||||
fetcher(userWalletId = userWalletId, currency = currency)
|
||||
advanceUntilIdle()
|
||||
|
||||
coVerify(exactly = 1) { balanceFetchingOperations.fetchAll(any(), any(), any()) }
|
||||
}
|
||||
}
|
||||
|
||||
@Nested
|
||||
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
|
||||
inner class MutexBehavior {
|
||||
|
||||
@Test
|
||||
fun `concurrent calls for same wallet are serialized`() = runTest {
|
||||
// Arrange
|
||||
val fetcher = createFetcher(this)
|
||||
val currency = cryptoCurrencyFactory.ethereum
|
||||
val callOrder = mutableListOf<Int>()
|
||||
val callCount = AtomicInteger(0)
|
||||
|
||||
coEvery {
|
||||
balanceFetchingOperations.fetchAll(any(), any(), any())
|
||||
} coAnswers {
|
||||
val currentCall = callCount.incrementAndGet()
|
||||
callOrder.add(currentCall)
|
||||
delay(100) // Simulate work
|
||||
callOrder.add(-currentCall) // Mark completion
|
||||
emptyMap()
|
||||
}
|
||||
|
||||
// Act - launch two concurrent calls for the same wallet
|
||||
fetcher(userWalletId = userWalletId, currency = currency)
|
||||
fetcher(userWalletId = userWalletId, currency = currency)
|
||||
advanceUntilIdle()
|
||||
|
||||
// Assert - calls should be serialized (1 starts, 1 finishes, 2 starts, 2 finishes)
|
||||
assertThat(callOrder).containsExactly(1, -1, 2, -2).inOrder()
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `concurrent calls for different wallets run in parallel`() = runTest {
|
||||
// Arrange
|
||||
val fetcher = createFetcher(this)
|
||||
val currency = cryptoCurrencyFactory.ethereum
|
||||
val callOrder = mutableListOf<String>()
|
||||
|
||||
coEvery {
|
||||
balanceFetchingOperations.fetchAll(eq(userWalletId), any(), any())
|
||||
} coAnswers {
|
||||
callOrder.add("wallet1-start")
|
||||
delay(100)
|
||||
callOrder.add("wallet1-end")
|
||||
emptyMap()
|
||||
}
|
||||
|
||||
coEvery {
|
||||
balanceFetchingOperations.fetchAll(eq(userWalletId2), any(), any())
|
||||
} coAnswers {
|
||||
callOrder.add("wallet2-start")
|
||||
delay(100)
|
||||
callOrder.add("wallet2-end")
|
||||
emptyMap()
|
||||
}
|
||||
|
||||
// Act - launch two concurrent calls for different wallets
|
||||
fetcher(userWalletId = userWalletId, currency = currency)
|
||||
fetcher(userWalletId = userWalletId2, currency = currency)
|
||||
advanceUntilIdle()
|
||||
|
||||
// Assert - both should start before either finishes (parallel execution)
|
||||
val wallet1StartIndex = callOrder.indexOf("wallet1-start")
|
||||
val wallet2StartIndex = callOrder.indexOf("wallet2-start")
|
||||
val wallet1EndIndex = callOrder.indexOf("wallet1-end")
|
||||
val wallet2EndIndex = callOrder.indexOf("wallet2-end")
|
||||
|
||||
// Both should start before both end
|
||||
assertThat(wallet1StartIndex).isLessThan(wallet1EndIndex)
|
||||
assertThat(wallet2StartIndex).isLessThan(wallet2EndIndex)
|
||||
// Both starts should happen before both ends (parallel)
|
||||
assertThat(maxOf(wallet1StartIndex, wallet2StartIndex))
|
||||
.isLessThan(minOf(wallet1EndIndex, wallet2EndIndex))
|
||||
}
|
||||
}
|
||||
|
||||
@Nested
|
||||
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
|
||||
inner class InvokeAndAwaitBehavior {
|
||||
|
||||
@Test
|
||||
fun `invokeAndAwait suspends until completion`() = runTest {
|
||||
// Arrange
|
||||
val fetcher = createFetcher(this)
|
||||
val currency = cryptoCurrencyFactory.ethereum
|
||||
var completed = false
|
||||
|
||||
coEvery {
|
||||
balanceFetchingOperations.fetchAll(any(), any(), any())
|
||||
} coAnswers {
|
||||
delay(100)
|
||||
completed = true
|
||||
emptyMap()
|
||||
}
|
||||
|
||||
// Act
|
||||
fetcher.invokeAndAwait(userWalletId = userWalletId, currency = currency)
|
||||
|
||||
// Assert - should be completed after invokeAndAwait returns
|
||||
assertThat(completed).isTrue()
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue