Add quote currency management feature

api/api-core/src/main/kotlin/co/nilin/opex/api/core/inout/QuoteCurrency.kt

@@ -0,0 +1,9 @@
+package co.nilin.opex.api.core.inout
+
+import java.time.LocalDateTime
+
+data class QuoteCurrency(
+    val currency: String,
+    val isActive: Boolean = false,
+    var lastUpdateDate: LocalDateTime = LocalDateTime.now(),
+)

api/api-core/src/main/kotlin/co/nilin/opex/api/core/spi/WalletProxy.kt

@@ -1,7 +1,6 @@
 package co.nilin.opex.api.core.spi
 
 import co.nilin.opex.api.core.inout.*
-import java.math.BigDecimal
 
 interface WalletProxy {
 
@@ -83,17 +82,19 @@ interface WalletProxy {
     suspend fun requestWithdraw(
         token: String,
         request: RequestWithdrawBody
-    ):WithdrawActionResult
+    ): WithdrawActionResult
 
     suspend fun cancelWithdraw(
         token: String,
         withdrawId: Long
-    ):Void?
+    ): Void?
 
     suspend fun findWithdraw(
         token: String,
         withdrawId: Long
     ): WithdrawResponse
 
-    suspend fun submitVoucher(code : String , token :String) : SubmitVoucherResponse
+    suspend fun submitVoucher(code: String, token: String): SubmitVoucherResponse
+
+    suspend fun getQuoteCurrencies(): List<QuoteCurrency>
 }

api/api-ports/api-opex-rest/src/main/kotlin/co/nilin/opex/api/ports/opex/controller/MarketController.kt

@@ -201,9 +201,7 @@ class MarketController(
 
     @GetMapping("/currencyInfo/quotes")
     suspend fun getQuoteCurrencies(): List<String> {
-        return accountantProxy.getPairConfigs()
-            .map { it.rightSideWalletSymbol }
-            .distinct()
+        return walletProxy.getQuoteCurrencies().map { it.currency }
     }
 
     @GetMapping("/klines")

api/api-ports/api-proxy-rest/src/main/kotlin/co/nilin/opex/api/ports/proxy/impl/WalletProxyImpl.kt

@@ -332,5 +332,20 @@ class WalletProxyImpl(private val webClient: WebClient) : WalletProxy {
             .bodyToMono<SubmitVoucherResponse>()
             .awaitFirstOrElse { throw OpexError.BadRequest.exception() }
     }
+
+    override suspend fun getQuoteCurrencies(): List<QuoteCurrency> {
+        return withContext(ProxyDispatchers.wallet) {
+            webClient.get()
+                .uri("$baseUrl/currency/quotes") {
+                    it.queryParam("isActive", true)
+                    it.build()
+                }.accept(MediaType.APPLICATION_JSON)
+                .retrieve()
+                .onStatus({ t -> t.isError }, { it.createException() })
+                .bodyToFlux<QuoteCurrency>()
+                .collectList()
+                .awaitFirstOrElse { emptyList() }
+        }
+    }
 }
 

docker-compose.yml

@@ -24,8 +24,6 @@ services:
       - KAFKA_PROCESS_ROLES=broker,controller
       - CLUSTER_ID=${KAFKA_CLUSTER_ID}
       - ALLOW_PLAINTEXT_LISTENER=yes
-      - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CLIENT:PLAINTEXT,EXTERNAL:PLAINTEXT
-      - KAFKA_LISTENERS=CLIENT://kafka-1:29092,EXTERNAL://kafka-1:9092
       - KAFKA_ADVERTISED_LISTENERS=CLIENT://kafka-1:29092,EXTERNAL://kafka-1:9092
       - KAFKA_INTER_BROKER_LISTENER_NAME=CLIENT
       - KAFKA_UNCLEAN_LEADER_ELECTION_ENABLE=false
