diff --git a/accountant/accountant-app/src/main/resources/application.yml b/accountant/accountant-app/src/main/resources/application.yml index 1f08d8860..f191ca34c 100644 --- a/accountant/accountant-app/src/main/resources/application.yml +++ b/accountant/accountant-app/src/main/resources/application.yml @@ -47,6 +47,7 @@ spring: instance-id: ${spring.application.name}:${server.port} healthCheckInterval: 20s prefer-ip-address: true + query-passing: true config: import: vault://secret/${spring.application.name} management: 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/accountant/accountant-ports/accountant-wallet-proxy/src/main/kotlin/co/nilin/opex/accountant/ports/walletproxy/config/WebClientConfig.kt b/accountant/accountant-ports/accountant-wallet-proxy/src/main/kotlin/co/nilin/opex/accountant/ports/walletproxy/config/WebClientConfig.kt index 2ba1c3c02..2f9bbe64c 100644 --- a/accountant/accountant-ports/accountant-wallet-proxy/src/main/kotlin/co/nilin/opex/accountant/ports/walletproxy/config/WebClientConfig.kt +++ b/accountant/accountant-ports/accountant-wallet-proxy/src/main/kotlin/co/nilin/opex/accountant/ports/walletproxy/config/WebClientConfig.kt @@ -1,25 +1,52 @@ package co.nilin.opex.accountant.ports.walletproxy.config +import io.netty.channel.ChannelOption import org.springframework.cloud.client.ServiceInstance import org.springframework.cloud.client.loadbalancer.reactive.ReactiveLoadBalancer import org.springframework.cloud.client.loadbalancer.reactive.ReactorLoadBalancerExchangeFilterFunction import org.springframework.context.annotation.Bean import org.springframework.context.annotation.Configuration +import org.springframework.http.client.reactive.ReactorClientHttpConnector import org.springframework.web.reactive.function.client.WebClient import org.zalando.logbook.Logbook import org.zalando.logbook.netty.LogbookClientHandler import reactor.netty.http.client.HttpClient +import reactor.netty.resources.ConnectionProvider +import java.time.Duration @Configuration class WebClientConfig { @Bean - fun webClient(loadBalancerFactory: ReactiveLoadBalancer.Factory, logbook: Logbook): WebClient { - val client = HttpClient.create().doOnConnected { it.addHandlerLast(LogbookClientHandler(logbook)) } + fun webClient( + loadBalancerFactory: ReactiveLoadBalancer.Factory, + logbook: Logbook + ): WebClient { + + val connectionProvider = ConnectionProvider.builder("accountant-wallet") + .maxIdleTime(Duration.ofSeconds(20)) + .maxLifeTime(Duration.ofMinutes(5)) + .pendingAcquireTimeout(Duration.ofSeconds(5)) + .evictInBackground(Duration.ofSeconds(30)) + .lifo() + .build() + + val client = HttpClient.create(connectionProvider) + .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000) + .responseTimeout(Duration.ofSeconds(10)) + .keepAlive(true) + .doOnConnected { + it.addHandlerLast(LogbookClientHandler(logbook)) + } + return WebClient.builder() - //.clientConnector(ReactorClientHttpConnector(client)) - .filter(ReactorLoadBalancerExchangeFilterFunction(loadBalancerFactory, emptyList())) + .clientConnector(ReactorClientHttpConnector(client)) + .filter( + ReactorLoadBalancerExchangeFilterFunction( + loadBalancerFactory, + emptyList() + ) + ) .build() } - -} +} \ No newline at end of file diff --git a/api/api-app/src/main/kotlin/co/nilin/opex/api/app/config/RateLimitConfig.kt b/api/api-app/src/main/kotlin/co/nilin/opex/api/app/config/RateLimitConfig.kt index fd9167451..054d639d8 100644 --- a/api/api-app/src/main/kotlin/co/nilin/opex/api/app/config/RateLimitConfig.kt +++ b/api/api-app/src/main/kotlin/co/nilin/opex/api/app/config/RateLimitConfig.kt @@ -53,12 +53,17 @@ class RateLimitConfig( return ReactiveSecurityContextHolder.getContext() .mapNotNull { it.authentication } .filter { it.isAuthenticated } - .flatMap { auth -> - applyRateLimit(auth.name, exchange, chain, groupId) + .map { auth -> + Mono.defer { + applyRateLimit(auth.name, exchange, chain, groupId) + } } - .switchIfEmpty( - chain.filter(exchange) + .defaultIfEmpty( + Mono.defer { + chain.filter(exchange) + } ) + .flatMap { it } } private fun applyRateLimit( diff --git a/api/api-app/src/main/kotlin/co/nilin/opex/api/app/config/WebClientConfig.kt b/api/api-app/src/main/kotlin/co/nilin/opex/api/app/config/WebClientConfig.kt index a89032080..c2d67d527 100644 --- a/api/api-app/src/main/kotlin/co/nilin/opex/api/app/config/WebClientConfig.kt +++ b/api/api-app/src/main/kotlin/co/nilin/opex/api/app/config/WebClientConfig.kt @@ -26,22 +26,58 @@ class WebClientConfig( private val logbook: Logbook, @Value("\${app.auth.url}") private val url: String, + @Value("\${app.http.client.wiretap.enabled:false}") + private val wiretapEnabled: Boolean, + @Value("\${app.http.client.general.max-connections:300}") + private val generalMaxConnections: Int, + @Value("\${app.http.client.general.pending-acquire-max-count:1000}") + private val generalPendingAcquireMaxCount: Int, + @Value("\${app.http.client.general.max-idle-seconds:30}") + private val generalMaxIdleSeconds: Long, + @Value("\${app.http.client.general.max-life-seconds:120}") + private val generalMaxLifeSeconds: Long, + @Value("\${app.http.client.general.pending-acquire-timeout-seconds:30}") + private val generalPendingAcquireTimeoutSeconds: Long, + @Value("\${app.http.client.general.connect-timeout-millis:5000}") + private val generalConnectTimeoutMillis: Int, + @Value("\${app.http.client.general.response-timeout-seconds:30}") + private val generalResponseTimeoutSeconds: Long, + @Value("\${app.http.client.keycloak.max-connections:150}") + private val keycloakMaxConnections: Int, + @Value("\${app.http.client.keycloak.pending-acquire-max-count:500}") + private val keycloakPendingAcquireMaxCount: Int, + @Value("\${app.http.client.keycloak.max-idle-seconds:30}") + private val keycloakMaxIdleSeconds: Long, + @Value("\${app.http.client.keycloak.max-life-seconds:120}") + private val keycloakMaxLifeSeconds: Long, + @Value("\${app.http.client.keycloak.pending-acquire-timeout-seconds:60}") + private val keycloakPendingAcquireTimeoutSeconds: Long, + @Value("\${app.http.client.keycloak.connect-timeout-millis:10000}") + private val keycloakConnectTimeoutMillis: Int, + @Value("\${app.http.client.keycloak.response-timeout-seconds:10}") + private val keycloakResponseTimeoutSeconds: Long, ) { private val provider = ConnectionProvider.builder("apiPool") - .maxConnections(150) - .pendingAcquireMaxCount(100) - .maxIdleTime(Duration.ofSeconds(30)) - .maxLifeTime(Duration.ofMinutes(2)) - .pendingAcquireTimeout(Duration.ofSeconds(10)) + .maxConnections(generalMaxConnections) + .pendingAcquireMaxCount(generalPendingAcquireMaxCount) + .maxIdleTime(Duration.ofSeconds(generalMaxIdleSeconds)) + .maxLifeTime(Duration.ofSeconds(generalMaxLifeSeconds)) + .pendingAcquireTimeout(Duration.ofSeconds(generalPendingAcquireTimeoutSeconds)) .evictInBackground(Duration.ofMinutes(1)) .build() - private val client = HttpClient.create(provider) - .wiretap("reactor.netty.http.client.HttpClient", LogLevel.DEBUG, AdvancedByteBufFormat.SIMPLE) - .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000) - .responseTimeout(Duration.ofSeconds(30)) - .keepAlive(true) - .doOnConnected { it.addHandlerLast(LogbookClientHandler(logbook)) } + private val client = HttpClient.create(provider).let { + val configured = if (wiretapEnabled) { + it.wiretap("reactor.netty.http.client.HttpClient", LogLevel.DEBUG, AdvancedByteBufFormat.SIMPLE) + } else { + it + } + configured + .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, generalConnectTimeoutMillis) + .responseTimeout(Duration.ofSeconds(generalResponseTimeoutSeconds)) + .keepAlive(true) + .doOnConnected { conn -> conn.addHandlerLast(LogbookClientHandler(logbook)) } + } @Bean("generalWebClient") @@ -63,16 +99,17 @@ class WebClientConfig( @Bean("keycloakWebClient") fun keycloakWebClient(logbook: Logbook): WebClient { val provider = ConnectionProvider.builder("keycloakPool") - .maxConnections(100) - .maxIdleTime(Duration.ofSeconds(30)) - .maxLifeTime(Duration.ofMinutes(2)) - .pendingAcquireTimeout(Duration.ofSeconds(60)) + .maxConnections(keycloakMaxConnections) + .pendingAcquireMaxCount(keycloakPendingAcquireMaxCount) + .maxIdleTime(Duration.ofSeconds(keycloakMaxIdleSeconds)) + .maxLifeTime(Duration.ofSeconds(keycloakMaxLifeSeconds)) + .pendingAcquireTimeout(Duration.ofSeconds(keycloakPendingAcquireTimeoutSeconds)) .evictInBackground(Duration.ofMinutes(1)) .build() val client = HttpClient.create(provider) - .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000) - .responseTimeout(Duration.ofSeconds(10)) + .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, keycloakConnectTimeoutMillis) + .responseTimeout(Duration.ofSeconds(keycloakResponseTimeoutSeconds)) .keepAlive(true) .doOnConnected { it.addHandlerLast(LogbookClientHandler(logbook)) } diff --git a/api/api-app/src/main/resources/application.yml b/api/api-app/src/main/resources/application.yml index ec0a0b34f..aa3606c00 100644 --- a/api/api-app/src/main/resources/application.yml +++ b/api/api-app/src/main/resources/application.yml @@ -107,10 +107,10 @@ logging: level: co.nilin: INFO org.zalando.logbook: TRACE - reactor.netty.pool: DEBUG - reactor.netty.http.client: DEBUG - org.springframework.web.reactive.function.client: DEBUG - co.nilin.opex.api.ports.proxy.impl: DEBUG + reactor.netty.pool: WARN + reactor.netty.http.client: WARN + org.springframework.web.reactive.function.client: WARN + co.nilin.opex.api.ports.proxy.impl: INFO app: base: @@ -152,6 +152,31 @@ app: api: crypto: key: ${api_crypto_key:0e1fd29572ec8c85970d76e3433e96ee} + http: + client: + wiretap: + enabled: ${HTTP_CLIENT_WIRETAP_ENABLED:false} + general: + max-connections: ${HTTP_CLIENT_GENERAL_MAX_CONNECTIONS:300} + pending-acquire-max-count: ${HTTP_CLIENT_GENERAL_PENDING_ACQUIRE_MAX_COUNT:1000} + max-idle-seconds: ${HTTP_CLIENT_GENERAL_MAX_IDLE_SECONDS:30} + max-life-seconds: ${HTTP_CLIENT_GENERAL_MAX_LIFE_SECONDS:120} + pending-acquire-timeout-seconds: ${HTTP_CLIENT_GENERAL_PENDING_ACQUIRE_TIMEOUT_SECONDS:30} + connect-timeout-millis: ${HTTP_CLIENT_GENERAL_CONNECT_TIMEOUT_MILLIS:5000} + response-timeout-seconds: ${HTTP_CLIENT_GENERAL_RESPONSE_TIMEOUT_SECONDS:30} + keycloak: + max-connections: ${HTTP_CLIENT_KEYCLOAK_MAX_CONNECTIONS:150} + pending-acquire-max-count: ${HTTP_CLIENT_KEYCLOAK_PENDING_ACQUIRE_MAX_COUNT:500} + max-idle-seconds: ${HTTP_CLIENT_KEYCLOAK_MAX_IDLE_SECONDS:30} + max-life-seconds: ${HTTP_CLIENT_KEYCLOAK_MAX_LIFE_SECONDS:120} + pending-acquire-timeout-seconds: ${HTTP_CLIENT_KEYCLOAK_PENDING_ACQUIRE_TIMEOUT_SECONDS:60} + connect-timeout-millis: ${HTTP_CLIENT_KEYCLOAK_CONNECT_TIMEOUT_MILLIS:10000} + response-timeout-seconds: ${HTTP_CLIENT_KEYCLOAK_RESPONSE_TIMEOUT_SECONDS:10} + proxy: + market: + max-concurrent-requests: ${API_PROXY_MARKET_MAX_CONCURRENT_REQUESTS:64} + matching: + max-concurrent-requests: ${API_PROXY_MATCHING_MAX_CONCURRENT_REQUESTS:64} cors: enabled: true allowed-origins: ${ALLOWED_ORIGINS:"http://localhost:8110"} @@ -170,4 +195,3 @@ springdoc: display-request-duration: true operations-sorter: method tags-sorter: alpha - diff --git a/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/config/ProxyDispatchers.kt b/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/config/ProxyDispatchers.kt index 1dbbda2aa..48e8aeecd 100644 --- a/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/config/ProxyDispatchers.kt +++ b/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/config/ProxyDispatchers.kt @@ -4,8 +4,30 @@ import kotlinx.coroutines.reactor.asCoroutineDispatcher import reactor.core.scheduler.Schedulers object ProxyDispatchers { + private fun envInt(name: String, default: Int): Int { + val value = System.getenv(name)?.toIntOrNull() ?: return default + return if (value > 0) value else default + } - val general = Schedulers.newBoundedElastic(8, 16, "general").asCoroutineDispatcher() - val market = Schedulers.newBoundedElastic(8, 16, "market").asCoroutineDispatcher() - val wallet = Schedulers.newBoundedElastic(10, 20, "wallet").asCoroutineDispatcher() + private val cpu = Runtime.getRuntime().availableProcessors().coerceAtLeast(4) + private val defaultThreads = cpu * 4 + private val defaultQueue = 10_000 + + val general = Schedulers.newBoundedElastic( + envInt("API_PROXY_GENERAL_THREADS", defaultThreads), + envInt("API_PROXY_GENERAL_QUEUE", defaultQueue), + "general" + ).asCoroutineDispatcher() + + val market = Schedulers.newBoundedElastic( + envInt("API_PROXY_MARKET_THREADS", defaultThreads), + envInt("API_PROXY_MARKET_QUEUE", defaultQueue), + "market" + ).asCoroutineDispatcher() + + val wallet = Schedulers.newBoundedElastic( + envInt("API_PROXY_WALLET_THREADS", defaultThreads), + envInt("API_PROXY_WALLET_QUEUE", defaultQueue), + "wallet" + ).asCoroutineDispatcher() } \ No newline at end of file diff --git a/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MarketUserDataProxyImpl.kt b/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MarketUserDataProxyImpl.kt index b40cfdc03..f67f0c113 100644 --- a/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MarketUserDataProxyImpl.kt +++ b/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MarketUserDataProxyImpl.kt @@ -34,6 +34,9 @@ class MarketUserDataProxyImpl(@Qualifier("generalWebClient") private val webClie @Value("\${app.market.url}") private lateinit var baseUrl: String + + @Value("\${app.proxy.market.max-concurrent-requests:64}") + private var marketMaxConcurrentRequests: Int = 64 private suspend fun retryOnce(backoffMs: Long = 200, block: suspend () -> T): T = try { block() @@ -41,7 +44,9 @@ class MarketUserDataProxyImpl(@Qualifier("generalWebClient") private val webClie delay(backoffMs); block() } - private val mgLimiter = Semaphore(permits = 16, acquiredPermits = 0) + private val mgLimiter by lazy { + Semaphore(permits = marketMaxConcurrentRequests, acquiredPermits = 0) + } override suspend fun queryOrder( token: String, diff --git a/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MatchingGatewayProxyImpl.kt b/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MatchingGatewayProxyImpl.kt index a0718dba0..58ef1533d 100644 --- a/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MatchingGatewayProxyImpl.kt +++ b/api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/MatchingGatewayProxyImpl.kt @@ -39,7 +39,13 @@ class MatchingGatewayProxyImpl(@Qualifier("generalWebClient") private val client @Value("\${app.matching-gateway.url}") private lateinit var baseUrl: String - private val mgLimiter = Semaphore(permits = 16, acquiredPermits = 0) // fair-like behavior + + @Value("\${app.proxy.matching.max-concurrent-requests:64}") + private var matchingMaxConcurrentRequests: Int = 64 + + private val mgLimiter by lazy { + Semaphore(permits = matchingMaxConcurrentRequests, acquiredPermits = 0) + } override suspend fun createNewOrder( uuid: String?, pair: String, diff --git a/device-management/pom.xml b/device-management/pom.xml index fc80e43f7..0923d7cdd 100644 --- a/device-management/pom.xml +++ b/device-management/pom.xml @@ -22,7 +22,7 @@ 2.1.0 3.4.2 2024.0.0 - 1.2.25 + 1.2.26 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 { diff --git a/pom.xml b/pom.xml index 853d3c8d5..e49130f86 100644 --- a/pom.xml +++ b/pom.xml @@ -16,7 +16,7 @@ 1.9.0 2.7.6 2021.0.5 - 1.2.25 + 1.2.26 1.0.8 true 1.0.1-beta.38 diff --git a/wallet/wallet-app/src/main/resources/application.yml b/wallet/wallet-app/src/main/resources/application.yml index 488d21278..ec67e2e77 100644 --- a/wallet/wallet-app/src/main/resources/application.yml +++ b/wallet/wallet-app/src/main/resources/application.yml @@ -8,6 +8,8 @@ management: endpoint: health: show-details: when_authorized + probes: + enabled: true metrics: enabled: true prometheus: @@ -66,9 +68,10 @@ spring: host: ${CONSUL_HOST:localhost} port: 8500 discovery: - #healthCheckPath: ${management.context-path}/health instance-id: ${spring.application.name}:${server.port} - healthCheckInterval: 20s + health-check-path: /actuator/health/liveness + health-check-interval: 10s + health-check-timeout: 5s prefer-ip-address: true config: import: vault://secret/${spring.application.name} @@ -174,6 +177,4 @@ logging: co.nilin: INFO reactor.netty.http.client: INFO org.zalando.logbook: TRACE - org.hibernate.SQL: DEBUG - logging.level.org.hibernate.type.descriptor.sql.BasicBinder: TRACE swagger.authUrl: ${SWAGGER_AUTH_URL:https://api.opex.dev/auth}/realms/opex/protocol/openid-connect/token