fix index_block_start and index_block_end

Author: canonbrother

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

@@ -32,16 +32,6 @@ impl Index for SetLoanToken {
     }
 }
 
-impl IndexBlockEnd for SetLoanToken {
-    fn index_block_end(self, services: &Arc<Services>, block: &BlockContext) -> Result<()> {
-        index_active_price(services, block)
-    }
-
-    fn invalidate_block_end(self, services: &Arc<Services>, block: &BlockContext) -> Result<()> {
-        invalidate_active_price(services, block)
-    }
-}
-
 fn is_aggregate_valid(aggregate: &OraclePriceAggregated, block: &BlockContext) -> bool {
     if (aggregate.block.time - block.time).abs() >= 3600 {
         return false;

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

@@ -19,6 +19,8 @@ use defichain_rpc::json::blockchain::{Block, Transaction, Vin, VinStandard, Vout
 use helper::check_if_evm_tx;
 use log::trace;
 pub use poolswap::PoolSwapAggregatedInterval;
+use loan_token::{index_active_price, invalidate_active_price};
+use poolswap::{index_pool_swap_aggregated, invalidate_pool_swap_aggregated};
 
 use crate::{
     error::{Error, IndexAction},
@@ -41,18 +43,6 @@ pub trait Index {
     fn invalidate(&self, services: &Arc<Services>, ctx: &Context) -> Result<()>;
 }
 
-pub trait IndexBlockStart: Index {
-    fn index_block_start(self, services: &Arc<Services>, block: &BlockContext) -> Result<()>;
-
-    fn invalidate_block_start(self, services: &Arc<Services>, block: &BlockContext) -> Result<()>;
-}
-
-pub trait IndexBlockEnd: Index {
-    fn index_block_end(self, services: &Arc<Services>, block: &BlockContext) -> Result<()>;
-
-    fn invalidate_block_end(self, services: &Arc<Services>, block: &BlockContext) -> Result<()>;
-}
-
 #[derive(Debug)]
 pub struct Context {
     block: BlockContext,
@@ -538,6 +528,22 @@ pub fn get_block_height(services: &Arc<Services>) -> Result<u32> {
         .map_or(0, |block| block.height))
 }
 
+pub fn index_block_start(services: &Arc<Services>, block: &BlockContext) -> Result<()> {
+    index_pool_swap_aggregated(services, block)
+}
+
+pub fn invalidate_block_start(services: &Arc<Services>, block: &BlockContext) -> Result<()> {
+    invalidate_pool_swap_aggregated(services, block)
+}
+
+pub fn index_block_end(services: &Arc<Services>, block: &BlockContext) -> Result<()> {
+    index_active_price(services, block)
+}
+
+pub fn invalidate_block_end(services: &Arc<Services>, block: &BlockContext) -> Result<()> {
+    invalidate_active_price(services, block)
+}
+
 pub fn index_block(services: &Arc<Services>, block: Block<Transaction>) -> Result<()> {
     trace!("[index_block] Indexing block...");
     let start = Instant::now();
@@ -586,12 +592,7 @@ pub fn index_block(services: &Arc<Services>, block: Block<Transaction>) -> Resul
         }
     }
 
-    // index_block_start
-    for (dftx, _) in &dftxs {
-        if let DfTx::PoolSwap(data) = dftx.clone() {
-            data.index_block_start(services, &block_ctx)?
-        }
-    }
+    index_block_start(services, &block_ctx)?;
 
     // index_dftx
     for (dftx, ctx) in &dftxs {
@@ -615,12 +616,7 @@ pub fn index_block(services: &Arc<Services>, block: Block<Transaction>) -> Resul
         log_elapsed(start, "Indexed dftx");
     }
 
-    // index_block_end
-    for (dftx, _) in dftxs {
-        if let DfTx::SetLoanToken(data) = dftx {
-            data.index_block_end(services, &block_ctx)?
-        }
-    }
+    index_block_end(services, &block_ctx)?;
 
     let block_mapper = BlockMapper {
         hash: block_hash,
@@ -699,12 +695,7 @@ pub fn invalidate_block(services: &Arc<Services>, block: Block<Transaction>) ->
         }
     }
 
-    // invalidate_block_end
-    for (dftx, _) in &dftxs {
-        if let DfTx::SetLoanToken(data) = dftx.clone() {
-            data.invalidate_block_end(services, &block_ctx)?
-        }
-    }
+    invalidate_block_end(services, &block_ctx)?;
 
     // invalidate_dftx
     for (dftx, ctx) in &dftxs {
@@ -727,12 +718,7 @@ pub fn invalidate_block(services: &Arc<Services>, block: Block<Transaction>) ->
         log_elapsed(start, "Invalidate dftx");
     }
 
-    // invalidate_block_start
-    for (dftx, _) in &dftxs {
-        if let DfTx::PoolSwap(data) = dftx.clone() {
-            data.invalidate_block_start(services, &block_ctx)?
-        }
-    }
+    invalidate_block_start(services, &block_ctx)?;
 
     // invalidate_block
     services.block.by_height.delete(&block.height)?;

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

@@ -182,67 +182,65 @@ fn create_new_bucket(
     Ok(())
 }
 
-impl IndexBlockStart for PoolSwap {
-    fn index_block_start(self, services: &Arc<Services>, block: &BlockContext) -> Result<()> {
-        let mut pool_pairs = ain_cpp_imports::get_pool_pairs();
-        pool_pairs.sort_by(|a, b| b.creation_height.cmp(&a.creation_height));
-
-        for interval in AGGREGATED_INTERVALS {
-            for pool_pair in &pool_pairs {
-                let repo = &services.pool_swap_aggregated;
-
-                let prev = repo
-                    .by_key
-                    .list(
-                        Some((pool_pair.id, interval, i64::MAX)),
-                        SortOrder::Descending,
-                    )?
-                    .take_while(|item| match item {
-                        Ok((k, _)) => k.0 == pool_pair.id && k.1 == interval,
-                        _ => true,
-                    })
-                    .next()
-                    .transpose()?;
-
-                let bucket = block.median_time - (block.median_time % interval as i64);
-
-                let Some((_, prev_id)) = prev else {
-                    create_new_bucket(repo, bucket, pool_pair.id, interval, block)?;
-                    continue;
-                };
+pub fn index_pool_swap_aggregated(services: &Arc<Services>, block: &BlockContext) -> Result<()> {
+    let mut pool_pairs = ain_cpp_imports::get_pool_pairs();
+    pool_pairs.sort_by(|a, b| b.creation_height.cmp(&a.creation_height));
 
-                let Some(prev) = repo.by_id.get(&prev_id)? else {
-                    create_new_bucket(repo, bucket, pool_pair.id, interval, block)?;
-                    continue;
-                };
-
-                if prev.bucket >= bucket {
-                    break;
-                }
+    for interval in AGGREGATED_INTERVALS {
+        for pool_pair in &pool_pairs {
+            let repo = &services.pool_swap_aggregated;
+
+            let prev = repo
+                .by_key
+                .list(
+                    Some((pool_pair.id, interval, i64::MAX)),
+                    SortOrder::Descending,
+                )?
+                .take_while(|item| match item {
+                    Ok((k, _)) => k.0 == pool_pair.id && k.1 == interval,
+                    _ => true,
+                })
+                .next()
+                .transpose()?;
+
+            let bucket = block.median_time - (block.median_time % interval as i64);
+
+            let Some((_, prev_id)) = prev else {
+                create_new_bucket(repo, bucket, pool_pair.id, interval, block)?;
+                continue;
+            };
 
+            let Some(prev) = repo.by_id.get(&prev_id)? else {
                 create_new_bucket(repo, bucket, pool_pair.id, interval, block)?;
+                continue;
+            };
+
+            if prev.bucket >= bucket {
+                break;
             }
-        }
 
-        Ok(())
+            create_new_bucket(repo, bucket, pool_pair.id, interval, block)?;
+        }
     }
 
-    fn invalidate_block_start(self, services: &Arc<Services>, block: &BlockContext) -> Result<()> {
-        let mut pool_pairs = ain_cpp_imports::get_pool_pairs();
-        pool_pairs.sort_by(|a, b| b.creation_height.cmp(&a.creation_height));
-
-        for interval in AGGREGATED_INTERVALS.into_iter().rev() {
-            for pool_pair in pool_pairs.iter().rev() {
-                let pool_swap_aggregated_id = (pool_pair.id, interval, block.hash);
-                services
-                    .pool_swap_aggregated
-                    .by_id
-                    .delete(&pool_swap_aggregated_id)?;
-            }
-        }
+    Ok(())
+}
 
-        Ok(())
+pub fn invalidate_pool_swap_aggregated(services: &Arc<Services>, block: &BlockContext) -> Result<()> {
+    let mut pool_pairs = ain_cpp_imports::get_pool_pairs();
+    pool_pairs.sort_by(|a, b| b.creation_height.cmp(&a.creation_height));
+
+    for interval in AGGREGATED_INTERVALS.into_iter().rev() {
+        for pool_pair in pool_pairs.iter().rev() {
+            let pool_swap_aggregated_id = (pool_pair.id, interval, block.hash);
+            services
+                .pool_swap_aggregated
+                .by_id
+                .delete(&pool_swap_aggregated_id)?;
+        }
     }
+
+    Ok(())
 }
 
 impl Index for PoolSwap {