Report user activity per day

accountant/accountant-app/src/main/kotlin/co/nilin/opex/accountant/app/config/AppConfig.kt

@@ -74,7 +74,7 @@ class AppConfig(
         feeCalculator: FeeCalculator,
         financialActionPublisher: FinancialActionPublisher,
         currencyRatePersister: CurrencyRatePersister,
-        userVolumePersister: UserVolumePersister
+        userVolumePersister: UserTradeVolumePersister
     ): TradeManager {
         return TradeManagerImpl(
             financeActionPersister,
@@ -157,4 +157,12 @@ class AppConfig(
         withdrawRequestKafkaListener.addListener(withdrawRequestEventListener)
     }
 
+
+    @Autowired
+    fun configureDepositEventListener(
+        depositKafkaListener: DepositKafkaListener,
+        depositEventListener: DepositEventListener
+    ) {
+        depositKafkaListener.addListener(depositEventListener)
+    }
 }

accountant/accountant-app/src/main/kotlin/co/nilin/opex/accountant/app/controller/UserDailyActivityController.kt

@@ -0,0 +1,46 @@
+package co.nilin.opex.accountant.app.controller
+
+import co.nilin.opex.accountant.core.api.DepositActivityManager
+import co.nilin.opex.accountant.core.api.TradeActivityManager
+import co.nilin.opex.accountant.core.api.WithdrawActivityManager
+import co.nilin.opex.accountant.core.model.DailyAmount
+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.RestController
+
+@RestController
+@RequestMapping("/user-activity")
+class UserDailyActivityController(
+    private val withdrawManager: WithdrawActivityManager,
+    private val depositManager: DepositActivityManager,
+    private val TradeManager: TradeActivityManager
+) {
+
+    @GetMapping("/withdraw/{userId}")
+    suspend fun getDailyWithdrawLast31Days(
+        @PathVariable userId: String
+    ): List<DailyAmount> {
+        return withdrawManager.getLastDaysWithdrawActivity(
+            userId = userId
+        )
+    }
+
+    @GetMapping("/deposit/{userId}")
+    suspend fun getDailyDepositLast31Days(
+        @PathVariable userId: String
+    ): List<DailyAmount> {
+        return depositManager.getLastDaysDepositActivity(
+            userId = userId
+        )
+    }
+
+    @GetMapping("/trade/{userId}")
+    suspend fun getDailyTradeLast31Days(
+        @PathVariable userId: String
+    ): List<DailyAmount> {
+        return TradeManager.getLastDaysTradeActivity(
+            userId = userId
+        )
+    }
+}

accountant/accountant-app/src/main/kotlin/co/nilin/opex/accountant/app/controller/UserDataController.kt

@@ -2,7 +2,7 @@ package co.nilin.opex.accountant.app.controller
 
 import co.nilin.opex.accountant.core.api.FeeCalculator
 import co.nilin.opex.accountant.core.model.UserFee
-import co.nilin.opex.accountant.core.spi.UserVolumePersister
+import co.nilin.opex.accountant.core.spi.UserTradeVolumePersister
 import co.nilin.opex.accountant.core.spi.UserWithdrawVolumePersister
 import co.nilin.opex.common.utils.Interval
 import org.springframework.beans.factory.annotation.Value
@@ -13,7 +13,7 @@ import java.time.LocalDateTime
 @RestController
 @RequestMapping("/user/data")
 class UserDataController(
-    private val userVolumePersister: UserVolumePersister,
+    private val userVolumePersister: UserTradeVolumePersister,
     private val feeCalculator: FeeCalculator,
     private val userWithdrawVolumePersister: UserWithdrawVolumePersister,
     @Value("\${app.trade-volume-calculation-currency}")
@@ -60,4 +60,5 @@ class UserDataController(
             (interval?.getLocalDateTime() ?: LocalDateTime.now())
         )
     }
+
 }

accountant/accountant-app/src/main/kotlin/co/nilin/opex/accountant/app/listener/DepositEventListener.kt

