Ocean: Indexer implementation

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

@@ -0,0 +1,52 @@
+use std::sync::Arc;
+
+use ain_dftx::vault::PlaceAuctionBid;
+use log::debug;
+
+use super::Context;
+use crate::{
+    indexer::{Index, Result},
+    model::VaultAuctionBatchHistory,
+    repository::RepositoryOps,
+    Services,
+};
+
+impl Index for PlaceAuctionBid {
+    fn index(self, services: &Arc<Services>, ctx: &Context) -> Result<()> {
+        debug!("[PlaceAuctionBid] Indexing...");
+
+        let auction = VaultAuctionBatchHistory {
+            id: format!("{}-{}-{}", self.vault_id, self.index, ctx.tx.txid),
+            key: format!("{}-{}", self.vault_id, self.index),
+            sort: format!("{}-{}", ctx.block.height, ctx.tx_idx),
+            vault_id: self.vault_id,
+            index: ctx.tx_idx,
+            from: self.from,
+            amount: self.token_amount.amount,
+            token_id: self.token_amount.token.0,
+            block: ctx.block.clone(),
+        };
+        debug!("auction : {:?}", auction);
+
+        let key = (self.vault_id, self.index, ctx.tx.txid);
+        services.auction.by_id.put(&key, &auction)?;
+        services.auction.by_height.put(
+            &(self.vault_id, self.index, ctx.block.height, ctx.tx_idx),
+            &key,
+        )
+    }
+
+    fn invalidate(&self, services: &Arc<Services>, ctx: &Context) -> Result<()> {
+        debug!("[PlaceAuctionBid] Invalidating...");
+        services
+            .auction
+            .by_id
+            .delete(&(self.vault_id, self.index, ctx.tx.txid))?;
+        services.auction.by_height.delete(&(
+            self.vault_id,
+            self.index,
+            ctx.block.height,
+            ctx.tx_idx,
+        ))
+    }
+}

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

