diff --git a/accountant/accountant-core/src/main/kotlin/co/nilin/opex/accountant/core/inout/RichOrder.kt b/accountant/accountant-core/src/main/kotlin/co/nilin/opex/accountant/core/inout/RichOrder.kt index 85f13c547..a09ebced1 100644 --- a/accountant/accountant-core/src/main/kotlin/co/nilin/opex/accountant/core/inout/RichOrder.kt +++ b/accountant/accountant-core/src/main/kotlin/co/nilin/opex/accountant/core/inout/RichOrder.kt @@ -4,6 +4,7 @@ import co.nilin.opex.matching.engine.core.model.MatchConstraint import co.nilin.opex.matching.engine.core.model.OrderDirection import co.nilin.opex.matching.engine.core.model.OrderType import java.math.BigDecimal +import java.time.LocalDateTime data class RichOrder( val orderId: Long? = 0, @@ -23,5 +24,6 @@ data class RichOrder( val quoteQuantity: BigDecimal, val executedQuantity: BigDecimal, val accumulativeQuoteQty: BigDecimal, - val status: Int = 0 + val status: Int = 0, + val createDate: LocalDateTime? ) : RichOrderEvent diff --git a/accountant/accountant-core/src/main/kotlin/co/nilin/opex/accountant/core/inout/RichOrderUpdate.kt b/accountant/accountant-core/src/main/kotlin/co/nilin/opex/accountant/core/inout/RichOrderUpdate.kt index 81f515e26..78588f0d1 100644 --- a/accountant/accountant-core/src/main/kotlin/co/nilin/opex/accountant/core/inout/RichOrderUpdate.kt +++ b/accountant/accountant-core/src/main/kotlin/co/nilin/opex/accountant/core/inout/RichOrderUpdate.kt @@ -1,13 +1,15 @@ package co.nilin.opex.accountant.core.inout import java.math.BigDecimal +import java.time.LocalDateTime data class RichOrderUpdate( val ouid: String, val price: BigDecimal, val quantity: BigDecimal, val remainedQuantity: BigDecimal, - val status: OrderStatus = OrderStatus.NEW + val status: OrderStatus = OrderStatus.NEW, + val updateDate: LocalDateTime?= LocalDateTime.now() ) : RichOrderEvent { fun executedQuantity(): BigDecimal = quantity.minus(remainedQuantity) diff --git a/accountant/accountant-core/src/main/kotlin/co/nilin/opex/accountant/core/service/OrderManagerImpl.kt b/accountant/accountant-core/src/main/kotlin/co/nilin/opex/accountant/core/service/OrderManagerImpl.kt index 34bf51f4b..942157f83 100644 --- a/accountant/accountant-core/src/main/kotlin/co/nilin/opex/accountant/core/service/OrderManagerImpl.kt +++ b/accountant/accountant-core/src/main/kotlin/co/nilin/opex/accountant/core/service/OrderManagerImpl.kt @@ -212,12 +212,13 @@ open class OrderManagerImpl( richOrderPublisher.publish( RichOrderUpdate( order.ouid, - order.price.toBigDecimal(), - order.quantity.toBigDecimal(), - cancelOrderEvent.remainedQuantity.toBigDecimal(), + order.price.toBigDecimal().multiply(order.rightSideFraction), + order.origQuantity, + cancelOrderEvent.remainedQuantity.toBigDecimal().multiply(order.leftSideFraction), OrderStatus.CANCELED ) ) + return financialActionPersister.persist(listOf(financialAction)) /*publishFinancialAction(financialAction) return fa*/ @@ -253,7 +254,8 @@ open class OrderManagerImpl( OrderStatus.NEW.code } else { OrderStatus.PARTIALLY_FILLED.code - } + }, + LocalDateTime.now() ) ) } diff --git a/accountant/accountant-ports/accountant-eventlistener-kafka/src/main/kotlin/co/nilin/opex/accountant/ports/kafka/listener/config/AccountantKafkaConfig.kt b/accountant/accountant-ports/accountant-eventlistener-kafka/src/main/kotlin/co/nilin/opex/accountant/ports/kafka/listener/config/AccountantKafkaConfig.kt index 6ae6f37dd..48d81d1f0 100644 --- a/accountant/accountant-ports/accountant-eventlistener-kafka/src/main/kotlin/co/nilin/opex/accountant/ports/kafka/listener/config/AccountantKafkaConfig.kt +++ b/accountant/accountant-ports/accountant-eventlistener-kafka/src/main/kotlin/co/nilin/opex/accountant/ports/kafka/listener/config/AccountantKafkaConfig.kt @@ -1,4 +1,4 @@ -package co.nilin.opex.accountant.ports.kafka.listener.config +package co.nilin.opex.accountant.ports.kafka.listener.config import co.nilin.opex.accountant.core.inout.KycLevelUpdatedEvent import co.nilin.opex.accountant.ports.kafka.listener.consumer.* @@ -9,7 +9,6 @@ import co.nilin.opex.matching.engine.core.eventh.events.CoreEvent import org.apache.kafka.clients.consumer.ConsumerConfig import org.apache.kafka.common.TopicPartition import org.apache.kafka.common.serialization.StringDeserializer -import org.springframework.beans.factory.annotation.Autowired import org.springframework.beans.factory.annotation.Qualifier import org.springframework.beans.factory.annotation.Value import org.springframework.boot.autoconfigure.condition.ConditionalOnBean @@ -61,85 +60,83 @@ class AccountantKafkaConfig { fun withdrawRequestConsumerFactory(@Qualifier("consumerConfig") consumerConfigs: Map): ConsumerFactory { return DefaultKafkaConsumerFactory(consumerConfigs) } + @Bean("depositConsumerFactory") fun depositConsumerFactory(@Qualifier("consumerConfig") consumerConfigs: Map): ConsumerFactory { return DefaultKafkaConsumerFactory(consumerConfigs) } - @Autowired + @Bean("tradeKafkaListenerContainer") @ConditionalOnBean(TradeKafkaListener::class) - fun configureTradeListener( + fun tradeListenerContainer( tradeListener: TradeKafkaListener, @Qualifier("accountantEventKafkaTemplate") template: KafkaTemplate, @Qualifier("accountantConsumerFactory") consumerFactory: ConsumerFactory - ) { + ): ConcurrentMessageListenerContainer { val containerProps = ContainerProperties(Pattern.compile("trades_.*")) containerProps.messageListener = tradeListener val container = ConcurrentMessageListenerContainer(consumerFactory, containerProps) container.setBeanName("TradeKafkaListenerContainer") container.commonErrorHandler = createConsumerErrorHandler(template, "trades.DLT") - container.start() + return container } - @Autowired + @Bean("eventKafkaListenerContainer") @ConditionalOnBean(EventKafkaListener::class) - fun configureEventListener( + fun eventListenerContainer( eventListener: EventKafkaListener, @Qualifier("accountantEventKafkaTemplate") template: KafkaTemplate, @Qualifier("accountantConsumerFactory") consumerFactory: ConsumerFactory - ) { + ): ConcurrentMessageListenerContainer { val containerProps = ContainerProperties(Pattern.compile("events_.*")) containerProps.messageListener = eventListener val container = ConcurrentMessageListenerContainer(consumerFactory, containerProps) container.setBeanName("EventKafkaListenerContainer") container.commonErrorHandler = createConsumerErrorHandler(template, "events.DLT") - container.start() + return container } - @Autowired + @Bean("orderKafkaListenerContainer") @ConditionalOnBean(OrderKafkaListener::class) - fun configureOrderListener( + fun orderListenerContainer( orderListener: OrderKafkaListener, @Qualifier("accountantEventKafkaTemplate") template: KafkaTemplate, @Qualifier("accountantConsumerFactory") consumerFactory: ConsumerFactory - ) { + ): ConcurrentMessageListenerContainer { val containerProps = ContainerProperties(Pattern.compile("orders_.*")) containerProps.messageListener = orderListener val container = ConcurrentMessageListenerContainer(consumerFactory, containerProps) container.setBeanName("OrderKafkaListenerContainer") container.commonErrorHandler = createConsumerErrorHandler(template, "orders.DLT") - container.start() + return container } - @Autowired + @Bean("tempEventKafkaListenerContainer") @ConditionalOnBean(TempEventKafkaListener::class) - fun configureTempEventListener( + fun tempEventListenerContainer( eventListener: TempEventKafkaListener, @Qualifier("accountantEventKafkaTemplate") template: KafkaTemplate, @Qualifier("accountantConsumerFactory") consumerFactory: ConsumerFactory - ) { + ): ConcurrentMessageListenerContainer { val containerProps = ContainerProperties(Pattern.compile("tempevents")) containerProps.messageListener = eventListener val container = ConcurrentMessageListenerContainer(consumerFactory, containerProps) container.setBeanName("TempEventKafkaListenerContainer") container.commonErrorHandler = createConsumerErrorHandler(template, "tempevents.DLT") - container.start() + return container } - @Autowired + @Bean("faResponseKafkaListenerContainer") @ConditionalOnBean(FAResponseKafkaListener::class) - fun configureEventListener( + fun faResponseListenerContainer( eventListener: FAResponseKafkaListener, - //@Qualifier("accountantEventKafkaTemplate") template: KafkaTemplate, @Qualifier("faResponseConsumerFactory") consumerFactory: ConsumerFactory - ) { + ): ConcurrentMessageListenerContainer { val containerProps = ContainerProperties(Pattern.compile("fiAction_response")) containerProps.messageListener = eventListener val container = ConcurrentMessageListenerContainer(consumerFactory, containerProps) container.setBeanName("FAResponseKafkaListenerContainer") - //TODO add error handler - //container.commonErrorHandler = createConsumerErrorHandler(template, "events.DLT") - container.start() + return container } @Bean("kycLevelUpdatedProducerFactory") @@ -152,69 +149,69 @@ class AccountantKafkaConfig { return KafkaTemplate(producerFactory) } - @Bean("withdrawRequestProducerFactory") - fun withdrawRequestProducerFactory(@Qualifier("consumerConfig") producerConfigs: Map): ProducerFactory { - return DefaultKafkaProducerFactory(producerConfigs) - } - - @Bean("withdrawRequestKafkaTemplate") - fun withdrawRequestKafkaTemplate(@Qualifier("withdrawRequestProducerFactory") producerFactory: ProducerFactory): KafkaTemplate { - return KafkaTemplate(producerFactory) - } - - @Bean("depositProducerFactory") - fun depositProducerFactory(@Qualifier("consumerConfig") producerConfigs: Map): ProducerFactory { - return DefaultKafkaProducerFactory(producerConfigs) - } - - @Bean("depositKafkaTemplate") - fun depositKafkaTemplate(@Qualifier("depositProducerFactory") producerFactory: ProducerFactory): KafkaTemplate { - return KafkaTemplate(producerFactory) - } - - @Autowired + @Bean("kycLevelUpdatedKafkaListenerContainer") @ConditionalOnBean(KycLevelUpdatedKafkaListener::class) - fun configureKycLevelUpdatedListener( + fun kycListenerContainer( listener: KycLevelUpdatedKafkaListener, @Qualifier("kycLevelUpdatedKafkaTemplate") template: KafkaTemplate, @Qualifier("KycConsumerFactory") consumerFactory: ConsumerFactory - ) { + ): ConcurrentMessageListenerContainer { val containerProps = ContainerProperties(Pattern.compile("kyc_level_updated")) containerProps.messageListener = listener val container = ConcurrentMessageListenerContainer(consumerFactory, containerProps) container.setBeanName("KycLevelUpdatedKafkaListenerContainer") container.commonErrorHandler = createConsumerErrorHandler(template, "kyc_level_updated.DLT") - container.start() + return container } - @Autowired + @Bean("withdrawRequestProducerFactory") + fun withdrawRequestProducerFactory(@Qualifier("consumerConfig") producerConfigs: Map): ProducerFactory { + return DefaultKafkaProducerFactory(producerConfigs) + } + + @Bean("withdrawRequestKafkaTemplate") + fun withdrawRequestKafkaTemplate(@Qualifier("withdrawRequestProducerFactory") producerFactory: ProducerFactory): KafkaTemplate { + return KafkaTemplate(producerFactory) + } + + @Bean("withdrawRequestKafkaListenerContainer") @ConditionalOnBean(WithdrawRequestKafkaListener::class) - fun configureWithdrawRequestEventListener( + fun withdrawRequestListenerContainer( listener: WithdrawRequestKafkaListener, @Qualifier("withdrawRequestKafkaTemplate") template: KafkaTemplate, @Qualifier("withdrawRequestConsumerFactory") consumerFactory: ConsumerFactory - ) { + ): ConcurrentMessageListenerContainer { val containerProps = ContainerProperties(Pattern.compile("withdraw_request")) containerProps.messageListener = listener val container = ConcurrentMessageListenerContainer(consumerFactory, containerProps) container.setBeanName("WithdrawRequestKafkaListenerContainer") container.commonErrorHandler = createConsumerErrorHandler(template, "withdraw_request.DLT") - container.start() + return container } - @Autowired + @Bean("depositProducerFactory") + fun depositProducerFactory(@Qualifier("consumerConfig") producerConfigs: Map): ProducerFactory { + return DefaultKafkaProducerFactory(producerConfigs) + } + + @Bean("depositKafkaTemplate") + fun depositKafkaTemplate(@Qualifier("depositProducerFactory") producerFactory: ProducerFactory): KafkaTemplate { + return KafkaTemplate(producerFactory) + } + + @Bean("depositKafkaListenerContainer") @ConditionalOnBean(DepositKafkaListener::class) - fun configureDepositRequestEventListener( + fun depositListenerContainer( listener: DepositKafkaListener, @Qualifier("depositKafkaTemplate") template: KafkaTemplate, @Qualifier("depositConsumerFactory") consumerFactory: ConsumerFactory - ) { + ): ConcurrentMessageListenerContainer { val containerProps = ContainerProperties(Pattern.compile("deposit")) containerProps.messageListener = listener val container = ConcurrentMessageListenerContainer(consumerFactory, containerProps) container.setBeanName("DepositKafkaListenerContainer") container.commonErrorHandler = createConsumerErrorHandler(template, "deposit.DLT") - container.start() + return container } private fun createConsumerErrorHandler(kafkaTemplate: KafkaTemplate<*, *>, dltTopic: String): CommonErrorHandler { @@ -224,5 +221,4 @@ class AccountantKafkaConfig { } return DefaultErrorHandler(recoverer, FixedBackOff(5_000, 20)) } - } \ No newline at end of file diff --git a/docker-compose.local.yml b/docker-compose.local.yml index 3ab397629..bcc4ed564 100644 --- a/docker-compose.local.yml +++ b/docker-compose.local.yml @@ -36,9 +36,33 @@ services: postgres-otp: ports: - "127.0.0.1:5462:5432" + postgres-accountant: + ports: + - "5432:5432" + postgres-eventlog: + ports: + - "5433:5432" + postgres-auth: + ports: + - "5434:5432" + postgres-wallet: + ports: + - "5435:5432" + postgres-api: + ports: + - "5436:5432" postgres-market: ports: - - "127.0.0.1:5438:5432" + - "5438:5432" + postgres-bc-gateway: + ports: + - "5437:5432" + postgres-matching-gateway: + ports: + - "5439:5432" + postgres-profile: + ports: + - "5440:5432" accountant: ports: - "127.0.0.1:8089:8080" diff --git a/market/market-core/src/main/kotlin/co/nilin/opex/market/core/event/RichOrderUpdate.kt b/market/market-core/src/main/kotlin/co/nilin/opex/market/core/event/RichOrderUpdate.kt index df048861e..98ce274fe 100644 --- a/market/market-core/src/main/kotlin/co/nilin/opex/market/core/event/RichOrderUpdate.kt +++ b/market/market-core/src/main/kotlin/co/nilin/opex/market/core/event/RichOrderUpdate.kt @@ -2,13 +2,15 @@ package co.nilin.opex.market.core.event import co.nilin.opex.market.core.inout.OrderStatus import java.math.BigDecimal +import java.time.LocalDateTime data class RichOrderUpdate( val ouid: String, val price: BigDecimal, val quantity: BigDecimal, val remainedQuantity: BigDecimal, - val status: OrderStatus = OrderStatus.NEW + val status: OrderStatus = OrderStatus.NEW, + val updateDate: LocalDateTime? = LocalDateTime.now() ) : RichOrderEvent { fun executedQuantity(): BigDecimal = quantity.minus(remainedQuantity) diff --git a/market/market-ports/market-eventlistener-kafka/src/main/kotlin/co/nilin/opex/market/ports/kafka/listener/config/KafkaConsumerConfig.kt b/market/market-ports/market-eventlistener-kafka/src/main/kotlin/co/nilin/opex/market/ports/kafka/listener/config/KafkaConsumerConfig.kt index 7a55d05a4..854c3fa7b 100644 --- a/market/market-ports/market-eventlistener-kafka/src/main/kotlin/co/nilin/opex/market/ports/kafka/listener/config/KafkaConsumerConfig.kt +++ b/market/market-ports/market-eventlistener-kafka/src/main/kotlin/co/nilin/opex/market/ports/kafka/listener/config/KafkaConsumerConfig.kt @@ -7,7 +7,6 @@ import co.nilin.opex.market.ports.kafka.listener.consumer.TradeKafkaListener import org.apache.kafka.clients.consumer.ConsumerConfig import org.apache.kafka.common.TopicPartition import org.apache.kafka.common.serialization.StringDeserializer -import org.springframework.beans.factory.annotation.Autowired import org.springframework.beans.factory.annotation.Qualifier import org.springframework.beans.factory.annotation.Value import org.springframework.boot.autoconfigure.condition.ConditionalOnBean @@ -52,34 +51,34 @@ class KafkaConsumerConfig { return DefaultKafkaConsumerFactory(consumerConfigs) } - @Autowired + @Bean("marketTradeKafkaListenerContainer") @ConditionalOnBean(TradeKafkaListener::class) - fun configureTradeListener( + fun tradeListenerContainer( tradeListener: TradeKafkaListener, template: KafkaTemplate, @Qualifier("richTradeConsumerFactory") consumerFactory: ConsumerFactory - ) { + ): ConcurrentMessageListenerContainer { val containerProps = ContainerProperties(Pattern.compile("richTrade")) containerProps.messageListener = tradeListener val container = ConcurrentMessageListenerContainer(consumerFactory, containerProps) container.setBeanName("marketTradeKafkaListenerContainer") container.commonErrorHandler = createConsumerErrorHandler(template, "richTrade.DLT") - container.start() + return container } - @Autowired + @Bean("marketOrderKafkaListenerContainer") @ConditionalOnBean(OrderKafkaListener::class) - fun configureOrderListener( + fun orderListenerContainer( orderListener: OrderKafkaListener, template: KafkaTemplate, @Qualifier("richOrderConsumerFactory") consumerFactory: ConsumerFactory - ) { + ): ConcurrentMessageListenerContainer { val containerProps = ContainerProperties(Pattern.compile("richOrder")) containerProps.messageListener = orderListener val container = ConcurrentMessageListenerContainer(consumerFactory, containerProps) container.setBeanName("marketOrderKafkaListenerContainer") container.commonErrorHandler = createConsumerErrorHandler(template, "richOrder.DLT") - container.start() + return container } private fun createConsumerErrorHandler(kafkaTemplate: KafkaTemplate<*, *>, dltTopic: String): CommonErrorHandler { 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 6f7bbd9a3..cc394b310 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 @@ -41,6 +41,14 @@ interface OrderRepository : ReactiveCrudRepository { origClientOrderId: String, ): Mono + @Query("update orders set update_date = :updateDate where ouid = :ouid") + fun touchUpdateDateByOuid( + @Param("ouid") + ouid: String, + @Param("updateDate") + updateDate: LocalDateTime = LocalDateTime.now() + ): Mono + @Query( """ select * from orders @@ -138,35 +146,55 @@ interface OrderRepository : ReactiveCrudRepository { @Query( """ -select o.symbol, - o.ouid, - o.order_type, - o.side, - o.price, - o.quantity, - o.quote_quantity, - os.executed_quantity, - o.taker_fee, - o.maker_fee, - os.status as status_code, - os.appearance, - o.create_date, - os.date as update_date, - o.uuid -from orders o - left join (select * - from order_status os1 - where os1.date = (select max(os2.date) - from order_status os2 - where os2.ouid = os1.ouid)) os on o.ouid = os.ouid - WHERE (:uuid is null or o.uuid = :uuid) - and (:symbol is null or o.symbol = :symbol) - and (:startTime is null or o.create_date >= :startTime) - and (:endTime is null or o.create_date <= :endTime) - and (:orderType is null or o.order_type = :orderType) - and (:direction is null or o.side = :direction) -order by create_date desc - limit :limit offset :offset; +with filtered_orders as ( + select o.symbol, + o.ouid, + o.order_type, + o.side, + o.price, + o.quantity, + o.quote_quantity, + o.taker_fee, + o.maker_fee, + o.create_date, + o.uuid + from orders o + where (:uuid is null or o.uuid = :uuid) + and (:symbol is null or o.symbol = :symbol) + and (:startTime is null or o.create_date >= :startTime) + and (:endTime is null or o.create_date <= :endTime) + and (:orderType is null or o.order_type = :orderType) + and (:direction is null or o.side = :direction) + order by o.create_date desc + limit :limit offset :offset +) +select fo.symbol, + fo.ouid, + fo.order_type, + fo.side, + fo.price, + fo.quantity, + fo.quote_quantity, + os.executed_quantity, + fo.taker_fee, + fo.maker_fee, + os.status as status_code, + os.appearance, + fo.create_date, + os.date as update_date, + fo.uuid +from filtered_orders fo +left join lateral ( + select s.executed_quantity, + s.status, + s.appearance, + s.date + from order_status s + where s.ouid = fo.ouid + order by s.appearance desc, s.executed_quantity desc nulls last, s.date desc, s.id desc + limit 1 +) os on true +order by fo.create_date desc; """ ) fun findByCriteria( @@ -228,14 +256,20 @@ SELECT FROM orders o LEFT JOIN ( - SELECT DISTINCT ON (ouid) - ouid, - executed_quantity, - status, - appearance, - date - FROM order_status - ORDER BY ouid, date DESC + SELECT ranked.ouid, + ranked.executed_quantity, + ranked.status, + ranked.appearance, + ranked.date + FROM ( + SELECT os.*, + ROW_NUMBER() OVER ( + PARTITION BY os.ouid + ORDER BY os.appearance DESC, os.executed_quantity DESC NULLS LAST, os.date DESC, os.id DESC + ) AS rnk + FROM order_status os + ) ranked + WHERE ranked.rnk = 1 ) os ON os.ouid = o.ouid diff --git a/market/market-ports/market-persister-postgres/src/main/kotlin/co/nilin/opex/market/ports/postgres/dao/OrderStatusRepository.kt b/market/market-ports/market-persister-postgres/src/main/kotlin/co/nilin/opex/market/ports/postgres/dao/OrderStatusRepository.kt index 72aa02548..b7632fc7b 100644 --- a/market/market-ports/market-persister-postgres/src/main/kotlin/co/nilin/opex/market/ports/postgres/dao/OrderStatusRepository.kt +++ b/market/market-ports/market-persister-postgres/src/main/kotlin/co/nilin/opex/market/ports/postgres/dao/OrderStatusRepository.kt @@ -30,7 +30,11 @@ interface OrderStatusRepository : ReactiveCrudRepository @Query( """ WITH ranked_order_status AS ( - SELECT *, ROW_NUMBER() OVER (PARTITION BY ouid ORDER BY appearance DESC, executed_quantity DESC) AS rnk + SELECT *, + ROW_NUMBER() OVER ( + PARTITION BY ouid + ORDER BY appearance DESC, executed_quantity DESC NULLS LAST, date DESC, id DESC + ) AS rnk FROM order_status WHERE ouid = :ouid ) diff --git a/market/market-ports/market-persister-postgres/src/main/kotlin/co/nilin/opex/market/ports/postgres/dao/TradeRepository.kt b/market/market-ports/market-persister-postgres/src/main/kotlin/co/nilin/opex/market/ports/postgres/dao/TradeRepository.kt index c0eacbe01..517db2d74 100644 --- a/market/market-ports/market-persister-postgres/src/main/kotlin/co/nilin/opex/market/ports/postgres/dao/TradeRepository.kt +++ b/market/market-ports/market-persister-postgres/src/main/kotlin/co/nilin/opex/market/ports/postgres/dao/TradeRepository.kt @@ -22,6 +22,14 @@ interface TradeRepository : ReactiveCrudRepository { @Query("select * from trades where symbol = :symbol order by create_date desc limit 1") fun findMostRecentBySymbol(symbol: String): Flux + @Query("select * from trades where symbol = :symbol and trade_id = :tradeId limit 1") + fun findBySymbolAndTradeId( + @Param("symbol") + symbol: String, + @Param("tradeId") + tradeId: Long + ): Mono + @Query("select * from trades where symbol = :symbol order by create_date desc limit :limit") fun findBySymbolSortDescendingByCreateDate( @Param("symbol") @@ -134,8 +142,8 @@ interface TradeRepository : ReactiveCrudRepository { CASE WHEN t.taker_uuid = :uuid - THEN (to2.side = 'ASK') - ELSE (mo.side = 'ASK') + THEN (to2.side = 'BID') + ELSE (mo.side = 'BID') END AS isBuyer, (t.maker_uuid = :uuid) AS isMaker 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 c6870c2cf..87a7d7a3d 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,6 +17,8 @@ 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 @@ -55,7 +57,15 @@ class OrderPersisterImpl( LocalDateTime.now(), LocalDateTime.now() ) - orderRepository.save(orderModel).awaitFirstOrNull() + try { + orderRepository.save(orderModel).awaitFirstOrNull() + } catch (e: DuplicateKeyException) { + logger.info("order ${order.ouid} is duplicate; skipping create flow") + return + } catch (e: DataIntegrityViolationException) { + logger.info("order ${order.ouid} is duplicate; skipping create flow") + return + } logger.info("order ${order.ouid} saved") orderStatusRepository.insert( @@ -83,12 +93,18 @@ class OrderPersisterImpl( @Transactional override suspend fun update(orderUpdate: RichOrderUpdate) { + + val updateTime = orderUpdate.updateDate ?: LocalDateTime.now() + + orderRepository.touchUpdateDateByOuid(orderUpdate.ouid, updateTime).awaitFirstOrNull() + orderStatusRepository.insert( orderUpdate.ouid, orderUpdate.executedQuantity(), orderUpdate.accumulativeQuoteQuantity(), orderUpdate.status.code, - orderUpdate.status.orderOfAppearance + orderUpdate.status.orderOfAppearance, + updateTime ).awaitFirstOrNull() logger.info("OrderStatus ${orderUpdate.ouid} updated with status of ${orderUpdate.status}") @@ -101,7 +117,11 @@ class OrderPersisterImpl( openOrderRepository.delete(orderUpdate.ouid).awaitSingleOrNull() logger.info("Order ${orderUpdate.ouid} deleted from open orders") } - val order = orderRepository.findByOuid(orderUpdate.ouid).awaitFirstOrNull() ?: return + val order = orderRepository.findByOuid(orderUpdate.ouid).awaitFirstOrNull() + ?: run { + logger.info("Order ${orderUpdate.ouid} not found for update event, SKIPPED") + return + } marketOrderProducer.openOrderUpdate(order.uuid, order.symbol) } 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 b229535a4..a257a1d99 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 @@ -9,11 +9,13 @@ import co.nilin.opex.market.ports.postgres.model.TradeModel import co.nilin.opex.market.ports.postgres.util.RedisCacheHelper import kotlinx.coroutines.reactive.awaitFirstOrNull 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 import java.time.ZoneId import java.util.* +import java.util.concurrent.atomic.AtomicLong @Component class TradePersisterImpl( @@ -23,33 +25,42 @@ class TradePersisterImpl( private val logger = LoggerFactory.getLogger(TradePersisterImpl::class.java) - @Transactional override suspend fun save(trade: RichTrade) { val pair = trade.pair.split("_") + val tradeModel = TradeModel( + null, + trade.id, + trade.pair, + pair[0].uppercase(), + pair[1].uppercase(), + trade.matchedPrice, + trade.matchedQuantity, + trade.takerPrice, + trade.makerPrice, + trade.takerCommision, + trade.makerCommision, + trade.takerCommisionAsset, + trade.makerCommisionAsset, + trade.tradeDateTime, + trade.makerOuid, + trade.takerOuid, + trade.makerUuid, + trade.takerUuid, + LocalDateTime.now() + ) - val tradeEntity = tradeRepository.save( - TradeModel( - null, - trade.id, - trade.pair, - pair[0].uppercase(), - pair[1].uppercase(), - trade.matchedPrice, - trade.matchedQuantity, - trade.takerPrice, - trade.makerPrice, - trade.takerCommision, - trade.makerCommision, - trade.takerCommisionAsset, - trade.makerCommisionAsset, - trade.tradeDateTime, - trade.makerOuid, - trade.takerOuid, - trade.makerUuid, - trade.takerUuid, - LocalDateTime.now() - ) - ).awaitFirstOrNull() + val tradeEntity = try { + tradeRepository.save(tradeModel).awaitFirstOrNull() + } catch (e: DuplicateKeyException) { + ensureNotCollision(tradeModel, trade) + return + } catch (e: DataIntegrityViolationException) { + if (!isDuplicateTradeViolation(e)) { + throw e + } + ensureNotCollision(tradeModel, trade) + return + } logger.info("RichTrade ${trade.id} saved") //calculateTradeVolume(trade, pair[0].uppercase(), pair[1].uppercase()) // Moved to accountant updateCache(trade, tradeEntity) @@ -85,4 +96,76 @@ class TradePersisterImpl( logger.info("Could not update recentTrades cache") } } + + private suspend fun ensureNotCollision(incomingTrade: TradeModel, originalTrade: RichTrade) { + val existingTrade = tradeRepository.findBySymbolAndTradeId( + incomingTrade.symbol, + incomingTrade.tradeId + ).awaitFirstOrNull() ?: throw IllegalStateException( + "Duplicate trade conflict detected but existing row not found for symbol=${incomingTrade.symbol}, tradeId=${incomingTrade.tradeId}" + ) + + if (isSameTradePayload(existingTrade, incomingTrade)) { + logger.info("RichTrade ${incomingTrade.tradeId} for ${incomingTrade.symbol} is duplicate delivery; skipping") + return + } + + // Real ID collision (e.g. Redis counter reset): persist under a new synthetic ID and continue + logger.error( + "Trade ID collision for symbol=${incomingTrade.symbol}, tradeId=${incomingTrade.tradeId}. " + + "Saving colliding trade under a new synthetic ID." + ) + //todo cast the tradeId to BigInteger + val newId = generateUniqueId() + logger.info("The old tradeId ${incomingTrade.tradeId} - The new tradeId: $newId") + val reassigned = TradeModel( + null, + newId, + incomingTrade.symbol, + incomingTrade.baseAsset, + incomingTrade.quoteAsset, + incomingTrade.matchedPrice, + incomingTrade.matchedQuantity, + incomingTrade.takerPrice, + incomingTrade.makerPrice, + incomingTrade.takerCommission, + incomingTrade.makerCommission, + incomingTrade.takerCommissionAsset, + incomingTrade.makerCommissionAsset, + incomingTrade.tradeDate, + incomingTrade.makerOuid, + incomingTrade.takerOuid, + incomingTrade.makerUuid, + incomingTrade.takerUuid, + incomingTrade.createDate + ) + val saved = tradeRepository.save(reassigned).awaitFirstOrNull() + updateCache(originalTrade, saved) + } + + private fun isSameTradePayload(existing: TradeModel, incoming: TradeModel): Boolean { + 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 && + existing.makerCommission == incoming.makerCommission && + existing.takerCommission == incoming.takerCommission && + existing.makerCommissionAsset == incoming.makerCommissionAsset && + existing.takerCommissionAsset == incoming.takerCommissionAsset + } + + private fun isDuplicateTradeViolation(exception: Throwable): Boolean { + val errorText = buildString { + append(exception.message.orEmpty()) + append(' ') + append(exception.cause?.message.orEmpty()) + } + return errorText.contains("uq_trades_symbol_trade_id", ignoreCase = true) || + errorText.contains("duplicate key value", ignoreCase = true) + } + + private fun generateUniqueId(): Long { + return AtomicLong(System.currentTimeMillis() * 1000).incrementAndGet() + } } \ No newline at end of file diff --git a/market/market-ports/market-persister-postgres/src/main/resources/schema.sql b/market/market-ports/market-persister-postgres/src/main/resources/schema.sql index 37cb1ce0c..294115909 100644 --- a/market/market-ports/market-persister-postgres/src/main/resources/schema.sql +++ b/market/market-ports/market-persister-postgres/src/main/resources/schema.sql @@ -33,6 +33,8 @@ CREATE TABLE IF NOT EXISTS order_status date TIMESTAMP NOT NULL, UNIQUE (ouid, status, appearance, executed_quantity) ); +CREATE INDEX IF NOT EXISTS idx_order_status_ouid_rank + ON order_status (ouid, appearance DESC, executed_quantity DESC, date DESC, id DESC); CREATE TABLE IF NOT EXISTS open_orders ( @@ -42,6 +44,9 @@ CREATE TABLE IF NOT EXISTS open_orders status INTEGER NOT NULL ); +CREATE INDEX IF NOT EXISTS idx_orders_uuid_create_date ON orders (uuid, create_date DESC); +CREATE INDEX IF NOT EXISTS idx_orders_uuid_symbol_create_date ON orders (uuid, symbol, create_date DESC); + CREATE TABLE IF NOT EXISTS trades ( id SERIAL PRIMARY KEY, @@ -67,6 +72,26 @@ CREATE TABLE IF NOT EXISTS trades CREATE INDEX IF NOT EXISTS idx_trades_symbol on trades (symbol); CREATE INDEX IF NOT EXISTS idx_trades_create_date on trades (create_date); +ALTER TABLE trades + ALTER COLUMN id TYPE BIGINT, + ALTER COLUMN trade_id TYPE BIGINT; +ALTER SEQUENCE trades_id_seq AS BIGINT; + +WITH duplicate_trades AS ( + SELECT id + FROM ( + SELECT id, + ROW_NUMBER() OVER (PARTITION BY symbol, trade_id ORDER BY id) AS rn + FROM trades + ) ranked + WHERE rn > 1 +) +DELETE +FROM trades t + USING duplicate_trades d +WHERE t.id = d.id; +CREATE UNIQUE INDEX IF NOT EXISTS uq_trades_symbol_trade_id on trades (symbol, trade_id); + CREATE OR REPLACE FUNCTION interval_generator( start_ts TIMESTAMP without TIME ZONE, end_ts TIMESTAMP without TIME ZONE, @@ -85,4 +110,3 @@ BEGIN END; $$ LANGUAGE 'plpgsql'; - 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 0a533641a..ba1077369 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 @@ -9,9 +9,12 @@ import co.nilin.opex.market.ports.postgres.util.RedisCacheHelper import io.mockk.coEvery import io.mockk.every import io.mockk.mockk +import io.mockk.verify import kotlinx.coroutines.runBlocking import org.assertj.core.api.Assertions.assertThatNoException import org.junit.jupiter.api.Test +import org.junit.jupiter.api.assertThrows +import org.springframework.dao.DuplicateKeyException import reactor.core.publisher.Mono class OrderPersisterTest { @@ -57,6 +60,9 @@ class OrderPersisterTest { @Test fun givenOrderRepo_whenUpdateRichOrder_thenSuccess(): Unit = runBlocking { + every { + orderRepository.touchUpdateDateByOuid(any(), any()) + } returns Mono.empty() every { orderStatusRepository.insert(any(), any(), any(), any(), any(), any()) } returns Mono.empty() @@ -76,4 +82,40 @@ class OrderPersisterTest { assertThatNoException().isThrownBy { runBlocking { orderPersister.update(VALID.RICH_ORDER_UPDATE) } } } + + @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) } } + + verify(exactly = 0) { + orderStatusRepository.insert(any(), any(), any(), any(), any(), any()) + } + } + + //To have race condition between RichOrder and UpdateRichOrder,we will temporarily skip this test + +// @Test +// fun givenMissingOrder_whenUpdateRichOrder_thenFailBeforeSideEffects(): Unit = runBlocking { +// every { +// orderRepository.findByOuid(any()) +// } returns Mono.empty() +// +// assertThrows { +// runBlocking { orderPersister.update(VALID.RICH_ORDER_UPDATE) } +// } +// +// verify(exactly = 0) { +// orderRepository.touchUpdateDateByOuid(any(), any()) +// } +// verify(exactly = 0) { +// orderStatusRepository.insert(any(), any(), any(), any(), any(), any()) +// } +// verify(exactly = 0) { +// openOrderRepository.insertOrUpdate(any(), any(), any()) +// } +// } } diff --git a/market/market-ports/market-persister-postgres/src/test/kotlin/co/nilin/opex/market/ports/postgres/impl/TradePersisterTest.kt b/market/market-ports/market-persister-postgres/src/test/kotlin/co/nilin/opex/market/ports/postgres/impl/TradePersisterTest.kt index 2a2f8f531..308f9f9cc 100644 --- a/market/market-ports/market-persister-postgres/src/test/kotlin/co/nilin/opex/market/ports/postgres/impl/TradePersisterTest.kt +++ b/market/market-ports/market-persister-postgres/src/test/kotlin/co/nilin/opex/market/ports/postgres/impl/TradePersisterTest.kt @@ -2,13 +2,18 @@ package co.nilin.opex.market.ports.postgres.impl import co.nilin.opex.market.ports.postgres.dao.TradeRepository import co.nilin.opex.market.ports.postgres.impl.sample.VALID +import co.nilin.opex.market.ports.postgres.model.TradeModel import co.nilin.opex.market.ports.postgres.util.RedisCacheHelper import io.mockk.every import io.mockk.mockk import kotlinx.coroutines.runBlocking import org.assertj.core.api.Assertions.assertThatNoException import org.junit.jupiter.api.Test +import org.junit.jupiter.api.assertThrows +import org.springframework.dao.DataIntegrityViolationException +import org.springframework.dao.DuplicateKeyException import reactor.core.publisher.Mono +import java.math.BigDecimal class TradePersisterTest { @@ -22,4 +27,46 @@ class TradePersisterTest { every { tradeRepository.save(any()) } returns Mono.just(VALID.TRADE_MODEL) assertThatNoException().isThrownBy { runBlocking { tradePersister.save(VALID.RICH_TRADE) } } } + + @Test + fun givenDuplicateTrade_whenSaveRichTrade_thenIgnoredAsIdempotent(): Unit = runBlocking { + every { tradeRepository.save(any()) } returnsMany listOf( + Mono.error(DuplicateKeyException("Duplicate key")), + Mono.just(VALID.TRADE_MODEL) + ) + every { tradeRepository.findBySymbolAndTradeId(any(), any()) } returns Mono.just(VALID.TRADE_MODEL) + assertThatNoException().isThrownBy { runBlocking { tradePersister.save(VALID.RICH_TRADE) } } + } + + @Test + fun givenTradeIdCollision_whenSaveRichTrade_thenThrow() { + every { tradeRepository.save(any()) } returns Mono.error(DuplicateKeyException("duplicate trade")) + every { tradeRepository.findBySymbolAndTradeId(any(), any()) } returns Mono.just( + TradeModel( + VALID.TRADE_MODEL.id, + VALID.TRADE_MODEL.tradeId, + VALID.TRADE_MODEL.symbol, + VALID.TRADE_MODEL.baseAsset, + VALID.TRADE_MODEL.quoteAsset, + VALID.TRADE_MODEL.matchedPrice, + VALID.TRADE_MODEL.matchedQuantity.add(BigDecimal.ONE), + VALID.TRADE_MODEL.takerPrice, + VALID.TRADE_MODEL.makerPrice, + VALID.TRADE_MODEL.takerCommission, + VALID.TRADE_MODEL.makerCommission, + VALID.TRADE_MODEL.takerCommissionAsset, + VALID.TRADE_MODEL.makerCommissionAsset, + VALID.TRADE_MODEL.tradeDate, + VALID.TRADE_MODEL.makerOuid, + VALID.TRADE_MODEL.takerOuid, + VALID.TRADE_MODEL.makerUuid, + VALID.TRADE_MODEL.takerUuid, + VALID.TRADE_MODEL.createDate + ) + ) + + assertThrows { + runBlocking { tradePersister.save(VALID.RICH_TRADE) } + } + } } diff --git a/matching-gateway/matching-gateway-app/src/main/resources/application.yml b/matching-gateway/matching-gateway-app/src/main/resources/application.yml index 860504b23..9d7103666 100644 --- a/matching-gateway/matching-gateway-app/src/main/resources/application.yml +++ b/matching-gateway/matching-gateway-app/src/main/resources/application.yml @@ -44,6 +44,8 @@ spring: instance-id: ${spring.application.name}:${server.port} healthCheckInterval: 20s prefer-ip-address: true + config: + import: vault://secret/${spring.application.name} management: endpoints: web: diff --git a/matching-gateway/matching-gateway-app/src/test/kotlin/co/nilin/opex/matching/gateway/app/service/sample/Samples.kt b/matching-gateway/matching-gateway-app/src/test/kotlin/co/nilin/opex/matching/gateway/app/service/sample/Samples.kt index ea9f1ab47..6f14f8ade 100644 --- a/matching-gateway/matching-gateway-app/src/test/kotlin/co/nilin/opex/matching/gateway/app/service/sample/Samples.kt +++ b/matching-gateway/matching-gateway-app/src/test/kotlin/co/nilin/opex/matching/gateway/app/service/sample/Samples.kt @@ -24,7 +24,7 @@ object VALID { val PAIR_CONFIG = PairConfig(ETH_USDT, ETH, USDT, BigDecimal.valueOf(0.01), BigDecimal.valueOf(0.0001)) - val PAIR_SETTING = PairSetting(ETH_USDT, true, 0.0000001.toBigDecimal(), 100.toBigDecimal(), "LIMIT_ORDER,MARKET_ORDER", null) + val PAIR_SETTING = PairSetting(ETH_USDT, true, 0.0000001.toBigDecimal(), 100.toBigDecimal(), "LIMIT_ORDER,MARKET_ORDER", null,true,true) val CREATE_ORDER_REQUEST_ASK = CreateOrderRequest( UUID, diff --git a/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/dao/PairCategoryRepository.kt b/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/dao/PairCategoryRepository.kt new file mode 100644 index 000000000..dd3c39a39 --- /dev/null +++ b/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/dao/PairCategoryRepository.kt @@ -0,0 +1,14 @@ +package co.nilin.opex.matching.gateway.ports.postgres.dao + +import co.nilin.opex.matching.gateway.ports.postgres.model.PairCategoryModel +import kotlinx.coroutines.flow.Flow +import org.springframework.data.repository.kotlin.CoroutineCrudRepository +import org.springframework.stereotype.Repository +import reactor.core.publisher.Mono + +@Repository +interface PairCategoryRepository : CoroutineCrudRepository { + fun findByPair(pair: String): Flow + + fun deleteByPair(pair: String): Mono +} \ No newline at end of file diff --git a/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/dao/PairSettingRepository.kt b/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/dao/PairSettingRepository.kt index 12fcff723..dc562879a 100644 --- a/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/dao/PairSettingRepository.kt +++ b/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/dao/PairSettingRepository.kt @@ -11,6 +11,14 @@ import java.math.BigDecimal interface PairSettingRepository : ReactiveCrudRepository { fun findByPair(pair: String): Mono - @Query("insert into pair_setting(pair,is_available,min_order,max_order,order_types) values(:pair,:isAvailable,:minOrder,:maxOrder,:orderTypes) ") - fun insert(pair: String, isAvailable: Boolean , minOrder : BigDecimal, maxOrder : BigDecimal,orderTypes : String): Mono + @Query("insert into pair_setting(pair,is_available,min_order,max_order,order_types,internal_chart,global_chart) values(:pair,:isAvailable,:minOrder,:maxOrder,:orderTypes,:internalChart,:globalChart) ") + fun insert( + pair: String, + isAvailable: Boolean, + minOrder: BigDecimal, + maxOrder: BigDecimal, + orderTypes: String, + internalChart: Boolean, + globalChart: Boolean + ): Mono } diff --git a/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/dto/PairSetting.kt b/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/dto/PairSetting.kt index 1fd568629..d13a53b62 100644 --- a/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/dto/PairSetting.kt +++ b/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/dto/PairSetting.kt @@ -1,13 +1,17 @@ package co.nilin.opex.matching.gateway.ports.postgres.dto +import co.nilin.opex.matching.gateway.ports.postgres.model.PairCategory import java.math.BigDecimal import java.time.LocalDateTime class PairSetting( val pair: String, val isAvailable: Boolean, - val minOrder : BigDecimal, - val maxOrder : BigDecimal, - val orderTypes : String, + val minOrder: BigDecimal, + val maxOrder: BigDecimal, + val orderTypes: String, val updateDate: LocalDateTime? = null, + val internalChart: Boolean, + val globalChart: Boolean, + val categories: List = emptyList() ) \ No newline at end of file diff --git a/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/impl/PairSettingServiceImpl.kt b/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/impl/PairSettingServiceImpl.kt index 288ef97c9..0c603e74a 100644 --- a/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/impl/PairSettingServiceImpl.kt +++ b/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/impl/PairSettingServiceImpl.kt @@ -2,10 +2,14 @@ package co.nilin.opex.matching.gateway.ports.postgres.impl import co.nilin.opex.common.OpexError import co.nilin.opex.common.utils.CacheManager +import co.nilin.opex.matching.gateway.ports.postgres.dao.PairCategoryRepository import co.nilin.opex.matching.gateway.ports.postgres.dao.PairSettingRepository import co.nilin.opex.matching.gateway.ports.postgres.dto.PairSetting +import co.nilin.opex.matching.gateway.ports.postgres.model.PairCategoryModel import co.nilin.opex.matching.gateway.ports.postgres.service.PairSettingService import co.nilin.opex.matching.gateway.ports.postgres.util.toPairSetting +import kotlinx.coroutines.flow.map +import kotlinx.coroutines.flow.toList import kotlinx.coroutines.reactive.awaitFirst import kotlinx.coroutines.reactive.awaitFirstOrNull import org.springframework.beans.factory.annotation.Qualifier @@ -16,14 +20,20 @@ import java.util.concurrent.TimeUnit @Service class PairSettingServiceImpl( private val pairSettingRepository: PairSettingRepository, + private val pairCategoryRepository: PairCategoryRepository, @Qualifier("appCacheManager") private val cacheManager: CacheManager ) : PairSettingService { override suspend fun load(pair: String): PairSetting { return cacheManager.get("pair-setting:$pair") ?: pairSettingRepository.findByPair(pair) - .awaitFirstOrNull() - ?.let { + .awaitFirstOrNull()?.let { pairSettingModel -> + val categories = pairCategoryRepository.findByPair(pairSettingModel.pair) + .map { it.category } + .toList() + pairSettingModel.categories = categories + pairSettingModel + }?.let { it.toPairSetting().also { cacheManager.put( "pair-setting:${it.pair}", @@ -36,28 +46,61 @@ class PairSettingServiceImpl( } override suspend fun loadAll(): List { - return pairSettingRepository.findAll() - .map { it.toPairSetting() } - .collectList() - .awaitFirstOrNull() ?: emptyList() + val pairSettings = pairSettingRepository.findAll().collectList().awaitFirst() + + if (pairSettings.isEmpty()) { + return emptyList() + } + + val categoriesByPair = pairCategoryRepository.findAll() + .toList() + .groupBy( + keySelector = { it.pair }, + valueTransform = { it.category } + ) + + return pairSettings.map { ps -> + ps.categories = categoriesByPair[ps.pair] ?: emptyList() + ps.toPairSetting() + } } override suspend fun update(pairSetting: PairSetting): PairSetting { - val pairSetting = - pairSettingRepository.findByPair(pairSetting.pair).awaitFirstOrNull() - ?: throw OpexError.PairNotFound.exception() - pairSetting.apply { - this.isAvailable = pairSetting.isAvailable - this.minOrder = pairSetting.minOrder - this.maxOrder = pairSetting.maxOrder - this.orderTypes = pairSetting.orderTypes - this.updateDate = LocalDateTime.now() + val existing = pairSettingRepository.findByPair(pairSetting.pair) + .awaitFirstOrNull() + ?: throw OpexError.PairNotFound.exception() + + existing.apply { + isAvailable = pairSetting.isAvailable + minOrder = pairSetting.minOrder + maxOrder = pairSetting.maxOrder + orderTypes = pairSetting.orderTypes + updateDate = LocalDateTime.now() + internalChart = pairSetting.internalChart + globalChart = pairSetting.globalChart + } + + val saved = pairSettingRepository.save(existing) + .awaitFirst() + + pairCategoryRepository.deleteByPair(pairSetting.pair).awaitFirstOrNull() + pairSetting.categories.forEach { category -> + pairCategoryRepository.save( + PairCategoryModel( + pair = pairSetting.pair, + category = category + ) + ) } - return pairSettingRepository.save(pairSetting).awaitFirst().toPairSetting().also { + + return saved.apply { + categories = pairSetting.categories + }.toPairSetting().also { cacheManager.put( "pair-setting:${it.pair}", it, - 5, TimeUnit.MINUTES + 5, + TimeUnit.MINUTES ) } } diff --git a/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/model/PairCategory.kt b/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/model/PairCategory.kt new file mode 100644 index 000000000..a3e2ad3f0 --- /dev/null +++ b/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/model/PairCategory.kt @@ -0,0 +1,7 @@ +package co.nilin.opex.matching.gateway.ports.postgres.model + +enum class PairCategory { + REAL_ASSET_TOKEN, + FIAT, + CRYPTO +} \ No newline at end of file diff --git a/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/model/PairCategoryModel.kt b/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/model/PairCategoryModel.kt new file mode 100644 index 000000000..7203802c9 --- /dev/null +++ b/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/model/PairCategoryModel.kt @@ -0,0 +1,11 @@ +package co.nilin.opex.matching.gateway.ports.postgres.model + +import org.springframework.data.annotation.Id +import org.springframework.data.relational.core.mapping.Table + +@Table("pair_category") +data class PairCategoryModel( + @Id val id: Long? = null, + val pair: String, + val category: PairCategory +) \ No newline at end of file diff --git a/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/model/PairSettingModel.kt b/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/model/PairSettingModel.kt index f62832c5d..f36bdf34b 100644 --- a/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/model/PairSettingModel.kt +++ b/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/model/PairSettingModel.kt @@ -1,6 +1,7 @@ package co.nilin.opex.matching.gateway.ports.postgres.model import org.springframework.data.annotation.Id +import org.springframework.data.annotation.Transient import org.springframework.data.relational.core.mapping.Table import java.math.BigDecimal import java.time.LocalDateTime @@ -10,8 +11,13 @@ data class PairSettingModel( @Id val pair: String, var isAvailable: Boolean, - var minOrder : BigDecimal, - var maxOrder : BigDecimal, - var orderTypes : String, + var minOrder: BigDecimal, + var maxOrder: BigDecimal, + var orderTypes: String, var updateDate: LocalDateTime? = null, -) \ No newline at end of file + var internalChart: Boolean, + var globalChart: Boolean +) { + @Transient + var categories: List = emptyList() +} \ No newline at end of file diff --git a/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/service/PairSettingInitializer.kt b/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/service/PairSettingInitializer.kt index 8bafa2161..da9bf47b0 100644 --- a/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/service/PairSettingInitializer.kt +++ b/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/service/PairSettingInitializer.kt @@ -46,7 +46,9 @@ class PairSettingInitializer( false, BigDecimal.ONE, BigDecimal.ONE, - "LIMIT_ORDER,MARKET_ORDER" + "LIMIT_ORDER,MARKET_ORDER", + true, + true ).then(pairSettingRepository.findByPair(pair)).awaitFirstOrNull() .also { if (it == null) logger.warn("Failed to insert pair: $pair") } ?: return@forEach diff --git a/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/util/Convertor.kt b/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/util/Convertor.kt index c039102d7..8737575f4 100644 --- a/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/util/Convertor.kt +++ b/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/kotlin/co/nilin/opex/matching/gateway/ports/postgres/util/Convertor.kt @@ -12,6 +12,9 @@ fun PairSettingModel.toPairSetting(): PairSetting { maxOrder, orderTypes, updateDate, + internalChart, + globalChart, + categories ) } diff --git a/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/resources/schema.sql b/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/resources/schema.sql index c8a595b04..eb9af6fa8 100644 --- a/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/resources/schema.sql +++ b/matching-gateway/matching-gateway-port/matching-gateway-persister-postgres/src/main/resources/schema.sql @@ -23,5 +23,24 @@ $$ WHERE table_name = 'pair_setting' AND column_name = 'order_types') THEN ALTER TABLE pair_setting ADD COLUMN order_types varchar(255) NOT NULL default 'LIMIT_ORDER, MARKET_ORDER' ; END IF; + IF NOT EXISTS (SELECT 1 + FROM information_schema.columns + WHERE table_name = 'pair_setting' + AND column_name = 'internal_chart') THEN ALTER TABLE pair_setting + ADD COLUMN internal_chart BOOLEAN NOT NULL default true; + END IF; + IF NOT EXISTS (SELECT 1 + FROM information_schema.columns + WHERE table_name = 'pair_setting' AND column_name = 'global_chart') THEN ALTER TABLE pair_setting + ADD COLUMN global_chart BOOLEAN NOT NULL default true; + END IF; END -$$; \ No newline at end of file +$$; + +CREATE TABLE IF NOT EXISTS pair_category +( + id SERIAL PRIMARY KEY, + pair VARCHAR(72) NOT NULL REFERENCES pair_setting (pair), + category VARCHAR(255) NOT NULL, + UNIQUE (pair, category) +); diff --git a/otp/otp-app/src/main/kotlin/co/nilin/opex/otp/app/proxy/SMSIRProxy.kt b/otp/otp-app/src/main/kotlin/co/nilin/opex/otp/app/proxy/SMSIRProxy.kt index 10156b8cd..df0f0bbdd 100644 --- a/otp/otp-app/src/main/kotlin/co/nilin/opex/otp/app/proxy/SMSIRProxy.kt +++ b/otp/otp-app/src/main/kotlin/co/nilin/opex/otp/app/proxy/SMSIRProxy.kt @@ -32,8 +32,8 @@ class SMSIRProxy( .queryParam("password", config.password) .queryParam("mobile", receiver) .queryParam("line", config.sender) - .queryParam("text", "otp code : $message") - .build(true) + .queryParam("text", "Your OTP code is $message") + .build() .toUri() return try {