@@ -50,8 +48,6 @@ services:
       - KAFKA_PROCESS_ROLES=broker,controller
       - CLUSTER_ID=${KAFKA_CLUSTER_ID}
       - ALLOW_PLAINTEXT_LISTENER=yes
-      - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CLIENT:PLAINTEXT,EXTERNAL:PLAINTEXT
-      - KAFKA_LISTENERS=CLIENT://kafka-2:29092,EXTERNAL://kafka-2:9092
       - KAFKA_ADVERTISED_LISTENERS=CLIENT://kafka-2:29092,EXTERNAL://kafka-2:9092
       - KAFKA_INTER_BROKER_LISTENER_NAME=CLIENT
       - KAFKA_UNCLEAN_LEADER_ELECTION_ENABLE=false
@@ -76,8 +72,6 @@ services:
       - KAFKA_PROCESS_ROLES=broker,controller
       - CLUSTER_ID=${KAFKA_CLUSTER_ID}
       - ALLOW_PLAINTEXT_LISTENER=yes
-      - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CLIENT:PLAINTEXT,EXTERNAL:PLAINTEXT
-      - KAFKA_LISTENERS=CLIENT://kafka-3:29092,EXTERNAL://kafka-3:9092
       - KAFKA_ADVERTISED_LISTENERS=CLIENT://kafka-3:29092,EXTERNAL://kafka-3:9092
       - KAFKA_INTER_BROKER_LISTENER_NAME=CLIENT
       - KAFKA_UNCLEAN_LEADER_ELECTION_ENABLE=false
@@ -419,6 +413,7 @@ services:
       - SWAGGER_AUTH_URL=$KEYCLOAK_FRONTEND_URL
       - DRIVE_FOLDER_ID=$DRIVE_FOLDER_ID
       - BACKUP_ENABLED=$WALLET_BACKUP_ENABLED
+      - SYMBOLS=BTC_USDT,ETH_USDT,BTC_IRT,ETH_IRT,USDT_IRT,ETH_BUSD,BTC_BUSD,BNB_BUSD
     depends_on:
       - kafka-1
       - kafka-2

wallet/wallet-app/src/main/kotlin/co/nilin/opex/wallet/app/controller/CurrencyController.kt

@@ -7,14 +7,17 @@ import co.nilin.opex.wallet.core.inout.CurrencyData
 import co.nilin.opex.wallet.core.inout.CurrencyGatewayCommand
 import co.nilin.opex.wallet.core.inout.GatewayType
 import co.nilin.opex.wallet.core.inout.TerminalCommand
+import co.nilin.opex.wallet.core.model.QuoteCurrency
 import co.nilin.opex.wallet.core.spi.GatewayTerminalManager
+import co.nilin.opex.wallet.core.spi.QuoteCurrencyManager
 import org.springframework.web.bind.annotation.*
 
 @RestController
 @RequestMapping("/currency")
 class CurrencyController(
     private val currencyService: CurrencyServiceV2,
     private val gatewayTerminalManager: GatewayTerminalManager,
+    private val quoteCurrencyManager: QuoteCurrencyManager,
 ) {
 
     @PostMapping("")
@@ -139,4 +142,19 @@ class CurrencyController(
         return gatewayTerminalManager.revokeTerminalsToGateway(gatewayUuid, terminal)
     }
 
+    @GetMapping("/quotes")
+    suspend fun getQuoteCurrencies(
+        @RequestParam isActive: Boolean?,
+    ): List<QuoteCurrency> {
+        return quoteCurrencyManager.getAll(isActive)
+    }
+
+    @PutMapping("/quote/{currency}")
+    suspend fun updateQuoteCurrency(
+        @PathVariable("currency") currency: String,
+        @RequestParam isActive: Boolean,
+    ) {
+        quoteCurrencyManager.update(currency, isActive)
+    }
+
 }

wallet/wallet-app/src/main/resources/application.yml

@@ -121,6 +121,7 @@ app:
     url: lb://opex-bc-gateway
   reserved-transfer:
     life-time: 15 #minutes