@@ -0,0 +1,38 @@
+package co.nilin.opex.accountant.app.listener
+
+import co.nilin.opex.accountant.core.spi.UserDepositVolumePersister
+import co.nilin.opex.accountant.ports.kafka.listener.inout.DepositEvent
+import co.nilin.opex.accountant.ports.kafka.listener.inout.WithdrawRequestEvent
+import co.nilin.opex.accountant.ports.kafka.listener.spi.DepositListener
+import co.nilin.opex.accountant.ports.kafka.listener.spi.WithdrawRequestListener
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.launch
+import org.slf4j.LoggerFactory
+import org.springframework.stereotype.Component
+
+@Component
+class DepositEventListener(private val userDepositVolumePersister: UserDepositVolumePersister) :
+    DepositListener {
+
+    private val logger = LoggerFactory.getLogger(DepositEventListener::class.java)
+    val scope = CoroutineScope(Dispatchers.IO)
+    override fun id(): String {
+        return "DepositEventListener"
+    }
+
+    override fun onEvent(
+        event: DepositEvent,
+        partition: Int,
+        offset: Long,
+        timestamp: Long
+    ) {
+        logger.info("==========================================================================")
+        logger.info("Incoming Deposit event: $event")
+        logger.info("==========================================================================")
+        scope.launch {
+            userDepositVolumePersister.update(event.uuid, event.currency, event.amount, event.createDate)
+        }
+    }
+
+}

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

@@ -108,6 +108,7 @@ app:
   zone-offset: +03:30
   trade-volume-calculation-currency: ${TRADE_VOLUME_CALCULATION_CURRENCY:USDT}
   withdraw-volume-calculation-currency: ${WITHDRAW_VOLUME_CALCULATION_CURRENCY:USDT}
+  deposit-volume-calculation-currency: ${DEPOSIT_VOLUME_CALCULATION_CURRENCY:USDT}
   custom-message:
     enabled: ${CUSTOM_MESSAGE_ENABLED:false}
     base-url: ${CUSTOM_MESSAGE_URL}

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

@@ -57,4 +57,5 @@ app:
       delay-multiplier: 3
   zone-offset: +03:30
   trade-volume-calculation-currency: USDT
-  withdraw-volume-calculation-currency: USDT
+  withdraw-volume-calculation-currency: USDT
+  deposit-volume-calculation-currency: USDT

accountant/accountant-core/pom.xml

@@ -66,5 +66,11 @@
             <artifactId>jackson-datatype-jsr310</artifactId>
             <version>2.15.2</version>
         </dependency>
+        <dependency>
+            <groupId>com.sun.mail</groupId>
+            <artifactId>mailapi</artifactId>
+            <version>1.6.2</version>
+            <scope>compile</scope>
+        </dependency>
     </dependencies>
 </project>

accountant/accountant-core/src/main/kotlin/co/nilin/opex/accountant/core/api/DepositActivityManager.kt

@@ -0,0 +1,11 @@
+package co.nilin.opex.accountant.core.api
+
+import co.nilin.opex.accountant.core.model.DailyAmount
+
+interface DepositActivityManager {
+    suspend fun getLastDaysDepositActivity(
+        userId: String,
+        quoteCurrency: String? = null,
+        n: Int = 31
+    ): List<DailyAmount>
+}

accountant/accountant-core/src/main/kotlin/co/nilin/opex/accountant/core/api/TradeActivityManager.kt

@@ -0,0 +1,11 @@
+package co.nilin.opex.accountant.core.api
+
+import co.nilin.opex.accountant.core.model.DailyAmount
+
+interface TradeActivityManager {
+    suspend fun getLastDaysTradeActivity(
+        userId: String,
+        quoteCurrency: String? = null,
+        n: Int = 31
+    ): List<DailyAmount>
+}

accountant/accountant-core/src/main/kotlin/co/nilin/opex/accountant/core/api/WithdrawActivityManager.kt

@@ -0,0 +1,11 @@
+package co.nilin.opex.accountant.core.api
+
+import co.nilin.opex.accountant.core.model.DailyAmount
+
+interface WithdrawActivityManager {
+    suspend fun getLastDaysWithdrawActivity(
+        userId: String,
+        quoteCurrency: String? = null,
+        n: Int = 31
+    ): List<DailyAmount>
+}

