diff --git a/common/test/src/main/java/com/tangem/common/test/data/staking/MockP2PEthPoolAccountResponseFactory.kt b/common/test/src/main/java/com/tangem/common/test/data/staking/MockP2PEthPoolAccountResponseFactory.kt new file mode 100644 index 0000000000..1e814c5d9f --- /dev/null +++ b/common/test/src/main/java/com/tangem/common/test/data/staking/MockP2PEthPoolAccountResponseFactory.kt @@ -0,0 +1,82 @@ +package com.tangem.common.test.data.staking + +import com.tangem.datasource.api.ethpool.models.response.P2PEthPoolAccountResponse +import com.tangem.datasource.api.ethpool.models.response.P2PEthPoolExitQueueDTO +import com.tangem.datasource.api.ethpool.models.response.P2PEthPoolStakeDTO +import com.tangem.domain.models.staking.StakingID +import com.tangem.domain.staking.model.ethpool.P2PEthPoolVault +import java.math.BigDecimal + +/** + * Factory for creating mock P2P ETH Pool account responses for testing + */ +object MockP2PEthPoolAccountResponseFactory { + + private val defaultStakingId = StakingID( + integrationId = "p2p-ethereum-pooled", + address = "0x5aa711F440Eb6d4361148bBD89d03464628ace84", + ) + + const val defaultVaultAddress = "0x7f39C581F595B53c5cb19bD0b3f8dA6c935E2Ca0" + + fun createWithBalance( + stakingId: StakingID = defaultStakingId, + vaultAddress: String = defaultVaultAddress, + stakedAmount: BigDecimal = BigDecimal("1.5"), + earnedAmount: BigDecimal = BigDecimal("0.05"), + ): P2PEthPoolAccountResponse { + return P2PEthPoolAccountResponse( + delegatorAddress = stakingId.address, + vaultAddress = vaultAddress, + stake = P2PEthPoolStakeDTO( + assets = stakedAmount, + totalEarnedAssets = earnedAmount, + ), + availableToUnstake = stakedAmount, + availableToWithdraw = BigDecimal.ZERO, + exitQueue = P2PEthPoolExitQueueDTO( + total = 0.0, + requests = emptyList(), + ), + ) + } + + fun createWithEmptyBalance( + stakingId: StakingID = defaultStakingId, + vaultAddress: String = defaultVaultAddress, + ): P2PEthPoolAccountResponse { + return P2PEthPoolAccountResponse( + delegatorAddress = stakingId.address, + vaultAddress = vaultAddress, + stake = P2PEthPoolStakeDTO( + assets = BigDecimal.ZERO, + totalEarnedAssets = BigDecimal.ZERO, + ), + availableToUnstake = BigDecimal.ZERO, + availableToWithdraw = BigDecimal.ZERO, + exitQueue = P2PEthPoolExitQueueDTO( + total = 0.0, + requests = emptyList(), + ), + ) + } + + fun createMockVault(vaultAddress: String = defaultVaultAddress): P2PEthPoolVault { + return P2PEthPoolVault( + vaultAddress = vaultAddress, + displayName = "Test Vault", + apy = BigDecimal("3.5"), + baseApy = BigDecimal("3.0"), + capacity = BigDecimal("10000"), + totalAssets = BigDecimal("5000"), + feePercent = BigDecimal("10"), + isPrivate = false, + isGenesis = false, + isSmoothingPool = false, + isErc20 = false, + tokenName = "Test Token", + tokenSymbol = "TT", + createdAt = 0L, + ) + } +} \ No newline at end of file diff --git a/core/datasource/src/main/java/com/tangem/datasource/di/StakingStoreModule.kt b/core/datasource/src/main/java/com/tangem/datasource/di/StakingStoreModule.kt index 695c1c8d45..aeb4d1b8d9 100644 --- a/core/datasource/src/main/java/com/tangem/datasource/di/StakingStoreModule.kt +++ b/core/datasource/src/main/java/com/tangem/datasource/di/StakingStoreModule.kt @@ -5,6 +5,7 @@ import androidx.datastore.core.DataStore import androidx.datastore.core.DataStoreFactory import androidx.datastore.dataStoreFile import com.squareup.moshi.Moshi +import com.tangem.datasource.api.ethpool.models.response.P2PEthPoolAccountResponse import com.tangem.datasource.api.stakekit.models.response.model.YieldBalanceWrapperDTO import com.tangem.datasource.api.stakekit.models.response.model.YieldDTO import com.tangem.datasource.local.datastore.RuntimeDataStore @@ -77,6 +78,24 @@ internal object StakingStoreModule { return DefaultStakingActionsStore(dataStore = RuntimeDataStore()) } + @Provides + @Singleton + fun provideP2PBalancesPersistenceStore( + @NetworkMoshi moshi: Moshi, + @ApplicationContext context: Context, + dispatchers: CoroutineDispatcherProvider, + ): DataStore>> { + return DataStoreFactory.create( + serializer = MoshiDataStoreSerializer( + moshi = moshi, + types = mapWithStringKeyTypes(valueTypes = setTypes()), + defaultValue = emptyMap(), + ), + produceFile = { context.dataStoreFile(fileName = "p2p_balances") }, + scope = CoroutineScope(context = dispatchers.io + SupervisorJob()), + ) + } + @Provides @Singleton fun provideP2PEthPoolVaultsStore( diff --git a/data/staking/build.gradle.kts b/data/staking/build.gradle.kts index a01398d354..fc7059b626 100644 --- a/data/staking/build.gradle.kts +++ b/data/staking/build.gradle.kts @@ -51,6 +51,7 @@ dependencies { implementation(deps.androidx.datastore) implementation(deps.jodatime) implementation(deps.kotlin.coroutines) + implementation(deps.kotlin.datetime) implementation(deps.kotlin.immutable.collections) implementation(deps.moshi) implementation(deps.moshi.kotlin) diff --git a/data/staking/src/main/java/com/tangem/data/staking/converters/ethpool/P2PYieldBalanceConverter.kt b/data/staking/src/main/java/com/tangem/data/staking/converters/ethpool/P2PYieldBalanceConverter.kt new file mode 100644 index 0000000000..373a5de22f --- /dev/null +++ b/data/staking/src/main/java/com/tangem/data/staking/converters/ethpool/P2PYieldBalanceConverter.kt @@ -0,0 +1,92 @@ +package com.tangem.data.staking.converters.ethpool + +import com.tangem.domain.models.StatusSource +import com.tangem.domain.models.staking.* +import com.tangem.domain.staking.model.ethpool.P2PEthPoolAccount +import com.tangem.domain.staking.model.ethpool.P2PEthPoolVault +import java.math.BigDecimal + +/** + * tmp solution before facade implementation + */ +internal object P2PYieldBalanceConverter { + + private const val ETH_DECIMALS = 18 + private const val ETH_SYMBOL = "ETH" + private const val ETH_NAME = "Ethereum" + private const val ETH_COINGECKO_ID = "ethereum" + + fun convert( + account: P2PEthPoolAccount, + vault: P2PEthPoolVault, + address: String, + source: StatusSource, + ): YieldBalance { + val integrationId = "p2p-ethereum-pooled" + val stakingId = StakingID( + integrationId = integrationId, + address = address, + ) + + val balanceItems = buildBalanceItems(account, vault) + + return if (balanceItems.isEmpty()) { + YieldBalance.Empty(stakingId = stakingId, source = source) + } else { + YieldBalance.Data( + stakingId = stakingId, + source = source, + balance = YieldBalanceItem( + items = balanceItems, + integrationId = integrationId, + ), + ) + } + } + + private fun buildBalanceItems(account: P2PEthPoolAccount, vault: P2PEthPoolVault): List = buildList { + if (account.stake.assets > BigDecimal.ZERO) { + add( + createBalanceItem( + groupId = "p2p-staked", + amount = account.stake.assets, + type = BalanceType.STAKED, + validatorAddress = vault.vaultAddress, + ), + ) + } + } + + private fun createBalanceItem( + groupId: String, + amount: BigDecimal, + type: BalanceType, + validatorAddress: String, + ): BalanceItem { + return BalanceItem( + groupId = groupId, + token = createEthToken(), + type = type, + amount = amount, + rawCurrencyId = ETH_COINGECKO_ID, + validatorAddress = validatorAddress, + date = null, + pendingActions = emptyList(), + pendingActionsConstraints = emptyList(), + isPending = false, + ) + } + + private fun createEthToken(): YieldToken { + return YieldToken( + name = ETH_NAME, + network = NetworkType.ETHEREUM, + symbol = ETH_SYMBOL, + decimals = ETH_DECIMALS, + address = null, + coinGeckoId = ETH_COINGECKO_ID, + logoURI = null, + isPoints = false, + ) + } +} \ No newline at end of file diff --git a/data/staking/src/main/java/com/tangem/data/staking/di/YieldBalanceSupplierModule.kt b/data/staking/src/main/java/com/tangem/data/staking/di/YieldBalanceSupplierModule.kt index 3ef90f5b74..69c635bb0c 100644 --- a/data/staking/src/main/java/com/tangem/data/staking/di/YieldBalanceSupplierModule.kt +++ b/data/staking/src/main/java/com/tangem/data/staking/di/YieldBalanceSupplierModule.kt @@ -1,15 +1,20 @@ package com.tangem.data.staking.di import androidx.datastore.core.DataStore +import com.tangem.data.staking.store.DefaultP2PBalancesStore import com.tangem.data.staking.store.DefaultYieldsBalancesStore +import com.tangem.data.staking.store.P2PBalancesStore import com.tangem.data.staking.store.YieldsBalancesStore +import com.tangem.datasource.api.ethpool.models.response.P2PEthPoolAccountResponse import com.tangem.datasource.api.stakekit.models.response.model.YieldBalanceWrapperDTO import com.tangem.datasource.local.datastore.RuntimeSharedStore +import com.tangem.datasource.local.token.P2PEthPoolVaultsStore import com.tangem.domain.staking.multi.MultiYieldBalanceProducer import com.tangem.domain.staking.multi.MultiYieldBalanceSupplier import com.tangem.domain.staking.single.SingleYieldBalanceProducer import com.tangem.domain.staking.single.SingleYieldBalanceSupplier import com.tangem.utils.coroutines.CoroutineDispatcherProvider +import com.tangem.utils.coroutines.runSuspendCatching import dagger.Module import dagger.Provides import dagger.hilt.InstallIn @@ -33,6 +38,21 @@ internal object YieldBalanceSupplierModule { ) } + @Provides + @Singleton + fun provideP2PBalancesStore( + persistenceStore: DataStore>>, + p2pVaultsStore: P2PEthPoolVaultsStore, + dispatchers: CoroutineDispatcherProvider, + ): P2PBalancesStore { + return DefaultP2PBalancesStore( + runtimeStore = RuntimeSharedStore(), + persistenceStore = persistenceStore, + vaultsProvider = { runSuspendCatching { p2pVaultsStore.getSync() }.getOrNull().orEmpty() }, + dispatchers = dispatchers, + ) + } + @Provides @Singleton fun provideSingleYieldBalanceSupplier(factory: SingleYieldBalanceProducer.Factory): SingleYieldBalanceSupplier { diff --git a/data/staking/src/main/java/com/tangem/data/staking/multi/DefaultMultiYieldBalanceFetcher.kt b/data/staking/src/main/java/com/tangem/data/staking/multi/DefaultMultiYieldBalanceFetcher.kt index eca2f69f33..a7bcd2cdbb 100644 --- a/data/staking/src/main/java/com/tangem/data/staking/multi/DefaultMultiYieldBalanceFetcher.kt +++ b/data/staking/src/main/java/com/tangem/data/staking/multi/DefaultMultiYieldBalanceFetcher.kt @@ -5,12 +5,17 @@ import arrow.core.left import arrow.core.right import arrow.core.toOption import com.tangem.data.common.api.safeApiCall +import com.tangem.data.staking.store.P2PBalancesStore import com.tangem.data.staking.store.YieldsBalancesStore import com.tangem.data.staking.utils.YieldBalanceRequestBodyFactory +import com.tangem.datasource.api.common.response.ApiResponse +import com.tangem.datasource.api.ethpool.P2PEthPoolApi +import com.tangem.datasource.api.ethpool.models.response.P2PEthPoolAccountResponse import com.tangem.datasource.api.stakekit.StakeKitApi import com.tangem.datasource.api.stakekit.models.request.YieldBalanceRequestBody import com.tangem.datasource.api.stakekit.models.response.model.YieldBalanceWrapperDTO import com.tangem.datasource.api.stakekit.models.response.model.YieldDTO +import com.tangem.datasource.local.token.P2PEthPoolVaultsStore import com.tangem.datasource.local.token.StakingYieldsStore import com.tangem.datasource.local.userwallet.UserWalletsStore import com.tangem.domain.core.utils.catchOn @@ -18,30 +23,43 @@ import com.tangem.domain.models.staking.StakingID import com.tangem.domain.models.wallet.UserWallet import com.tangem.domain.models.wallet.UserWalletId import com.tangem.domain.models.wallet.isMultiCurrency +import com.tangem.domain.staking.model.StakingIntegrationID +import com.tangem.domain.staking.model.ethpool.P2PEthPoolNetwork import com.tangem.domain.staking.multi.MultiYieldBalanceFetcher import com.tangem.utils.coroutines.CoroutineDispatcherProvider +import com.tangem.utils.coroutines.runSuspendCatching import kotlinx.coroutines.async import kotlinx.coroutines.awaitAll import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.launch import timber.log.Timber import javax.inject.Inject /** * Default implementation of [MultiYieldBalanceFetcher] * - * @property userWalletsStore user wallets store - * @property stakingYieldsStore staking yields store - * @property yieldsBalancesStore yields balances store - * @property stakeKitApi stake kit API - * @property dispatchers dispatchers + * Supports both StakeKit and P2P staking providers. + * + * @property userWalletsStore user wallets store + * @property stakingYieldsStore staking yields store + * @property yieldsBalancesStore yields balances store (StakeKit) + * @property p2pBalancesStore P2P balances store + * @property stakeKitApi stake kit API + * @property p2pApi P2P ETH Pool API + * @property p2pVaultsStore P2P vaults store + * @property dispatchers dispatchers * [REDACTED_AUTHOR] */ +@Suppress("LongParameterList") internal class DefaultMultiYieldBalanceFetcher @Inject constructor( private val userWalletsStore: UserWalletsStore, private val stakingYieldsStore: StakingYieldsStore, private val yieldsBalancesStore: YieldsBalancesStore, + private val p2pBalancesStore: P2PBalancesStore, private val stakeKitApi: StakeKitApi, + private val p2pApi: P2PEthPoolApi, + private val p2pVaultsStore: P2PEthPoolVaultsStore, private val dispatchers: CoroutineDispatcherProvider, ) : MultiYieldBalanceFetcher { @@ -57,25 +75,150 @@ internal class DefaultMultiYieldBalanceFetcher @Inject constructor( return it.left() } - Timber.i("Staking IDs to fetch:\n${stakingIds.joinToString("\n")}") + val (stakeKitIds, p2pIds) = stakingIds.partition { stakingId -> + val stakingIntegrationID = StakingIntegrationID.entries.find { + it.value == stakingId.integrationId + } + stakingIntegrationID is StakingIntegrationID.StakeKit + } + + Timber.i( + """ + Staking IDs to fetch: + - StakeKit: ${stakeKitIds.joinToString()} + - P2P: ${p2pIds.joinToString()} + """.trimIndent(), + ) return Either.catchOn(dispatchers.default) { - yieldsBalancesStore.refresh(userWalletId = params.userWalletId, stakingIds = stakingIds) + coroutineScope { + if (stakeKitIds.isNotEmpty()) { + launch { fetchStakeKitBalances(params.userWalletId, stakeKitIds.toSet()) } + } - val availableStakingIds = getAvailableStakingIds( - userWalletId = params.userWalletId, - stakingIds = stakingIds, - ) - - fetch(userWalletId = params.userWalletId, stakingIds = availableStakingIds) + if (p2pIds.isNotEmpty()) { + launch { fetchP2PBalances(params.userWalletId, p2pIds.toSet()) } + } + } } .onLeft { throwable -> Timber.e(throwable, "Unable to fetch yield balances $params") - yieldsBalancesStore.storeError(userWalletId = params.userWalletId, stakingIds = stakingIds) + if (stakeKitIds.isNotEmpty()) { + yieldsBalancesStore.storeError(userWalletId = params.userWalletId, stakingIds = stakeKitIds.toSet()) + } + if (p2pIds.isNotEmpty()) { + p2pBalancesStore.storeError(userWalletId = params.userWalletId, stakingIds = p2pIds.toSet()) + } } } + private suspend fun fetchStakeKitBalances(userWalletId: UserWalletId, stakingIds: Set) { + yieldsBalancesStore.refresh(userWalletId = userWalletId, stakingIds = stakingIds) + + val availableStakingIds = getAvailableStakingIds( + userWalletId = userWalletId, + stakingIds = stakingIds, + ) + + fetchFromStakeKit(userWalletId = userWalletId, stakingIds = availableStakingIds) + } + + private suspend fun fetchP2PBalances(userWalletId: UserWalletId, stakingIds: Set) { + p2pBalancesStore.refresh(userWalletId = userWalletId, stakingIds = stakingIds) + + val vaults = runSuspendCatching { p2pVaultsStore.getSync() }.getOrNull().orEmpty() + if (vaults.isEmpty()) { + Timber.w("No P2P vaults available for $userWalletId") + p2pBalancesStore.storeError(userWalletId = userWalletId, stakingIds = stakingIds) + return + } + + fetchFromP2P(userWalletId = userWalletId, stakingIds = stakingIds, vaults = vaults) + } + + private suspend fun fetchFromP2P( + userWalletId: UserWalletId, + stakingIds: Set, + vaults: List, + ) { + safeApiCall( + call = { + val addresses = stakingIds.map { it.address }.toSet() + + val responses = mutableSetOf() + + for (vault in vaults) { + for (address in addresses) { + runSuspendCatching { + val response = p2pApi.getAccountInfo( + network = P2PEthPoolNetwork.MAINNET.value, + delegatorAddress = address, + vaultAddress = vault.vaultAddress, + ) + + when (response) { + is ApiResponse.Success -> { + val data = response.data + if (data.error != null) { + Timber.w( + "P2P API returned error for vault ${vault.vaultAddress}, " + + "address $address: ${data.error ?: "error"}", + ) + } else { + val result = requireNotNull(data.result) { + "Result is null in successful response" + } + responses.add(result) + } + } + is ApiResponse.Error -> { + Timber.w( + response.cause, + "Failed to fetch P2P balance for vault ${vault.vaultAddress}, " + + "address $address", + ) + } + } + }.onFailure { error -> + Timber.w( + error, + "Failed to fetch P2P balance for vault ${vault.vaultAddress}, address $address", + ) + } + } + } + + Timber.i("Successfully fetched ${responses.size} P2P balances for $userWalletId") + + if (responses.isNotEmpty()) { + p2pBalancesStore.storeActual(userWalletId = userWalletId, values = responses) + + val missingStakingIds = stakingIds.filter { stakingId -> + responses.none { response -> + response.delegatorAddress.equals(stakingId.address, ignoreCase = true) + } + } + + if (missingStakingIds.isNotEmpty()) { + Timber.i("Missing responses for ${missingStakingIds.size} staking IDs: $missingStakingIds") + p2pBalancesStore.storeError(userWalletId = userWalletId, stakingIds = missingStakingIds.toSet()) + } + } else { + Timber.i("No P2P responses received for $userWalletId") + p2pBalancesStore.storeError(userWalletId = userWalletId, stakingIds = stakingIds) + } + }, + onError = { throwable -> + Timber.e(throwable, "Unable to fetch P2P balances $userWalletId") + + p2pBalancesStore.storeError(userWalletId = userWalletId, stakingIds = stakingIds) + + throw throwable + }, + ) + } + private inline fun checkIsSupportedByWalletOrElse(userWalletId: UserWalletId, ifNotSupported: (Throwable) -> Unit) { val maybeUserWallet = userWalletsStore.getSyncOrNull(key = userWalletId).toOption() @@ -139,7 +282,7 @@ internal class DefaultMultiYieldBalanceFetcher @Inject constructor( return yieldsIds } - private suspend fun fetch(userWalletId: UserWalletId, stakingIds: Set) { + private suspend fun fetchFromStakeKit(userWalletId: UserWalletId, stakingIds: Set) { safeApiCall( call = { val requests = stakingIds.map(YieldBalanceRequestBodyFactory::create) diff --git a/data/staking/src/main/java/com/tangem/data/staking/multi/DefaultMultiYieldBalanceProducer.kt b/data/staking/src/main/java/com/tangem/data/staking/multi/DefaultMultiYieldBalanceProducer.kt index f66ec71196..9d4f4917b0 100644 --- a/data/staking/src/main/java/com/tangem/data/staking/multi/DefaultMultiYieldBalanceProducer.kt +++ b/data/staking/src/main/java/com/tangem/data/staking/multi/DefaultMultiYieldBalanceProducer.kt @@ -2,6 +2,7 @@ package com.tangem.data.staking.multi import arrow.core.Option import arrow.core.some +import com.tangem.data.staking.store.P2PBalancesStore import com.tangem.data.staking.store.YieldsBalancesStore import com.tangem.domain.models.staking.YieldBalance import com.tangem.domain.staking.multi.MultiYieldBalanceProducer @@ -10,6 +11,7 @@ import dagger.assisted.Assisted import dagger.assisted.AssistedFactory import dagger.assisted.AssistedInject import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.combine import kotlinx.coroutines.flow.distinctUntilChanged import kotlinx.coroutines.flow.flowOn import kotlinx.coroutines.flow.onEmpty @@ -17,8 +19,11 @@ import kotlinx.coroutines.flow.onEmpty /** * Default implementation of [MultiYieldBalanceProducer] * + * Combines yield balances from both StakeKit and P2P providers. + * * @property params params - * @property yieldsBalancesStore yields balances store + * @property yieldsBalancesStore StakeKit yields balances store + * @property p2pBalancesStore P2P balances store * @property dispatchers dispatchers * [REDACTED_AUTHOR] @@ -26,13 +31,19 @@ import kotlinx.coroutines.flow.onEmpty internal class DefaultMultiYieldBalanceProducer @AssistedInject constructor( @Assisted val params: MultiYieldBalanceProducer.Params, private val yieldsBalancesStore: YieldsBalancesStore, + private val p2pBalancesStore: P2PBalancesStore, private val dispatchers: CoroutineDispatcherProvider, ) : MultiYieldBalanceProducer { override val fallback: Option> = emptySet().some() override fun produce(): Flow> { - return yieldsBalancesStore.get(userWalletId = params.userWalletId) + val stakeKitFlow = yieldsBalancesStore.get(userWalletId = params.userWalletId) + val p2pFlow = p2pBalancesStore.get(userWalletId = params.userWalletId) + + return combine(stakeKitFlow, p2pFlow) { stakeKitBalances, p2pBalances -> + stakeKitBalances + p2pBalances + } .distinctUntilChanged() .onEmpty { emit(value = hashSetOf()) } .flowOn(dispatchers.default) diff --git a/data/staking/src/main/java/com/tangem/data/staking/store/DefaultP2PBalancesStore.kt b/data/staking/src/main/java/com/tangem/data/staking/store/DefaultP2PBalancesStore.kt new file mode 100644 index 0000000000..24ff66cec1 --- /dev/null +++ b/data/staking/src/main/java/com/tangem/data/staking/store/DefaultP2PBalancesStore.kt @@ -0,0 +1,213 @@ +package com.tangem.data.staking.store + +import androidx.datastore.core.DataStore +import com.tangem.data.staking.converters.ethpool.P2PEthPoolAccountConverter +import com.tangem.data.staking.converters.ethpool.P2PYieldBalanceConverter +import com.tangem.datasource.api.ethpool.models.response.P2PEthPoolAccountResponse +import com.tangem.datasource.local.datastore.RuntimeSharedStore +import com.tangem.domain.models.StatusSource +import com.tangem.domain.models.staking.StakingID +import com.tangem.domain.models.staking.YieldBalance +import com.tangem.domain.models.wallet.UserWalletId +import com.tangem.domain.staking.model.ethpool.P2PEthPoolVault +import com.tangem.utils.coroutines.CoroutineDispatcherProvider +import com.tangem.utils.extensions.addOrReplace +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.firstOrNull +import kotlinx.coroutines.flow.map +import kotlinx.coroutines.launch +import timber.log.Timber + +internal typealias WalletIdWithP2PBalances = Map> +internal typealias WalletIdWithP2PResponses = Map> + +/** + * Default implementation of [P2PBalancesStore] + * + * Stores P2P ETH Pool staking balances with persistence support. + * + * @property runtimeStore runtime store + * @property persistenceStore persistence store + * @property vaultsProvider provider for vaults + * @param dispatchers coroutine dispatchers + */ +internal class DefaultP2PBalancesStore( + private val runtimeStore: RuntimeSharedStore, + private val persistenceStore: DataStore, + private val vaultsProvider: suspend () -> List, + dispatchers: CoroutineDispatcherProvider, +) : P2PBalancesStore { + + private val scope = CoroutineScope(context = SupervisorJob() + dispatchers.io) + + init { + scope.launch { + val cachedData = persistenceStore.data.firstOrNull() ?: return@launch + val vaults = vaultsProvider() + + runtimeStore.store( + value = cachedData.map { (stringWalletId, responses) -> + val key = UserWalletId(stringWalletId) + val value = responses.mapNotNull { response -> + val vault = vaults.firstOrNull { it.vaultAddress == response.vaultAddress } + ?: return@mapNotNull null + val account = P2PEthPoolAccountConverter.convert(response) + P2PYieldBalanceConverter.convert( + account = account, + vault = vault, + address = account.delegatorAddress, + source = StatusSource.CACHE, + ) + }.toSet() + + key to value + }.toMap(), + ) + } + } + + override fun get(userWalletId: UserWalletId): Flow> { + return runtimeStore.get().map { it[userWalletId].orEmpty() } + } + + override suspend fun getSyncOrNull(userWalletId: UserWalletId, stakingId: StakingID): YieldBalance? { + return runtimeStore.getSyncOrNull() + ?.get(userWalletId) + ?.firstOrNull { it.stakingId == stakingId } + } + + override suspend fun getAllSyncOrNull(userWalletId: UserWalletId): Set? { + return runtimeStore.getSyncOrNull()?.get(userWalletId) + } + + override suspend fun refresh(userWalletId: UserWalletId, stakingId: StakingID) { + refresh(userWalletId = userWalletId, stakingIds = setOf(stakingId)) + } + + override suspend fun refresh(userWalletId: UserWalletId, stakingIds: Set) { + updateInRuntime(userWalletId = userWalletId, stakingIds = stakingIds) { + it.copySealed(source = StatusSource.CACHE) + } + } + + override suspend fun storeActual(userWalletId: UserWalletId, values: Set) { + coroutineScope { + launch { storeInRuntime(userWalletId = userWalletId, values = values) } + launch { storeInPersistence(userWalletId = userWalletId, values = values) } + } + } + + override suspend fun storeError(userWalletId: UserWalletId, stakingIds: Set) { + updateInRuntime( + userWalletId = userWalletId, + stakingIds = stakingIds, + ifNotFound = ::createErrorYieldBalance, + update = { it.copySealed(source = StatusSource.ONLY_CACHE) }, + ) + } + + override suspend fun clear(userWalletId: UserWalletId, stakingIds: Set) { + coroutineScope { + launch { clearInRuntime(userWalletId = userWalletId, stakingIds = stakingIds) } + launch { clearInPersistence(userWalletId = userWalletId, stakingIds = stakingIds) } + } + } + + private suspend fun storeInRuntime(userWalletId: UserWalletId, values: Set) { + val vaults = vaultsProvider() + + val newBalances = values.mapNotNull { response -> + val vault = vaults.firstOrNull { it.vaultAddress == response.vaultAddress } + if (vault == null) { + Timber.w("Vault not found for ${response.vaultAddress}") + return@mapNotNull null + } + + val account = P2PEthPoolAccountConverter.convert(response) + P2PYieldBalanceConverter.convert( + account = account, + vault = vault, + address = account.delegatorAddress, + source = StatusSource.ACTUAL, + ) + }.toSet() + + runtimeStore.update(default = emptyMap()) { saved -> + saved.toMutableMap().apply { + this[userWalletId] = saved[userWalletId] + ?.addOrReplace(newBalances) { old, new -> old.stakingId == new.stakingId } + ?: newBalances + } + } + } + + private suspend fun storeInPersistence(userWalletId: UserWalletId, values: Set) { + persistenceStore.updateData { current -> + current.toMutableMap().apply { + this[userWalletId.stringValue] = this[userWalletId.stringValue] + ?.addOrReplace(values) { old, new -> + old.delegatorAddress == new.delegatorAddress && old.vaultAddress == new.vaultAddress + } + ?: values + } + } + } + + private suspend fun clearInRuntime(userWalletId: UserWalletId, stakingIds: Set) { + runtimeStore.update(default = emptyMap()) { stored -> + stored.toMutableMap().apply { + this[userWalletId] = this[userWalletId].orEmpty() + .filterNot { it.stakingId in stakingIds } + .toSet() + } + } + } + + private suspend fun clearInPersistence(userWalletId: UserWalletId, stakingIds: Set) { + val integrationIds = stakingIds.map { it.integrationId }.toSet() + + persistenceStore.updateData { current -> + current.toMutableMap().apply { + this[userWalletId.stringValue] = this[userWalletId.stringValue].orEmpty() + .filterNot { response -> + val responseIntegrationId = "p2p-ethereum-pooled:${response.vaultAddress}" + responseIntegrationId in integrationIds + } + .toSet() + } + } + } + + private suspend fun updateInRuntime( + userWalletId: UserWalletId, + stakingIds: Set, + ifNotFound: (StakingID) -> YieldBalance? = { null }, + update: (YieldBalance) -> YieldBalance, + ) { + runtimeStore.update(default = emptyMap()) { stored -> + stored.toMutableMap().apply { + val portfolioBalances = stored[userWalletId].orEmpty() + + val balances = stakingIds.mapNotNullTo(hashSetOf()) { stakingId -> + val balance = portfolioBalances + .firstOrNull { it.stakingId == stakingId } + ?: ifNotFound(stakingId) + ?: return@mapNotNullTo null + + update(balance) + } + + val updatedBalances = portfolioBalances.addOrReplace(items = balances) { old, new -> + old.stakingId == new.stakingId + } + + put(key = userWalletId, value = updatedBalances) + } + } + } + + private fun createErrorYieldBalance(id: StakingID): YieldBalance = YieldBalance.Error(stakingId = id) +} \ No newline at end of file diff --git a/data/staking/src/main/java/com/tangem/data/staking/store/P2PBalancesStore.kt b/data/staking/src/main/java/com/tangem/data/staking/store/P2PBalancesStore.kt new file mode 100644 index 0000000000..7dbf4b1abe --- /dev/null +++ b/data/staking/src/main/java/com/tangem/data/staking/store/P2PBalancesStore.kt @@ -0,0 +1,29 @@ +package com.tangem.data.staking.store + +import com.tangem.datasource.api.ethpool.models.response.P2PEthPoolAccountResponse +import com.tangem.domain.models.staking.StakingID +import com.tangem.domain.models.staking.YieldBalance +import com.tangem.domain.models.wallet.UserWalletId +import kotlinx.coroutines.flow.Flow + +/** + * Store for P2P ETH Pool staking balances + */ +interface P2PBalancesStore { + + fun get(userWalletId: UserWalletId): Flow> + + suspend fun getSyncOrNull(userWalletId: UserWalletId, stakingId: StakingID): YieldBalance? + + suspend fun getAllSyncOrNull(userWalletId: UserWalletId): Set? + + suspend fun refresh(userWalletId: UserWalletId, stakingId: StakingID) + + suspend fun refresh(userWalletId: UserWalletId, stakingIds: Set) + + suspend fun storeActual(userWalletId: UserWalletId, values: Set) + + suspend fun storeError(userWalletId: UserWalletId, stakingIds: Set) + + suspend fun clear(userWalletId: UserWalletId, stakingIds: Set) +} \ No newline at end of file diff --git a/data/staking/src/test/kotlin/com/tangem/data/staking/StakingBalanceExt.kt b/data/staking/src/test/kotlin/com/tangem/data/staking/StakingBalanceExt.kt new file mode 100644 index 0000000000..968fb21321 --- /dev/null +++ b/data/staking/src/test/kotlin/com/tangem/data/staking/StakingBalanceExt.kt @@ -0,0 +1,28 @@ +package com.tangem.data.staking + +import com.tangem.common.test.data.staking.MockP2PEthPoolAccountResponseFactory +import com.tangem.data.staking.converters.ethpool.P2PEthPoolAccountConverter +import com.tangem.data.staking.converters.ethpool.P2PYieldBalanceConverter +import com.tangem.datasource.api.ethpool.models.response.P2PEthPoolAccountResponse +import com.tangem.datasource.api.stakekit.models.response.model.YieldBalanceWrapperDTO +import com.tangem.datasource.local.token.converter.YieldBalanceConverter +import com.tangem.domain.models.StatusSource +import com.tangem.domain.models.staking.YieldBalance +import com.tangem.domain.staking.model.ethpool.P2PEthPoolVault + +internal fun YieldBalanceWrapperDTO.toDomain(source: StatusSource = StatusSource.CACHE): YieldBalance { + return YieldBalanceConverter(source = source).convert(this)!! +} + +internal fun P2PEthPoolAccountResponse.toDomain( + vault: P2PEthPoolVault = MockP2PEthPoolAccountResponseFactory.createMockVault(vaultAddress = vaultAddress), + source: StatusSource = StatusSource.CACHE, +): YieldBalance { + val account = P2PEthPoolAccountConverter.convert(this) + return P2PYieldBalanceConverter.convert( + account = account, + vault = vault, + address = account.delegatorAddress, + source = source, + ) +} \ No newline at end of file diff --git a/data/staking/src/test/kotlin/com/tangem/data/staking/YieldBalanceExt.kt b/data/staking/src/test/kotlin/com/tangem/data/staking/YieldBalanceExt.kt deleted file mode 100644 index 09b56e3b9c..0000000000 --- a/data/staking/src/test/kotlin/com/tangem/data/staking/YieldBalanceExt.kt +++ /dev/null @@ -1,10 +0,0 @@ -package com.tangem.data.staking - -import com.tangem.datasource.api.stakekit.models.response.model.YieldBalanceWrapperDTO -import com.tangem.datasource.local.token.converter.YieldBalanceConverter -import com.tangem.domain.models.StatusSource -import com.tangem.domain.models.staking.YieldBalance - -internal fun YieldBalanceWrapperDTO.toDomain(source: StatusSource = StatusSource.CACHE): YieldBalance { - return YieldBalanceConverter(source = source).convert(this)!! -} \ No newline at end of file diff --git a/data/staking/src/test/kotlin/com/tangem/data/staking/multi/DefaultMultiYieldBalanceFetcherTest.kt b/data/staking/src/test/kotlin/com/tangem/data/staking/multi/DefaultMultiYieldBalanceFetcherTest.kt index f7c56f3dc3..3e0c59cd8b 100644 --- a/data/staking/src/test/kotlin/com/tangem/data/staking/multi/DefaultMultiYieldBalanceFetcherTest.kt +++ b/data/staking/src/test/kotlin/com/tangem/data/staking/multi/DefaultMultiYieldBalanceFetcherTest.kt @@ -4,12 +4,15 @@ import arrow.core.toOption import com.tangem.common.test.data.staking.MockYieldBalanceWrapperDTOFactory import com.tangem.common.test.data.staking.MockYieldDTOFactory import com.tangem.common.test.domain.wallet.MockUserWalletFactory +import com.tangem.data.staking.store.P2PBalancesStore import com.tangem.data.staking.store.YieldsBalancesStore import com.tangem.data.staking.utils.YieldBalanceRequestBodyFactory import com.tangem.datasource.api.common.response.ApiResponse import com.tangem.datasource.api.common.response.ApiResponseError +import com.tangem.datasource.api.ethpool.P2PEthPoolApi import com.tangem.datasource.api.stakekit.StakeKitApi import com.tangem.datasource.api.stakekit.models.response.model.YieldBalanceWrapperDTO +import com.tangem.datasource.local.token.P2PEthPoolVaultsStore import com.tangem.datasource.local.token.StakingYieldsStore import com.tangem.datasource.local.userwallet.UserWalletsStore import com.tangem.domain.models.staking.StakingID @@ -33,13 +36,19 @@ internal class DefaultMultiYieldBalanceFetcherTest { private val userWalletsStore: UserWalletsStore = mockk() private val stakingYieldsStore: StakingYieldsStore = mockk() private val yieldsBalancesStore: YieldsBalancesStore = mockk(relaxUnitFun = true) + private val p2pBalancesStore: P2PBalancesStore = mockk(relaxUnitFun = true) private val stakeKitApi: StakeKitApi = mockk() + private val p2pApi: P2PEthPoolApi = mockk() + private val p2pVaultsStore: P2PEthPoolVaultsStore = mockk() private val fetcher = DefaultMultiYieldBalanceFetcher( userWalletsStore = userWalletsStore, stakingYieldsStore = stakingYieldsStore, yieldsBalancesStore = yieldsBalancesStore, + p2pBalancesStore = p2pBalancesStore, stakeKitApi = stakeKitApi, + p2pApi = p2pApi, + p2pVaultsStore = p2pVaultsStore, dispatchers = TestingCoroutineDispatcherProvider(), ) diff --git a/data/staking/src/test/kotlin/com/tangem/data/staking/multi/DefaultMultiYieldBalanceProducerTest.kt b/data/staking/src/test/kotlin/com/tangem/data/staking/multi/DefaultMultiYieldBalanceProducerTest.kt index 08fada9524..3435a0e034 100644 --- a/data/staking/src/test/kotlin/com/tangem/data/staking/multi/DefaultMultiYieldBalanceProducerTest.kt +++ b/data/staking/src/test/kotlin/com/tangem/data/staking/multi/DefaultMultiYieldBalanceProducerTest.kt @@ -2,10 +2,12 @@ package com.tangem.data.staking.multi import com.google.common.truth.Truth import com.tangem.common.test.data.staking.MockYieldBalanceWrapperDTOFactory +import com.tangem.common.test.data.staking.MockP2PEthPoolAccountResponseFactory +import com.tangem.data.staking.store.P2PBalancesStore import com.tangem.data.staking.store.YieldsBalancesStore import com.tangem.data.staking.toDomain -import com.tangem.domain.models.staking.StakingID -import com.tangem.domain.models.staking.YieldBalance +import com.tangem.domain.models.StatusSource +import com.tangem.domain.models.staking.* import com.tangem.domain.models.wallet.UserWalletId import com.tangem.domain.staking.multi.MultiYieldBalanceProducer import com.tangem.test.core.getEmittedValues @@ -25,11 +27,13 @@ internal class DefaultMultiYieldBalanceProducerTest { private val params = MultiYieldBalanceProducer.Params(userWalletId = UserWalletId("011")) private val yieldsBalancesStore = mockk() + private val p2pBalancesStore = mockk() private val dispatchers = TestingCoroutineDispatcherProvider() private val producer = DefaultMultiYieldBalanceProducer( params = params, yieldsBalancesStore = yieldsBalancesStore, + p2pBalancesStore = p2pBalancesStore, dispatchers = dispatchers, ) @@ -43,11 +47,13 @@ internal class DefaultMultiYieldBalanceProducerTest { val networksStatusesFlow = flowOf(balances) every { yieldsBalancesStore.get(params.userWalletId) } returns networksStatusesFlow + every { p2pBalancesStore.get(params.userWalletId) } returns flowOf(emptySet()) val actual = producer.produce() // check after producer.produce() verify { yieldsBalancesStore.get(params.userWalletId) } + verify { p2pBalancesStore.get(params.userWalletId) } val values = getEmittedValues(flow = actual) @@ -60,11 +66,13 @@ internal class DefaultMultiYieldBalanceProducerTest { val networksStatusesFlow = MutableSharedFlow>(replay = 2) every { yieldsBalancesStore.get(params.userWalletId) } returns networksStatusesFlow + every { p2pBalancesStore.get(params.userWalletId) } returns flowOf(emptySet()) val actual = producer.produce() // check after producer.produce() verify { yieldsBalancesStore.get(params.userWalletId) } + verify { p2pBalancesStore.get(params.userWalletId) } // first emit val balances = setOf( @@ -99,11 +107,13 @@ internal class DefaultMultiYieldBalanceProducerTest { val networksStatusesFlow = MutableSharedFlow>(replay = 2) every { yieldsBalancesStore.get(params.userWalletId) } returns networksStatusesFlow + every { p2pBalancesStore.get(params.userWalletId) } returns flowOf(emptySet()) val actual = producer.produce() // check after producer.produce() verify { yieldsBalancesStore.get(params.userWalletId) } + verify { p2pBalancesStore.get(params.userWalletId) } // first emit val wrappers = setOf( @@ -146,11 +156,13 @@ internal class DefaultMultiYieldBalanceProducerTest { .buffer(capacity = 5) every { yieldsBalancesStore.get(params.userWalletId) } returns networksStatusesFlow + every { p2pBalancesStore.get(params.userWalletId) } returns flowOf(emptySet()) val actual = producer.produceWithFallback() // check after producer.produce() verify { yieldsBalancesStore.get(params.userWalletId) } + verify { p2pBalancesStore.get(params.userWalletId) } val values1 = getEmittedValues(flow = actual) @@ -168,11 +180,13 @@ internal class DefaultMultiYieldBalanceProducerTest { @Test fun `test that flow is empty`() = runTest { every { yieldsBalancesStore.get(params.userWalletId) } returns emptyFlow() + every { p2pBalancesStore.get(params.userWalletId) } returns emptyFlow() val actual = producer.produce() // check after producer.produce() verify { yieldsBalancesStore.get(params.userWalletId) } + verify { p2pBalancesStore.get(params.userWalletId) } val values = getEmittedValues(flow = actual) @@ -180,6 +194,58 @@ internal class DefaultMultiYieldBalanceProducerTest { Truth.assertThat(values).isEqualTo(listOf(emptySet())) } + @Test + fun `test that StakeKit and P2P balances are combined`() = runTest { + val stakeKitBalances = createStakeKitBalances() + val p2pBalances = createP2PBalances() + + every { yieldsBalancesStore.get(params.userWalletId) } returns flowOf(stakeKitBalances) + every { p2pBalancesStore.get(params.userWalletId) } returns flowOf(p2pBalances) + + val actual = producer.produce() + + // check after producer.produce() + verify { yieldsBalancesStore.get(params.userWalletId) } + verify { p2pBalancesStore.get(params.userWalletId) } + + val values = getEmittedValues(flow = actual) + + Truth.assertThat(values.size).isEqualTo(1) + Truth.assertThat(values.first()).isEqualTo(stakeKitBalances + p2pBalances) + } + + @Test + fun `test that P2P balances are updated independently from StakeKit`() = runTest { + val stakeKitBalances = createStakeKitBalancesWithTonOnly() + val p2pFlow = MutableSharedFlow>(replay = 2) + + every { yieldsBalancesStore.get(params.userWalletId) } returns flowOf(stakeKitBalances) + every { p2pBalancesStore.get(params.userWalletId) } returns p2pFlow + + val actual = producer.produce() + + // check after producer.produce() + verify { yieldsBalancesStore.get(params.userWalletId) } + verify { p2pBalancesStore.get(params.userWalletId) } + + // first emit - empty P2P + p2pFlow.emit(emptySet()) + + val values1 = getEmittedValues(flow = actual) + + Truth.assertThat(values1.size).isEqualTo(1) + Truth.assertThat(values1.first()).isEqualTo(stakeKitBalances) + + // second emit - with P2P balance + val p2pBalances = createP2PBalances() + p2pFlow.emit(p2pBalances) + + val values2 = getEmittedValues(flow = actual) + + Truth.assertThat(values2.size).isEqualTo(2) + Truth.assertThat(values2.last()).isEqualTo(stakeKitBalances + p2pBalances) + } + private companion object { val tonId = MockYieldBalanceWrapperDTOFactory.defaultStakingId @@ -187,5 +253,30 @@ internal class DefaultMultiYieldBalanceProducerTest { integrationId = "solana-sol-native-multivalidator-staking", address = "0x1", ) + val p2pEthereumId = StakingID( + integrationId = "p2p-ethereum-pooled", + address = "0x5aa711F440Eb6d4361148bBD89d03464628ace84", + ) + + fun createStakeKitBalances(): Set { + return setOf( + MockYieldBalanceWrapperDTOFactory.createWithBalance(tonId).toDomain(), + MockYieldBalanceWrapperDTOFactory.createWithBalance(solanaId).toDomain(), + ) + } + + fun createStakeKitBalancesWithTonOnly(): Set { + return setOf( + MockYieldBalanceWrapperDTOFactory.createWithBalance(tonId).toDomain(), + ) + } + + fun createP2PBalances(): Set { + return setOf( + MockP2PEthPoolAccountResponseFactory.createWithBalance(stakingId = p2pEthereumId).toDomain( + source = StatusSource.ACTUAL, + ), + ) + } } } \ No newline at end of file diff --git a/domain/models/src/main/kotlin/com/tangem/domain/models/staking/StakingID.kt b/domain/models/src/main/kotlin/com/tangem/domain/models/staking/StakingID.kt index 5ed3a0b3ce..e684f70531 100644 --- a/domain/models/src/main/kotlin/com/tangem/domain/models/staking/StakingID.kt +++ b/domain/models/src/main/kotlin/com/tangem/domain/models/staking/StakingID.kt @@ -2,5 +2,13 @@ package com.tangem.domain.models.staking import kotlinx.serialization.Serializable +enum class StakingProvider { + STAKEKIT, + P2P, +} + @Serializable -data class StakingID(val integrationId: String, val address: String) \ No newline at end of file +data class StakingID( + val integrationId: String, + val address: String, +) \ No newline at end of file diff --git a/domain/staking/models/src/main/kotlin/com/tangem/domain/staking/model/StakingOption.kt b/domain/staking/models/src/main/kotlin/com/tangem/domain/staking/model/StakingOption.kt index 1a4563d0b5..d199edb836 100644 --- a/domain/staking/models/src/main/kotlin/com/tangem/domain/staking/model/StakingOption.kt +++ b/domain/staking/models/src/main/kotlin/com/tangem/domain/staking/model/StakingOption.kt @@ -41,7 +41,7 @@ sealed interface StakingOption { */ data class P2P(val vault: P2PEthPoolVault) : StakingOption { override val integrationId: String = - "p2p-ethereum-pooled:${vault.vaultAddress}" + "p2p-ethereum-pooled" override val apy: SerializedBigDecimal = vault.apy override val token: YieldToken = createEthToken() override val isAvailable: Boolean = !vault.isPrivate