Remove flow database dependency

CHANGELOG.md

@@ -34,6 +34,7 @@ Ref: https://keepachangelog.com/en/1.0.0/
 ## Unreleased
 
 ### Features
+* Remove flow dependency and collect on-chain navs in local database [#574](https://github.com/provenance-io/explorer-service/pull/574) 
 * Add hash price support for nav values from on chain events [#543](https://github.com/provenance-io/explorer-service/pull/543)
 * Integrate on-chain NAV data into historical hash price calculations [#555](https://github.com/provenance-io/explorer-service/pull/555) 
 * Use on-chain NAV data for latest asset pricing on markers [#556](https://github.com/provenance-io/explorer-service/pull/556)

buildSrc/src/main/kotlin/Dependencies.kt

@@ -60,7 +60,7 @@ object Versions {
     const val Jupiter = "5.9.1"
     const val SpringMockk = "3.1.1"
     const val Kotest = "5.5.4"
-    const val H2Database = "1.4.200"
+    const val H2Database = "2.1.214"
 }
 
 object Libraries {

database/src/main/resources/db/migration/V1_96__Add_nav_event_table.sql

@@ -0,0 +1,21 @@
+SELECT 'Add nav events table' AS comment;
+
+CREATE TABLE IF NOT EXISTS nav_events (
+    id SERIAL PRIMARY KEY,
+    block_height INT NOT NULL,
+    block_time TIMESTAMP NOT NULL,
+    tx_hash TEXT,
+    event_order INT,
+    event_type TEXT,
+    scope_id TEXT,
+    denom TEXT,
+    price_amount BIGINT,
+    price_denom TEXT,
+    volume BIGINT,
+    source TEXT,
+    UNIQUE (block_height, tx_hash, event_order)
+);
+
+CREATE INDEX IF NOT EXISTS idx_nav_events_block_time ON nav_events(block_time);
+CREATE INDEX IF NOT EXISTS idx_nav_events_denom ON nav_events(denom);
+CREATE INDEX IF NOT EXISTS idx_nav_events_scope_id ON nav_events(scope_id);

service/src/main/kotlin/io/provenance/explorer/domain/entities/NavEvents.kt

@@ -0,0 +1,261 @@
+package io.provenance.explorer.domain.entities
+
+import io.provenance.explorer.domain.extensions.execAndMap
+import io.provenance.explorer.domain.extensions.toDateTime
+import org.jetbrains.exposed.dao.IntEntity
+import org.jetbrains.exposed.dao.IntEntityClass
+import org.jetbrains.exposed.dao.id.EntityID
+import org.jetbrains.exposed.dao.id.IntIdTable
+import org.jetbrains.exposed.sql.ColumnType
+import org.jetbrains.exposed.sql.VarCharColumnType
+import org.jetbrains.exposed.sql.insertIgnore
+import org.jetbrains.exposed.sql.jodatime.DateColumnType
+import org.jetbrains.exposed.sql.jodatime.datetime
+import org.jetbrains.exposed.sql.transactions.transaction
+import org.joda.time.DateTime
+import java.sql.ResultSet
+
+object NavEventsTable : IntIdTable(name = "nav_events") {
+    val blockHeight = integer("block_height")
+    val blockTime = datetime("block_time")
+    val txHash = text("tx_hash")
+    val eventOrder = integer("event_order")
+    val eventType = text("event_type")
+    val scopeId = text("scope_id").nullable()
+    val denom = text("denom").nullable()
+    val priceAmount = long("price_amount").nullable()
+    val priceDenom = text("price_denom").nullable()
+    val volume = long("volume")
+    val dataSource = text("source")
+
+    init {
+        uniqueIndex("nav_events_unique_idx", blockHeight, txHash, eventOrder)
+    }
+}
+
+class NavEventsRecord(id: EntityID<Int>) : IntEntity(id) {
+    companion object : IntEntityClass<NavEventsRecord>(NavEventsTable) {
+        fun insert(
+            blockHeight: Int,
+            blockTime: DateTime,
+            txHash: String,
+            eventOrder: Int,
+            eventType: String,
+            scopeId: String?,
+            denom: String?,
+            priceAmount: Long?,
+            priceDenom: String?,
+            volume: Long,
+            source: String
+        ) = transaction {
+            NavEventsTable.insertIgnore {
+                it[this.blockHeight] = blockHeight
+                it[this.blockTime] = blockTime
+                it[this.txHash] = txHash
+                it[this.eventOrder] = eventOrder
+                it[this.eventType] = eventType
+                it[this.scopeId] = scopeId
+                it[this.denom] = denom
+                it[this.priceAmount] = priceAmount
+                it[this.priceDenom] = priceDenom
+                it[this.volume] = volume
+                it[this.dataSource] = source
+            }
+        }
+
+        fun getNavEvents(
+            denom: String? = null,
+            scopeId: String? = null,
+            fromDate: DateTime? = null,
+            toDate: DateTime? = null,
+            priceDenoms: List<String>? = null
+        ) = transaction {
+            var query = """
+            SELECT block_height, block_time, tx_hash, event_order, 
+            event_type, scope_id, denom, price_amount, price_denom, volume, source
+            FROM nav_events
+            WHERE 1=1
+            """.trimIndent()
+
+            val args = mutableListOf<Pair<ColumnType, *>>()
+
+            denom?.let {
+                query += " AND denom = ?"
+                args.add(Pair(VarCharColumnType(), it))
+            } ?: scopeId?.let {
+                query += " AND scope_id = ?"
+                args.add(Pair(VarCharColumnType(), it))
+            }
+
+            fromDate?.let {
+                query += " AND block_time >= ?"
+                args.add(Pair(DateColumnType(true), it))
+            }
+
+            toDate?.let {
+                query += " AND block_time <= ?"
+                args.add(Pair(DateColumnType(true), it))
+            }
+
+            priceDenoms?.let {
+                if (it.isNotEmpty()) {
+                    val placeholders = it.joinToString(", ") { "?" }
+                    query += " AND price_denom IN ($placeholders)"
+                    it.forEach { denom ->
+                        args.add(Pair(VarCharColumnType(), denom))
+                    }
+                }
+            }
+
+            query += " ORDER BY block_height DESC, event_order DESC"
+
+            query.execAndMap(args) {
+                NavEvent(
+                    it.getInt("block_height"),
+                    it.getTimestamp("block_time").toDateTime(),
+                    it.getString("tx_hash"),
+                    it.getInt("event_order"),
+                    it.getString("event_type"),
+                    it.getString("scope_id"),
+                    it.getString("denom"),
+                    it.getLong("price_amount"),
+                    it.getString("price_denom"),
+                    it.getLong("volume"),
+                    it.getString("source")
+                )
+            }
+        }
+
+        fun getLatestNavEvents(
+            priceDenom: String,
+            includeMarkers: Boolean,
+            includeScopes: Boolean,
+            fromDate: DateTime? = null
+        ) = transaction {
+            require(includeMarkers || includeScopes) { "Either includeMarkers or includeScope must be true" }
+
+            var query = """
+            SELECT DISTINCT ON (denom, scope_id)
+                block_height, block_time, tx_hash, event_order, event_type, 
+                scope_id, denom, price_amount, price_denom, volume, source
+            FROM nav_events
+            WHERE price_denom = ?
+            """.trimIndent()
+
+            val args = mutableListOf<Pair<ColumnType, *>>(
+                Pair(VarCharColumnType(), priceDenom)
+            )
+
+            fromDate?.let {
+                query += " AND block_time >= ?"
+                args.add(Pair(DateColumnType(true), it))
+            }
+
+            when {
+                includeMarkers && includeScopes -> query += " AND (denom IS NOT NULL OR scope_id IS NOT NULL)"
+                includeMarkers -> query += " AND denom IS NOT NULL"
+                includeScopes -> query += " AND scope_id IS NOT NULL"
+            }
+
+            query += " ORDER BY denom, scope_id, block_height DESC, event_order DESC"
+
+            query.execAndMap(args) {
+                NavEvent(
+                    it.getInt("block_height"),
+                    it.getTimestamp("block_time").toDateTime(),
+                    it.getString("tx_hash"),
+                    it.getInt("event_order"),
+                    it.getString("event_type"),
+                    it.getString("scope_id"),
+                    it.getString("denom"),
+                    it.getLong("price_amount"),
+                    it.getString("price_denom"),
+                    it.getLong("volume"),
+                    it.getString("source")
+                )
+            }
+        }
+    }
+
+    var blockHeight by NavEventsTable.blockHeight
+    var blockTime by NavEventsTable.blockTime
+    var txHash by NavEventsTable.txHash
+    var eventOrder by NavEventsTable.eventOrder
+    var eventType by NavEventsTable.eventType
+    var scopeId by NavEventsTable.scopeId
+    var denom by NavEventsTable.denom
+    var priceAmount by NavEventsTable.priceAmount
+    var priceDenom by NavEventsTable.priceDenom
+    var volume by NavEventsTable.volume
+    var source by NavEventsTable.dataSource
+}
+
+data class NavPrice(
+    val blockHeight: Int,
+    val blockTime: DateTime,
+    val txHash: String,
+    val eventOrder: Int,
+    val eventType: String,
+    val scopeId: String?,
+    val denom: String,
+    val priceAmount: Long,
+    val priceDenom: String,
+    val volume: Long,
+    val source: String
+) {
+    constructor(rs: ResultSet) : this(
+        rs.getInt("block_height"),
+        rs.getTimestamp("block_time").toDateTime(),
+        rs.getString("tx_hash"),
+        rs.getInt("event_order"),
+        rs.getString("event_type"),
+        rs.getString("scope_id"),
+        rs.getString("denom"),
+        rs.getLong("price_amount"),
+        rs.getString("price_denom"),
+        rs.getLong("volume"),
+        rs.getString("source")
+    )
+
+    constructor(record: NavEventsRecord) : this(
+        record.blockHeight,
+        record.blockTime,
+        record.txHash,
+        record.eventOrder,
+        record.eventType,
+        record.scopeId,
+        record.denom!!,
+        record.priceAmount!!,
+        record.priceDenom!!,
+        record.volume,
+        record.source
+    )
+}
+
+data class NavEvent(
+    val blockHeight: Int,
+    val blockTime: DateTime,
+    val txHash: String?,
+    val eventOrder: Int,
+    val eventType: String,
+    val scopeId: String?,
+    val denom: String?,
+    val priceAmount: Long?,
+    val priceDenom: String?,
+    val volume: Long,
+    val source: String
+) {
+    constructor(record: NavEventsRecord) : this(
+        record.blockHeight,
+        record.blockTime,
+        record.txHash,
+        record.eventOrder,
+        record.eventType,
+        record.scopeId,
+        record.denom,
+        record.priceAmount,
+        record.priceDenom,
+        record.volume,
+        record.source
+    )
+}

service/src/main/kotlin/io/provenance/explorer/domain/extensions/NavEventExtensions.kt

@@ -1,10 +1,17 @@
 package io.provenance.explorer.domain.extensions
 
 import io.provenance.explorer.model.base.USD_LOWER
-import io.provlabs.flow.api.NavEvent
 import java.math.BigDecimal
 import java.math.RoundingMode
 
+fun io.provlabs.flow.api.NavEvent.calculateUsdPricePerUnit(): BigDecimal {
+    return calculateUsdPrice(this.priceDenom, this.priceAmount, this.volume)
+}
+
+fun io.provenance.explorer.domain.entities.NavEvent.calculateUsdPricePerUnit(): BigDecimal {
+    return calculateUsdPrice(this.priceDenom, this.priceAmount, this.volume)
+}
+
 /**
  * Calculates the USD price per unit for a NAV event.
  *
@@ -13,17 +20,17 @@ import java.math.RoundingMode
  *
  * @return The USD price per unit or `BigDecimal.ZERO` if not a USD event or volume is 0.
  */
-fun NavEvent.calculateUsdPricePerUnit(): BigDecimal {
-    if (this.priceDenom != USD_LOWER) {
+private fun calculateUsdPrice(priceDenom: String?, priceAmount: Long?, volume: Long): BigDecimal {
+    if (priceDenom != USD_LOWER) {
         return BigDecimal.ZERO
     }
 
-    if (this.volume == 0L) {
+    if (volume == 0L) {
         return BigDecimal.ZERO
     }
 
-    return BigDecimal(this.priceAmount)
+    return BigDecimal(priceAmount ?: 0)
         .setScale(3, RoundingMode.DOWN)
         .divide(BigDecimal(1000), RoundingMode.DOWN)
-        .divide(BigDecimal(this.volume), 3, RoundingMode.DOWN)
+        .divide(BigDecimal(volume), 3, RoundingMode.DOWN)
 }

service/src/main/kotlin/io/provenance/explorer/service/AssetService.kt

@@ -12,6 +12,7 @@ import io.provenance.explorer.domain.entities.AssetPricingRecord
 import io.provenance.explorer.domain.entities.BaseDenomType
 import io.provenance.explorer.domain.entities.MarkerCacheRecord
 import io.provenance.explorer.domain.entities.MarkerUnitRecord
+import io.provenance.explorer.domain.entities.NavEventsRecord
 import io.provenance.explorer.domain.entities.TxMarkerJoinRecord
 import io.provenance.explorer.domain.exceptions.requireNotNullToMessage
 import io.provenance.explorer.domain.extensions.calculateUsdPricePerUnit
@@ -244,22 +245,22 @@ class AssetService(
         val now = OffsetDateTime.now().withOffsetSameInstant(ZoneOffset.UTC)
         logger.info("Updating asset pricing, last run at: $assetPricinglastRun")
 
-        val latestPrices = flowApiGrpcClient.getAllLatestNavPrices(
+        val latestPrices = NavEventsRecord.getLatestNavEvents(
             priceDenom = USD_LOWER,
             includeMarkers = true,
             includeScopes = false,
             fromDate = assetPricinglastRun?.toDateTime()
         )
         latestPrices.forEach { price ->
             if (price.denom != UTILITY_TOKEN) {
-                val marker = getAssetRaw(price.denom)
+                val marker = getAssetRaw(price.denom!!)
                 insertAssetPricing(
                     marker = marker,
                     markerDenom = price.denom,
                     markerAddress = marker.second.markerAddress,
-                    pricingDenom = price.priceDenom,
+                    pricingDenom = price.priceDenom!!,
                     pricingAmount = price.calculateUsdPricePerUnit(),
-                    timestamp = DateTime(price.blockTime * 1000)
+                    timestamp = price.blockTime
                 )
             }
         }