accountant/accountant-core/src/main/kotlin/co/nilin/opex/accountant/core/model/DailyAmount.kt

@@ -0,0 +1,9 @@
+package co.nilin.opex.accountant.core.model
+
+import java.math.BigDecimal
+import java.time.LocalDate
+
+data class DailyAmount(
+    val date: LocalDate,
+    val totalAmount: BigDecimal
+)

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

@@ -0,0 +1,73 @@
+package co.nilin.opex.accountant.core.service
+
+import co.nilin.opex.accountant.core.api.DepositActivityManager
+import co.nilin.opex.accountant.core.model.DailyAmount
+import co.nilin.opex.accountant.core.spi.UserDepositVolumePersister
+import co.nilin.opex.common.utils.CacheManager
+import org.springframework.beans.factory.annotation.Value
+import org.springframework.stereotype.Service
+import java.math.BigDecimal
+import java.time.LocalDate
+import java.time.ZoneOffset
+import java.util.concurrent.TimeUnit
+import java.util.stream.Collectors
+
+@Service
+class DepositActivityManagerImpl(
+    private val cacheManager: CacheManager<String, BigDecimal>,
+    private val depositVolumePersister: UserDepositVolumePersister,
+    @Value("\${app.zone-offset}") private val zoneOffsetString: String,
+) : DepositActivityManager {
+    override suspend fun getLastDaysDepositActivity(
+        userId: String,
+        quoteCurrency: String?,
+        n: Int
+    ): List<DailyAmount> {
+        val today = LocalDate.now(ZoneOffset.of(zoneOffsetString))
+        val dates = (0..n - 1).map { today.minusDays(it.toLong()) }
+
+        val result = mutableMapOf<LocalDate, BigDecimal>()
+        val missingDates = mutableListOf<LocalDate>()
+
+        for (date in dates) {
+            val cacheKey = "deposit:daily:$userId:$date"
+            val cached = cacheManager.get(cacheKey)
+
+            if (cached != null) {
+                result[date] = cached
+            } else {
+                missingDates.add(date)
+            }
+        }
+
+        if (missingDates.isNotEmpty()) {
+            val startDate = missingDates.minOrNull()!!
+
+            val dbData = depositVolumePersister.getLastDaysDeposit(userId, startDate, quoteCurrency)
+                .stream().collect(Collectors.toMap(DailyAmount::date, DailyAmount::totalAmount));
+
+            for (date in missingDates) {
+                val value = dbData[date] ?: BigDecimal.ZERO
+                val (ttl, unit) = ttlFor(date, today)
+                val cacheKey = "deposit:daily:$userId:$date"
+
+                cacheManager.put(cacheKey, value, ttl, unit)
+                result[date] = value
+            }
+        }
+
+        return result
+            .map { DailyAmount(it.key, it.value) }
+            .sortedBy { it.date }
+
+    }
+
+
+    private fun ttlFor(date: LocalDate, today: LocalDate): Pair<Long, TimeUnit> =
+        if (date == today) {
+            15L to TimeUnit.MINUTES
+        } else {
+            100L to TimeUnit.DAYS
+        }
+
+}

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

@@ -4,7 +4,7 @@ import co.nilin.opex.accountant.core.api.FeeCalculator
 import co.nilin.opex.accountant.core.model.*
 import co.nilin.opex.accountant.core.spi.FeeConfigService
 import co.nilin.opex.accountant.core.spi.JsonMapper
-import co.nilin.opex.accountant.core.spi.UserVolumePersister
+import co.nilin.opex.accountant.core.spi.UserTradeVolumePersister
 import co.nilin.opex.accountant.core.spi.WalletProxy
 import co.nilin.opex.common.utils.CacheManager
 import co.nilin.opex.matching.engine.core.eventh.events.TradeEvent
