Dev

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

@@ -0,0 +1,40 @@
+package co.nilin.opex.accountant.app.scheduler
+
+import co.nilin.opex.accountant.core.spi.FinancialActionPersister
+import kotlinx.coroutines.runBlocking
+import org.slf4j.LoggerFactory
+import org.springframework.beans.factory.annotation.Value
+import org.springframework.context.annotation.Profile
+import org.springframework.scheduling.annotation.Scheduled
+import org.springframework.stereotype.Service
+import java.time.LocalDateTime
+
+@Service
+@Profile("scheduled")
+class FinancialActionsArchiveJob(
+    private val financialActionPersister: FinancialActionPersister
+) {
+    private val log = LoggerFactory.getLogger(FinancialActionsArchiveJob::class.java)
+
+    @Value("\${app.fi-action.archive.enabled:true}")
+    private var enabled: Boolean = true
+
+    @Value("\${app.fi-action.archive.retention-days:30}")
+    private var retentionDays: Long = 30
+
+    @Value("\${app.fi-action.archive.batch-size:1000}")
+    private var batchSize: Int = 1000
+
+    @Scheduled(fixedDelayString = "\${app.fi-action.archive.fixed-delay-ms:300000}", initialDelay = 60000)
+    fun archiveProcessedActions() {
+        if (!enabled || batchSize <= 0 || retentionDays <= 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")
+            }
+        }
+    }
+}

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

@@ -101,11 +101,21 @@ app:
   address: 1
   wallet:
     url: lb://opex-wallet/
+    http:
+      retry:
+        count: 2
+        delay-millis: 250
+      timeout-seconds: 10
   fi-action:
     retry:
       count: 10
       delay-seconds: 5
       delay-multiplier: 3
+    archive:
+      enabled: true
+      retention-days: 30
+      batch-size: 1000
+      fixed-delay-ms: 300000
   zone-offset: +03:30
   trade-volume-calculation-currency: ${TRADE_VOLUME_CALCULATION_CURRENCY:USDT}
   withdraw-volume-calculation-currency: ${WITHDRAW_VOLUME_CALCULATION_CURRENCY:USDT}

accountant/accountant-app/src/test/kotlin/co/nilin/opex/accountant/app/scheduler/FinancialActionJobManagerIT.kt

@@ -14,7 +14,6 @@ import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Test
 import org.mockito.InOrder
 import org.mockito.Mockito
-import org.mockito.Mockito.any
 import org.mockito.Mockito.`when`
 import org.mockito.kotlin.eq
 import org.springframework.beans.factory.annotation.Autowired
@@ -248,7 +247,7 @@ class FinancialActionJobManagerIT : KafkaEnabledTest() {
             eq(fi.receiver),
             eq(fi.amount),
             eq(fi.eventType + fi.pointer),
-            any(),
+            eq("accountant:fiActions:${fi.uuid}"),
             eq(fi.category.toString()),
         )
     }

accountant/accountant-core/src/main/kotlin/co/nilin/opex/accountant/core/service/FinancialActionJobManagerImpl.kt