@@ -0,0 +1,180 @@
+use std::sync::Arc;
+
+use ain_dftx::masternode::*;
+use bitcoin::{hashes::Hash, PubkeyHash, ScriptBuf, WPubkeyHash};
+use log::debug;
+use rust_decimal::{prelude::FromPrimitive, Decimal};
+
+use super::Context;
+use crate::{
+    error::Error,
+    indexer::{Index, Result},
+    model::{HistoryItem, Masternode, MasternodeStats, MasternodeStatsData, TimelockStats},
+    repository::RepositoryOps,
+    Services,
+};
+
+fn get_operator_script(hash: &PubkeyHash, r#type: u8) -> Result<ScriptBuf> {
+    match r#type {
+        0x1 => Ok(ScriptBuf::new_p2pkh(hash)),
+        0x4 => Ok(ScriptBuf::new_p2wpkh(&WPubkeyHash::hash(
+            hash.as_byte_array(),
+        ))),
+        _ => Err("Unsupported type".into()),
+    }
+}
+
+impl Index for CreateMasternode {
+    fn index(self, services: &Arc<Services>, ctx: &Context) -> Result<()> {
+        debug!("[CreateMasternode] Indexing...");
+        let txid = ctx.tx.txid;
+        let Some(ref addresses) = ctx.tx.vout[1].script_pub_key.addresses else {
+            return Err("Missing owner address".into());
+        };
+        let collateral = Decimal::from_f64(ctx.tx.vout[1].value).ok_or(Error::DecimalError)?;
+
+        let masternode = Masternode {
+            id: txid,
+            owner_address: addresses[0].clone(),
+            operator_address: get_operator_script(&self.operator_pub_key_hash, self.operator_type)?
+                .to_hex_string(),
+            creation_height: ctx.block.height,
+            resign_height: None,
+            resign_tx: None,
+            minted_blocks: 0,
+            timelock: self.timelock.0.unwrap_or_default(),
+            block: ctx.block.clone(),
+            collateral,
+            history: Vec::new(),
+        };
+
+        services.masternode.by_id.put(&txid, &masternode)?;
+        services
+            .masternode
+            .by_height
+            .put(&(ctx.block.height, txid), &0)?;
+
+        index_stats(&self, services, ctx, collateral)
+    }
+
+    fn invalidate(&self, services: &Arc<Services>, ctx: &Context) -> Result<()> {
+        debug!("[CreateMasternode] Invalidating...");
+        services.masternode.by_id.delete(&ctx.tx.txid)?;
+        services
+            .masternode
+            .by_height
+            .delete(&(ctx.block.height, ctx.tx.txid))
+    }
+}
+
+fn index_stats(
+    data: &CreateMasternode,
+    services: &Arc<Services>,
+    ctx: &Context,
+    collateral: Decimal,
+) -> Result<()> {
+    let mut stats = services
+        .masternode
+        .stats
+        .get_latest()?
+        .map_or(MasternodeStatsData::default(), |mn| mn.stats);
+
+    let count = stats.count + 1;
+    let tvl = stats.tvl + collateral;
+    let locked = stats
+        .locked
+        .entry(data.timelock.0.unwrap_or_default())
+        .or_insert_with(TimelockStats::default);
+
+    locked.count += 1;
+    locked.tvl += collateral;
+
+    services.masternode.stats.put(
+        &ctx.block.height,
+        &MasternodeStats {
+            stats: MasternodeStatsData {
+                count,
+                tvl,
+                locked: stats.clone().locked,
+            },
+            block: ctx.block.clone(),
+        },
+    )
+}
+
+impl Index for UpdateMasternode {
+    fn index(self, services: &Arc<Services>, ctx: &Context) -> Result<()> {
+        debug!("[UpdateMasternode] Indexing...");
+        if let Some(mut mn) = services.masternode.by_id.get(&self.node_id)? {
+            mn.history.push(HistoryItem {
+                owner_address: mn.owner_address.clone(),
+                operator_address: mn.operator_address.clone(),
+            });
+
+            for update in self.updates.as_ref() {
+                debug!("update : {:?}", update);
+                match update.r#type {
+                    0x1 => {
+                        if let Some(ref addresses) = ctx.tx.vout[1].script_pub_key.addresses {
+                            mn.owner_address = addresses[0].clone()
+                        }
+                    }
+                    0x2 => {
+                        if let Some(hash) = update.address.address_pub_key_hash {
+                            mn.operator_address =
+                                get_operator_script(&hash, update.address.r#type)?.to_hex_string()
+                        }
+                    }
+                    _ => (),
+                }
+            }
+
+            services.masternode.by_id.put(&self.node_id, &mn)?;
+        }
+        Ok(())
+    }
+
+    fn invalidate(&self, services: &Arc<Services>, _ctx: &Context) -> Result<()> {
+        debug!("[UpdateMasternode] Invalidating...");
+        if let Some(mut mn) = services.masternode.by_id.get(&self.node_id)? {
+            if let Some(history_item) = mn.history.pop() {
+                mn.owner_address = history_item.owner_address;
+                mn.operator_address = history_item.operator_address;
+            }
+
+            services.masternode.by_id.put(&self.node_id, &mn)?;
+        }
+        Ok(())
+    }
+}
+
+impl Index for ResignMasternode {
+    fn index(self, services: &Arc<Services>, ctx: &Context) -> Result<()> {
+        debug!("[ResignMasternode] Indexing...");
+        if let Some(mn) = services.masternode.by_id.get(&self.node_id)? {
+            services.masternode.by_id.put(
+                &self.node_id,
+                &Masternode {
+                    resign_height: Some(ctx.block.height),
+                    resign_tx: Some(ctx.tx.txid),
+                    ..mn
+                },
+            )?;
+        }
+        Ok(())
+    }
+
+    fn invalidate(&self, services: &Arc<Services>, _ctx: &Context) -> Result<()> {
+        debug!("[ResignMasternode] Invalidating...");
+        if let Some(mn) = services.masternode.by_id.get(&self.node_id)? {
+            services.masternode.by_id.put(
+                &self.node_id,
+                &Masternode {
+                    resign_height: None,
+                    ..mn
+                },
+            )?;
+        }
+        Ok(())
+    }
+}

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

@@ -1,9 +1,128 @@
+mod auction;
+mod masternode;
+mod oracle;
+mod pool;
+pub mod transaction;
+pub mod tx_result;
+
+use std::{sync::Arc, time::Instant};
+
+use ain_dftx::{deserialize, DfTx, Stack};
 use defichain_rpc::json::blockchain::{Block, Transaction};
