Chore/optimize market accountant

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<OrderModel, Long> {
         updateDate: LocalDateTime = LocalDateTime.now()
     ): Mono<Void>
 
+    @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<String>
+
     @Query(
         """
         select * from orders

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
         }

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 &&

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