Ocean: fix loan token indexer

lib/ain-ocean/src/api/prices.rs

@@ -21,7 +21,7 @@ use crate::{
     error::{ApiError, Error, NotFoundKind},
     model::{
         ApiResponseOraclePriceFeed, OracleIntervalSeconds, OraclePriceActive,
-        OraclePriceAggregated, OraclePriceAggregatedAggregated, OraclePriceAggregatedApi,
+        OraclePriceActiveNext, OraclePriceAggregated, OraclePriceAggregatedApi,
         OraclePriceAggregatedInterval, OraclePriceAggregatedIntervalAggregated,
         OracleTokenCurrency, PriceOracles, PriceTickerApi,
     },
@@ -77,7 +77,7 @@ async fn list_prices(
                     sort: price_ticker.price.sort,
                     token: price_ticker.price.token,
                     currency: price_ticker.price.currency,
-                    aggregated: OraclePriceAggregatedAggregated {
+                    aggregated: OraclePriceActiveNext {
                         amount: amount.to_string(),
                         weightage: price_ticker.price.aggregated.weightage,
                         oracles: price_ticker.price.aggregated.oracles,
@@ -120,7 +120,7 @@ async fn get_price(
                     sort: price_ticker.price.sort,
                     token: price_ticker.price.token,
                     currency: price_ticker.price.currency,
-                    aggregated: OraclePriceAggregatedAggregated {
+                    aggregated: OraclePriceActiveNext {
                         amount: amount.to_string(),
                         weightage: price_ticker.price.aggregated.weightage,
                         oracles: price_ticker.price.aggregated.oracles,

lib/ain-ocean/src/indexer/loan_token.rs

@@ -6,10 +6,7 @@ use rust_decimal_macros::dec;
 
 use crate::{
     indexer::{Context, Index, Result},
-    model::{
-        BlockContext, OraclePriceActive, OraclePriceActiveActive, OraclePriceActiveActiveOracles,
-        OraclePriceActiveNext, OraclePriceActiveNextOracles, OraclePriceAggregated,
-    },
+    model::{BlockContext, OraclePriceActive, OraclePriceActiveNext, OraclePriceAggregated},
     network::Network,
     repository::RepositoryOps,
     storage::SortOrder,
@@ -55,7 +52,7 @@ fn is_aggregate_valid(aggregate: &OraclePriceAggregated, block: &BlockContext) -
     true
 }
 
-fn is_live(active: Option<OraclePriceActiveActive>, next: Option<OraclePriceActiveNext>) -> bool {
+fn is_live(active: Option<OraclePriceActiveNext>, next: Option<OraclePriceActiveNext>) -> bool {
     let Some(active) = active else {
         return false;
     };
@@ -90,15 +87,16 @@ fn is_live(active: Option<OraclePriceActiveActive>, next: Option<OraclePriceActi
 }
 
 pub fn index_active_price(services: &Arc<Services>, block: &BlockContext) -> Result<()> {
-    let block_interval = match Network::Regtest {
+    let network = ain_cpp_imports::get_network();
+    let block_interval = match Network::from_str(&network)? {
         Network::Regtest => 6,
         _ => 120,
     };
     if block.height % block_interval == 0 {
         let pt = services
             .price_ticker
             .by_id
-            .list(None, SortOrder::Ascending)?
+            .list(None, SortOrder::Descending)?
             .map(|item| {
                 let (_, priceticker) = item?;
                 Ok(priceticker)
@@ -112,124 +110,89 @@ pub fn index_active_price(services: &Arc<Services>, block: &BlockContext) -> Res
     Ok(())
 }
 
+fn map_active_price(
+    block: &BlockContext,
+    ticker_id: (String, String),
+    aggregated_price: OraclePriceAggregated,
+    prev_price: OraclePriceActive,
+) -> OraclePriceActive {
+    let next_price = if is_aggregate_valid(&aggregated_price, block) {
+        Some(aggregated_price.aggregated)
+    } else {
+        None
+    };
+
+    let active_price = if let Some(next) = prev_price.next {
+        Some(next)
+    } else {
+        prev_price.active
+    };
+
+    OraclePriceActive {
+        id: (ticker_id.0.clone(), ticker_id.1.clone(), block.height),
+        key: ticker_id,
+        sort: hex::encode(block.height.to_be_bytes()),
+        active: active_price.clone(),
+        next: next_price.clone(),
+        is_live: is_live(active_price, next_price),
+        block: block.clone(),
+    }
+}
+
 pub fn perform_active_price_tick(
     services: &Arc<Services>,
     ticker_id: (String, String),
     block: &BlockContext,
 ) -> Result<()> {
-    let aggregated_prices = services
-        .oracle_price_aggregated
+    let repo = &services.oracle_price_aggregated;
+    let prev_keys = repo
         .by_key
         .list(Some(ticker_id.clone()), SortOrder::Descending)?
-        .map(|item| {
-            let (_, id) = item?;
-            let aggregated = services
-                .oracle_price_aggregated
-                .by_id
-                .get(&id)?
-                .ok_or("Missing oracle previous history index")?;
-
-            Ok(aggregated)
-        })
-        .collect::<Result<Vec<_>>>()?;
-
-    log::debug!(
-        "set_loan_token indexing aggregated_price: {:?}",
-        aggregated_prices
-    );
-
-    if aggregated_prices.is_empty() {
+        .take(1)
+        .flatten() // return empty vec if none
+        .collect::<Vec<_>>();
+
+    if prev_keys.is_empty() {
         return Ok(());
     }
-    let aggregated_price = aggregated_prices.first().unwrap();
 
-    let previous_prices = services
-        .oracle_price_active
+    let Some((_, prev_id)) = prev_keys.first() else {
+        return Ok(());
+    };
+
+    let aggregated_price = repo.by_id.get(prev_id)?;
+
+    let Some(aggregated_price) = aggregated_price else {
+        return Ok(());
+    };
+
+    let repo = &services.oracle_price_active;
+    let prev_keys = repo
         .by_key
         .list(Some(ticker_id.clone()), SortOrder::Descending)?
         .take(1)
-        .map(|item| {
-            let (_, id) = item?;
-            let price = services
-                .oracle_price_active
-                .by_id
-                .get(&id)?
-                .ok_or("Missing oracle previous history index")?;
-            Ok(price)
-        })
-        .collect::<Result<Vec<_>>>()?;
-
-    let active_price = if previous_prices.first().is_some() {
-        if previous_prices[0].next.is_some() {
-            let price = previous_prices[0].next.clone().unwrap();
-            Some(OraclePriceActiveActive {
-                amount: price.amount,
-                weightage: price.weightage,
-                oracles: OraclePriceActiveActiveOracles {
-                    active: price.oracles.active,
-                    total: price.oracles.total,
-                },
-            })
-        } else if previous_prices[0].active.is_some() {
-            let price = previous_prices[0].active.clone().unwrap();
-            Some(OraclePriceActiveActive {
-                amount: price.amount,
-                weightage: price.weightage,
-                oracles: OraclePriceActiveActiveOracles {
-                    active: price.oracles.active,
-                    total: price.oracles.total,
-                },
-            })
-        } else {
-            None
-        }
-    } else {
-        None
-    };
+        .flatten()
+        .collect::<Vec<_>>();
 
-    let price_active_id = (
-        ticker_id.0.clone(),
-        ticker_id.1.clone(),
-        aggregated_price.block.height,
-    );
-
-    let next_price = if is_aggregate_valid(aggregated_price, block) {
-        Some(OraclePriceActiveNext {
-            amount: aggregated_price.aggregated.amount.clone(),
-            weightage: aggregated_price.aggregated.weightage,
-            oracles: OraclePriceActiveNextOracles {
-                active: aggregated_price.aggregated.oracles.active,
-                total: aggregated_price.aggregated.oracles.total,
-            },
-        })
-    } else {
-        None
+    if prev_keys.is_empty() {
+        return Ok(());
+    }
+
+    let Some((_, prev_id)) = prev_keys.first() else {
+        return Ok(());
     };
 
-    let oracle_price_active = OraclePriceActive {
-        id: price_active_id.clone(),
-        key: ticker_id,
-        sort: hex::encode(block.height.to_be_bytes()),
-        active: active_price.clone(),
-        next: next_price.clone(),
-        is_live: is_live(active_price, next_price),
-        block: block.clone(),
+    let prev_price = repo.by_id.get(prev_id)?;
+
+    let Some(prev_price) = prev_price else {
+        return Ok(());
     };
 
-    services
-        .oracle_price_active
-        .by_id
-        .put(&price_active_id, &oracle_price_active)?;
+    let active_price = map_active_price(block, ticker_id, aggregated_price, prev_price);
 
-    services
-        .oracle_price_active
-        .by_key
-        .put(&oracle_price_active.key, &oracle_price_active.id)?;
+    repo.by_id.put(&active_price.id, &active_price)?;
 
-    log::debug!(
-        "set_loan_token indexing oracle_price_active: {:?}",
-        oracle_price_active
-    );
+    repo.by_key.put(&active_price.key, &active_price.id)?;
 
     Ok(())
 }

lib/ain-ocean/src/indexer/oracle.rs

@@ -13,11 +13,10 @@ use rust_decimal_macros::dec;
 use crate::{
     indexer::{Context, Index, Result},
     model::{
-        BlockContext, Oracle, OracleHistory, OracleIntervalSeconds, OraclePriceAggregated,
-        OraclePriceAggregatedAggregated, OraclePriceAggregatedAggregatedOracles,
-        OraclePriceAggregatedInterval, OraclePriceAggregatedIntervalAggregated,
-        OraclePriceAggregatedIntervalAggregatedOracles, OraclePriceFeed, OracleTokenCurrency,
-        PriceFeedsItem, PriceTicker,
+        BlockContext, Oracle, OracleHistory, OracleIntervalSeconds, OraclePriceActiveNext,
+        OraclePriceActiveNextOracles, OraclePriceAggregated, OraclePriceAggregatedInterval,
+        OraclePriceAggregatedIntervalAggregated, OraclePriceAggregatedIntervalAggregatedOracles,
+        OraclePriceFeed, OracleTokenCurrency, PriceFeedsItem, PriceTicker,
     },
     repository::RepositoryOps,
     storage::SortOrder,
@@ -472,10 +471,10 @@ fn map_price_aggregated(
         ),
         token,
         currency,
-        aggregated: OraclePriceAggregatedAggregated {
+        aggregated: OraclePriceActiveNext {
             amount: format!("{:.8}", aggregated_amount),
             weightage: aggregated_weightage,
-            oracles: OraclePriceAggregatedAggregatedOracles {
+            oracles: OraclePriceActiveNextOracles {
                 active: aggregated_count,
                 total: oracles_len as i32,
             },
@@ -504,7 +503,7 @@ fn index_set_oracle_data(
             price_aggregated.currency.clone(),
         );
         let id = (key.0.clone(), key.1.clone(), price_aggregated.block.height);
-        // oracle_repo.by_key.put(&key, &id)?;
+        oracle_repo.by_key.put(&key, &id)?;
         oracle_repo.by_id.put(&id, &price_aggregated)?;
 
         let id = (

lib/ain-ocean/src/lib.rs

@@ -19,7 +19,7 @@ use repository::{
     OracleHistoryRepository, OracleHistoryRepositoryKey, OraclePriceActiveKeyRepository,
     OraclePriceActiveRepository, OraclePriceAggregatedIntervalKeyRepository,
     OraclePriceAggregatedIntervalRepository, OraclePriceAggregatedRepository,
-    OraclePriceAggregatedRepositorykey, OraclePriceFeedKeyRepository, OraclePriceFeedRepository,
+    OraclePriceAggregatedRepositoryKey, OraclePriceFeedKeyRepository, OraclePriceFeedRepository,
     OracleRepository, OracleTokenCurrencyKeyRepository, OracleTokenCurrencyRepository,
     PoolPairByHeightRepository, PoolPairRepository, PoolSwapAggregatedKeyRepository,
     PoolSwapAggregatedRepository, PoolSwapRepository, PriceTickerKeyRepository,
@@ -102,7 +102,7 @@ pub struct OraclePriceAggregatedIntervalService {
     by_id: OraclePriceAggregatedIntervalRepository,
 }
 pub struct OraclePriceAggregatedService {
-    by_key: OraclePriceAggregatedRepositorykey,
+    by_key: OraclePriceAggregatedRepositoryKey,
     by_id: OraclePriceAggregatedRepository,
 }
 
@@ -217,7 +217,7 @@ impl Services {
                 by_id: OraclePriceAggregatedIntervalRepository::new(Arc::clone(&store)),
             },
             oracle_price_aggregated: OraclePriceAggregatedService {
-                by_key: OraclePriceAggregatedRepositorykey::new(Arc::clone(&store)),
+                by_key: OraclePriceAggregatedRepositoryKey::new(Arc::clone(&store)),
                 by_id: OraclePriceAggregatedRepository::new(Arc::clone(&store)),
             },
             oracle_token_currency: OracleTokenCurrencyService {

lib/ain-ocean/src/model/oracle_price_active.rs

@@ -10,20 +10,12 @@ pub struct OraclePriceActive {
     pub id: OraclePriceActiveId,
     pub key: OraclePriceActiveKey,
     pub sort: String, //height
-    pub active: Option<OraclePriceActiveActive>,
+    pub active: Option<OraclePriceActiveNext>,
     pub next: Option<OraclePriceActiveNext>,
     pub is_live: bool,
     pub block: BlockContext,
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone)]
-#[serde(rename_all = "camelCase")]
-pub struct OraclePriceActiveActive {
-    pub amount: String,
-    pub weightage: u8,
-    pub oracles: OraclePriceActiveActiveOracles,
-}
-
 #[derive(Serialize, Deserialize, Debug, Clone, Default)]
 #[serde(rename_all = "camelCase")]
 pub struct OraclePriceActiveNext {
@@ -32,13 +24,6 @@ pub struct OraclePriceActiveNext {
     pub oracles: OraclePriceActiveNextOracles,
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone, Default)]
-#[serde(rename_all = "camelCase")]
-pub struct OraclePriceActiveActiveOracles {
-    pub active: i32,
-    pub total: i32,
-}
-
 #[derive(Serialize, Deserialize, Debug, Clone, Default)]
 #[serde(rename_all = "camelCase")]
 pub struct OraclePriceActiveNextOracles {