-use std::sync::Arc;
+use log::debug;
+
+use crate::{
+    model::{Block as BlockMapper, BlockContext},
+    repository::RepositoryOps,
+    Result, Services,
+};
+use transaction::index_transaction;
 
-use crate::{Result, Services};
+pub(crate) trait Index {
+    fn index(self, services: &Arc<Services>, ctx: &Context) -> Result<()>;
+
+    fn invalidate(&self, services: &Arc<Services>, ctx: &Context) -> Result<()>;
+}
+
+#[derive(Debug)]
+pub struct Context {
+    block: BlockContext,
+    tx: Transaction,
+    tx_idx: usize,
+}
+
+fn log_elapsed(previous: Instant, msg: &str) {
+    let now = Instant::now();
+    debug!("{} in {} ms", msg, now.duration_since(previous).as_millis());
+}
 
 pub fn index_block(services: &Arc<Services>, block: Block<Transaction>) -> Result<()> {
+    debug!("[index_block] Indexing block...");
+    let start = Instant::now();
+
+    let block_hash = block.hash;
+    let transaction_count = block.tx.len();
+    let block_ctx = BlockContext {
+        height: block.height,
+        hash: block_hash,
+        time: block.time,
+        median_time: block.mediantime,
+    };
+
+    for (tx_idx, tx) in block.tx.into_iter().enumerate() {
+        let start = Instant::now();
+        let ctx = Context {
+            block: block_ctx.clone(),
+            tx,
+            tx_idx,
+        };
+
+        let bytes = &ctx.tx.vout[0].script_pub_key.hex;
+        if bytes.len() > 6 && bytes[0] == 0x6a && bytes[1] <= 0x4e {
+            let offset = 1 + match bytes[1] {
+                0x4c => 2,
+                0x4d => 3,
+                0x4e => 4,
+                _ => 1,
+            };
+
+            let raw_tx = &bytes[offset..];
+            match deserialize::<Stack>(raw_tx) {
+                Err(bitcoin::consensus::encode::Error::ParseFailed("Invalid marker")) => {
+                    println!("Discarding invalid marker");
+                }
+                Err(e) => return Err(e.into()),
+                Ok(Stack { dftx, .. }) => {
+                    match dftx {
+                        DfTx::CreateMasternode(data) => data.index(services, &ctx)?,
+                        DfTx::UpdateMasternode(data) => data.index(services, &ctx)?,
+                        DfTx::ResignMasternode(data) => data.index(services, &ctx)?,
+                        // DfTx::AppointOracle(data) => data.index(services,&ctx)?,
+                        // DfTx::RemoveOracle(data) => data.index(services,&ctx)?,
+                        // DfTx::UpdateOracle(data) => data.index(services,&ctx)?,
+                        // DfTx::SetOracleData(data) => data.index(services,&ctx)?,
+                        DfTx::PoolSwap(data) => data.index(services, &ctx)?,
+                        DfTx::CompositeSwap(data) => data.index(services, &ctx)?,
+                        DfTx::PlaceAuctionBid(data) => data.index(services, &ctx)?,
+                        _ => (),
+                    }
+                    log_elapsed(start, "Indexed dftx");
+                }
+            }
+        }
+
+        index_transaction(services, ctx)?;
+    }
+
+    log_elapsed(start, "Indexed block");
+
+    let block_mapper = BlockMapper {
+        hash: block_hash,
+        id: block_hash,
+        previous_hash: block.previousblockhash,
+        height: block.height,
+        version: block.version,
+        time: block.time,
+        median_time: block.mediantime,
+        transaction_count,
+        difficulty: block.difficulty,
+        masternode: block.masternode,
+        minter: block.minter,
+        minter_block_count: block.minted_blocks,
+        stake_modifier: block.stake_modifier.to_owned(),
+        merkleroot: block.merkleroot,
+        size: block.size,
+        size_stripped: block.strippedsize,
+        weight: block.weight,
+    };
+
+    // services.block.raw.put(&ctx.hash, &encoded_block)?; TODO
+    services.block.by_id.put(&block_ctx.hash, &block_mapper)?;
+    services
+        .block
+        .by_height
+        .put(&block_ctx.height, &block_hash)?;
+
     Ok(())
 }