@@ -23,13 +23,6 @@ class FinancialActionJobManagerImpl(
             .also { if (it.isNotEmpty()) logger.info("Processing ${it.size} financial actions") }
             .forEach {
                 try {
-                    if (it.parent != null) {
-                        val reloadParent = financialActionLoader.loadFinancialAction(it.parent.id)!!
-                        if (reloadParent.status != FinancialActionStatus.PROCESSED) {
-                            logger.warn("Financial job (uuid=${it.uuid}) skipped because of parent status: uuid=${reloadParent.uuid}, status=${reloadParent.status}")
-                            return@forEach
-                        }
-                    }
                     walletProxy.transfer(
                         it.symbol,
                         it.senderWalletType,
@@ -70,7 +63,7 @@ class FinancialActionJobManagerImpl(
                         it.receiver,
                         it.amount,
                         it.eventType + it.pointer,
-                        "accountant:fiActions:${it.id.toString()}",
+                        "accountant:fiActions:${it.uuid}",
                         it.category.toString()
                     )
                     with(financialActionPersister) {

accountant/accountant-core/src/main/kotlin/co/nilin/opex/accountant/core/spi/FinancialActionPersister.kt

@@ -2,6 +2,7 @@ package co.nilin.opex.accountant.core.spi
 
 import co.nilin.opex.accountant.core.model.FinancialAction
 import co.nilin.opex.accountant.core.model.FinancialActionStatus
+import java.time.LocalDateTime
 
 interface FinancialActionPersister {
 
@@ -20,4 +21,6 @@ interface FinancialActionPersister {
     suspend fun updateStatusNewTx(financialAction: FinancialAction, status: FinancialActionStatus)
 
     suspend fun retrySuccessful(financialAction: FinancialAction)
+
+    suspend fun archiveProcessedActions(before: LocalDateTime, limit: Int): Int
 }

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

@@ -10,6 +10,7 @@ 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
 interface FinancialActionRepository : ReactiveCrudRepository<FinancialActionModel, Long> {
@@ -45,8 +46,83 @@ interface FinancialActionRepository : ReactiveCrudRepository<FinancialActionMode
         """
         select * from fi_actions fi 
         where status = 'CREATED'
-            and (parent_id is null or 'ERROR' != (select pfi.status from fi_actions pfi where pfi.id = fi.parent_id))
+            and (
+                parent_id is null
+                or exists(
+                    select 1 from fi_actions pfi
+                    where pfi.id = fi.parent_id and pfi.status = 'PROCESSED'
+                )
+            )
+        order by create_date
     """
     )
     fun findReadyToProcess(of: Pageable): Flow<FinancialActionModel>
+
+    @Query(
+        """
+        with candidates as (
+            select id
+            from fi_actions
+            where status = 'PROCESSED'
+              and create_date < :before
+              and not exists (
+                select 1 from fi_action_retry far
+                where far.fa_id = fi_actions.id and far.is_resolved = false
+              )
+            order by create_date
+            limit :limit
+        ),
+        moved_actions as (
+            insert into fi_actions_archive (
+                id, uuid, parent_id, event_type, pointer, symbol, amount, sender, sender_wallet_type,
+                receiver, receiver_wallet_type, agent, ip, create_date, status, category_name
+            )
+            select fa.id, fa.uuid, fa.parent_id, fa.event_type, fa.pointer, fa.symbol, fa.amount, fa.sender, fa.sender_wallet_type,
+                   fa.receiver, fa.receiver_wallet_type, fa.agent, fa.ip, fa.create_date, fa.status, fa.category_name
+            from fi_actions fa
+                     join candidates c on c.id = fa.id
+            on conflict (id) do nothing
+            returning id
+        ),
+        moved_retries as (
+            insert into fi_action_retry_archive (id, fa_id, retries, next_run_time, is_resolved, has_given_up)
+            select far.id, far.fa_id, far.retries, far.next_run_time, far.is_resolved, far.has_given_up
+            from fi_action_retry far
+                     join moved_actions ma on ma.id = far.fa_id
+            on conflict (id) do nothing
+            returning id
+        ),
+        moved_errors as (
+            insert into fi_action_error_archive (id, fa_id, error, message, body, retry_id, date)
+            select fae.id, fae.fa_id, fae.error, fae.message, fae.body, fae.retry_id, fae.date
+            from fi_action_error fae
+                     join moved_actions ma on ma.id = fae.fa_id
+            on conflict (id) do nothing
+            returning id
+        ),
+        deleted_errors as (
+            delete from fi_action_error fae
+                using moved_actions ma
+            where fae.fa_id = ma.id
+            returning fae.id
+        ),
+        deleted_retries as (
+            delete from fi_action_retry far
+                using moved_actions ma
+            where far.fa_id = ma.id
+            returning far.id
+        ),
+        deleted_actions as (
+            delete from fi_actions fa
+                using moved_actions ma
+            where fa.id = ma.id
+            returning fa.id
+        )
+        select count(1) from deleted_actions
+        """
+    )
+    fun archiveProcessedActions(
+        @Param("before") before: LocalDateTime,
+        @Param("limit") limit: Int
+    ): Mono<Long>
 }

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

@@ -6,6 +6,7 @@ import co.nilin.opex.accountant.core.spi.FinancialActionLoader
 import co.nilin.opex.accountant.ports.postgres.dao.FinancialActionErrorRepository
 import co.nilin.opex.accountant.ports.postgres.dao.FinancialActionRepository
 import co.nilin.opex.accountant.ports.postgres.dao.FinancialActionRetryRepository
+import co.nilin.opex.accountant.ports.postgres.model.FinancialActionModel
 import kotlinx.coroutines.flow.firstOrNull
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.toList
@@ -35,7 +36,7 @@ class FinancialActionLoaderImpl(
     override suspend fun loadReadyToProcess(offset: Long, size: Long): List<FinancialAction> {
         return financialActionRepository.findReadyToProcess(
             PageRequest.of(offset.toInt(), size.toInt(), Sort.by(Sort.Direction.ASC, "createDate"))
-        ).map { loadFinancialAction(it.id)!! }
+        ).map { mapToFinancialAction(it) }
             .toList()
     }
 
@@ -80,25 +81,27 @@ class FinancialActionLoaderImpl(
 
     override suspend fun loadRetries(limit: Int): List<FinancialAction> {
         return faRetryRepository.findAllRetries(LocalDateTime.now(), limit)
-            .map {
-                FinancialAction(
-                    null, // Skipping parent. If it's in retry, it means its parent is already processed
-                    it.eventType,
-                    it.pointer,
-                    it.symbol,
-                    it.amount,
-                    it.sender,
-                    it.senderWalletType,
-                    it.receiver,
-                    it.receiverWalletType,
-                    it.createDate,
-                    it.categoryName,
-                    it.status,
-                    it.uuid,
-                    it.id
-                )
-            }
+            .map { mapToFinancialAction(it) }
             .collectList()
             .awaitFirstOrElse { emptyList() }
     }
+
+    private fun mapToFinancialAction(financialAction: FinancialActionModel): FinancialAction {
+        return FinancialAction(
+            null,
+            financialAction.eventType,
+            financialAction.pointer,
+            financialAction.symbol,
+            financialAction.amount,
+            financialAction.sender,
+            financialAction.senderWalletType,
+            financialAction.receiver,
+            financialAction.receiverWalletType,
+            financialAction.createDate,
+            financialAction.categoryName,
+            financialAction.status,
+            financialAction.uuid,
+            financialAction.id
+        )
+    }
 }

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

@@ -131,6 +131,10 @@ class FinancialActionPersisterImpl(
         faRetryRepository.updateResolvedTrue(financialAction.id!!).awaitSingleOrNull()
     }
 
+    override suspend fun archiveProcessedActions(before: LocalDateTime, limit: Int): Int {
+        return (repository.archiveProcessedActions(before, limit).awaitSingleOrNull() ?: 0).toInt()
+    }
+
     override suspend fun updateStatus(faUuid: String, status: FinancialActionStatus) {
         repository.updateStatus(faUuid, status).awaitSingleOrNull()
     }

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

@@ -49,6 +49,8 @@ CREATE INDEX IF NOT EXISTS idx_fi_actions_symbol ON fi_actions (symbol);
 CREATE INDEX IF NOT EXISTS idx_fi_event_type ON fi_actions (event_type);
 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);
 
 ALTER TABLE fi_actions
     ADD COLUMN IF NOT EXISTS category_name VARCHAR(36);
@@ -63,6 +65,11 @@ CREATE TABLE IF NOT EXISTS fi_action_retry
     has_given_up  BOOLEAN   NOT NULL DEFAULT false
 );
 
+CREATE INDEX IF NOT EXISTS idx_fi_action_retry_due
+    ON fi_action_retry (next_run_time)
+    WHERE has_given_up = false
+      AND is_resolved = false;
+
 CREATE TABLE IF NOT EXISTS fi_action_error
 (
     id       SERIAL PRIMARY KEY,
@@ -74,6 +81,58 @@ CREATE TABLE IF NOT EXISTS fi_action_error
     date     TIMESTAMP NOT NULL DEFAULT CURRENT_DATE
 );
 
+CREATE INDEX IF NOT EXISTS idx_fi_action_error_fa_id_date ON fi_action_error (fa_id, date);
+
+CREATE TABLE IF NOT EXISTS fi_actions_archive
+(
+    id                   INTEGER PRIMARY KEY,
+    uuid                 VARCHAR(72) NOT NULL UNIQUE,
+    parent_id            INTEGER,
+    event_type           VARCHAR(72) NOT NULL,
+    pointer              VARCHAR(72) NOT NULL,
+    symbol               VARCHAR(36) NOT NULL,
+    amount               DECIMAL     NOT NULL,
+    sender               VARCHAR(36) NOT NULL,
+    sender_wallet_type   VARCHAR(36) NOT NULL,
+    receiver             VARCHAR(36) NOT NULL,
+    receiver_wallet_type VARCHAR(36) NOT NULL,
+    agent                VARCHAR(20),
+    ip                   VARCHAR(11),
+    create_date          TIMESTAMP   NOT NULL,
+    status               VARCHAR(20),
+    category_name        VARCHAR(36),
+    archived_at          TIMESTAMP   NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE INDEX IF NOT EXISTS idx_fi_actions_archive_create_date ON fi_actions_archive (create_date);
+
+CREATE TABLE IF NOT EXISTS fi_action_retry_archive
+(
+    id            INTEGER PRIMARY KEY,
+    fa_id         INTEGER   NOT NULL UNIQUE,
+    retries       INTEGER   NOT NULL DEFAULT 0,
+    next_run_time TIMESTAMP NOT NULL,
+    is_resolved   BOOLEAN   NOT NULL DEFAULT false,
+    has_given_up  BOOLEAN   NOT NULL DEFAULT false,
+    archived_at   TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE INDEX IF NOT EXISTS idx_fi_action_retry_archive_fa_id ON fi_action_retry_archive (fa_id);
+
+CREATE TABLE IF NOT EXISTS fi_action_error_archive
+(
+    id          INTEGER PRIMARY KEY,
+    fa_id       INTEGER   NOT NULL,
+    error       TEXT      NOT NULL,
+    message     TEXT      NOT NULL,
+    body        TEXT,
+    retry_id    INTEGER,
+    date        TIMESTAMP NOT NULL,
+    archived_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE INDEX IF NOT EXISTS idx_fi_action_error_archive_fa_id ON fi_action_error_archive (fa_id);
+
 CREATE TABLE IF NOT EXISTS pair_config
 (
     pair                     VARCHAR(72) PRIMARY KEY,