Dev

accountant/accountant-app/src/main/kotlin/co/nilin/opex/accountant/app/scheduler/FinancialActionsArchiveJob.kt

@@ -25,15 +25,26 @@ class FinancialActionsArchiveJob(
     @Value("\${app.fi-action.archive.batch-size:1000}")
     private var batchSize: Int = 1000
 
+    @Value("\${app.fi-action.archive.max-batches-per-run:20}")
+    private var maxBatchesPerRun: Int = 20
+
     @Scheduled(fixedDelayString = "\${app.fi-action.archive.fixed-delay-ms:300000}", initialDelay = 60000)
     fun archiveProcessedActions() {
-        if (!enabled || batchSize <= 0 || retentionDays <= 0) return
+        if (!enabled || batchSize <= 0 || retentionDays <= 0 || maxBatchesPerRun <= 0) return
 
         runBlocking {
             val before = LocalDateTime.now().minusDays(retentionDays)
-            val archived = financialActionPersister.archiveProcessedActions(before, batchSize)
-            if (archived > 0) {
-                log.info("Archived $archived processed financial actions older than $before")
+            var totalArchived = 0
+            var shouldContinue = true
+            repeat(maxBatchesPerRun) {
+                if (!shouldContinue) return@repeat
+                val archived = financialActionPersister.archiveProcessedActions(before, batchSize)
+                totalArchived += archived
+                if (archived < batchSize) shouldContinue = false
+            }
+
+            if (totalArchived > 0) {
+                log.info("Archived $totalArchived processed financial actions older than $before")
             }
         }
     }

accountant/accountant-ports/accountant-persister-postgres/src/main/kotlin/co/nilin/opex/accountant/ports/postgres/dao/FinancialActionRepository.kt

@@ -9,7 +9,6 @@ import org.springframework.data.repository.query.Param
 import org.springframework.data.repository.reactive.ReactiveCrudRepository
 import org.springframework.stereotype.Repository
 import reactor.core.publisher.Mono
-import java.math.BigDecimal
 import java.time.LocalDateTime
 
 @Repository
@@ -22,13 +21,23 @@ interface FinancialActionRepository : ReactiveCrudRepository<FinancialActionMode
         paging: Pageable
     ): Flow<FinancialActionModel>
 
-    @Query("select count(1) from fi_actions fi where fi.sender = :uuid and fi.symbol = :symbol and fi.event_type = :eventType and fi.status != :status")
-    fun countByUuidAndSymbolAndEventTypeAndStatusNot(
+    @Query(
+        """
+        select exists(
+            select 1
+            from fi_actions fi
+            where fi.sender = :uuid
+              and fi.symbol = :symbol
+              and fi.event_type = :eventType
+              and fi.status <> 'PROCESSED'
+        )
+        """
+    )
+    fun existsUnprocessedBySenderAndSymbolAndEventType(
         @Param("uuid") uuid: String,
         @Param("symbol") symbol: String,
-        @Param("eventType") eventType: String,
-        @Param("status") financialActionStatus: FinancialActionStatus
-    ): Mono<BigDecimal>
+        @Param("eventType") eventType: String
+    ): Mono<Boolean>
 
     @Query("select * from fi_actions fi where status != :status")
     fun findByStatusNot(@Param("status") status: String, paging: Pageable): Flow<FinancialActionModel>
@@ -69,6 +78,11 @@ interface FinancialActionRepository : ReactiveCrudRepository<FinancialActionMode
                 select 1 from fi_action_retry far
                 where far.fa_id = fi_actions.id and far.is_resolved = false
               )
+              and not exists (
+                select 1 from fi_actions child
+                where child.parent_id = fi_actions.id
+                  and child.status <> 'PROCESSED'
+              )
             order by create_date
             limit :limit
         ),

accountant/accountant-ports/accountant-persister-postgres/src/main/kotlin/co/nilin/opex/accountant/ports/postgres/impl/FinancialActionLoaderImpl.kt

@@ -15,7 +15,6 @@ import kotlinx.coroutines.reactive.awaitFirstOrElse
 import org.springframework.data.domain.PageRequest
 import org.springframework.data.domain.Sort
 import org.springframework.stereotype.Component
-import java.math.BigDecimal
 import java.time.LocalDateTime
 
 @Component
