WIP: Ocean API database implementation to all modules (#2748)

* added block indexing

* added block indexing and query method

* added query methods and extra table

* added query methods to modules

* added oracle price active and prive feed module

* added oracle price active and prive feed module

* rebased indexing ocean-archive

* rebased indexing ocean-archive and updated query methods all modules

* rebased indexing ocean-archive and updated query methods all modules

* fixed white space

* fixed white space

* fixed query method for masternode and masterstats

lib/ain-evm/src/storage/block_store.rs

@@ -116,7 +116,6 @@ impl BlockStorage for BlockStore {
 
     fn put_block(&self, block: &BlockAny) -> Result<()> {
         self.extend_transactions_from_block(block)?;
-
         let block_number = block.header.number;
         let hash = block.header.hash();
         let blocks_cf = self.column::<columns::Blocks>();

lib/ain-ocean/Cargo.toml

@@ -4,6 +4,8 @@ version = "0.1.0"
 edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+[profile.test]
+debug = true
 
 [dependencies]
 ain-cpp-imports.workspace = true
@@ -31,4 +33,5 @@ bitcoin_hashes = "0.12.0"
 structopt = { version = "0.3", default-features = false }
 tempfile = "3.8.1"
 anyhow.workspace = true
+chrono = "0.4.31"
 cached.workspace = true

lib/ain-ocean/src/data_acces/block.rs

@@ -1,43 +1,171 @@
-use anyhow::{anyhow, Result};
-use serde::{Deserialize, Serialize};
-
 use crate::{
-    database::db_manager::{ColumnFamilyOperations, RocksDB},
+    database::db_manager::{ColumnFamilyOperations, RocksDB, SortOrder},
     model::block::Block,
 };
 
+use anyhow::{anyhow, Result};
+use rocksdb::IteratorMode;
+use std::convert::TryInto;
+
 #[derive(Debug)]
 pub struct BlockDb {
     pub db: RocksDB,
 }
 
 impl BlockDb {
-    pub async fn get_by_hash(&self) -> Result<Block> {
-        todo!()
+    pub async fn get_by_hash(&self, hash: String) -> Result<Option<Block>> {
+        let number = match self.db.get("block_map", hash.as_bytes()) {
+            Ok(Some(value)) => {
+                // Convert the stored bytes to a block number
+                let block_number_bytes: [u8; 4] = match value.try_into() {
+                    Ok(bytes) => bytes,
+                    Err(e) => {
+                        return Err(anyhow!("Error converting bytes to block number: {:?}", e))
+                    }
+                };
+                let block_number = i32::from_be_bytes(block_number_bytes);
+                Some(block_number)
+            }
+            Ok(None) => None,
+            Err(e) => return Err(anyhow!("Error retrieving block number: {:?}", e)),
+        };
+
+        if let Some(block_number) = number {
+            let block_key = block_number.to_be_bytes();
+            match self.db.get("block", &block_key) {
+                Ok(Some(value)) => {
+                    let block: Block = serde_json::from_slice(&value).map_err(|e| anyhow!(e))?;
+                    Ok(Some(block))
+                }
+                Ok(None) => Ok(None),
+                Err(e) => Err(anyhow!(e)),
+            }
+        } else {
+            Ok(None)
+        }
     }
-    pub async fn get_by_height(&self) -> Result<Block> {
-        todo!()
+
+    pub async fn get_by_height(&self, height: i32) -> Result<Option<Block>> {
+        match self.db.get("block", &height.to_be_bytes()) {
+            Ok(Some(value)) => {
+                let block: Block = serde_json::from_slice(&value).map_err(|e| anyhow!(e))?;
+                Ok(Some(block))
+            }
+            Ok(None) => Ok(None),
+            Err(e) => Err(anyhow!(e)),
+        }
     }
-    pub async fn get_heighest(&self) -> Result<Block> {
-        todo!()
+
+    pub async fn get_highest(&self) -> Result<Option<Block>> {
+        // Retrieve the latest block height
+        let latest_height_bytes = match self.db.get("latest_block_height", b"latest_block_height") {
+            Ok(Some(value)) => value,
+            Ok(None) => return Ok(None), // No latest block height set
+            Err(e) => return Err(anyhow!(e)),
+        };
+
+        // Convert the latest height bytes back to an integer
+        let latest_height = i32::from_be_bytes(
+            latest_height_bytes
+                .as_slice()
+                .try_into()
+                .map_err(|_| anyhow!("Byte length mismatch for latest height"))?,
+        );
+
+        // Retrieve the block with the latest height
+        match self.db.get("block", &latest_height.to_be_bytes()) {
+            Ok(Some(value)) => {
+                let block: Block = serde_json::from_slice(&value).map_err(|e| anyhow!(e))?;
+                Ok(Some(block))
+            }
+            Ok(None) => Ok(None), // No block found for the latest height
+            Err(e) => Err(anyhow!(e)),
+        }
     }
-    pub async fn query_by_height(&self, limit: i32, lt: i32) -> Result<Vec<Block>> {
-        todo!()
+
+    pub async fn query_by_height(
+        &self,
+        limit: i32,
+        lt: i32,
+        sort_order: SortOrder,
+    ) -> Result<Vec<Block>> {
+        let mut blocks: Vec<Block> = Vec::new();
+
+        let iterator = self.db.iterator("block", IteratorMode::End)?;
+        let collected_blocks: Vec<_> = iterator.collect();
+
+        for result in collected_blocks.into_iter().rev() {
+            let (key, value) = match result {
+                Ok((key, value)) => (key, value),
+                Err(err) => return Err(anyhow!("Error during iteration: {}", err)),
+            };
+
+            let block: Block = serde_json::from_slice(&value)?;
+
+            if block.height < lt {
+                blocks.push(block);
+
+                if blocks.len() == limit as usize {
+                    break;
+                }
+            }
+        }
+
+        // Sort blocks based on the specified sort order
+        match sort_order {
+            SortOrder::Ascending => blocks.sort_by(|a, b| a.height.cmp(&b.height)),
+            SortOrder::Descending => blocks.sort_by(|a, b| b.height.cmp(&a.height)),
+        }
+
+        Ok(blocks)
     }
-    pub async fn store_block(&self, block: Block) -> Result<()> {
+
+    pub async fn put_block(&self, block: Block) -> Result<()> {
         match serde_json::to_string(&block) {
             Ok(value) => {
-                let key = block.id.clone();
-                self.db.put("raw_block", key.as_bytes(), value.as_bytes())?;
+                let block_number = block.height;
+                self.db
+                    .put("block", &block_number.to_be_bytes(), value.as_bytes())?;
+                let block_map_key = block.hash.as_bytes();
+                self.db
+                    .put("block_map", block_map_key, &block_number.to_be_bytes())?;
+                self.db
+                    .delete("latest_block_height", b"latest_block_height")?;
+                self.db.put(
+                    "latest_block_height",
+                    b"latest_block_height",
+                    &block_number.to_be_bytes(),
+                )?;
                 Ok(())
             }
             Err(e) => Err(anyhow!(e)),
         }
     }
     pub async fn delete_block(&self, hash: String) -> Result<()> {
-        match self.db.delete("raw_block", hash.as_bytes()) {
-            Ok(_) => Ok(()),
-            Err(e) => Err(anyhow!(e)),
+        let number = match self.db.get("block_map", hash.as_bytes()) {
+            Ok(Some(value)) => {
+                // Convert the stored bytes to a block number
+                let block_number_bytes: [u8; 4] = match value.try_into() {
+                    Ok(bytes) => bytes,
+                    Err(e) => {
+                        return Err(anyhow!("Error converting bytes to block number: {:?}", e))
+                    }
+                };
+                let block_number = i32::from_be_bytes(block_number_bytes);
+                Some(block_number)
+            }
+            Ok(None) => None,
+            Err(e) => return Err(anyhow!("Error retrieving block number: {:?}", e)),
+        };
+
+        if let Some(block_number) = number {
+            let block_key = block_number.to_be_bytes();
+            match self.db.delete("block", &block_key) {
+                Ok(_) => Ok(()),
+                Err(e) => Err(anyhow!(e)),
+            }
+        } else {
+            Ok(())
         }
     }
 }

lib/ain-ocean/src/data_acces/masternode.rs

@@ -1,19 +1,47 @@
-use anyhow::{anyhow, Result};
-use serde::{Deserialize, Serialize};
-use serde_json;
-
 use crate::{
-    database::db_manager::{ColumnFamilyOperations, RocksDB},
+    database::db_manager::{ColumnFamilyOperations, RocksDB, SortOrder},
     model::masternode::Masternode,
 };
+use anyhow::{anyhow, Result};
+use rocksdb::IteratorMode;
+use serde_json;
 
 pub struct MasterNodeDB {
     pub db: RocksDB,
 }
 
 impl MasterNodeDB {
-    pub async fn query(&self, limit: i32, lt: i32) -> Result<Vec<Masternode>> {
-        todo!()
+    pub async fn query(
+        &self,
+        limit: i32,
+        lt: u32,
+        sort_order: SortOrder,
+    ) -> Result<Vec<Masternode>> {
+        let iter_mode: IteratorMode = sort_order.into();
+        let master_node: Result<Vec<_>> = self
+            .db
+            .iterator("masternode", iter_mode)?
+            .into_iter()
+            .take(limit as usize)
+            .map(|result| {
+                result
+                    .map_err(|e| {
+                        anyhow!("Error during iteration: {}", e)
+                            .context("error master_node query error")
+                    })
+                    .and_then(|(_key, value)| {
+                        let stats: Masternode = serde_json::from_slice(&value)?;
+                        if stats.block.height < lt {
+                            Ok(stats)
+                        } else {
+                            Err(anyhow!("Value is not less than lt")
+                                .context("Contextual error message"))
+                        }
+                    })
+            })
+            .collect();
+
+        master_node.and_then(|result| Ok(result))
     }
     pub async fn get(&self, id: String) -> Result<Option<Masternode>> {
         match self.db.get("masternode", id.as_bytes()) {