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