@@ -48,12 +47,11 @@ class FinancialActionLoaderImpl(
     }
 
     override suspend fun countUnprocessed(userUuid: String, symbol: String, eventType: String): Long {
-        return financialActionRepository.countByUuidAndSymbolAndEventTypeAndStatusNot(
+        return if (financialActionRepository.existsUnprocessedBySenderAndSymbolAndEventType(
             userUuid,
             symbol,
-            eventType,
-            FinancialActionStatus.PROCESSED
-        ).awaitFirstOrElse { BigDecimal.ZERO }.toLong()
+            eventType
+        ).awaitFirstOrElse { false }) 1L else 0L
     }
 
     override suspend fun loadFinancialAction(id: Long?): FinancialAction? {

accountant/accountant-ports/accountant-persister-postgres/src/main/kotlin/co/nilin/opex/accountant/ports/postgres/impl/FinancialActionPersisterImpl.kt

@@ -103,7 +103,7 @@ class FinancialActionPersisterImpl(
                 faRetryRepository.scheduleNext(
                     id!!,
                     retries + 1,
-                    LocalDateTime.now().plusSeconds(retries * delayMultiplier * delaySeconds),
+                    LocalDateTime.now().plusSeconds((retries + 1L) * delayMultiplier * delaySeconds),
                     giveUp
                 ).awaitSingleOrNull()
 

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

@@ -51,6 +51,15 @@ CREATE INDEX IF NOT EXISTS idx_fi_actions_status ON fi_actions (status);
 CREATE INDEX IF NOT EXISTS idx_fi_actions_pointer ON fi_actions (pointer);
 CREATE INDEX IF NOT EXISTS idx_fi_actions_status_create_date ON fi_actions (status, create_date);
 CREATE INDEX IF NOT EXISTS idx_fi_actions_parent_status ON fi_actions (parent_id, status);
+CREATE INDEX IF NOT EXISTS idx_fi_actions_unprocessed_lookup
+    ON fi_actions (sender, symbol, event_type)
+    WHERE status <> 'PROCESSED';
+CREATE INDEX IF NOT EXISTS idx_fi_actions_archive_candidates
+    ON fi_actions (create_date, id)
+    WHERE status = 'PROCESSED';
+CREATE INDEX IF NOT EXISTS idx_fi_actions_unprocessed_children_by_parent
+    ON fi_actions (parent_id)
+    WHERE status <> 'PROCESSED';
 
 ALTER TABLE fi_actions
     ADD COLUMN IF NOT EXISTS category_name VARCHAR(36);

accountant/accountant-ports/accountant-persister-postgres/src/test/kotlin/co/nilin/opex/accountant/ports/postgres/FAPersisterImplTest.kt

@@ -9,10 +9,14 @@ import co.nilin.opex.accountant.ports.postgres.model.FinancialActionModel
 import io.mockk.coEvery
 import io.mockk.coVerify
 import io.mockk.mockk
+import io.mockk.slot
+import co.nilin.opex.accountant.ports.postgres.model.FinancialActionRetryModel
 import kotlinx.coroutines.runBlocking
+import org.junit.jupiter.api.Assertions.assertTrue
 import org.junit.jupiter.api.Test
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
+import java.time.LocalDateTime
 
 @Suppress("ReactiveStreamsUnusedPublisher")
 class FAPersisterImplTest {
@@ -48,4 +52,29 @@ class FAPersisterImplTest {
         }
     }
 
+    @Test
+    fun givenRetryableAction_whenUpdateWithError_thenScheduleUsesBackoffDelay(): Unit = runBlocking {
+        val retryModel = FinancialActionRetryModel(
+            faId = Valid.fa.id!!,
+            nextRunTime = LocalDateTime.now(),
+            retries = 0,
+            isResolved = false,
+            hasGivenUp = false,
+            id = 10
+        )
+        val nextRunSlot = slot<LocalDateTime>()
+
+        coEvery { faRetryRepository.findByFaId(Valid.fa.id!!) } returns Mono.just(retryModel)
+        coEvery { faRetryRepository.scheduleNext(eq(10), eq(1), capture(nextRunSlot), eq(false)) } returns Mono.empty()
+        coEvery { financialActionRepository.updateStatus(eq(Valid.fa.id!!), eq(FinancialActionStatus.RETRYING)) } returns Mono.empty()
+        coEvery { faErrorRepository.save(any()) } returns Mono.empty()
+
+        val before = LocalDateTime.now()
+        faPersister.updateWithError(Valid.fa, "ERR", "message", null)
+
+        coVerify(exactly = 1) { faRetryRepository.scheduleNext(eq(10), eq(1), any(), eq(false)) }
+        assertTrue(nextRunSlot.isCaptured)
+        assertTrue(nextRunSlot.captured.isAfter(before.plusSeconds(10)))
+    }
+
 }

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

@@ -3,6 +3,7 @@ package co.nilin.opex.api.core.inout
 data class ChainInfo(
     val name: String,
     val addressTypes: String?,
-    val externalChainScannerUrl: String? = null,
-    val addressRegex: String? = null
+    val addressRegex: String? = null,
+    val transactionScannerUrl: String? = null,
+    val addressScannerUrl: String? = null
 )

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

@@ -2,5 +2,5 @@ package co.nilin.opex.api.core.inout
 
 enum class OTPType {
 
-    SMS, EMAIL,
+    SMS, EMAIL, TOTP, NONE
 }

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

@@ -0,0 +1,7 @@
+package co.nilin.opex.api.core.inout
+
+enum class PairCategory {
+    REAL_ASSET_TOKEN,
+    FIAT,
+    CRYPTO
+}

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

@@ -10,4 +10,7 @@ data class PairInfoResponse(
     val minOrder : BigDecimal,
     val maxOrder : BigDecimal,
     val orderTypes : String,
+    val internalChart: Boolean,
+    val globalChart: Boolean,
+    val categories: List<PairCategory> = emptyList()
 )

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

@@ -10,4 +10,8 @@ class PairSetting(
     val maxOrder : BigDecimal,
     val orderTypes : String,
     val updateDate: LocalDateTime? = null,
+    val internalChart: Boolean,
+    val globalChart: Boolean,
+    val categories: List<PairCategory> = emptyList()
+
 )

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

@@ -0,0 +1,28 @@
+package co.nilin.opex.api.core.inout
+
+data class SetupTOTPRequest(
+    val userId: String,
+    val label: String?
+)
+
+data class SetupTOTPResponse(
+    val uri: String
+)
+
+data class VerifyTOTPRequest(
+    val userId: String,
+    val code: String
+)
+
+data class VerifyTOTPResponse(val result: Boolean)
+
+data class TOTPQueryResponse(
+    val userId: String,
+    val isEnabled: Boolean,
+    val isActivated: Boolean,
+    val uri : String
+)
+
+data class TOTPCode(
+    val code: String
+)

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

@@ -0,0 +1,15 @@
+package co.nilin.opex.api.core.inout
+
+import com.fasterxml.jackson.annotation.JsonInclude
+
+data class TwoFactorRequest(
+    val method: OTPType,
+)
+
+data class ConfirmTwoFactorRequest(
+    val method: OTPType,
+    val otp: String,
+)
+
+@JsonInclude(JsonInclude.Include.NON_NULL)
+data class TwoFactorResponse(val otp: String?, val otpReceiver: OTPReceiver?)

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

@@ -1,5 +1,11 @@
 package co.nilin.opex.api.core.spi
 
+import co.nilin.opex.api.core.inout.ConfirmTwoFactorRequest
+import co.nilin.opex.api.core.inout.OTPType
+import co.nilin.opex.api.core.inout.SetupTOTPResponse
+import co.nilin.opex.api.core.inout.TOTPCode
+import co.nilin.opex.api.core.inout.TwoFactorRequest
+import co.nilin.opex.api.core.inout.TwoFactorResponse
 import co.nilin.opex.api.core.inout.auth.*
 
 interface AuthProxy {
@@ -22,4 +28,12 @@ interface AuthProxy {
     suspend fun logoutOthers(token: String)
     suspend fun logoutAll(token: String)
 
+    suspend fun getTwoFactorConfig(token: String): OTPType
+    suspend fun requestEnableTwoFactor(request: TwoFactorRequest, token: String): TwoFactorResponse
+    suspend fun confirmEnableTwoFactor(request: ConfirmTwoFactorRequest, token: String): OTPVerifyResponse
+    suspend fun requestDisableTwoFactor(request: TwoFactorRequest, token: String): TwoFactorResponse
+    suspend fun confirmDisableTwoFactor(request: ConfirmTwoFactorRequest, token: String): OTPVerifyResponse
+    suspend fun setupTOTP(token: String): SetupTOTPResponse
+    suspend fun verifyTOTPSetup(request: TOTPCode, token: String)
+
 }

api/api-ports/api-binance-rest/src/main/kotlin/co/nilin/opex/api/ports/binance/config/SecurityConfig.kt

@@ -106,6 +106,7 @@ class SecurityConfig(
                     // Opex endpoints
                     .pathMatchers("/opex/v1/oauth/protocol/openid-connect/**").permitAll()
                     .pathMatchers("/opex/v1/oauth.***").permitAll()
+                    .pathMatchers("/opex/v1/user/2fa/**").authenticated()
                     .pathMatchers("/opex/v1/user/public/**").permitAll()
                     .pathMatchers("/opex/v1/user/update/**").permitAll()
                     .pathMatchers("/v1/deposit/webhook").permitAll()

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

@@ -18,11 +18,7 @@ import io.swagger.v3.oas.annotations.tags.Tag
 import kotlinx.coroutines.async
 import kotlinx.coroutines.coroutineScope
 import org.springframework.beans.factory.annotation.Value
-import org.springframework.web.bind.annotation.GetMapping
-import org.springframework.web.bind.annotation.PathVariable
-import org.springframework.web.bind.annotation.RequestMapping
-import org.springframework.web.bind.annotation.RequestParam
-import org.springframework.web.bind.annotation.RestController
+import org.springframework.web.bind.annotation.*
 import java.math.BigDecimal
 import java.time.ZoneId
 
@@ -104,7 +100,11 @@ Response body:
                     isAvailable = isAvailable,
                     minOrder = minOrder,
                     maxOrder = maxOrder,
-                    orderTypes = orderTypes
+                    orderTypes = orderTypes,
+                    internalChart = internalChart,
+                    globalChart = globalChart,
+                    categories = categories
+
                 )
             }
         }