+  symbols: ${SYMBOLS}
 logging:
   level:
     root: INFO

wallet/wallet-app/src/test/resources/application.yml

@@ -48,6 +48,7 @@ spring:
     vault:
       enabled: false
 app:
+  symbols: BTC_USDT,ETH_USDT,BTC_IRT,ETH_IRT,USDT_IRT,ETH_BUSD,BTC_BUSD,BNB_BUSD
   auth:
     url: none
     cert-url: none

wallet/wallet-core/src/main/kotlin/co/nilin/opex/wallet/core/model/QuoteCurrency.kt

@@ -0,0 +1,9 @@
+package co.nilin.opex.wallet.core.model
+
+import java.time.LocalDateTime
+
+data class QuoteCurrency(
+    val currency: String,
+    val isActive: Boolean = false,
+    var lastUpdateDate: LocalDateTime = LocalDateTime.now(),
+)

wallet/wallet-core/src/main/kotlin/co/nilin/opex/wallet/core/spi/QuoteCurrencyManager.kt

@@ -0,0 +1,9 @@
+package co.nilin.opex.wallet.core.spi
+
+import co.nilin.opex.wallet.core.model.QuoteCurrency
+
+interface QuoteCurrencyManager {
+
+    suspend fun getAll(isActive: Boolean?): List<QuoteCurrency>
+    suspend fun update(currency: String, isActive: Boolean)
+}

wallet/wallet-ports/wallet-persister-postgres/src/main/kotlin/co/nilin/opex/wallet/ports/postgres/dao/QuoteCurrencyRepository.kt

@@ -0,0 +1,20 @@
+package co.nilin.opex.wallet.ports.postgres.dao
+
+import co.nilin.opex.wallet.core.model.QuoteCurrency
+import co.nilin.opex.wallet.ports.postgres.model.QuoteCurrencyModel
+import kotlinx.coroutines.flow.Flow
+import org.springframework.data.r2dbc.repository.Query
+import org.springframework.data.repository.reactive.ReactiveCrudRepository
+import org.springframework.stereotype.Repository
+import reactor.core.publisher.Mono
+import java.time.LocalDateTime
+
+@Repository
+interface QuoteCurrencyRepository : ReactiveCrudRepository<QuoteCurrencyModel, Long> {
+
+    fun findByCurrency(currency: String): Mono<QuoteCurrencyModel>
+
+    @Query("SELECT * FROM quote_currency WHERE (:isActive IS NULL OR is_active = :isActive)")
+    fun findAllByActive(isActive: Boolean?): Flow<QuoteCurrency>
+
+}

wallet/wallet-ports/wallet-persister-postgres/src/main/kotlin/co/nilin/opex/wallet/ports/postgres/impl/QuoteCurrencyManagerImpl.kt

@@ -0,0 +1,34 @@
+package co.nilin.opex.wallet.ports.postgres.impl
+
+import co.nilin.opex.wallet.core.model.QuoteCurrency
+import co.nilin.opex.wallet.core.spi.QuoteCurrencyManager
+import co.nilin.opex.wallet.ports.postgres.dao.QuoteCurrencyRepository
+import co.nilin.opex.wallet.ports.postgres.model.QuoteCurrencyModel
+import kotlinx.coroutines.flow.toList
+import kotlinx.coroutines.reactive.awaitFirstOrNull
+import org.springframework.stereotype.Service
+import java.time.LocalDateTime
+
+@Service
+class QuoteCurrencyManagerImpl(
+    private val quoteCurrencyRepository: QuoteCurrencyRepository
+) : QuoteCurrencyManager {
+
+    override suspend fun getAll(isActive: Boolean?): List<QuoteCurrency> {
+        return quoteCurrencyRepository.findAllByActive(isActive).toList()
+    }
+
+    override suspend fun update(currency: String, isActive: Boolean) {
+        quoteCurrencyRepository.findByCurrency(currency).awaitFirstOrNull()?.let { quoteCurrency ->
+            quoteCurrencyRepository.save(
+                QuoteCurrencyModel(
+                    quoteCurrency.id,
+                    currency,
+                    isActive,
+                    LocalDateTime.now(),
+                )
+            ).awaitFirstOrNull()
+        }
+
+    }
+}

wallet/wallet-ports/wallet-persister-postgres/src/main/kotlin/co/nilin/opex/wallet/ports/postgres/model/QuoteCurrencyModel.kt

@@ -0,0 +1,14 @@
+package co.nilin.opex.wallet.ports.postgres.model
+
+import org.springframework.data.annotation.Id
+import org.springframework.data.relational.core.mapping.Table
+import java.time.LocalDateTime
+
+@Table("quote_currency")
+data class QuoteCurrencyModel(
+    @Id
+    val id: Long? = null,
+    val currency: String,
+    val isActive: Boolean = false,
+    var lastUpdateDate: LocalDateTime = LocalDateTime.now(),
+)

wallet/wallet-ports/wallet-persister-postgres/src/main/kotlin/co/nilin/opex/wallet/ports/postgres/service/QuoteCurrencyInitializer.kt

@@ -0,0 +1,66 @@
+package co.nilin.opex.wallet.ports.postgres.service
+
+import co.nilin.opex.wallet.ports.postgres.dao.QuoteCurrencyRepository
+import co.nilin.opex.wallet.ports.postgres.model.QuoteCurrencyModel
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.reactive.awaitFirstOrNull
+import org.slf4j.LoggerFactory
+import org.springframework.beans.factory.annotation.Value
+import org.springframework.stereotype.Service
+import java.time.LocalDateTime
+import javax.annotation.PostConstruct
+
+@Service
+class QuoteCurrencyInitializer(
+    private val quoteCurrencyRepository: QuoteCurrencyRepository,
+    @Value("\${app.symbols}")
+    private val symbols: String
+) {
+
+    private val logger = LoggerFactory.getLogger(QuoteCurrencyInitializer::class.java)
+    val scope = CoroutineScope(Dispatchers.IO)
+
+    @PostConstruct
+    fun initialize() {
+        logger.info(
+            """
+================================================================================================
+                                Initialize Quote Currencies
+================================================================================================
+        """
+        )
+        scope.launch {
+            try {
+
+                val quoteSymbols = symbols.split(",")
+                    .mapNotNull { symbol ->
+                        val quote = symbol.substringAfter('_', "")
+                        if (quote.isNotBlank()) quote else null
+                    }
+                    .toSet()
+
+                quoteSymbols.forEach { quote ->
+                    val existing = quoteCurrencyRepository.findByCurrency(quote).awaitFirstOrNull()
+                    if (existing == null) {
+                        quoteCurrencyRepository.save(
+                            QuoteCurrencyModel(null, quote, false, LocalDateTime.now())
+                        ).awaitFirstOrNull()
+                        logger.info("Quote currency inserted: $quote")
+                    }
+                }
+                logger.info(
+                    """
+================================================================================================
+                                 Completed Successfully
+================================================================================================
+            """
+                )
+            } catch (e: Exception) {
+                logger.error("Error initializing Quote Currencies: ${e.message}")
+                throw e
+            }
+        }
+    }
+}

wallet/wallet-ports/wallet-persister-postgres/src/main/resources/schema.sql

@@ -487,4 +487,12 @@ $$
             ADD COLUMN transfer_method VARCHAR(255);
         END IF;
     END
-$$;
+$$;
+
+CREATE TABLE IF NOT EXISTS quote_currency
+(
+    id          SERIAL PRIMARY KEY,
+    currency    VARCHAR(25) NOT NULL UNIQUE REFERENCES currency (symbol),
+    is_active         BOOLEAN      NOT NULL DEFAULT false,
+    last_update_date TIMESTAMP
+);