@@ -22,7 +22,7 @@ import java.util.concurrent.TimeUnit
 class FeeCalculatorImpl(
     private val walletProxy: WalletProxy,
     private val feeConfigService: FeeConfigService,
-    private val userVolumePersister: UserVolumePersister,
+    private val userVolumePersister: UserTradeVolumePersister,
     @Qualifier("appCacheManager") private val cacheManager: CacheManager<String, UserFee>,
     @Value("\${app.address}") private val platformAddress: String,
     @Value("\${app.zone-offset}") private val zoneOffsetString: String,

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

@@ -0,0 +1,73 @@
+package co.nilin.opex.accountant.core.service
+
+import co.nilin.opex.accountant.core.api.TradeActivityManager
+import co.nilin.opex.accountant.core.api.WithdrawActivityManager
+import co.nilin.opex.accountant.core.model.DailyAmount
+import co.nilin.opex.accountant.core.spi.UserTradeVolumePersister
+import co.nilin.opex.common.utils.CacheManager
+import org.springframework.beans.factory.annotation.Value
+import org.springframework.stereotype.Service
+import java.math.BigDecimal
+import java.time.LocalDate
+import java.time.ZoneOffset
+import java.util.concurrent.TimeUnit
+import java.util.stream.Collectors
+@Service
+class TradeActivityManagerImpl(
+    private val cacheManager: CacheManager<String, BigDecimal>,
+    private val tradeVolumePersister: UserTradeVolumePersister,
+    @Value("\${app.zone-offset}") private val zoneOffsetString: String,
+) : TradeActivityManager {
+    override suspend fun getLastDaysTradeActivity(
+        userId: String,
+        quoteCurrency: String?,
+        n: Int
+    ): List<DailyAmount> {
+        val today = LocalDate.now(ZoneOffset.of(zoneOffsetString))
+        val dates = (0..n - 1).map { today.minusDays(it.toLong()) }
+
+        val result = mutableMapOf<LocalDate, BigDecimal>()
+        val missingDates = mutableListOf<LocalDate>()
+
+        for (date in dates) {
+            val cacheKey = "trade:daily:$userId:$date"
+            val cached = cacheManager.get(cacheKey)
+
+            if (cached != null) {
+                result[date] = cached
+            } else {
+                missingDates.add(date)
+            }
+        }
+
+        if (missingDates.isNotEmpty()) {
+            val startDate = missingDates.minOrNull()!!
+
+            val dbData = tradeVolumePersister.getLastDaysTrade(userId, startDate,quoteCurrency)
+                .stream().collect(Collectors.toMap(DailyAmount::date, DailyAmount::totalAmount));
+
+            for (date in missingDates) {
+                val value = dbData[date] ?: BigDecimal.ZERO
+                val (ttl, unit) = ttlFor(date, today)
+                val cacheKey = "trade:daily:$userId:$date"
+
+                cacheManager.put(cacheKey, value, ttl, unit)
+                result[date] = value
+            }
+        }
+
+        return result
+            .map { DailyAmount(it.key, it.value) }
+            .sortedBy { it.date }
+
+    }
+
+
+    private fun ttlFor(date: LocalDate, today: LocalDate): Pair<Long, TimeUnit> =
+        if (date == today) {
+            15L to TimeUnit.MINUTES
+        } else {
+            100L to TimeUnit.DAYS
+        }
+
+}

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

@@ -9,7 +9,6 @@ import co.nilin.opex.accountant.core.model.*
 import co.nilin.opex.accountant.core.spi.*
 import co.nilin.opex.matching.engine.core.eventh.events.TradeEvent
 import org.slf4j.LoggerFactory
-import org.springframework.beans.factory.annotation.Value
 import org.springframework.transaction.annotation.Transactional
 import java.math.BigDecimal
 import java.time.LocalDateTime
@@ -25,7 +24,7 @@ open class TradeManagerImpl(
     private val feeCalculator: FeeCalculator,
     private val financialActionPublisher: FinancialActionPublisher,
     private val currencyRatePersister: CurrencyRatePersister,
-    private val userVolumePersister: UserVolumePersister,
+    private val userVolumePersister: UserTradeVolumePersister,
     private val tradeVolumeCalculationCurrency: String,
     private val zoneOffsetString: String,
 ) : TradeManager {