Updated on 2026-08-14

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,
+        )
+    }
+}

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<Map<String, Set<P2PEthPoolAccountResponse>>> {
+        return DataStoreFactory.create(
+            serializer = MoshiDataStoreSerializer(
+                moshi = moshi,
+                types = mapWithStringKeyTypes(valueTypes = setTypes<P2PEthPoolAccountResponse>()),
+                defaultValue = emptyMap(),
+            ),
+            produceFile = { context.dataStoreFile(fileName = "p2p_balances") },
+            scope = CoroutineScope(context = dispatchers.io + SupervisorJob()),
+        )
+    }
+
     @Provides
     @Singleton
     fun provideP2PEthPoolVaultsStore(

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)

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<BalanceItem> = 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,
+        )
+    }
+}

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<Map<String, Set<P2PEthPoolAccountResponse>>>,
+        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 {

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
+ * Supports both StakeKit and P2P staking providers.
- * @property stakingYieldsStore  staking yields store
+ *
- * @property yieldsBalancesStore yields balances store
+ * @property userWalletsStore      user wallets store
- * @property stakeKitApi         stake kit API
+ * @property stakingYieldsStore    staking yields store
- * @property dispatchers         dispatchers
+ * @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(
+                if (p2pIds.isNotEmpty()) {
-                userWalletId = params.userWalletId,
+                    launch { fetchP2PBalances(params.userWalletId, p2pIds.toSet()) }
-                stakingIds = stakingIds,
+                }
-            )
+            }
-
-            fetch(userWalletId = params.userWalletId, stakingIds = availableStakingIds)
         }
             .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<StakingID>) {
+        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<StakingID>) {
+        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<StakingID>,
+        vaults: List<com.tangem.domain.staking.model.ethpool.P2PEthPoolVault>,
+    ) {
+        safeApiCall(
+            call = {
+                val addresses = stakingIds.map { it.address }.toSet()
+
+                val responses = mutableSetOf<P2PEthPoolAccountResponse>()
+
+                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<StakingID>) {
+    private suspend fun fetchFromStakeKit(userWalletId: UserWalletId, stakingIds: Set<StakingID>) {
         safeApiCall(
             call = {
                 val requests = stakingIds.map(YieldBalanceRequestBodyFactory::create)

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<Set<YieldBalance>> = emptySet<YieldBalance>().some()
 
     override fun produce(): Flow<Set<YieldBalance>> {
-        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)

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<UserWalletId, Set<YieldBalance>>
+internal typealias WalletIdWithP2PResponses = Map<String, Set<P2PEthPoolAccountResponse>>
+
+/**
+ * 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<WalletIdWithP2PBalances>,
+    private val persistenceStore: DataStore<WalletIdWithP2PResponses>,
+    private val vaultsProvider: suspend () -> List<P2PEthPoolVault>,
+    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<Set<YieldBalance>> {
+        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<YieldBalance>? {
+        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<StakingID>) {
+        updateInRuntime(userWalletId = userWalletId, stakingIds = stakingIds) {
+            it.copySealed(source = StatusSource.CACHE)
+        }
+    }
+
+    override suspend fun storeActual(userWalletId: UserWalletId, values: Set<P2PEthPoolAccountResponse>) {
+        coroutineScope {
+            launch { storeInRuntime(userWalletId = userWalletId, values = values) }
+            launch { storeInPersistence(userWalletId = userWalletId, values = values) }
+        }
+    }
+
+    override suspend fun storeError(userWalletId: UserWalletId, stakingIds: Set<StakingID>) {
+        updateInRuntime(
+            userWalletId = userWalletId,
+            stakingIds = stakingIds,
+            ifNotFound = ::createErrorYieldBalance,
+            update = { it.copySealed(source = StatusSource.ONLY_CACHE) },
+        )
+    }
+
+    override suspend fun clear(userWalletId: UserWalletId, stakingIds: Set<StakingID>) {
+        coroutineScope {
+            launch { clearInRuntime(userWalletId = userWalletId, stakingIds = stakingIds) }
+            launch { clearInPersistence(userWalletId = userWalletId, stakingIds = stakingIds) }
+        }
+    }
+
+    private suspend fun storeInRuntime(userWalletId: UserWalletId, values: Set<P2PEthPoolAccountResponse>) {
+        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<P2PEthPoolAccountResponse>) {
+        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<StakingID>) {
+        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<StakingID>) {
+        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<StakingID>,
+        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)
+}

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<Set<YieldBalance>>
+
+    suspend fun getSyncOrNull(userWalletId: UserWalletId, stakingId: StakingID): YieldBalance?
+
+    suspend fun getAllSyncOrNull(userWalletId: UserWalletId): Set<YieldBalance>?
+
+    suspend fun refresh(userWalletId: UserWalletId, stakingId: StakingID)
+
+    suspend fun refresh(userWalletId: UserWalletId, stakingIds: Set<StakingID>)
+
+    suspend fun storeActual(userWalletId: UserWalletId, values: Set<P2PEthPoolAccountResponse>)
+
+    suspend fun storeError(userWalletId: UserWalletId, stakingIds: Set<StakingID>)
+
+    suspend fun clear(userWalletId: UserWalletId, stakingIds: Set<StakingID>)
+}

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,
+    )
+}

data/staking/src/test/kotlin/com/tangem/data/staking/YieldBalanceExt.kt

@@ -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)!!
-}

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(),
     )
 

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.StatusSource
-import com.tangem.domain.models.staking.YieldBalance
+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<YieldsBalancesStore>()
+    private val p2pBalancesStore = mockk<P2PBalancesStore>()
     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<Set<YieldBalance>>(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<Set<YieldBalance>>(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<YieldBalance>()))
     }
 
+    @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<Set<YieldBalance>>(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<YieldBalance> {
+            return setOf(
+                MockYieldBalanceWrapperDTOFactory.createWithBalance(tonId).toDomain(),
+                MockYieldBalanceWrapperDTOFactory.createWithBalance(solanaId).toDomain(),
+            )
+        }
+
+        fun createStakeKitBalancesWithTonOnly(): Set<YieldBalance> {
+            return setOf(
+                MockYieldBalanceWrapperDTOFactory.createWithBalance(tonId).toDomain(),
+            )
+        }
+
+        fun createP2PBalances(): Set<YieldBalance> {
+            return setOf(
+                MockP2PEthPoolAccountResponseFactory.createWithBalance(stakingId = p2pEthereumId).toDomain(
+                    source = StatusSource.ACTUAL,
+                ),
+            )
+        }
     }
 }

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)
+data class StakingID(
+    val integrationId: String,
+    val address: String,
+)

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