diff --git a/accountant/accountant-app/src/main/kotlin/co/nilin/opex/accountant/app/scheduler/FinancialActionsArchiveJob.kt b/accountant/accountant-app/src/main/kotlin/co/nilin/opex/accountant/app/scheduler/FinancialActionsArchiveJob.kt new file mode 100644 index 000000000..f0d26804b --- /dev/null +++ b/accountant/accountant-app/src/main/kotlin/co/nilin/opex/accountant/app/scheduler/FinancialActionsArchiveJob.kt @@ -0,0 +1,40 @@ +package co.nilin.opex.accountant.app.scheduler + +import co.nilin.opex.accountant.core.spi.FinancialActionPersister +import kotlinx.coroutines.runBlocking +import org.slf4j.LoggerFactory +import org.springframework.beans.factory.annotation.Value +import org.springframework.context.annotation.Profile +import org.springframework.scheduling.annotation.Scheduled +import org.springframework.stereotype.Service +import java.time.LocalDateTime + +@Service +@Profile("scheduled") +class FinancialActionsArchiveJob( + private val financialActionPersister: FinancialActionPersister +) { + private val log = LoggerFactory.getLogger(FinancialActionsArchiveJob::class.java) + + @Value("\${app.fi-action.archive.enabled:true}") + private var enabled: Boolean = true + + @Value("\${app.fi-action.archive.retention-days:30}") + private var retentionDays: Long = 30 + + @Value("\${app.fi-action.archive.batch-size:1000}") + private var batchSize: Int = 1000 + + @Scheduled(fixedDelayString = "\${app.fi-action.archive.fixed-delay-ms:300000}", initialDelay = 60000) + fun archiveProcessedActions() { + if (!enabled || batchSize <= 0 || retentionDays <= 0) return + + runBlocking { + val before = LocalDateTime.now().minusDays(retentionDays) + val archived = financialActionPersister.archiveProcessedActions(before, batchSize) + if (archived > 0) { + log.info("Archived $archived processed financial actions older than $before") + } + } + } +} diff --git a/accountant/accountant-app/src/main/resources/application.yml b/accountant/accountant-app/src/main/resources/application.yml index f191ca34c..3c8b8c1d4 100644 --- a/accountant/accountant-app/src/main/resources/application.yml +++ b/accountant/accountant-app/src/main/resources/application.yml @@ -101,11 +101,21 @@ app: address: 1 wallet: url: lb://opex-wallet/ + http: + retry: + count: 2 + delay-millis: 250 + timeout-seconds: 10 fi-action: retry: count: 10 delay-seconds: 5 delay-multiplier: 3 + archive: + enabled: true + retention-days: 30 + batch-size: 1000 + fixed-delay-ms: 300000 zone-offset: +03:30 trade-volume-calculation-currency: ${TRADE_VOLUME_CALCULATION_CURRENCY:USDT} withdraw-volume-calculation-currency: ${WITHDRAW_VOLUME_CALCULATION_CURRENCY:USDT} diff --git a/accountant/accountant-app/src/test/kotlin/co/nilin/opex/accountant/app/scheduler/FinancialActionJobManagerIT.kt b/accountant/accountant-app/src/test/kotlin/co/nilin/opex/accountant/app/scheduler/FinancialActionJobManagerIT.kt index c29f3c13b..2081965d6 100644 --- a/accountant/accountant-app/src/test/kotlin/co/nilin/opex/accountant/app/scheduler/FinancialActionJobManagerIT.kt +++ b/accountant/accountant-app/src/test/kotlin/co/nilin/opex/accountant/app/scheduler/FinancialActionJobManagerIT.kt @@ -14,7 +14,6 @@ import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.Test import org.mockito.InOrder import org.mockito.Mockito -import org.mockito.Mockito.any import org.mockito.Mockito.`when` import org.mockito.kotlin.eq import org.springframework.beans.factory.annotation.Autowired @@ -248,7 +247,7 @@ class FinancialActionJobManagerIT : KafkaEnabledTest() { eq(fi.receiver), eq(fi.amount), eq(fi.eventType + fi.pointer), - any(), + eq("accountant:fiActions:${fi.uuid}"), eq(fi.category.toString()), ) } diff --git a/accountant/accountant-core/src/main/kotlin/co/nilin/opex/accountant/core/service/FinancialActionJobManagerImpl.kt b/accountant/accountant-core/src/main/kotlin/co/nilin/opex/accountant/core/service/FinancialActionJobManagerImpl.kt index 0dce40930..125ee32d2 100644 --- a/accountant/accountant-core/src/main/kotlin/co/nilin/opex/accountant/core/service/FinancialActionJobManagerImpl.kt +++ b/accountant/accountant-core/src/main/kotlin/co/nilin/opex/accountant/core/service/FinancialActionJobManagerImpl.kt @@ -23,13 +23,6 @@ class FinancialActionJobManagerImpl( .also { if (it.isNotEmpty()) logger.info("Processing ${it.size} financial actions") } .forEach { try { - if (it.parent != null) { - val reloadParent = financialActionLoader.loadFinancialAction(it.parent.id)!! - if (reloadParent.status != FinancialActionStatus.PROCESSED) { - logger.warn("Financial job (uuid=${it.uuid}) skipped because of parent status: uuid=${reloadParent.uuid}, status=${reloadParent.status}") - return@forEach - } - } walletProxy.transfer( it.symbol, it.senderWalletType, @@ -70,7 +63,7 @@ class FinancialActionJobManagerImpl( it.receiver, it.amount, it.eventType + it.pointer, - "accountant:fiActions:${it.id.toString()}", + "accountant:fiActions:${it.uuid}", it.category.toString() ) with(financialActionPersister) { diff --git a/accountant/accountant-core/src/main/kotlin/co/nilin/opex/accountant/core/spi/FinancialActionPersister.kt b/accountant/accountant-core/src/main/kotlin/co/nilin/opex/accountant/core/spi/FinancialActionPersister.kt index d8747ced3..a4e4afa2e 100644 --- a/accountant/accountant-core/src/main/kotlin/co/nilin/opex/accountant/core/spi/FinancialActionPersister.kt +++ b/accountant/accountant-core/src/main/kotlin/co/nilin/opex/accountant/core/spi/FinancialActionPersister.kt @@ -2,6 +2,7 @@ package co.nilin.opex.accountant.core.spi import co.nilin.opex.accountant.core.model.FinancialAction import co.nilin.opex.accountant.core.model.FinancialActionStatus +import java.time.LocalDateTime interface FinancialActionPersister { @@ -20,4 +21,6 @@ interface FinancialActionPersister { suspend fun updateStatusNewTx(financialAction: FinancialAction, status: FinancialActionStatus) suspend fun retrySuccessful(financialAction: FinancialAction) + + suspend fun archiveProcessedActions(before: LocalDateTime, limit: Int): Int } \ No newline at end of file diff --git a/accountant/accountant-ports/accountant-persister-postgres/src/main/kotlin/co/nilin/opex/accountant/ports/postgres/dao/FinancialActionRepository.kt b/accountant/accountant-ports/accountant-persister-postgres/src/main/kotlin/co/nilin/opex/accountant/ports/postgres/dao/FinancialActionRepository.kt index 3139f647d..16dbeddf4 100644 --- a/accountant/accountant-ports/accountant-persister-postgres/src/main/kotlin/co/nilin/opex/accountant/ports/postgres/dao/FinancialActionRepository.kt +++ b/accountant/accountant-ports/accountant-persister-postgres/src/main/kotlin/co/nilin/opex/accountant/ports/postgres/dao/FinancialActionRepository.kt @@ -10,6 +10,7 @@ import org.springframework.data.repository.reactive.ReactiveCrudRepository import org.springframework.stereotype.Repository import reactor.core.publisher.Mono import java.math.BigDecimal +import java.time.LocalDateTime @Repository interface FinancialActionRepository : ReactiveCrudRepository { @@ -45,8 +46,83 @@ interface FinancialActionRepository : ReactiveCrudRepository + + @Query( + """ + with candidates as ( + select id + from fi_actions + where status = 'PROCESSED' + and create_date < :before + and not exists ( + select 1 from fi_action_retry far + where far.fa_id = fi_actions.id and far.is_resolved = false + ) + order by create_date + limit :limit + ), + moved_actions as ( + insert into fi_actions_archive ( + id, uuid, parent_id, event_type, pointer, symbol, amount, sender, sender_wallet_type, + receiver, receiver_wallet_type, agent, ip, create_date, status, category_name + ) + select fa.id, fa.uuid, fa.parent_id, fa.event_type, fa.pointer, fa.symbol, fa.amount, fa.sender, fa.sender_wallet_type, + fa.receiver, fa.receiver_wallet_type, fa.agent, fa.ip, fa.create_date, fa.status, fa.category_name + from fi_actions fa + join candidates c on c.id = fa.id + on conflict (id) do nothing + returning id + ), + moved_retries as ( + insert into fi_action_retry_archive (id, fa_id, retries, next_run_time, is_resolved, has_given_up) + select far.id, far.fa_id, far.retries, far.next_run_time, far.is_resolved, far.has_given_up + from fi_action_retry far + join moved_actions ma on ma.id = far.fa_id + on conflict (id) do nothing + returning id + ), + moved_errors as ( + insert into fi_action_error_archive (id, fa_id, error, message, body, retry_id, date) + select fae.id, fae.fa_id, fae.error, fae.message, fae.body, fae.retry_id, fae.date + from fi_action_error fae + join moved_actions ma on ma.id = fae.fa_id + on conflict (id) do nothing + returning id + ), + deleted_errors as ( + delete from fi_action_error fae + using moved_actions ma + where fae.fa_id = ma.id + returning fae.id + ), + deleted_retries as ( + delete from fi_action_retry far + using moved_actions ma + where far.fa_id = ma.id + returning far.id + ), + deleted_actions as ( + delete from fi_actions fa + using moved_actions ma + where fa.id = ma.id + returning fa.id + ) + select count(1) from deleted_actions + """ + ) + fun archiveProcessedActions( + @Param("before") before: LocalDateTime, + @Param("limit") limit: Int + ): Mono } \ No newline at end of file diff --git a/accountant/accountant-ports/accountant-persister-postgres/src/main/kotlin/co/nilin/opex/accountant/ports/postgres/impl/FinancialActionLoaderImpl.kt b/accountant/accountant-ports/accountant-persister-postgres/src/main/kotlin/co/nilin/opex/accountant/ports/postgres/impl/FinancialActionLoaderImpl.kt index 32bf2cfd5..1fb65e088 100644 --- a/accountant/accountant-ports/accountant-persister-postgres/src/main/kotlin/co/nilin/opex/accountant/ports/postgres/impl/FinancialActionLoaderImpl.kt +++ b/accountant/accountant-ports/accountant-persister-postgres/src/main/kotlin/co/nilin/opex/accountant/ports/postgres/impl/FinancialActionLoaderImpl.kt @@ -6,6 +6,7 @@ import co.nilin.opex.accountant.core.spi.FinancialActionLoader import co.nilin.opex.accountant.ports.postgres.dao.FinancialActionErrorRepository import co.nilin.opex.accountant.ports.postgres.dao.FinancialActionRepository import co.nilin.opex.accountant.ports.postgres.dao.FinancialActionRetryRepository +import co.nilin.opex.accountant.ports.postgres.model.FinancialActionModel import kotlinx.coroutines.flow.firstOrNull import kotlinx.coroutines.flow.map import kotlinx.coroutines.flow.toList @@ -35,7 +36,7 @@ class FinancialActionLoaderImpl( override suspend fun loadReadyToProcess(offset: Long, size: Long): List { return financialActionRepository.findReadyToProcess( PageRequest.of(offset.toInt(), size.toInt(), Sort.by(Sort.Direction.ASC, "createDate")) - ).map { loadFinancialAction(it.id)!! } + ).map { mapToFinancialAction(it) } .toList() } @@ -80,25 +81,27 @@ class FinancialActionLoaderImpl( override suspend fun loadRetries(limit: Int): List { return faRetryRepository.findAllRetries(LocalDateTime.now(), limit) - .map { - FinancialAction( - null, // Skipping parent. If it's in retry, it means its parent is already processed - it.eventType, - it.pointer, - it.symbol, - it.amount, - it.sender, - it.senderWalletType, - it.receiver, - it.receiverWalletType, - it.createDate, - it.categoryName, - it.status, - it.uuid, - it.id - ) - } + .map { mapToFinancialAction(it) } .collectList() .awaitFirstOrElse { emptyList() } } + + private fun mapToFinancialAction(financialAction: FinancialActionModel): FinancialAction { + return FinancialAction( + null, + financialAction.eventType, + financialAction.pointer, + financialAction.symbol, + financialAction.amount, + financialAction.sender, + financialAction.senderWalletType, + financialAction.receiver, + financialAction.receiverWalletType, + financialAction.createDate, + financialAction.categoryName, + financialAction.status, + financialAction.uuid, + financialAction.id + ) + } } \ No newline at end of file diff --git a/accountant/accountant-ports/accountant-persister-postgres/src/main/kotlin/co/nilin/opex/accountant/ports/postgres/impl/FinancialActionPersisterImpl.kt b/accountant/accountant-ports/accountant-persister-postgres/src/main/kotlin/co/nilin/opex/accountant/ports/postgres/impl/FinancialActionPersisterImpl.kt index 86d9dcd89..12292c2e6 100644 --- a/accountant/accountant-ports/accountant-persister-postgres/src/main/kotlin/co/nilin/opex/accountant/ports/postgres/impl/FinancialActionPersisterImpl.kt +++ b/accountant/accountant-ports/accountant-persister-postgres/src/main/kotlin/co/nilin/opex/accountant/ports/postgres/impl/FinancialActionPersisterImpl.kt @@ -131,6 +131,10 @@ class FinancialActionPersisterImpl( faRetryRepository.updateResolvedTrue(financialAction.id!!).awaitSingleOrNull() } + override suspend fun archiveProcessedActions(before: LocalDateTime, limit: Int): Int { + return (repository.archiveProcessedActions(before, limit).awaitSingleOrNull() ?: 0).toInt() + } + override suspend fun updateStatus(faUuid: String, status: FinancialActionStatus) { repository.updateStatus(faUuid, status).awaitSingleOrNull() } diff --git a/accountant/accountant-ports/accountant-persister-postgres/src/main/resources/schema.sql b/accountant/accountant-ports/accountant-persister-postgres/src/main/resources/schema.sql index c4d6999c6..8b7b1b772 100644 --- a/accountant/accountant-ports/accountant-persister-postgres/src/main/resources/schema.sql +++ b/accountant/accountant-ports/accountant-persister-postgres/src/main/resources/schema.sql @@ -49,6 +49,8 @@ CREATE INDEX IF NOT EXISTS idx_fi_actions_symbol ON fi_actions (symbol); CREATE INDEX IF NOT EXISTS idx_fi_event_type ON fi_actions (event_type); CREATE INDEX IF NOT EXISTS idx_fi_actions_status ON fi_actions (status); CREATE INDEX IF NOT EXISTS idx_fi_actions_pointer ON fi_actions (pointer); +CREATE INDEX IF NOT EXISTS idx_fi_actions_status_create_date ON fi_actions (status, create_date); +CREATE INDEX IF NOT EXISTS idx_fi_actions_parent_status ON fi_actions (parent_id, status); ALTER TABLE fi_actions ADD COLUMN IF NOT EXISTS category_name VARCHAR(36); @@ -63,6 +65,11 @@ CREATE TABLE IF NOT EXISTS fi_action_retry has_given_up BOOLEAN NOT NULL DEFAULT false ); +CREATE INDEX IF NOT EXISTS idx_fi_action_retry_due + ON fi_action_retry (next_run_time) + WHERE has_given_up = false + AND is_resolved = false; + CREATE TABLE IF NOT EXISTS fi_action_error ( id SERIAL PRIMARY KEY, @@ -74,6 +81,58 @@ CREATE TABLE IF NOT EXISTS fi_action_error date TIMESTAMP NOT NULL DEFAULT CURRENT_DATE ); +CREATE INDEX IF NOT EXISTS idx_fi_action_error_fa_id_date ON fi_action_error (fa_id, date); + +CREATE TABLE IF NOT EXISTS fi_actions_archive +( + id INTEGER PRIMARY KEY, + uuid VARCHAR(72) NOT NULL UNIQUE, + parent_id INTEGER, + event_type VARCHAR(72) NOT NULL, + pointer VARCHAR(72) NOT NULL, + symbol VARCHAR(36) NOT NULL, + amount DECIMAL NOT NULL, + sender VARCHAR(36) NOT NULL, + sender_wallet_type VARCHAR(36) NOT NULL, + receiver VARCHAR(36) NOT NULL, + receiver_wallet_type VARCHAR(36) NOT NULL, + agent VARCHAR(20), + ip VARCHAR(11), + create_date TIMESTAMP NOT NULL, + status VARCHAR(20), + category_name VARCHAR(36), + archived_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP +); + +CREATE INDEX IF NOT EXISTS idx_fi_actions_archive_create_date ON fi_actions_archive (create_date); + +CREATE TABLE IF NOT EXISTS fi_action_retry_archive +( + id INTEGER PRIMARY KEY, + fa_id INTEGER NOT NULL UNIQUE, + retries INTEGER NOT NULL DEFAULT 0, + next_run_time TIMESTAMP NOT NULL, + is_resolved BOOLEAN NOT NULL DEFAULT false, + has_given_up BOOLEAN NOT NULL DEFAULT false, + archived_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP +); + +CREATE INDEX IF NOT EXISTS idx_fi_action_retry_archive_fa_id ON fi_action_retry_archive (fa_id); + +CREATE TABLE IF NOT EXISTS fi_action_error_archive +( + id INTEGER PRIMARY KEY, + fa_id INTEGER NOT NULL, + error TEXT NOT NULL, + message TEXT NOT NULL, + body TEXT, + retry_id INTEGER, + date TIMESTAMP NOT NULL, + archived_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP +); + +CREATE INDEX IF NOT EXISTS idx_fi_action_error_archive_fa_id ON fi_action_error_archive (fa_id); + CREATE TABLE IF NOT EXISTS pair_config ( pair VARCHAR(72) PRIMARY KEY, diff --git a/accountant/accountant-ports/accountant-wallet-proxy/src/main/kotlin/co/nilin/opex/accountant/ports/walletproxy/proxy/WalletProxyImpl.kt b/accountant/accountant-ports/accountant-wallet-proxy/src/main/kotlin/co/nilin/opex/accountant/ports/walletproxy/proxy/WalletProxyImpl.kt index ff7078afd..c3701d421 100644 --- a/accountant/accountant-ports/accountant-wallet-proxy/src/main/kotlin/co/nilin/opex/accountant/ports/walletproxy/proxy/WalletProxyImpl.kt +++ b/accountant/accountant-ports/accountant-wallet-proxy/src/main/kotlin/co/nilin/opex/accountant/ports/walletproxy/proxy/WalletProxyImpl.kt @@ -11,15 +11,24 @@ import kotlinx.coroutines.reactive.awaitFirstOrNull import org.springframework.beans.factory.annotation.Value import org.springframework.http.MediaType import org.springframework.stereotype.Component +import org.springframework.web.reactive.function.client.WebClientRequestException import org.springframework.web.reactive.function.client.WebClient import org.springframework.web.reactive.function.client.bodyToMono +import reactor.core.publisher.Mono +import reactor.util.retry.Retry +import java.io.IOException +import java.net.ConnectException import java.math.BigDecimal +import java.time.Duration +import java.util.concurrent.TimeoutException @Component class WalletProxyImpl( private val webClient: WebClient, - @Value("\${app.wallet.url}") - private val walletBaseUrl: String + @Value("\${app.wallet.url}") private val walletBaseUrl: String, + @Value("\${app.wallet.http.retry.count:2}") private val retryCount: Long = 2, + @Value("\${app.wallet.http.retry.delay-millis:250}") private val retryDelayMillis: Long = 250, + @Value("\${app.wallet.http.timeout-seconds:10}") private val timeoutSeconds: Long = 10 ) : WalletProxy { data class TransferBody( @@ -39,14 +48,15 @@ class WalletProxyImpl( transferRef: String?, transferCategory: String ) { - webClient.post() - .uri("$walletBaseUrl/v2/transfer/${amount}_$symbol/from/${senderUuid}_$senderWalletType/to/${receiverUuid}_$receiverWalletType") - .contentType(MediaType.APPLICATION_JSON) - .bodyValue(TransferBody(description, transferRef, transferCategory)) - .retrieve() - .onStatus({ t -> t.isError }, { it.createException() }) - .bodyToMono() - .awaitFirst() + withTransientRetry { + webClient.post() + .uri("$walletBaseUrl/v2/transfer/${amount}_$symbol/from/${senderUuid}_$senderWalletType/to/${receiverUuid}_$receiverWalletType") + .contentType(MediaType.APPLICATION_JSON) + .bodyValue(TransferBody(description, transferRef, transferCategory)) + .retrieve() + .onStatus({ t -> t.isError }, { it.createException() }) + .bodyToMono() + }.awaitFirst() } override suspend fun canFulfil(symbol: String, walletType: WalletType, uuid: String, amount: BigDecimal): Boolean { @@ -56,6 +66,7 @@ class WalletProxyImpl( .retrieve() .onStatus({ t -> t.isError }, { it.createException() }) .bodyToMono() + .timeout(Duration.ofSeconds(timeoutSeconds)) .awaitFirst() .result } @@ -69,6 +80,7 @@ class WalletProxyImpl( .retrieve() .onStatus({ t -> t.isError }, { it.createException() }) .bodyToMono() + .timeout(Duration.ofSeconds(timeoutSeconds)) .awaitFirstOrNull() } @@ -79,6 +91,27 @@ class WalletProxyImpl( .retrieve() .onStatus({ t -> t.isError }, { it.createException() }) .bodyToMono>() + .timeout(Duration.ofSeconds(timeoutSeconds)) .awaitFirst() } + + private fun withTransientRetry(request: () -> Mono): Mono { + return request() + .timeout(Duration.ofSeconds(timeoutSeconds)) + .retryWhen( + Retry.backoff(retryCount, Duration.ofMillis(retryDelayMillis)) + .filter { error -> + when { + error is WebClientRequestException -> true + error is TimeoutException -> true + error is ConnectException -> true + error is IOException -> true + error.cause is TimeoutException -> true + error.cause is ConnectException -> true + else -> false + } + } + .onRetryExhaustedThrow { _, signal -> signal.failure() } + ) + } } \ No newline at end of file diff --git a/accountant/accountant-ports/accountant-wallet-proxy/src/test/kotlin/co/nilin/opex/accountant/ports/walletproxy/proxy/WalletProxyImplTest.kt b/accountant/accountant-ports/accountant-wallet-proxy/src/test/kotlin/co/nilin/opex/accountant/ports/walletproxy/proxy/WalletProxyImplTest.kt index 107c2b794..0fe54ce0d 100644 --- a/accountant/accountant-ports/accountant-wallet-proxy/src/test/kotlin/co/nilin/opex/accountant/ports/walletproxy/proxy/WalletProxyImplTest.kt +++ b/accountant/accountant-ports/accountant-wallet-proxy/src/test/kotlin/co/nilin/opex/accountant/ports/walletproxy/proxy/WalletProxyImplTest.kt @@ -22,7 +22,10 @@ class WalletProxyImplTest { private lateinit var mockServer: MockServerClient private val walletProxyImpl = WalletProxyImpl( WebClient.builder().build(), - "http://localhost:8089" + "http://localhost:8089", + 1, + 10, + 5 ) private val objectMapper = ObjectMapper() diff --git a/market/market-ports/market-persister-postgres/src/main/kotlin/co/nilin/opex/market/ports/postgres/dao/OrderRepository.kt b/market/market-ports/market-persister-postgres/src/main/kotlin/co/nilin/opex/market/ports/postgres/dao/OrderRepository.kt index cc394b310..0b54cbb32 100644 --- a/market/market-ports/market-persister-postgres/src/main/kotlin/co/nilin/opex/market/ports/postgres/dao/OrderRepository.kt +++ b/market/market-ports/market-persister-postgres/src/main/kotlin/co/nilin/opex/market/ports/postgres/dao/OrderRepository.kt @@ -13,6 +13,7 @@ import org.springframework.stereotype.Repository import reactor.core.publisher.Flux import reactor.core.publisher.Mono +import java.math.BigDecimal import java.time.LocalDateTime import java.util.* @@ -49,6 +50,62 @@ interface OrderRepository : ReactiveCrudRepository { updateDate: LocalDateTime = LocalDateTime.now() ): Mono + @Query( + """ + insert into orders ( + ouid, uuid, client_order_id, symbol, order_id, + maker_fee, taker_fee, left_side_fraction, right_side_fraction, + user_level, side, match_constraint, order_type, + price, quantity, quote_quantity, create_date, update_date + ) values ( + :ouid, :uuid, :clientOrderId, :symbol, :orderId, + :makerFee, :takerFee, :leftSideFraction, :rightSideFraction, + :userLevel, :side, :matchConstraint, :orderType, + :price, :quantity, :quoteQuantity, :createDate, :updateDate + ) + on conflict (ouid) do nothing + returning ouid + """ + ) + fun insertOrderIfAbsent( + @Param("ouid") + ouid: String, + @Param("uuid") + uuid: String, + @Param("clientOrderId") + clientOrderId: String?, + @Param("symbol") + symbol: String, + @Param("orderId") + orderId: Long?, + @Param("makerFee") + makerFee: BigDecimal?, + @Param("takerFee") + takerFee: BigDecimal?, + @Param("leftSideFraction") + leftSideFraction: BigDecimal?, + @Param("rightSideFraction") + rightSideFraction: BigDecimal?, + @Param("userLevel") + userLevel: String?, + @Param("side") + side: String?, + @Param("matchConstraint") + matchConstraint: String?, + @Param("orderType") + orderType: String?, + @Param("price") + price: BigDecimal?, + @Param("quantity") + quantity: BigDecimal?, + @Param("quoteQuantity") + quoteQuantity: BigDecimal?, + @Param("createDate") + createDate: LocalDateTime?, + @Param("updateDate") + updateDate: LocalDateTime + ): Mono + @Query( """ select * from orders diff --git a/market/market-ports/market-persister-postgres/src/main/kotlin/co/nilin/opex/market/ports/postgres/impl/OrderPersisterImpl.kt b/market/market-ports/market-persister-postgres/src/main/kotlin/co/nilin/opex/market/ports/postgres/impl/OrderPersisterImpl.kt index 87a7d7a3d..a4fd7dda7 100644 --- a/market/market-ports/market-persister-postgres/src/main/kotlin/co/nilin/opex/market/ports/postgres/impl/OrderPersisterImpl.kt +++ b/market/market-ports/market-persister-postgres/src/main/kotlin/co/nilin/opex/market/ports/postgres/impl/OrderPersisterImpl.kt @@ -17,8 +17,6 @@ import kotlinx.coroutines.reactive.awaitFirstOrNull import kotlinx.coroutines.reactor.awaitSingle import kotlinx.coroutines.reactor.awaitSingleOrNull import org.slf4j.LoggerFactory -import org.springframework.dao.DataIntegrityViolationException -import org.springframework.dao.DuplicateKeyException import org.springframework.stereotype.Component import org.springframework.transaction.annotation.Transactional import java.time.LocalDateTime @@ -57,12 +55,27 @@ class OrderPersisterImpl( LocalDateTime.now(), LocalDateTime.now() ) - try { - orderRepository.save(orderModel).awaitFirstOrNull() - } catch (e: DuplicateKeyException) { - logger.info("order ${order.ouid} is duplicate; skipping create flow") - return - } catch (e: DataIntegrityViolationException) { + val inserted = orderRepository.insertOrderIfAbsent( + ouid = orderModel.ouid, + uuid = orderModel.uuid, + clientOrderId = orderModel.clientOrderId, + symbol = orderModel.symbol, + orderId = orderModel.orderId, + makerFee = orderModel.makerFee, + takerFee = orderModel.takerFee, + leftSideFraction = orderModel.leftSideFraction, + rightSideFraction = orderModel.rightSideFraction, + userLevel = orderModel.userLevel, + side = orderModel.direction?.name, + matchConstraint = orderModel.constraint?.name, + orderType = orderModel.type?.name, + price = orderModel.price, + quantity = orderModel.quantity, + quoteQuantity = orderModel.quoteQuantity, + createDate = orderModel.createDate, + updateDate = orderModel.updateDate + ).awaitFirstOrNull() != null + if (!inserted) { logger.info("order ${order.ouid} is duplicate; skipping create flow") return } diff --git a/market/market-ports/market-persister-postgres/src/main/kotlin/co/nilin/opex/market/ports/postgres/impl/TradePersisterImpl.kt b/market/market-ports/market-persister-postgres/src/main/kotlin/co/nilin/opex/market/ports/postgres/impl/TradePersisterImpl.kt index a257a1d99..d3daa09f7 100644 --- a/market/market-ports/market-persister-postgres/src/main/kotlin/co/nilin/opex/market/ports/postgres/impl/TradePersisterImpl.kt +++ b/market/market-ports/market-persister-postgres/src/main/kotlin/co/nilin/opex/market/ports/postgres/impl/TradePersisterImpl.kt @@ -14,8 +14,10 @@ import org.springframework.dao.DuplicateKeyException import org.springframework.stereotype.Component import java.time.LocalDateTime import java.time.ZoneId +import java.time.temporal.ChronoUnit import java.util.* import java.util.concurrent.atomic.AtomicLong +import kotlin.math.abs @Component class TradePersisterImpl( @@ -144,11 +146,12 @@ class TradePersisterImpl( } private fun isSameTradePayload(existing: TradeModel, incoming: TradeModel): Boolean { + val tradeDateDeltaSeconds = abs(ChronoUnit.SECONDS.between(existing.tradeDate, incoming.tradeDate)) return existing.makerOuid == incoming.makerOuid && existing.takerOuid == incoming.takerOuid && existing.matchedPrice.compareTo(incoming.matchedPrice) == 0 && existing.matchedQuantity.compareTo(incoming.matchedQuantity) == 0 && - existing.tradeDate == incoming.tradeDate && + tradeDateDeltaSeconds <= 5 && existing.makerCommission == incoming.makerCommission && existing.takerCommission == incoming.takerCommission && existing.makerCommissionAsset == incoming.makerCommissionAsset && diff --git a/market/market-ports/market-persister-postgres/src/test/kotlin/co/nilin/opex/market/ports/postgres/impl/OrderPersisterTest.kt b/market/market-ports/market-persister-postgres/src/test/kotlin/co/nilin/opex/market/ports/postgres/impl/OrderPersisterTest.kt index ba1077369..9ab3a2a9f 100644 --- a/market/market-ports/market-persister-postgres/src/test/kotlin/co/nilin/opex/market/ports/postgres/impl/OrderPersisterTest.kt +++ b/market/market-ports/market-persister-postgres/src/test/kotlin/co/nilin/opex/market/ports/postgres/impl/OrderPersisterTest.kt @@ -35,11 +35,31 @@ class OrderPersisterTest { @Test fun givenOrderRepo_whenSaveRichOrder_thenSuccess(): Unit = runBlocking { every { - orderRepository.save(any()) - } returns Mono.just(VALID.MAKER_ORDER_MODEL) + orderRepository.insertOrderIfAbsent( + ouid = any(), + uuid = any(), + clientOrderId = any(), + symbol = any(), + orderId = any(), + makerFee = any(), + takerFee = any(), + leftSideFraction = any(), + rightSideFraction = any(), + userLevel = any(), + side = any(), + matchConstraint = any(), + orderType = any(), + price = any(), + quantity = any(), + quoteQuantity = any(), + createDate = any(), + updateDate = any() + ) + } returns Mono.just(VALID.RICH_ORDER.ouid) every { orderStatusRepository.insert(any(), any(), any(), any(), any(), any()) } returns Mono.empty() + every { orderStatusRepository.findMostRecentByOUID(any()) } returns Mono.just(VALID.MAKER_ORDER_STATUS_MODEL) @@ -86,10 +106,27 @@ class OrderPersisterTest { @Test fun givenDuplicateOrderCreate_whenSaveRichOrder_thenIgnoredAsIdempotent(): Unit = runBlocking { every { - orderRepository.save(any()) - } returns Mono.error(DuplicateKeyException("duplicate order")) - - assertThatNoException().isThrownBy { runBlocking { orderPersister.save(VALID.RICH_ORDER) } } + orderRepository.insertOrderIfAbsent( + ouid = any(), + uuid = any(), + clientOrderId = any(), + symbol = any(), + orderId = any(), + makerFee = any(), + takerFee = any(), + leftSideFraction = any(), + rightSideFraction = any(), + userLevel = any(), + side = any(), + matchConstraint = any(), + orderType = any(), + price = any(), + quantity = any(), + quoteQuantity = any(), + createDate = any(), + updateDate = any() + ) + } returns Mono.empty() verify(exactly = 0) { orderStatusRepository.insert(any(), any(), any(), any(), any(), any())