Updated on 2026-08-14
This commit is contained in:
Tangem
parent
4e05219c25
commit
f172215393
17 changed files with 550 additions and 145 deletions
|
|
@ -2,7 +2,7 @@
|
|||
"formatVersion": 1,
|
||||
"database": {
|
||||
"version": 1,
|
||||
"identityHash": "942246bf975439606ad20e05b930827c",
|
||||
"identityHash": "36ff7cc1634c100cadc4b04fc2eba1c9",
|
||||
"entities": [
|
||||
{
|
||||
"tableName": "express_provider",
|
||||
|
|
@ -44,7 +44,7 @@
|
|||
},
|
||||
{
|
||||
"tableName": "express_exchange",
|
||||
"createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`tx_id` TEXT NOT NULL, `owner_address` TEXT NOT NULL, `provider_id` TEXT NOT NULL, `from_address` TEXT NOT NULL, `payin_address` TEXT NOT NULL, `payin_extra_id` TEXT, `payout_address` TEXT NOT NULL, `refund_address` TEXT, `refund_extra_id` TEXT, `rate_type` TEXT NOT NULL, `status` TEXT NOT NULL, `external_tx_id` TEXT, `external_tx_status` TEXT, `external_tx_url` TEXT, `payin_hash` TEXT, `payout_hash` TEXT, `refund_network` TEXT, `refund_contract_address` TEXT, `created_at` TEXT NOT NULL, `pay_till` TEXT, `average_duration` INTEGER, `from_contract_address` TEXT NOT NULL, `from_network` TEXT NOT NULL, `from_decimals` INTEGER NOT NULL, `from_amount` TEXT NOT NULL, `from_actual_amount` TEXT, `to_contract_address` TEXT NOT NULL, `to_network` TEXT NOT NULL, `to_decimals` INTEGER NOT NULL, `to_amount` TEXT NOT NULL, `to_actual_amount` TEXT, PRIMARY KEY(`tx_id`), FOREIGN KEY(`provider_id`) REFERENCES `express_provider`(`id`) ON UPDATE NO ACTION ON DELETE RESTRICT )",
|
||||
"createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`tx_id` TEXT NOT NULL, `owner_address` TEXT NOT NULL, `provider_id` TEXT NOT NULL, `from_address` TEXT NOT NULL, `payin_address` TEXT NOT NULL, `payin_extra_id` TEXT, `payout_address` TEXT NOT NULL, `refund_address` TEXT, `refund_extra_id` TEXT, `rate_type` TEXT NOT NULL, `status` TEXT NOT NULL, `external_tx_id` TEXT, `external_tx_status` TEXT, `external_tx_url` TEXT, `payin_hash` TEXT, `payout_hash` TEXT, `refund_network` TEXT, `refund_contract_address` TEXT, `created_at` TEXT NOT NULL, `pay_till` TEXT, `average_duration` INTEGER, `from_contract_address` TEXT NOT NULL, `from_network` TEXT NOT NULL, `from_decimals` INTEGER NOT NULL, `from_amount` TEXT NOT NULL, `from_actual_amount` TEXT, `to_contract_address` TEXT NOT NULL, `to_network` TEXT NOT NULL, `to_decimals` INTEGER NOT NULL, `to_amount` TEXT NOT NULL, `to_actual_amount` TEXT, PRIMARY KEY(`tx_id`))",
|
||||
"fields": [
|
||||
{
|
||||
"fieldPath": "txId",
|
||||
|
|
@ -270,25 +270,22 @@
|
|||
],
|
||||
"orders": [],
|
||||
"createSql": "CREATE INDEX IF NOT EXISTS `index_express_exchange_owner_address_payout_hash` ON `${TABLE_NAME}` (`owner_address`, `payout_hash`)"
|
||||
}
|
||||
],
|
||||
"foreignKeys": [
|
||||
},
|
||||
{
|
||||
"table": "express_provider",
|
||||
"onDelete": "RESTRICT",
|
||||
"onUpdate": "NO ACTION",
|
||||
"columns": [
|
||||
"name": "index_express_exchange_provider_id",
|
||||
"unique": false,
|
||||
"columnNames": [
|
||||
"provider_id"
|
||||
],
|
||||
"referencedColumns": [
|
||||
"id"
|
||||
]
|
||||
"orders": [],
|
||||
"createSql": "CREATE INDEX IF NOT EXISTS `index_express_exchange_provider_id` ON `${TABLE_NAME}` (`provider_id`)"
|
||||
}
|
||||
]
|
||||
],
|
||||
"foreignKeys": []
|
||||
},
|
||||
{
|
||||
"tableName": "express_onramp",
|
||||
"createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`tx_id` TEXT NOT NULL, `owner_address` TEXT NOT NULL, `provider_id` TEXT NOT NULL, `from_address` TEXT NOT NULL, `payin_address` TEXT NOT NULL, `payin_extra_id` TEXT, `payout_address` TEXT NOT NULL, `refund_address` TEXT, `refund_extra_id` TEXT, `rate_type` TEXT NOT NULL, `status` TEXT NOT NULL, `external_tx_id` TEXT, `external_tx_status` TEXT, `external_tx_url` TEXT, `payin_hash` TEXT, `payout_hash` TEXT, `refund_network` TEXT, `refund_contract_address` TEXT, `created_at` TEXT NOT NULL, `pay_till` TEXT, `average_duration` INTEGER, `from_contract_address` TEXT NOT NULL, `from_network` TEXT NOT NULL, `from_decimals` INTEGER NOT NULL, `from_amount` TEXT NOT NULL, `from_actual_amount` TEXT, `to_contract_address` TEXT NOT NULL, `to_network` TEXT NOT NULL, `to_decimals` INTEGER NOT NULL, `to_amount` TEXT NOT NULL, `to_actual_amount` TEXT, PRIMARY KEY(`tx_id`), FOREIGN KEY(`provider_id`) REFERENCES `express_provider`(`id`) ON UPDATE NO ACTION ON DELETE RESTRICT )",
|
||||
"createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`tx_id` TEXT NOT NULL, `owner_address` TEXT NOT NULL, `provider_id` TEXT NOT NULL, `from_address` TEXT NOT NULL, `payin_address` TEXT NOT NULL, `payin_extra_id` TEXT, `payout_address` TEXT NOT NULL, `refund_address` TEXT, `refund_extra_id` TEXT, `rate_type` TEXT NOT NULL, `status` TEXT NOT NULL, `external_tx_id` TEXT, `external_tx_status` TEXT, `external_tx_url` TEXT, `payin_hash` TEXT, `payout_hash` TEXT, `refund_network` TEXT, `refund_contract_address` TEXT, `created_at` TEXT NOT NULL, `pay_till` TEXT, `average_duration` INTEGER, `from_contract_address` TEXT NOT NULL, `from_network` TEXT NOT NULL, `from_decimals` INTEGER NOT NULL, `from_amount` TEXT NOT NULL, `from_actual_amount` TEXT, `to_contract_address` TEXT NOT NULL, `to_network` TEXT NOT NULL, `to_decimals` INTEGER NOT NULL, `to_amount` TEXT NOT NULL, `to_actual_amount` TEXT, PRIMARY KEY(`tx_id`))",
|
||||
"fields": [
|
||||
{
|
||||
"fieldPath": "txId",
|
||||
|
|
@ -504,21 +501,18 @@
|
|||
],
|
||||
"orders": [],
|
||||
"createSql": "CREATE INDEX IF NOT EXISTS `index_express_onramp_owner_address_payout_hash` ON `${TABLE_NAME}` (`owner_address`, `payout_hash`)"
|
||||
}
|
||||
],
|
||||
"foreignKeys": [
|
||||
},
|
||||
{
|
||||
"table": "express_provider",
|
||||
"onDelete": "RESTRICT",
|
||||
"onUpdate": "NO ACTION",
|
||||
"columns": [
|
||||
"name": "index_express_onramp_provider_id",
|
||||
"unique": false,
|
||||
"columnNames": [
|
||||
"provider_id"
|
||||
],
|
||||
"referencedColumns": [
|
||||
"id"
|
||||
]
|
||||
"orders": [],
|
||||
"createSql": "CREATE INDEX IF NOT EXISTS `index_express_onramp_provider_id` ON `${TABLE_NAME}` (`provider_id`)"
|
||||
}
|
||||
]
|
||||
],
|
||||
"foreignKeys": []
|
||||
},
|
||||
{
|
||||
"tableName": "express_sync_state",
|
||||
|
|
@ -569,7 +563,7 @@
|
|||
"views": [],
|
||||
"setupQueries": [
|
||||
"CREATE TABLE IF NOT EXISTS room_master_table (id INTEGER PRIMARY KEY,identity_hash TEXT)",
|
||||
"INSERT OR REPLACE INTO room_master_table (id,identity_hash) VALUES(42, '942246bf975439606ad20e05b930827c')"
|
||||
"INSERT OR REPLACE INTO room_master_table (id,identity_hash) VALUES(42, '36ff7cc1634c100cadc4b04fc2eba1c9')"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
|
@ -89,6 +89,7 @@ interface TangemExpressApi {
|
|||
|
||||
@GET("history/exchange")
|
||||
suspend fun getHistory(
|
||||
@Header("user-id") userWalletId: String,
|
||||
@Query("fromAddress") fromAddress: String,
|
||||
@Query("afterCursor") cursor: String?,
|
||||
@Query("limit") limit: Int = 100,
|
||||
|
|
@ -96,6 +97,7 @@ interface TangemExpressApi {
|
|||
|
||||
@GET("history/delta/exchange")
|
||||
suspend fun getHistoryDelta(
|
||||
@Header("user-id") userWalletId: String,
|
||||
@Query("fromAddress") fromAddress: String,
|
||||
@Query("beforeCursor") cursor: String?,
|
||||
@Query("limit") limit: Int = 100,
|
||||
|
|
|
|||
|
|
@ -9,7 +9,7 @@ data class ExpressPagination(
|
|||
val endCursor: String?,
|
||||
@Json(name = "startDeltaCursor")
|
||||
val startDeltaCursor: String?,
|
||||
@Json(name = "hasMore")
|
||||
@Json(name = "hasNextPage")
|
||||
val hasMore: Boolean,
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -91,6 +91,7 @@ interface OnrampApi {
|
|||
|
||||
@GET("history/onramp")
|
||||
suspend fun getHistory(
|
||||
@Header("user-id") userWalletId: String,
|
||||
@Query("payoutAddress") payoutAddress: String,
|
||||
@Query("afterCursor") afterCursor: String?,
|
||||
@Query("limit") limit: Int = 100,
|
||||
|
|
@ -98,6 +99,7 @@ interface OnrampApi {
|
|||
|
||||
@GET("history/delta/onramp")
|
||||
suspend fun getHistoryDelta(
|
||||
@Header("user-id") userWalletId: String,
|
||||
@Query("payoutAddress") payoutAddress: String,
|
||||
@Query("beforeCursor") cursor: String?,
|
||||
@Query("limit") limit: Int = 100,
|
||||
|
|
|
|||
|
|
@ -13,6 +13,27 @@ interface ExpressSyncStateDao {
|
|||
@Insert(onConflict = OnConflictStrategy.REPLACE)
|
||||
suspend fun upsert(item: ExpressSyncStateEntity)
|
||||
|
||||
@Query(
|
||||
"""
|
||||
UPDATE express_sync_state
|
||||
SET after_cursor = :afterCursor,
|
||||
is_initial_completed = :isInitialCompleted
|
||||
WHERE type = :type
|
||||
AND address = :address
|
||||
""",
|
||||
)
|
||||
suspend fun updateHistoryCursor(type: String, address: String, afterCursor: String?, isInitialCompleted: Boolean)
|
||||
|
||||
@Query(
|
||||
"""
|
||||
UPDATE express_sync_state
|
||||
SET delta_cursor = :deltaCursor
|
||||
WHERE type = :type
|
||||
AND address = :address
|
||||
""",
|
||||
)
|
||||
suspend fun updateDeltaCursor(type: String, address: String, deltaCursor: String)
|
||||
|
||||
@Query(
|
||||
"""
|
||||
SELECT *
|
||||
|
|
|
|||
|
|
@ -9,18 +9,11 @@ import androidx.room.*
|
|||
*/
|
||||
@Entity(
|
||||
tableName = "express_exchange",
|
||||
foreignKeys = [
|
||||
ForeignKey(
|
||||
entity = ExpressProviderEntity::class,
|
||||
parentColumns = ["id"],
|
||||
childColumns = ["provider_id"],
|
||||
onDelete = ForeignKey.RESTRICT,
|
||||
),
|
||||
],
|
||||
indices = [
|
||||
Index(value = ["owner_address", "from_network", "created_at"]),
|
||||
Index(value = ["owner_address", "payin_hash"]),
|
||||
Index(value = ["owner_address", "payout_hash"]),
|
||||
Index(value = ["provider_id"]),
|
||||
],
|
||||
)
|
||||
data class ExpressExchangeEntity(
|
||||
|
|
|
|||
|
|
@ -9,17 +9,10 @@ import androidx.room.*
|
|||
*/
|
||||
@Entity(
|
||||
tableName = "express_onramp",
|
||||
foreignKeys = [
|
||||
ForeignKey(
|
||||
entity = ExpressProviderEntity::class,
|
||||
parentColumns = ["id"],
|
||||
childColumns = ["provider_id"],
|
||||
onDelete = ForeignKey.RESTRICT,
|
||||
),
|
||||
],
|
||||
indices = [
|
||||
Index(value = ["owner_address", "to_network", "created_at"]),
|
||||
Index(value = ["owner_address", "payout_hash"]),
|
||||
Index(value = ["provider_id"]),
|
||||
],
|
||||
)
|
||||
data class ExpressOnrampEntity(
|
||||
|
|
|
|||
|
|
@ -7,7 +7,9 @@ import com.tangem.data.txhistory.fetcher.TxHistoryFetcherUtils.Companion.receive
|
|||
import com.tangem.domain.account.supplier.SingleAccountSupplier
|
||||
import com.tangem.domain.models.account.Account
|
||||
import com.tangem.domain.models.account.AccountId
|
||||
import com.tangem.domain.models.account.AccountStatus
|
||||
import com.tangem.domain.models.currency.CryptoCurrency
|
||||
import com.tangem.domain.models.currency.CryptoCurrencyStatus
|
||||
import com.tangem.domain.models.network.Network
|
||||
import com.tangem.domain.models.wallet.UserWalletId
|
||||
import com.tangem.domain.pay.usecase.GetPaymentAccountCryptoCurrencyStatusUseCase
|
||||
|
|
@ -50,15 +52,24 @@ internal class DefaultAccountTxHistoryFetcher @AssistedInject constructor(
|
|||
private fun buildFlow(): Flow<Unit> = channelFlow {
|
||||
val accountFlow = singleAccountSupplier(accountId).stateIn(this)
|
||||
|
||||
val controlFetchersFlow = when (accountFlow.value) {
|
||||
is Account.CryptoPortfolio -> accountFlow
|
||||
.filterIsInstance<Account.CryptoPortfolio>()
|
||||
.controlFetchersForCryptoAccount()
|
||||
is Account.Payment -> controlFetchersForPaymentAccount()
|
||||
when (val account = accountFlow.value) {
|
||||
is Account.CryptoPortfolio -> {
|
||||
account.getExpressKeys().createExpressFetcher()
|
||||
accountFlow
|
||||
.filterIsInstance<Account.CryptoPortfolio>()
|
||||
.controlFetchersForCryptoAccount()
|
||||
.launchIn(this)
|
||||
}
|
||||
is Account.Payment -> {
|
||||
paymentAccountCurrency.invokeSync(walletId)
|
||||
.getOrNull()
|
||||
?.controlFetchersForPaymentAccount()
|
||||
controlFetchersForPaymentAccount()
|
||||
.launchIn(this)
|
||||
}
|
||||
// Virtual account tx-history isn't wired yet (separate task) — no express fetchers for now.
|
||||
is Account.Virtual -> emptyFlow()
|
||||
is Account.Virtual -> Unit
|
||||
}
|
||||
controlFetchersFlow.launchIn(this)
|
||||
|
||||
receiveTrigger().onEach { trigger ->
|
||||
when (trigger) {
|
||||
|
|
@ -76,33 +87,43 @@ internal class DefaultAccountTxHistoryFetcher @AssistedInject constructor(
|
|||
|
||||
private fun controlFetchersForPaymentAccount(): Flow<Unit> {
|
||||
return paymentAccountCurrency(walletId)
|
||||
.map { (_, paymentCurrency) ->
|
||||
val paymentNetwork = paymentCurrency.currency.network
|
||||
val address = getAddress(walletId, paymentCurrency.currency)
|
||||
if (paymentNetwork.isSupportExpressTxHistory() && !address.isNullOrBlank()) {
|
||||
getOrPutExpressFetcher(address, accountId)
|
||||
} else {
|
||||
// single currency for payment account, so we can close all(one)
|
||||
expressFetchers.forEach { (_, fetcher) -> fetcher.close() }
|
||||
expressFetchers.clear()
|
||||
}
|
||||
}
|
||||
.map { pair -> pair.controlFetchersForPaymentAccount() }
|
||||
}
|
||||
|
||||
private suspend fun Pair<AccountStatus.Payment, CryptoCurrencyStatus>?.controlFetchersForPaymentAccount() {
|
||||
val (_, paymentCurrency) = this ?: return
|
||||
val paymentNetwork = paymentCurrency.currency.network
|
||||
val address = getAddress(walletId, paymentCurrency.currency)
|
||||
if (paymentNetwork.isSupportExpressTxHistory() && !address.isNullOrBlank()) {
|
||||
getOrPutExpressFetcher(address)
|
||||
} else {
|
||||
// single currency for payment account, so we can close all(one)
|
||||
expressFetchers.forEach { (_, fetcher) -> fetcher.close() }
|
||||
expressFetchers.clear()
|
||||
}
|
||||
}
|
||||
|
||||
private fun Flow<Account.CryptoPortfolio>.controlFetchersForCryptoAccount(): Flow<Unit> {
|
||||
return map { account -> account.cryptoCurrencies }
|
||||
.map { currencies ->
|
||||
val onlyCoins = currencies.filterIsInstance<CryptoCurrency.Coin>()
|
||||
val networks = onlyCoins.map { coin -> coin.network }
|
||||
val newExpressKeys = networks
|
||||
.filter { net -> net.isSupportExpressTxHistory() }
|
||||
.mapNotNull { net -> getAddress(walletId, net) }
|
||||
.toSet()
|
||||
val previousExpressKeys = expressFetchers.keys
|
||||
val removed = previousExpressKeys - newExpressKeys
|
||||
newExpressKeys.forEach { address -> getOrPutExpressFetcher(address, accountId) }
|
||||
removed.forEach { address -> expressFetchers.remove(address)?.close() }
|
||||
}
|
||||
return map { account ->
|
||||
val newExpressKeys = account.getExpressKeys()
|
||||
val previousExpressKeys = expressFetchers.keys
|
||||
val removed = previousExpressKeys - newExpressKeys
|
||||
newExpressKeys.createExpressFetcher()
|
||||
removed.forEach { address -> expressFetchers.remove(address)?.close() }
|
||||
}
|
||||
}
|
||||
|
||||
private fun Set<String>.createExpressFetcher() = this.forEach { address -> getOrPutExpressFetcher(address) }
|
||||
|
||||
private suspend fun Account.CryptoPortfolio.getExpressKeys(): Set<String> {
|
||||
val currencies = this.cryptoCurrencies
|
||||
val onlyCoins = currencies.filterIsInstance<CryptoCurrency.Coin>()
|
||||
val networks = onlyCoins.map { coin -> coin.network }
|
||||
val newExpressKeys = networks
|
||||
.filter { net -> net.isSupportExpressTxHistory() }
|
||||
.mapNotNull { net -> getAddress(walletId, net) }
|
||||
.toSet()
|
||||
return newExpressKeys
|
||||
}
|
||||
|
||||
@Suppress("FunctionOnlyReturningConstant") // todo txhistory check
|
||||
|
|
@ -116,8 +137,8 @@ internal class DefaultAccountTxHistoryFetcher @AssistedInject constructor(
|
|||
private suspend fun getAddress(userWalletId: UserWalletId, network: Network): String? =
|
||||
walletManagersFacade.getDefaultAddress(userWalletId, network)
|
||||
|
||||
private fun getOrPutExpressFetcher(address: String, id: AccountId): ExpressTxHistoryFetcher {
|
||||
return expressFetchers.computeIfAbsent(address) { expressFetcherFactory.create(address, id) }
|
||||
private fun getOrPutExpressFetcher(address: String): ExpressTxHistoryFetcher {
|
||||
return expressFetchers.computeIfAbsent(address) { expressFetcherFactory.create(address, accountId) }
|
||||
}
|
||||
|
||||
@AssistedFactory
|
||||
|
|
|
|||
|
|
@ -45,6 +45,8 @@ internal class DefaultAppTxHistoryFetcher @Inject constructor(
|
|||
.invokeAsMap(isOnlyMultiCurrency = true, filterLocked = true)
|
||||
.stateIn(this)
|
||||
|
||||
walletsFlow.value.keys.createForNewWallets()
|
||||
|
||||
selectedWalletUseCase.selectedFlow()
|
||||
.filter { wallet -> wallet.isMultiCurrency }
|
||||
// todo txhistory some init trigger?
|
||||
|
|
@ -69,8 +71,9 @@ internal class DefaultAppTxHistoryFetcher @Inject constructor(
|
|||
.collect {}
|
||||
}
|
||||
|
||||
private fun Flow<Set<UserWalletId>>.createForNewWallets() =
|
||||
onEach { ids -> ids.forEach { walletId -> getOrPutFetcher(walletId) } }
|
||||
private fun Flow<Set<UserWalletId>>.createForNewWallets() = onEach { ids -> ids.createForNewWallets() }
|
||||
|
||||
private fun Set<UserWalletId>.createForNewWallets() = this.forEach { walletId -> getOrPutFetcher(walletId) }
|
||||
|
||||
private fun Flow<Set<UserWalletId>>.closeForRemovedWallets() = runningReduce { previousIds, newIds ->
|
||||
val removedWallets = previousIds.subtract(newIds)
|
||||
|
|
|
|||
|
|
@ -1,28 +1,162 @@
|
|||
package com.tangem.data.txhistory.fetcher
|
||||
|
||||
import com.tangem.data.txhistory.fetcher.TxHistoryFetcherUtils.Companion.cancelScope
|
||||
import com.tangem.data.txhistory.fetcher.TxHistoryFetcherUtils.Companion.defaultLaunchIn
|
||||
import com.tangem.data.txhistory.fetcher.TxHistoryFetcherUtils.Companion.receiveTriggerInstance
|
||||
import com.tangem.data.txhistory.fetcher.TxHistoryFetcherUtils.Companion.retryThreeTimes
|
||||
import com.tangem.data.txhistory.repository.ExpressHistoryRepository
|
||||
import com.tangem.datasource.api.express.models.response.ExchangeHistoryDeltaResponse
|
||||
import com.tangem.datasource.api.express.models.response.ExchangeHistoryResponse
|
||||
import com.tangem.datasource.api.onramp.models.response.OnrampHistoryDeltaResponse
|
||||
import com.tangem.datasource.api.onramp.models.response.OnrampHistoryResponse
|
||||
import com.tangem.datasource.local.txhistory.db.dao.ExpressSyncStateDao
|
||||
import com.tangem.datasource.local.txhistory.db.entity.express.ExpressSyncStateEntity
|
||||
import com.tangem.domain.models.account.AccountId
|
||||
import com.tangem.domain.models.wallet.UserWalletId
|
||||
import com.tangem.domain.txhistory.fetcher.ExpressTxHistoryFetcher
|
||||
import com.tangem.domain.txhistory.fetcher.TxHistoryExpressTrigger
|
||||
import com.tangem.domain.txhistory.fetcher.TxHistoryFetchTrigger
|
||||
import dagger.assisted.Assisted
|
||||
import dagger.assisted.AssistedFactory
|
||||
import dagger.assisted.AssistedInject
|
||||
import kotlinx.coroutines.Job
|
||||
import kotlinx.coroutines.flow.first
|
||||
import kotlinx.coroutines.flow.firstOrNull
|
||||
import kotlinx.coroutines.flow.flow
|
||||
import kotlinx.coroutines.flow.onEach
|
||||
import kotlinx.coroutines.launch
|
||||
|
||||
internal class DefaultExpressTxHistoryFetcher @AssistedInject constructor(
|
||||
@Assisted override val address: String,
|
||||
@Assisted private val accountId: AccountId,
|
||||
private val utils: TxHistoryFetcherUtils,
|
||||
private val expressSyncStateDao: ExpressSyncStateDao,
|
||||
private val expressHistoryRepository: ExpressHistoryRepository,
|
||||
) : ExpressTxHistoryFetcher, TxHistoryFetcherUtils by utils {
|
||||
|
||||
private val userWalletId: UserWalletId get() = accountId.userWalletId
|
||||
|
||||
private var exchangeInitialPaginationJob: Job? = null
|
||||
private var exchangeDeltaPaginationJob: Job? = null
|
||||
|
||||
private var onrampInitialPaginationJob: Job? = null
|
||||
private var onrampDeltaPaginationJob: Job? = null
|
||||
|
||||
init {
|
||||
val receiveFlow = receiveTriggerInstance<TxHistoryExpressTrigger>()
|
||||
.onEach { trigger ->
|
||||
when (trigger) {
|
||||
is TxHistoryFetchTrigger.TokenDetailsOpen,
|
||||
is TxHistoryFetchTrigger.TokenDetailsPTR,
|
||||
-> {
|
||||
fetchExchange()
|
||||
fetchOnramp()
|
||||
}
|
||||
}
|
||||
}
|
||||
defaultLaunchIn(receiveFlow)
|
||||
}
|
||||
|
||||
override suspend fun invoke(params: TxHistoryExpressTrigger) {
|
||||
utils.sendTrigger(params)
|
||||
accountId
|
||||
}
|
||||
|
||||
override fun close() {
|
||||
cancelScope()
|
||||
}
|
||||
|
||||
private fun fetchExchange() {
|
||||
if (exchangeDeltaPaginationJob?.isActive == true) return
|
||||
exchangeDeltaPaginationJob = fetcherScope.launch {
|
||||
val isFirstFetch = expressSyncState() == null
|
||||
|
||||
if (isFirstFetch) {
|
||||
flow { emit(expressHistoryRepository.fetchExchangeHistory(address, userWalletId)) }
|
||||
.retryThreeTimes()
|
||||
.firstOrNull() ?: return@launch
|
||||
}
|
||||
|
||||
if (exchangeInitialPaginationJob?.isActive != true) {
|
||||
exchangeInitialPaginationJob = launch { expressInitialPagination() }
|
||||
}
|
||||
|
||||
expressDeltaPagination()
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun expressInitialPagination() {
|
||||
if (expressSyncState()?.isInitialCompleted == true) return
|
||||
var hasMore = true
|
||||
while (hasMore) {
|
||||
val pageResult: ExchangeHistoryResponse =
|
||||
flow { emit(expressHistoryRepository.fetchExchangeHistory(address, userWalletId)) }
|
||||
.retryThreeTimes()
|
||||
.firstOrNull() ?: return
|
||||
hasMore = pageResult.pagination.hasMore
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun expressDeltaPagination() {
|
||||
var hasMore = true
|
||||
while (hasMore) {
|
||||
val pageResult: ExchangeHistoryDeltaResponse =
|
||||
flow { emit(expressHistoryRepository.fetchExchangeHistoryDelta(address, userWalletId)) }
|
||||
.retryThreeTimes()
|
||||
.firstOrNull() ?: return
|
||||
hasMore = pageResult.pagination.hasMore
|
||||
}
|
||||
}
|
||||
|
||||
private fun fetchOnramp() {
|
||||
if (onrampDeltaPaginationJob?.isActive == true) return
|
||||
onrampDeltaPaginationJob = fetcherScope.launch {
|
||||
val isFirstFetch = onrampSyncState() == null
|
||||
|
||||
if (isFirstFetch) {
|
||||
flow { emit(expressHistoryRepository.fetchOnrampHistory(address, userWalletId)) }
|
||||
.retryThreeTimes()
|
||||
.firstOrNull() ?: return@launch
|
||||
}
|
||||
|
||||
if (onrampInitialPaginationJob?.isActive != true) {
|
||||
onrampInitialPaginationJob = launch { onrampInitialPagination() }
|
||||
}
|
||||
|
||||
onrampDeltaPagination()
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun onrampInitialPagination() {
|
||||
if (onrampSyncState()?.isInitialCompleted == true) return
|
||||
var hasMore = true
|
||||
while (hasMore) {
|
||||
val pageResult: OnrampHistoryResponse =
|
||||
flow { emit(expressHistoryRepository.fetchOnrampHistory(address, userWalletId)) }
|
||||
.retryThreeTimes()
|
||||
.firstOrNull() ?: return
|
||||
hasMore = pageResult.pagination.hasMore
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun onrampDeltaPagination() {
|
||||
var hasMore = true
|
||||
while (hasMore) {
|
||||
val pageResult: OnrampHistoryDeltaResponse =
|
||||
flow { emit(expressHistoryRepository.fetchOnrampHistoryDelta(address, userWalletId)) }
|
||||
.retryThreeTimes()
|
||||
.firstOrNull() ?: return
|
||||
hasMore = pageResult.pagination.hasMore
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun expressSyncState(): ExpressSyncStateEntity? = expressSyncStateDao
|
||||
.observe(ExpressSyncStateEntity.Type.EXCHANGE.name, address)
|
||||
.first()
|
||||
|
||||
private suspend fun onrampSyncState(): ExpressSyncStateEntity? = expressSyncStateDao
|
||||
.observe(ExpressSyncStateEntity.Type.ONRAMP.name, address)
|
||||
.first()
|
||||
|
||||
@AssistedFactory
|
||||
internal interface Factory {
|
||||
fun create(address: String, accountId: AccountId): DefaultExpressTxHistoryFetcher
|
||||
|
|
|
|||
|
|
@ -48,6 +48,9 @@ internal class DefaultWalletTxHistoryFetcher @AssistedInject constructor(
|
|||
.stateIn(this)
|
||||
|
||||
fun accountList(): AccountList = accountListFlow.value
|
||||
accountList().accounts
|
||||
.mapTo(mutableSetOf()) { it.accountId }
|
||||
.createForNewAccounts()
|
||||
|
||||
accountListFlow
|
||||
.map { accountList -> accountList.accounts.mapTo(mutableSetOf()) { account -> account.accountId } }
|
||||
|
|
@ -68,8 +71,9 @@ internal class DefaultWalletTxHistoryFetcher @AssistedInject constructor(
|
|||
.collect {}
|
||||
}
|
||||
|
||||
private fun Flow<Set<AccountId>>.createForNewAccounts() =
|
||||
onEach { ids -> ids.forEach { id -> getOrPutFetcher(id) } }
|
||||
private fun Flow<Set<AccountId>>.createForNewAccounts() = onEach { ids -> ids.createForNewAccounts() }
|
||||
|
||||
private fun Set<AccountId>.createForNewAccounts() = this.forEach { id -> getOrPutFetcher(id) }
|
||||
|
||||
private fun Flow<Set<AccountId>>.closeForRemovedAccounts() = runningReduce { previousIds, newIds ->
|
||||
val removedWallets = previousIds.subtract(newIds)
|
||||
|
|
|
|||
|
|
@ -5,9 +5,7 @@ import com.tangem.core.analytics.api.AnalyticsExceptionHandler
|
|||
import com.tangem.domain.txhistory.fetcher.TxHistoryFetchTrigger
|
||||
import com.tangem.utils.coroutines.AppCoroutineScope
|
||||
import com.tangem.utils.logging.TangemLogger
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.SupervisorJob
|
||||
import kotlinx.coroutines.cancel
|
||||
import kotlinx.coroutines.*
|
||||
import kotlinx.coroutines.channels.Channel
|
||||
import kotlinx.coroutines.flow.*
|
||||
import kotlinx.coroutines.plus
|
||||
|
|
@ -36,6 +34,13 @@ internal interface TxHistoryFetcherUtils {
|
|||
}
|
||||
.launchIn(fetcherScope)
|
||||
|
||||
@Suppress("MagicNumber")
|
||||
fun <T> Flow<T>.retryThreeTimes() = retry(3) { error ->
|
||||
logError(error)
|
||||
delay(1000)
|
||||
true
|
||||
}.catch { e -> logError(e) }
|
||||
|
||||
fun TxHistoryFetcherUtils.receiveTrigger(): Flow<TxHistoryFetchTrigger> {
|
||||
return triggersBuffer.receiveAsFlow()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -6,6 +6,8 @@ import com.tangem.datasource.api.express.TangemExpressApi
|
|||
import com.tangem.datasource.api.express.models.response.ExchangeHistoryDeltaResponse
|
||||
import com.tangem.datasource.api.express.models.response.ExchangeHistoryResponse
|
||||
import com.tangem.datasource.api.express.models.response.ExchangeItemResponse
|
||||
import com.tangem.datasource.api.express.models.response.ExpressPagination
|
||||
import com.tangem.datasource.api.express.models.response.ExpressPaginationDelta
|
||||
import com.tangem.datasource.api.onramp.OnrampApi
|
||||
import com.tangem.datasource.api.onramp.models.response.OnrampHistoryDeltaResponse
|
||||
import com.tangem.datasource.api.onramp.models.response.OnrampHistoryResponse
|
||||
|
|
@ -13,6 +15,7 @@ import com.tangem.datasource.api.onramp.models.response.OnrampItemResponse
|
|||
import com.tangem.datasource.local.txhistory.db.dao.ExpressHistoryDao
|
||||
import com.tangem.datasource.local.txhistory.db.dao.ExpressSyncStateDao
|
||||
import com.tangem.datasource.local.txhistory.db.entity.express.ExpressSyncStateEntity
|
||||
import com.tangem.domain.models.wallet.UserWalletId
|
||||
import kotlinx.coroutines.flow.first
|
||||
import javax.inject.Inject
|
||||
|
||||
|
|
@ -27,61 +30,97 @@ internal class ExpressHistoryRepository @Inject constructor(
|
|||
private val expressSyncStateDao: ExpressSyncStateDao,
|
||||
) {
|
||||
|
||||
suspend fun fetchExchangeHistory(fromAddress: String, limit: Int = DEFAULT_LIMIT): ExchangeHistoryResponse {
|
||||
suspend fun fetchExchangeHistory(
|
||||
fromAddress: String,
|
||||
userWalletId: UserWalletId,
|
||||
limit: Int = DEFAULT_LIMIT,
|
||||
): ExchangeHistoryResponse {
|
||||
val state = syncState(ExpressSyncStateEntity.Type.EXCHANGE, fromAddress)
|
||||
|
||||
val response = exchangeApi.getHistory(
|
||||
userWalletId = userWalletId.stringValue,
|
||||
fromAddress = fromAddress,
|
||||
cursor = state?.afterCursor,
|
||||
limit = limit,
|
||||
).getOrThrow()
|
||||
|
||||
saveExchanges(ownerAddress = fromAddress, items = response.items)
|
||||
persistHistoryState(
|
||||
type = ExpressSyncStateEntity.Type.EXCHANGE,
|
||||
address = fromAddress,
|
||||
previous = state,
|
||||
pagination = response.pagination,
|
||||
)
|
||||
return response
|
||||
}
|
||||
|
||||
suspend fun fetchExchangeHistoryDelta(
|
||||
fromAddress: String,
|
||||
userWalletId: UserWalletId,
|
||||
limit: Int = DEFAULT_LIMIT,
|
||||
): ExchangeHistoryDeltaResponse {
|
||||
val state = syncState(ExpressSyncStateEntity.Type.EXCHANGE, fromAddress)
|
||||
|
||||
val response = exchangeApi.getHistoryDelta(
|
||||
userWalletId = userWalletId.stringValue,
|
||||
fromAddress = fromAddress,
|
||||
cursor = state?.deltaCursor,
|
||||
limit = limit,
|
||||
).getOrThrow()
|
||||
|
||||
saveExchanges(ownerAddress = fromAddress, items = response.items)
|
||||
persistDeltaState(
|
||||
type = ExpressSyncStateEntity.Type.EXCHANGE,
|
||||
address = fromAddress,
|
||||
pagination = response.pagination,
|
||||
)
|
||||
return response
|
||||
}
|
||||
|
||||
suspend fun fetchOnrampHistory(payoutAddress: String, limit: Int = DEFAULT_LIMIT): OnrampHistoryResponse {
|
||||
suspend fun fetchOnrampHistory(
|
||||
payoutAddress: String,
|
||||
userWalletId: UserWalletId,
|
||||
limit: Int = DEFAULT_LIMIT,
|
||||
): OnrampHistoryResponse {
|
||||
val state = syncState(ExpressSyncStateEntity.Type.ONRAMP, payoutAddress)
|
||||
|
||||
val response = onrampApi.getHistory(
|
||||
userWalletId = userWalletId.stringValue,
|
||||
payoutAddress = payoutAddress,
|
||||
afterCursor = state?.afterCursor,
|
||||
limit = limit,
|
||||
).getOrThrow()
|
||||
|
||||
saveOnramps(ownerAddress = payoutAddress, items = response.items)
|
||||
persistHistoryState(
|
||||
type = ExpressSyncStateEntity.Type.ONRAMP,
|
||||
address = payoutAddress,
|
||||
previous = state,
|
||||
pagination = response.pagination,
|
||||
)
|
||||
return response
|
||||
}
|
||||
|
||||
suspend fun fetchOnrampHistoryDelta(
|
||||
payoutAddress: String,
|
||||
userWalletId: UserWalletId,
|
||||
limit: Int = DEFAULT_LIMIT,
|
||||
): OnrampHistoryDeltaResponse {
|
||||
val state = syncState(ExpressSyncStateEntity.Type.ONRAMP, payoutAddress)
|
||||
|
||||
val response = onrampApi.getHistoryDelta(
|
||||
userWalletId = userWalletId.stringValue,
|
||||
payoutAddress = payoutAddress,
|
||||
cursor = state?.deltaCursor,
|
||||
limit = limit,
|
||||
).getOrThrow()
|
||||
|
||||
saveOnramps(ownerAddress = payoutAddress, items = response.items)
|
||||
persistDeltaState(
|
||||
type = ExpressSyncStateEntity.Type.ONRAMP,
|
||||
address = payoutAddress,
|
||||
pagination = response.pagination,
|
||||
)
|
||||
return response
|
||||
}
|
||||
|
||||
|
|
@ -97,6 +136,41 @@ internal class ExpressHistoryRepository @Inject constructor(
|
|||
expressHistoryDao.upsertOnramps(items.map { it.toEntity(ownerAddress) })
|
||||
}
|
||||
|
||||
private suspend fun persistHistoryState(
|
||||
type: ExpressSyncStateEntity.Type,
|
||||
address: String,
|
||||
previous: ExpressSyncStateEntity?,
|
||||
pagination: ExpressPagination,
|
||||
) {
|
||||
if (previous == null) {
|
||||
expressSyncStateDao.upsert(
|
||||
ExpressSyncStateEntity(
|
||||
type = type.name,
|
||||
address = address,
|
||||
isInitialCompleted = !pagination.hasMore,
|
||||
afterCursor = pagination.endCursor,
|
||||
deltaCursor = pagination.startDeltaCursor,
|
||||
),
|
||||
)
|
||||
} else {
|
||||
expressSyncStateDao.updateHistoryCursor(
|
||||
type = type.name,
|
||||
address = address,
|
||||
afterCursor = pagination.endCursor,
|
||||
isInitialCompleted = !pagination.hasMore,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun persistDeltaState(
|
||||
type: ExpressSyncStateEntity.Type,
|
||||
address: String,
|
||||
pagination: ExpressPaginationDelta,
|
||||
) {
|
||||
val cursor = pagination.startCursor ?: return
|
||||
expressSyncStateDao.updateDeltaCursor(type = type.name, address = address, deltaCursor = cursor)
|
||||
}
|
||||
|
||||
private companion object {
|
||||
const val DEFAULT_LIMIT = 100
|
||||
}
|
||||
|
|
|
|||
|
|
@ -5,9 +5,7 @@ import com.tangem.test.core.TestAppCoroutineScope
|
|||
import com.tangem.common.test.domain.token.MockCryptoCurrencyFactory
|
||||
import com.tangem.domain.models.account.Account
|
||||
import com.tangem.domain.models.account.AccountId
|
||||
import com.tangem.domain.models.account.AccountStatus
|
||||
import com.tangem.domain.models.currency.CryptoCurrency
|
||||
import com.tangem.domain.models.currency.CryptoCurrencyStatus
|
||||
import com.tangem.domain.pay.usecase.GetPaymentAccountCryptoCurrencyStatusUseCase
|
||||
import com.tangem.domain.txhistory.fetcher.TxHistoryFetchTrigger
|
||||
import com.tangem.domain.account.supplier.SingleAccountSupplier
|
||||
|
|
@ -16,7 +14,6 @@ import com.tangem.test.mock.MockAccounts
|
|||
import io.mockk.*
|
||||
import kotlinx.coroutines.ExperimentalCoroutinesApi
|
||||
import kotlinx.coroutines.flow.MutableStateFlow
|
||||
import kotlinx.coroutines.flow.flowOf
|
||||
import kotlinx.coroutines.job
|
||||
import kotlinx.coroutines.test.*
|
||||
import org.junit.jupiter.api.BeforeEach
|
||||
|
|
@ -106,29 +103,6 @@ internal class DefaultAccountTxHistoryFetcherTest {
|
|||
coVerify(exactly = 1) { expressFetcher.invoke(trigger) }
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `creates express fetcher for a payment account currency`() = runTest {
|
||||
val utils = createUtils()
|
||||
val paymentAccountId = AccountId.forPaymentAccount(WALLET_ID)
|
||||
val accountFlow = MutableStateFlow<Account>(Account.Payment(WALLET_ID))
|
||||
every { singleAccountSupplier.invoke(paymentAccountId) } returns accountFlow
|
||||
|
||||
val paymentStatus = mockk<AccountStatus.Payment>(relaxed = true)
|
||||
val currencyStatus = mockk<CryptoCurrencyStatus> { every { currency } returns coin }
|
||||
every { paymentAccountCurrency.invoke(WALLET_ID) } returns flowOf(paymentStatus to currencyStatus)
|
||||
coEvery { walletManagersFacade.getDefaultAddress(WALLET_ID, coin.network) } returns ADDRESS
|
||||
val expressFetcher = relaxedExpressFetcher()
|
||||
every { expressFetcherFactory.create(ADDRESS, paymentAccountId) } returns expressFetcher
|
||||
|
||||
// Act
|
||||
val fetcher = createFetcher(paymentAccountId, utils)
|
||||
advanceUntilIdle()
|
||||
|
||||
// Assert
|
||||
assertThat(fetcher.expressFetchers.keys).containsExactly(ADDRESS)
|
||||
verify(exactly = 1) { expressFetcherFactory.create(ADDRESS, paymentAccountId) }
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `close cancels scope and closes all express fetchers`() = runTest {
|
||||
val utils = createUtils()
|
||||
|
|
|
|||
|
|
@ -0,0 +1,176 @@
|
|||
package com.tangem.data.txhistory.fetcher
|
||||
|
||||
import com.google.common.truth.Truth.assertThat
|
||||
import com.tangem.test.core.TestAppCoroutineScope
|
||||
import com.tangem.common.test.domain.token.MockCryptoCurrencyFactory
|
||||
import com.tangem.data.txhistory.repository.ExpressHistoryRepository
|
||||
import com.tangem.datasource.api.express.models.response.ExchangeHistoryDeltaResponse
|
||||
import com.tangem.datasource.api.express.models.response.ExchangeHistoryResponse
|
||||
import com.tangem.datasource.api.express.models.response.ExpressPagination
|
||||
import com.tangem.datasource.api.express.models.response.ExpressPaginationDelta
|
||||
import com.tangem.datasource.api.onramp.models.response.OnrampHistoryDeltaResponse
|
||||
import com.tangem.datasource.api.onramp.models.response.OnrampHistoryResponse
|
||||
import com.tangem.datasource.local.txhistory.db.dao.ExpressSyncStateDao
|
||||
import com.tangem.datasource.local.txhistory.db.entity.express.ExpressSyncStateEntity
|
||||
import com.tangem.domain.models.account.AccountId
|
||||
import com.tangem.domain.models.currency.CryptoCurrency
|
||||
import com.tangem.domain.txhistory.fetcher.TxHistoryFetchTrigger
|
||||
import com.tangem.test.mock.MockAccounts
|
||||
import io.mockk.*
|
||||
import kotlinx.coroutines.ExperimentalCoroutinesApi
|
||||
import kotlinx.coroutines.flow.flowOf
|
||||
import kotlinx.coroutines.job
|
||||
import kotlinx.coroutines.test.*
|
||||
import org.junit.jupiter.api.BeforeEach
|
||||
import org.junit.jupiter.api.Test
|
||||
import org.junit.jupiter.api.TestInstance
|
||||
|
||||
@OptIn(ExperimentalCoroutinesApi::class)
|
||||
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
|
||||
internal class DefaultExpressTxHistoryFetcherTest {
|
||||
|
||||
private val expressSyncStateDao: ExpressSyncStateDao = mockk()
|
||||
private val expressHistoryRepository: ExpressHistoryRepository = mockk()
|
||||
|
||||
private val coin: CryptoCurrency = MockCryptoCurrencyFactory().ethereum
|
||||
|
||||
@BeforeEach
|
||||
fun setup() {
|
||||
clearMocks(expressSyncStateDao, expressHistoryRepository)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `exposes the address it was created with`() = runTest {
|
||||
val fetcher = createFetcher(createUtils())
|
||||
|
||||
assertThat(fetcher.address).isEqualTo(ADDRESS)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `on first trigger fetches initial exchange and onramp history for the address`() = runTest {
|
||||
stubAllSuccess(hasMore = false)
|
||||
val fetcher = createFetcher(createUtils())
|
||||
|
||||
// Act
|
||||
fetcher.invoke(TxHistoryFetchTrigger.TokenDetailsOpen(walletId = WALLET_ID, currency = coin))
|
||||
advanceUntilIdle()
|
||||
|
||||
// Assert
|
||||
coVerify(atLeast = 1) { expressHistoryRepository.fetchExchangeHistory(ADDRESS, any()) }
|
||||
coVerify(atLeast = 1) { expressHistoryRepository.fetchOnrampHistory(ADDRESS, any()) }
|
||||
coVerify(exactly = 1) { expressHistoryRepository.fetchExchangeHistoryDelta(ADDRESS, any()) }
|
||||
coVerify(exactly = 1) { expressHistoryRepository.fetchOnrampHistoryDelta(ADDRESS, any()) }
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `continues exchange initial pagination while hasMore is true`() = runTest {
|
||||
every { expressSyncStateDao.observe(any(), ADDRESS) } returns flowOf(null)
|
||||
// 1st call: initial fetch in fetchExchange (pagination ignored)
|
||||
// 2nd call: pagination loop, hasMore = true -> continue
|
||||
// 3rd call: pagination loop, hasMore = false -> stop
|
||||
coEvery { expressHistoryRepository.fetchExchangeHistory(ADDRESS, any()) } returnsMany listOf(
|
||||
exchangeResponse(hasMore = true),
|
||||
exchangeResponse(hasMore = true),
|
||||
exchangeResponse(hasMore = false),
|
||||
)
|
||||
coEvery { expressHistoryRepository.fetchExchangeHistoryDelta(ADDRESS, any()) } returns
|
||||
exchangeDeltaResponse(hasMore = false)
|
||||
coEvery { expressHistoryRepository.fetchOnrampHistory(ADDRESS, any()) } returns onrampResponse(hasMore = false)
|
||||
coEvery { expressHistoryRepository.fetchOnrampHistoryDelta(ADDRESS, any()) } returns
|
||||
onrampDeltaResponse(hasMore = false)
|
||||
val fetcher = createFetcher(createUtils())
|
||||
|
||||
// Act
|
||||
fetcher.invoke(TxHistoryFetchTrigger.TokenDetailsOpen(walletId = WALLET_ID, currency = coin))
|
||||
advanceUntilIdle()
|
||||
|
||||
// Assert
|
||||
coVerify(exactly = 3) { expressHistoryRepository.fetchExchangeHistory(ADDRESS, any()) }
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `skips initial pagination when it is already completed`() = runTest {
|
||||
every { expressSyncStateDao.observe(any(), ADDRESS) } returns flowOf(completedSyncState())
|
||||
coEvery { expressHistoryRepository.fetchExchangeHistoryDelta(ADDRESS, any()) } returns
|
||||
exchangeDeltaResponse(hasMore = false)
|
||||
coEvery { expressHistoryRepository.fetchOnrampHistoryDelta(ADDRESS, any()) } returns
|
||||
onrampDeltaResponse(hasMore = false)
|
||||
val fetcher = createFetcher(createUtils())
|
||||
|
||||
// Act
|
||||
fetcher.invoke(TxHistoryFetchTrigger.TokenDetailsPTR(walletId = WALLET_ID, currency = coin))
|
||||
advanceUntilIdle()
|
||||
|
||||
// Assert: no initial history fetch, only the delta pagination runs
|
||||
coVerify(exactly = 0) { expressHistoryRepository.fetchExchangeHistory(any(), any()) }
|
||||
coVerify(exactly = 0) { expressHistoryRepository.fetchOnrampHistory(any(), any()) }
|
||||
coVerify(exactly = 1) { expressHistoryRepository.fetchExchangeHistoryDelta(ADDRESS, any()) }
|
||||
coVerify(exactly = 1) { expressHistoryRepository.fetchOnrampHistoryDelta(ADDRESS, any()) }
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `close cancels the fetcher scope`() = runTest {
|
||||
val utils = createUtils()
|
||||
val fetcher = createFetcher(utils)
|
||||
|
||||
// Act
|
||||
fetcher.close()
|
||||
|
||||
// Assert
|
||||
assertThat(utils.fetcherScope.coroutineContext.job.isActive).isFalse()
|
||||
}
|
||||
|
||||
private fun stubAllSuccess(hasMore: Boolean) {
|
||||
every { expressSyncStateDao.observe(any(), ADDRESS) } returns flowOf(null)
|
||||
coEvery { expressHistoryRepository.fetchExchangeHistory(ADDRESS, any()) } returns exchangeResponse(hasMore)
|
||||
coEvery { expressHistoryRepository.fetchExchangeHistoryDelta(ADDRESS, any()) } returns
|
||||
exchangeDeltaResponse(hasMore)
|
||||
coEvery { expressHistoryRepository.fetchOnrampHistory(ADDRESS, any()) } returns onrampResponse(hasMore)
|
||||
coEvery { expressHistoryRepository.fetchOnrampHistoryDelta(ADDRESS, any()) } returns onrampDeltaResponse(hasMore)
|
||||
}
|
||||
|
||||
private fun TestScope.createUtils(): DefaultTxHistoryFetcherUtils = DefaultTxHistoryFetcherUtils(
|
||||
appScope = TestAppCoroutineScope(testScope = this),
|
||||
analyticsEventHandler = mockk(relaxed = true),
|
||||
analyticsExceptionHandler = mockk(relaxed = true),
|
||||
)
|
||||
|
||||
private fun createFetcher(utils: DefaultTxHistoryFetcherUtils) = DefaultExpressTxHistoryFetcher(
|
||||
address = ADDRESS,
|
||||
accountId = ACCOUNT_ID,
|
||||
utils = utils,
|
||||
expressSyncStateDao = expressSyncStateDao,
|
||||
expressHistoryRepository = expressHistoryRepository,
|
||||
)
|
||||
|
||||
private fun exchangeResponse(hasMore: Boolean) =
|
||||
ExchangeHistoryResponse(items = emptyList(), pagination = pagination(hasMore))
|
||||
|
||||
private fun exchangeDeltaResponse(hasMore: Boolean) =
|
||||
ExchangeHistoryDeltaResponse(items = emptyList(), pagination = paginationDelta(hasMore))
|
||||
|
||||
private fun onrampResponse(hasMore: Boolean) =
|
||||
OnrampHistoryResponse(items = emptyList(), pagination = pagination(hasMore))
|
||||
|
||||
private fun onrampDeltaResponse(hasMore: Boolean) =
|
||||
OnrampHistoryDeltaResponse(items = emptyList(), pagination = paginationDelta(hasMore))
|
||||
|
||||
private fun pagination(hasMore: Boolean) =
|
||||
ExpressPagination(endCursor = null, startDeltaCursor = null, hasMore = hasMore)
|
||||
|
||||
private fun paginationDelta(hasMore: Boolean) = ExpressPaginationDelta(startCursor = null, hasMore = hasMore)
|
||||
|
||||
private fun completedSyncState() = ExpressSyncStateEntity(
|
||||
type = ExpressSyncStateEntity.Type.EXCHANGE.name,
|
||||
address = ADDRESS,
|
||||
isInitialCompleted = true,
|
||||
afterCursor = null,
|
||||
deltaCursor = null,
|
||||
)
|
||||
|
||||
private companion object {
|
||||
val WALLET_ID = MockAccounts.userWalletId
|
||||
val ACCOUNT_ID = AccountId.forMainCryptoPortfolio(WALLET_ID)
|
||||
const val ADDRESS = "0xEthAddress"
|
||||
}
|
||||
}
|
||||
|
|
@ -18,6 +18,7 @@ import com.tangem.datasource.local.txhistory.db.dao.ExpressHistoryDao
|
|||
import com.tangem.datasource.local.txhistory.db.dao.ExpressSyncStateDao
|
||||
import com.tangem.datasource.local.txhistory.db.entity.express.ExpressExchangeEntity
|
||||
import com.tangem.datasource.local.txhistory.db.entity.express.ExpressSyncStateEntity
|
||||
import com.tangem.domain.models.wallet.UserWalletId
|
||||
import io.mockk.clearMocks
|
||||
import io.mockk.coEvery
|
||||
import io.mockk.coVerify
|
||||
|
|
@ -35,7 +36,7 @@ internal class ExpressHistoryRepositoryTest {
|
|||
private val exchangeApi: TangemExpressApi = mockk()
|
||||
private val onrampApi: OnrampApi = mockk()
|
||||
private val expressHistoryDao: ExpressHistoryDao = mockk(relaxUnitFun = true)
|
||||
private val expressSyncStateDao: ExpressSyncStateDao = mockk()
|
||||
private val expressSyncStateDao: ExpressSyncStateDao = mockk(relaxUnitFun = true)
|
||||
|
||||
private val repository = ExpressHistoryRepository(
|
||||
exchangeApi = exchangeApi,
|
||||
|
|
@ -58,16 +59,16 @@ internal class ExpressHistoryRepositoryTest {
|
|||
val response = ExchangeHistoryResponse(items = listOf(item), pagination = pagination())
|
||||
stubSyncState(ExpressSyncStateEntity.Type.EXCHANGE, ADDRESS, syncState(afterCursor = AFTER_CURSOR))
|
||||
coEvery {
|
||||
exchangeApi.getHistory(fromAddress = ADDRESS, cursor = AFTER_CURSOR, limit = any())
|
||||
exchangeApi.getHistory(userWalletId = USER_WALLET_ID_VALUE, fromAddress = ADDRESS, cursor = AFTER_CURSOR, limit = any())
|
||||
} returns ApiResponse.Success(response)
|
||||
|
||||
// WHEN
|
||||
val result = repository.fetchExchangeHistory(fromAddress = ADDRESS)
|
||||
val result = repository.fetchExchangeHistory(fromAddress = ADDRESS, userWalletId = USER_WALLET_ID)
|
||||
|
||||
// THEN
|
||||
assertThat(result).isEqualTo(response)
|
||||
coVerify(exactly = 1) {
|
||||
exchangeApi.getHistory(fromAddress = ADDRESS, cursor = AFTER_CURSOR, limit = DEFAULT_LIMIT)
|
||||
exchangeApi.getHistory(userWalletId = USER_WALLET_ID_VALUE, fromAddress = ADDRESS, cursor = AFTER_CURSOR, limit = DEFAULT_LIMIT)
|
||||
}
|
||||
coVerify(exactly = 1) { expressHistoryDao.upsertExchanges(listOf(item.toEntity(ADDRESS))) }
|
||||
}
|
||||
|
|
@ -78,15 +79,15 @@ internal class ExpressHistoryRepositoryTest {
|
|||
val response = ExchangeHistoryResponse(items = emptyList(), pagination = pagination())
|
||||
stubSyncState(ExpressSyncStateEntity.Type.EXCHANGE, ADDRESS, state = null)
|
||||
coEvery {
|
||||
exchangeApi.getHistory(fromAddress = ADDRESS, cursor = null, limit = any())
|
||||
exchangeApi.getHistory(userWalletId = USER_WALLET_ID_VALUE, fromAddress = ADDRESS, cursor = null, limit = any())
|
||||
} returns ApiResponse.Success(response)
|
||||
|
||||
// WHEN
|
||||
repository.fetchExchangeHistory(fromAddress = ADDRESS)
|
||||
repository.fetchExchangeHistory(fromAddress = ADDRESS, userWalletId = USER_WALLET_ID)
|
||||
|
||||
// THEN
|
||||
coVerify(exactly = 1) {
|
||||
exchangeApi.getHistory(fromAddress = ADDRESS, cursor = null, limit = DEFAULT_LIMIT)
|
||||
exchangeApi.getHistory(userWalletId = USER_WALLET_ID_VALUE, fromAddress = ADDRESS, cursor = null, limit = DEFAULT_LIMIT)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -96,15 +97,15 @@ internal class ExpressHistoryRepositoryTest {
|
|||
val response = ExchangeHistoryResponse(items = emptyList(), pagination = pagination())
|
||||
stubSyncState(ExpressSyncStateEntity.Type.EXCHANGE, ADDRESS, syncState(afterCursor = AFTER_CURSOR))
|
||||
coEvery {
|
||||
exchangeApi.getHistory(fromAddress = ADDRESS, cursor = AFTER_CURSOR, limit = any())
|
||||
exchangeApi.getHistory(userWalletId = USER_WALLET_ID_VALUE, fromAddress = ADDRESS, cursor = AFTER_CURSOR, limit = any())
|
||||
} returns ApiResponse.Success(response)
|
||||
|
||||
// WHEN
|
||||
repository.fetchExchangeHistory(fromAddress = ADDRESS, limit = 25)
|
||||
repository.fetchExchangeHistory(fromAddress = ADDRESS, userWalletId = USER_WALLET_ID, limit = 25)
|
||||
|
||||
// THEN
|
||||
coVerify(exactly = 1) {
|
||||
exchangeApi.getHistory(fromAddress = ADDRESS, cursor = AFTER_CURSOR, limit = 25)
|
||||
exchangeApi.getHistory(userWalletId = USER_WALLET_ID_VALUE, fromAddress = ADDRESS, cursor = AFTER_CURSOR, limit = 25)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -114,11 +115,11 @@ internal class ExpressHistoryRepositoryTest {
|
|||
stubSyncState(ExpressSyncStateEntity.Type.EXCHANGE, ADDRESS, syncState(afterCursor = AFTER_CURSOR))
|
||||
val error = httpError()
|
||||
coEvery {
|
||||
exchangeApi.getHistory(fromAddress = ADDRESS, cursor = AFTER_CURSOR, limit = any())
|
||||
exchangeApi.getHistory(userWalletId = USER_WALLET_ID_VALUE, fromAddress = ADDRESS, cursor = AFTER_CURSOR, limit = any())
|
||||
} returns ApiResponse.Error(error).cast()
|
||||
|
||||
// WHEN
|
||||
val thrown = runCatching { repository.fetchExchangeHistory(fromAddress = ADDRESS) }.exceptionOrNull()
|
||||
val thrown = runCatching { repository.fetchExchangeHistory(fromAddress = ADDRESS, userWalletId = USER_WALLET_ID) }.exceptionOrNull()
|
||||
|
||||
// THEN
|
||||
assertThat(thrown).isEqualTo(error)
|
||||
|
|
@ -132,16 +133,16 @@ internal class ExpressHistoryRepositoryTest {
|
|||
val response = ExchangeHistoryDeltaResponse(items = listOf(item), pagination = paginationDelta())
|
||||
stubSyncState(ExpressSyncStateEntity.Type.EXCHANGE, ADDRESS, syncState(deltaCursor = DELTA_CURSOR))
|
||||
coEvery {
|
||||
exchangeApi.getHistoryDelta(fromAddress = ADDRESS, cursor = DELTA_CURSOR, limit = any())
|
||||
exchangeApi.getHistoryDelta(userWalletId = USER_WALLET_ID_VALUE, fromAddress = ADDRESS, cursor = DELTA_CURSOR, limit = any())
|
||||
} returns ApiResponse.Success(response)
|
||||
|
||||
// WHEN
|
||||
val result = repository.fetchExchangeHistoryDelta(fromAddress = ADDRESS)
|
||||
val result = repository.fetchExchangeHistoryDelta(fromAddress = ADDRESS, userWalletId = USER_WALLET_ID)
|
||||
|
||||
// THEN
|
||||
assertThat(result).isEqualTo(response)
|
||||
coVerify(exactly = 1) {
|
||||
exchangeApi.getHistoryDelta(fromAddress = ADDRESS, cursor = DELTA_CURSOR, limit = DEFAULT_LIMIT)
|
||||
exchangeApi.getHistoryDelta(userWalletId = USER_WALLET_ID_VALUE, fromAddress = ADDRESS, cursor = DELTA_CURSOR, limit = DEFAULT_LIMIT)
|
||||
}
|
||||
coVerify(exactly = 1) { expressHistoryDao.upsertExchanges(listOf(item.toEntity(ADDRESS))) }
|
||||
}
|
||||
|
|
@ -157,16 +158,16 @@ internal class ExpressHistoryRepositoryTest {
|
|||
val response = OnrampHistoryResponse(items = listOf(item), pagination = pagination())
|
||||
stubSyncState(ExpressSyncStateEntity.Type.ONRAMP, ADDRESS, syncState(afterCursor = AFTER_CURSOR))
|
||||
coEvery {
|
||||
onrampApi.getHistory(payoutAddress = ADDRESS, afterCursor = AFTER_CURSOR, limit = any())
|
||||
onrampApi.getHistory(userWalletId = USER_WALLET_ID_VALUE, payoutAddress = ADDRESS, afterCursor = AFTER_CURSOR, limit = any())
|
||||
} returns ApiResponse.Success(response)
|
||||
|
||||
// WHEN
|
||||
val result = repository.fetchOnrampHistory(payoutAddress = ADDRESS)
|
||||
val result = repository.fetchOnrampHistory(payoutAddress = ADDRESS, userWalletId = USER_WALLET_ID)
|
||||
|
||||
// THEN
|
||||
assertThat(result).isEqualTo(response)
|
||||
coVerify(exactly = 1) {
|
||||
onrampApi.getHistory(payoutAddress = ADDRESS, afterCursor = AFTER_CURSOR, limit = DEFAULT_LIMIT)
|
||||
onrampApi.getHistory(userWalletId = USER_WALLET_ID_VALUE, payoutAddress = ADDRESS, afterCursor = AFTER_CURSOR, limit = DEFAULT_LIMIT)
|
||||
}
|
||||
coVerify(exactly = 1) { expressHistoryDao.upsertOnramps(listOf(item.toEntity(ADDRESS))) }
|
||||
}
|
||||
|
|
@ -177,15 +178,15 @@ internal class ExpressHistoryRepositoryTest {
|
|||
val response = OnrampHistoryResponse(items = emptyList(), pagination = pagination())
|
||||
stubSyncState(ExpressSyncStateEntity.Type.ONRAMP, ADDRESS, state = null)
|
||||
coEvery {
|
||||
onrampApi.getHistory(payoutAddress = ADDRESS, afterCursor = null, limit = any())
|
||||
onrampApi.getHistory(userWalletId = USER_WALLET_ID_VALUE, payoutAddress = ADDRESS, afterCursor = null, limit = any())
|
||||
} returns ApiResponse.Success(response)
|
||||
|
||||
// WHEN
|
||||
repository.fetchOnrampHistory(payoutAddress = ADDRESS)
|
||||
repository.fetchOnrampHistory(payoutAddress = ADDRESS, userWalletId = USER_WALLET_ID)
|
||||
|
||||
// THEN
|
||||
coVerify(exactly = 1) {
|
||||
onrampApi.getHistory(payoutAddress = ADDRESS, afterCursor = null, limit = DEFAULT_LIMIT)
|
||||
onrampApi.getHistory(userWalletId = USER_WALLET_ID_VALUE, payoutAddress = ADDRESS, afterCursor = null, limit = DEFAULT_LIMIT)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -196,16 +197,16 @@ internal class ExpressHistoryRepositoryTest {
|
|||
val response = OnrampHistoryDeltaResponse(items = listOf(item), pagination = paginationDelta())
|
||||
stubSyncState(ExpressSyncStateEntity.Type.ONRAMP, ADDRESS, syncState(deltaCursor = DELTA_CURSOR))
|
||||
coEvery {
|
||||
onrampApi.getHistoryDelta(payoutAddress = ADDRESS, cursor = DELTA_CURSOR, limit = any())
|
||||
onrampApi.getHistoryDelta(userWalletId = USER_WALLET_ID_VALUE, payoutAddress = ADDRESS, cursor = DELTA_CURSOR, limit = any())
|
||||
} returns ApiResponse.Success(response)
|
||||
|
||||
// WHEN
|
||||
val result = repository.fetchOnrampHistoryDelta(payoutAddress = ADDRESS)
|
||||
val result = repository.fetchOnrampHistoryDelta(payoutAddress = ADDRESS, userWalletId = USER_WALLET_ID)
|
||||
|
||||
// THEN
|
||||
assertThat(result).isEqualTo(response)
|
||||
coVerify(exactly = 1) {
|
||||
onrampApi.getHistoryDelta(payoutAddress = ADDRESS, cursor = DELTA_CURSOR, limit = DEFAULT_LIMIT)
|
||||
onrampApi.getHistoryDelta(userWalletId = USER_WALLET_ID_VALUE, payoutAddress = ADDRESS, cursor = DELTA_CURSOR, limit = DEFAULT_LIMIT)
|
||||
}
|
||||
coVerify(exactly = 1) { expressHistoryDao.upsertOnramps(listOf(item.toEntity(ADDRESS))) }
|
||||
}
|
||||
|
|
@ -216,11 +217,11 @@ internal class ExpressHistoryRepositoryTest {
|
|||
stubSyncState(ExpressSyncStateEntity.Type.ONRAMP, ADDRESS, syncState(afterCursor = AFTER_CURSOR))
|
||||
val error = httpError()
|
||||
coEvery {
|
||||
onrampApi.getHistory(payoutAddress = ADDRESS, afterCursor = AFTER_CURSOR, limit = any())
|
||||
onrampApi.getHistory(userWalletId = USER_WALLET_ID_VALUE, payoutAddress = ADDRESS, afterCursor = AFTER_CURSOR, limit = any())
|
||||
} returns ApiResponse.Error(error).cast()
|
||||
|
||||
// WHEN
|
||||
val thrown = runCatching { repository.fetchOnrampHistory(payoutAddress = ADDRESS) }.exceptionOrNull()
|
||||
val thrown = runCatching { repository.fetchOnrampHistory(payoutAddress = ADDRESS, userWalletId = USER_WALLET_ID) }.exceptionOrNull()
|
||||
|
||||
// THEN
|
||||
assertThat(thrown).isEqualTo(error)
|
||||
|
|
@ -254,13 +255,13 @@ internal class ExpressHistoryRepositoryTest {
|
|||
val response = ExchangeHistoryResponse(items = items, pagination = pagination())
|
||||
stubSyncState(ExpressSyncStateEntity.Type.EXCHANGE, ADDRESS, syncState(afterCursor = AFTER_CURSOR))
|
||||
coEvery {
|
||||
exchangeApi.getHistory(fromAddress = ADDRESS, cursor = AFTER_CURSOR, limit = any())
|
||||
exchangeApi.getHistory(userWalletId = USER_WALLET_ID_VALUE, fromAddress = ADDRESS, cursor = AFTER_CURSOR, limit = any())
|
||||
} returns ApiResponse.Success(response)
|
||||
val saved = slot<List<ExpressExchangeEntity>>()
|
||||
coEvery { expressHistoryDao.upsertExchanges(capture(saved)) } returns Unit
|
||||
|
||||
// WHEN
|
||||
repository.fetchExchangeHistory(fromAddress = ADDRESS)
|
||||
repository.fetchExchangeHistory(fromAddress = ADDRESS, userWalletId = USER_WALLET_ID)
|
||||
|
||||
// THEN
|
||||
assertThat(saved.captured).isEqualTo(items.map { it.toEntity(ADDRESS) })
|
||||
|
|
@ -360,6 +361,8 @@ internal class ExpressHistoryRepositoryTest {
|
|||
|
||||
private companion object {
|
||||
const val ADDRESS = "0xowner"
|
||||
val USER_WALLET_ID = UserWalletId("0123456789abcdef")
|
||||
val USER_WALLET_ID_VALUE = USER_WALLET_ID.stringValue
|
||||
const val AFTER_CURSOR = "after-cursor"
|
||||
const val DELTA_CURSOR = "delta-cursor"
|
||||
const val DEFAULT_LIMIT = 100
|
||||
|
|
|
|||
|
|
@ -45,15 +45,21 @@ class GetPaymentAccountCryptoCurrencyStatusUseCase(
|
|||
userWalletId: UserWalletId,
|
||||
cryptoCurrency: CryptoCurrency,
|
||||
): Option<Pair<Account.Payment, CryptoCurrencyStatus>> {
|
||||
val accountStatus = paymentAccountStatusSupplier.invoke(userWalletId).firstOrNull() ?: return none()
|
||||
val cryptoCurrencyStatus = when (val statusValue = accountStatus.value) {
|
||||
is PaymentAccountStatusValue.Loaded -> statusValue.cryptoCurrencyStatus
|
||||
else -> return none()
|
||||
}
|
||||
val (accountStatus, cryptoCurrencyStatus) = invokeSync(userWalletId)
|
||||
.getOrNull() ?: return none()
|
||||
return if (cryptoCurrencyStatus.currency == cryptoCurrency) {
|
||||
(accountStatus.account to cryptoCurrencyStatus).some()
|
||||
} else {
|
||||
none()
|
||||
}
|
||||
}
|
||||
|
||||
suspend fun invokeSync(userWalletId: UserWalletId): Option<Pair<AccountStatus.Payment, CryptoCurrencyStatus>> {
|
||||
val accountStatus = paymentAccountStatusSupplier.invoke(userWalletId).firstOrNull() ?: return none()
|
||||
val cryptoCurrencyStatus = when (val statusValue = accountStatus.value) {
|
||||
is PaymentAccountStatusValue.Loaded -> statusValue.cryptoCurrencyStatus
|
||||
else -> return none()
|
||||
}
|
||||
return (accountStatus to cryptoCurrencyStatus).some()
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue