add nbkz source

README.md

@@ -1,13 +1,29 @@
-# rate-boss![]
-![Rate-boss scheme](doc/rate-boss.png)
+# rate-boss
 
-The essence of the service is quite simple. It collects exchange rate data from some source and stores it in a kafka topic
+Rate-boss periodically pulls exchange rates from external sources, stores them in the database, and optionally publishes rate events to Kafka. It also exposes API endpoints to read the latest or historical rates and to convert amounts by timestamp.
 
-Currency exchange rate is collected according to a specific schedule cron, which we specify in the configuration file.
-We use quartz cluster mode to support the work of our service in multi node mode.
+Supported sources today: Fixer, CBR, NBKZ.
 
-![Rate-boss scheme](doc/rate-boss-quartz.png)
+## Service flow (actual)
 
-Quartz execute like this. First, we have `ExchangeGrabberMasterJob` its job is to take the list of currencies from the configuration
-and create jobs `ExchangeGrabberJob` for each currency. `ExchangeGrabberJob` makes a request to a certain source of exchange rates
-and send the received rates into kafka topic.
+```mermaid
+flowchart TD
+  subgraph Quartz Scheduler
+    M[ExchangeGrabberMasterJob<br/>per source] -->|per currency| J[ExchangeGrabberJob]
+  end
+
+  J -->|HTTP fetch| S[ExchangeRateSource<br/>Fixer / CBR / NBKZ]
+  S -->|rates + timestamp| D[ExchangeDaoService]
+  D -->|save batch| DB[(PostgreSQL<br/>rb.ex_rate)]
+  J -->|optional publish| E[ExchangeEventService]
+  E --> K[(Kafka topic)]
+
+  API[ExchangeRateServiceSrv] -->|getExchangeRateData / getConvertedAmount| DB
+```
+
+## Quartz execution
+
+For each configured source:
+- `ExchangeGrabberMasterJob` reads the list of base currencies from `rates.*Job.currencies`.
+- It spawns `ExchangeGrabberJob` per currency.
+- Each `ExchangeGrabberJob` calls the respective `ExchangeRateSource`, saves the rates, and (for some sources) emits Kafka events.

src/main/kotlin/dev/vality/rateboss/client/nbkz/NbkzApiClient.kt

@@ -0,0 +1,33 @@
+package dev.vality.rateboss.client.nbkz
+
+import dev.vality.rateboss.config.properties.RatesProperties
+import org.springframework.http.HttpEntity
+import org.springframework.http.HttpMethod
+import org.springframework.stereotype.Component
+import org.springframework.web.client.RestTemplate
+import org.springframework.web.client.exchange
+import org.springframework.web.util.UriComponentsBuilder
+import java.time.LocalDate
+import java.time.format.DateTimeFormatter
+
+@Component
+class NbkzApiClient(
+    private val restTemplate: RestTemplate,
+    private val ratesProperties: RatesProperties,
+) {
+    fun getExchangeRates(date: LocalDate): String {
+        val url = buildUrl(ratesProperties.source.nbkz.rootUrl, date, ratesProperties.source.nbkz.dateFormat)
+        return restTemplate.exchange<String>(url, HttpMethod.GET, HttpEntity(null, null)).body!!
+    }
+
+    private fun buildUrl(
+        endpoint: String,
+        date: LocalDate,
+        dateFormat: String,
+    ): String =
+        UriComponentsBuilder
+            .fromUriString(endpoint)
+            .queryParam("fdate", date.format(DateTimeFormatter.ofPattern(dateFormat)))
+            .build()
+            .toUriString()
+}

src/main/kotlin/dev/vality/rateboss/config/JobConfig.kt

@@ -3,6 +3,7 @@ package dev.vality.rateboss.config
 import dev.vality.rateboss.config.properties.RatesProperties
 import dev.vality.rateboss.job.CbrExchangeGrabberMasterJob
 import dev.vality.rateboss.job.FixerExchangeGrabberMasterJob
+import dev.vality.rateboss.job.NbkzExchangeGrabberMasterJob
 import org.quartz.*
 import org.quartz.impl.JobDetailImpl
 import org.springframework.beans.factory.annotation.Autowired
@@ -33,6 +34,13 @@ class JobConfig {
                 schedulerFactoryBean.scheduleJob(cbrExchangeRateGrabberMasterTrigger())
             }
         }
+        val nbkzJobTriggerName = ratesProperties.nbkzJob.jobTriggerName
+        if (nbkzJobTriggerName.isNotEmpty()) {
+            schedulerFactoryBean.addJob(nbkzExchangeRateGrabberMasterJob(), true, true)
+            if (!schedulerFactoryBean.checkExists(TriggerKey(ratesProperties.nbkzJob.jobTriggerName))) {
+                schedulerFactoryBean.scheduleJob(nbkzExchangeRateGrabberMasterTrigger())
+            }
+        }
     }
 
     fun fixerExchangeRateGrabberMasterJob(): JobDetailImpl {
@@ -64,4 +72,19 @@ class JobConfig {
             .withIdentity(ratesProperties.cbrJob.jobTriggerName)
             .withSchedule(CronScheduleBuilder.cronSchedule(ratesProperties.cbrJob.jobCron))
             .build()
+
+    fun nbkzExchangeRateGrabberMasterJob(): JobDetailImpl {
+        val jobDetail = JobDetailImpl()
+        jobDetail.key = JobKey(ratesProperties.nbkzJob.jobKey)
+        jobDetail.jobClass = NbkzExchangeGrabberMasterJob::class.java
+        return jobDetail
+    }
+
+    fun nbkzExchangeRateGrabberMasterTrigger(): CronTrigger =
+        TriggerBuilder
+            .newTrigger()
+            .forJob(nbkzExchangeRateGrabberMasterJob())
+            .withIdentity(ratesProperties.nbkzJob.jobTriggerName)
+            .withSchedule(CronScheduleBuilder.cronSchedule(ratesProperties.nbkzJob.jobCron))
+            .build()
 }

src/main/kotlin/dev/vality/rateboss/config/properties/RatesProperties.kt

@@ -9,6 +9,7 @@ import java.time.ZoneId
 data class RatesProperties(
     val fixerJob: JobDescription,
     val cbrJob: JobDescription,
+    val nbkzJob: JobDescription,
     val source: RatesSourceProperties,
 )
 
@@ -27,6 +28,7 @@ data class CurrencyProperties(
 data class RatesSourceProperties(
     val fixer: FixerProperties,
     val cbr: CbrProperties,
+    val nbkz: NbkzProperties,
 )
 
 data class FixerProperties(
@@ -38,3 +40,9 @@ data class CbrProperties(
     val rootUrl: String,
     val timeZone: ZoneId,
 )
+
+data class NbkzProperties(
+    val rootUrl: String,
+    val dateFormat: String,
+    val timeZone: ZoneId,
+)

src/main/kotlin/dev/vality/rateboss/job/NbkzExchangeGrabberJob.kt

@@ -0,0 +1,37 @@
+package dev.vality.rateboss.job
+
+import dev.vality.rateboss.extensions.getApplicationContext
+import dev.vality.rateboss.source.ExchangeRateSource
+import dev.vality.rateboss.source.ExchangeRateSourceException
+import dev.vality.rateboss.source.impl.NbkzExchangeRateSource
+import dev.vality.rateboss.source.model.ExchangeRates
+import mu.KotlinLogging
+import org.quartz.JobExecutionContext
+import org.springframework.context.ApplicationContext
+import org.springframework.retry.support.RetryTemplate
+
+private val log = KotlinLogging.logger {}
+
+class NbkzExchangeGrabberJob : AbstractExchangeGrabberJob() {
+    override fun executeInternal(context: JobExecutionContext) {
+        val applicationContext = context.getApplicationContext()
+        val currencySymbolCode = context.jobDetail.jobDataMap["currencySymbolCode"] as String
+        val currencyExponent = context.jobDetail.jobDataMap["currencyExponent"] as Int
+        val exchangeRateSource = applicationContext.getBean(NbkzExchangeRateSource::class.java)
+        val sourceId = exchangeRateSource.getSourceId()
+        log.info { "Process NbkzExchangeGrabberJob for $sourceId" }
+        val exchangeRates = getExchangeRates(applicationContext, exchangeRateSource, currencySymbolCode)
+        saveExchangeRates(applicationContext, currencySymbolCode, currencyExponent, exchangeRates, sourceId)
+    }
+
+    private fun getExchangeRates(
+        applicationContext: ApplicationContext,
+        exchangeRateSource: ExchangeRateSource,
+        currencySymbolCode: String,
+    ): ExchangeRates {
+        val retryTemplate = applicationContext.getBean(RetryTemplate::class.java)
+        return retryTemplate.execute<ExchangeRates, ExchangeRateSourceException> {
+            exchangeRateSource.getExchangeRate(currencySymbolCode)
+        }
+    }
+}

src/main/kotlin/dev/vality/rateboss/job/NbkzExchangeGrabberMasterJob.kt

@@ -0,0 +1,18 @@
+package dev.vality.rateboss.job
+
+import dev.vality.rateboss.config.properties.RatesProperties
+import dev.vality.rateboss.extensions.getApplicationContext
+import org.quartz.JobExecutionContext
+import org.quartz.Scheduler
+
+class NbkzExchangeGrabberMasterJob : AbstractExchangeGrabberMasterJob() {
+    override fun executeInternal(context: JobExecutionContext) {
+        val applicationContext = context.getApplicationContext()
+        val ratesProperties = applicationContext.getBean(RatesProperties::class.java)
+        val currencies = ratesProperties.nbkzJob.currencies
+        val schedulerFactoryBean = applicationContext.getBean(Scheduler::class.java)
+        launchJob(currencies, schedulerFactoryBean, NbkzExchangeGrabberJob::class.java, getJobName())
+    }
+
+    override fun getJobName(): String = "nbkzJob"
+}

src/main/kotlin/dev/vality/rateboss/job/constant/ExRateSources.kt

@@ -3,4 +3,5 @@ package dev.vality.rateboss.job.constant
 object ExRateSources {
     const val FIXER = "fixer"
     const val CBR = "cbr"
+    const val NBKZ = "nbkz"
 }

src/main/kotlin/dev/vality/rateboss/source/impl/NbkzExchangeRateSource.kt

@@ -0,0 +1,79 @@
+package dev.vality.rateboss.source.impl
+
+import dev.vality.rateboss.client.nbkz.NbkzApiClient
+import dev.vality.rateboss.config.properties.RatesProperties
+import dev.vality.rateboss.job.constant.ExRateSources
+import dev.vality.rateboss.source.ExchangeRateSource
+import dev.vality.rateboss.source.ExchangeRateSourceException
+import dev.vality.rateboss.source.model.ExchangeRates
+import mu.KotlinLogging
+import org.springframework.stereotype.Component
+import org.w3c.dom.Element
+import org.w3c.dom.Node
+import java.io.ByteArrayInputStream
+import java.math.BigDecimal
+import java.nio.charset.StandardCharsets
+import java.time.LocalDate
+import java.time.ZoneOffset
+import javax.xml.parsers.DocumentBuilderFactory
+
+private val log = KotlinLogging.logger {}
+
+@Component
+class NbkzExchangeRateSource(
+    private val nbkzApiClient: NbkzApiClient,
+    private val ratesProperties: RatesProperties,
+) : ExchangeRateSource {
+    override fun getExchangeRate(currencySymbolCode: String): ExchangeRates {
+        val timeZone = ratesProperties.source.nbkz.timeZone
+        val date = LocalDate.now(timeZone)
+        log.info { "Trying to get exchange rates from nbkz for currency=$currencySymbolCode, date=$date" }
+        val response =
+            try {
+                nbkzApiClient.getExchangeRates(date)
+            } catch (e: Exception) {
+                throw ExchangeRateSourceException("Remote client exception", e)
+            }
+        val rates =
+            try {
+                parseRates(response)
+            } catch (e: Exception) {
+                throw ExchangeRateSourceException("Failed to parse response from NbkzApi", e)
+            }
+        if (rates.isEmpty()) {
+            throw ExchangeRateSourceException("Unsuccessful response from NbkzApi")
+        }
+        val nextDayTimestamp = date.plusDays(1).atStartOfDay().toEpochSecond(ZoneOffset.UTC)
+        log.info { "Exchange rates from nbkz have been retrieved, date=$date, exchangeRates=$rates, targetTimestamp=$nextDayTimestamp" }
+        return ExchangeRates(
+            rates = rates,
+            timestamp = nextDayTimestamp,
+        )
+    }
+
+    override fun getSourceId(): String = ExRateSources.NBKZ
+
+    private fun parseRates(xmlContent: String): Map<String, BigDecimal> {
+        val factory = DocumentBuilderFactory.newInstance()
+        val builder = factory.newDocumentBuilder()
+        val document = builder.parse(ByteArrayInputStream(xmlContent.toByteArray(StandardCharsets.UTF_8)))
+        val items = document.getElementsByTagName("item")
+        val rates = mutableMapOf<String, BigDecimal>()
+        for (i in 0 until items.length) {
+            val itemNode = items.item(i)
+            if (itemNode.nodeType != Node.ELEMENT_NODE) {
+                continue
+            }
+            val item = itemNode as Element
+            val titleNodes = item.getElementsByTagName("title")
+            val descriptionNodes = item.getElementsByTagName("description")
+            if (titleNodes.length == 0 || descriptionNodes.length == 0) {
+                continue
+            }
+            val currencyCode = titleNodes.item(0).textContent.trim { it <= ' ' }
+            val rateStr = descriptionNodes.item(0).textContent.trim { it <= ' ' }
+            rates[currencyCode] = BigDecimal(rateStr)
+        }
+        return rates
+    }
+}

src/main/resources/application.yml

@@ -90,10 +90,21 @@ rates:
     currencies:
       - symbolCode: "RUB"
         exponent: 2
+  nbkzJob:
+    jobCron: '0 0 0/1 * * ?'
+    jobKey: 'nbkz-exchange-rate-grabber-master-job'
+    jobTriggerName: 'nbkz-exchange-rate-grabber-master-job-trigger'
+    currencies:
+      - symbolCode: "KZT"
+        exponent: 2
   source:
     fixer:
       rootUrl: https://api.apilayer.com/fixer/
       apiKey: CJjxZVaxXZCJCfTzkKFQQP0GzP1ytmB2
     cbr:
       rootUrl: https://www.cbr.ru/scripts/XML_daily.asp
       timeZone: Europe/Moscow
+    nbkz:
+      rootUrl: https://nationalbank.kz/rss/get_rates.cfm
+      dateFormat: dd.MM.yyyy
+      timeZone: Asia/Almaty

src/test/kotlin/dev/vality/rateboss/client/nbkz/NbkzApiClientTest.kt

@@ -0,0 +1,43 @@
+package dev.vality.rateboss.client.nbkz
+
+import dev.vality.rateboss.config.TestConfig
+import dev.vality.rateboss.source.ExchangeRateSource
+import dev.vality.rateboss.source.impl.NbkzExchangeRateSource
+import org.junit.jupiter.api.Assertions.assertNotNull
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Disabled
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.extension.ExtendWith
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.context.annotation.Import
+import org.springframework.test.context.ContextConfiguration
+import org.springframework.test.context.junit.jupiter.SpringExtension
+import java.time.LocalDate
+
+@Disabled("integration test")
+@ExtendWith(SpringExtension::class)
+@ContextConfiguration(classes = [NbkzApiClient::class, NbkzExchangeRateSource::class])
+@Import(TestConfig::class)
+class NbkzApiClientTest {
+    @Autowired
+    lateinit var nbkzApiClient: NbkzApiClient
+
+    @Autowired
+    lateinit var nbkzExchangeRateSource: ExchangeRateSource
+
+    @Test
+    fun getExchangeRates() {
+        val response = nbkzApiClient.getExchangeRates(LocalDate.now())
+
+        assertNotNull(response)
+        assertTrue(response.isNotBlank())
+    }
+
+    @Test
+    fun getExchangeRatesViaSource() {
+        val exchangeRates = nbkzExchangeRateSource.getExchangeRate("KZT")
+
+        assertNotNull(exchangeRates)
+        assertTrue(exchangeRates.rates.isNotEmpty())
+    }
+}

src/test/kotlin/dev/vality/rateboss/config/TestConfig.kt

@@ -26,9 +26,16 @@ class TestConfig {
                 "cbr-name",
                 listOf(CurrencyProperties("RUB", 2)),
             ),
+            JobDescription(
+                "nbkz-cron",
+                "nbkz-key",
+                "nbkz-name",
+                listOf(CurrencyProperties("KZT", 2)),
+            ),
             RatesSourceProperties(
                 FixerProperties("url", "key"),
                 CbrProperties("https://www.cbr.ru/scripts/XML_daily.asp", ZoneId.of("Europe/Moscow")),
+                NbkzProperties("https://nationalbank.kz/rss/get_rates.cfm", "dd.MM.yyyy", ZoneId.of("Asia/Almaty")),
             ),
         )
 }