riftenlabs-indexer/src/index.rs

// Copyright (C) 2024 Riften Labs AS
//
// This software is licensed under the GNU Affero General Public License (AGPL), version 3.0 or later.
// A copy of the license can be found in the LICENSE file or at https://www.gnu.org/licenses/agpl-3.0.html

use std::{
    collections::HashSet,
    sync::{mpsc::sync_channel, Arc, Mutex},
};

use bitcoin_hashes::hex::{FromHex, ToHex};
use bitcoincash::{consensus::deserialize, Block, BlockHash, Transaction, Txid};
use electrum_client::{Client, Param};
use log::{debug, info, warn};
use rayon::prelude::*;
use riftenlabs_defi::cauldron::{parse_cauldron, ParsedContract};

use crate::{
    bcmr::index_bcmr,
    chain::{get_new_headers, Chain, StoreBlockUndoer},
    db::{
        self,
        cauldron::{
            config::{config_get, config_set},
            header::{db_get_header, store_headers},
            tx::insert_block_tx,
            user::insert_user_action,
            utxo_funding::insert_utxo_funding,
            utxo_spending::insert_utxo_spending,
        },
        DBPool, DB,
    },
    electrum::{electrum_fetch_mempool, electrum_get_tip, electrum_get_tx},
    CASHTOKEN_ACTIVATION_HEIGHT, KEY_LAST_INDEXED,
};
use anyhow::{Context, Result};
use electrum_client::ElectrumApi;

fn parse_cauldrons(tx: &Transaction) -> Vec<ParsedContract> {
    tx.input
        .par_iter()
        .enumerate()
        .filter_map(move |(i, _)| parse_cauldron(i, tx))
        .collect()
}

pub fn update_mempool(db: DBPool, electrum: Arc<Mutex<Client>>) -> Result<()> {
    let our_mempool_txs: HashSet<Txid> = db::cauldron::mempool::load_mempool(&db.get().unwrap())?;
    let node_mempool: HashSet<Txid> = electrum_fetch_mempool(&electrum.lock().unwrap())?;

    let txs_to_delete = our_mempool_txs.difference(&node_mempool);
    let txs_to_add = node_mempool.difference(&our_mempool_txs);

    let txs_to_add: Vec<Transaction> = txs_to_add
        .into_iter()
        .filter_map(
            |txid| match electrum_get_tx(&electrum.lock().unwrap(), txid) {
                Ok(tx) => Some(tx),
                Err(e) => {
                    info!("Failed to get mempool tx {}: {}", txid, e);
                    None
                }
            },
        )
        .collect();

    let mut db_conn = db.get().unwrap();
    let db_tx = db_conn.transaction()?;

    let mut all_cauldrons = vec![];

    for txid in txs_to_delete {
        debug!("mempool remove {}", txid.to_hex());
        db::cauldron::mempool::delete_mempool_tx(&db_tx, txid)?;
    }

    for tx in txs_to_add {
        let txid = tx.txid();
        debug!("mempool add {}", txid.to_hex());
        db::cauldron::tx::insert_mempool_tx(&db_tx, &txid)?;
        let cauldrons: Vec<ParsedContract> = tx
            .input
            .iter()
            .enumerate()
            .filter_map(|(i, _)| parse_cauldron(i, &tx))
            .collect();

        insert_utxo_funding(&db_tx, &cauldrons, &txid, false)?;
        insert_utxo_spending(&db_tx, &cauldrons, &txid, false)?;
        insert_user_action(&db_tx, &cauldrons, &tx, true)?;

        all_cauldrons.extend(cauldrons);
    }

    db::cauldron::pool::update_pool_history(&db_tx, all_cauldrons)
        .context("update pool history")?;

    Ok(db_tx.commit()?)
}

pub fn index_blocks(
    chain: Arc<Mutex<Chain>>,
    db: DB,
    client: Arc<Mutex<Client>>,
    bcmr_enabled: bool,
) -> Result<BlockHash> {
    let (tip_header, _) = electrum_get_tip(&client.lock().unwrap())?;

    let (block_send, block_recv) = sync_channel::<Option<(u64, u64, Block)>>(10);

    // Update header chain (and undo any blocks that may have reorged away)
    {
        let chain = chain.lock().unwrap();
        if tip_header.block_hash() != chain.tip_hash() {
            debug!(
                "Updating header chain from {} to {}",
                chain.tip_hash().to_hex(),
                tip_header.block_hash().to_hex()
            );
            let new_headers =
                get_new_headers(&client.lock().unwrap(), &chain, &tip_header.block_hash())?;
            debug!("Storing headers");
            {
                let mut db_conn = db.cauldron_w.get().unwrap();
                for chunk in new_headers.chunks(100000) {
                    let db_tx = db_conn.transaction()?;
                    store_headers(&db_tx, chunk)?;
                    db_tx.commit()?;
                }
            }
            let undoer = StoreBlockUndoer::new(db.clone())?;
            chain.update(undoer, new_headers, None)?;
            debug!("Header update done");
        }
    }

    let db_cpy = db.clone();

    std::thread::spawn(move || {
        let db = db_cpy;

        let last_indexed = config_get(&db.cauldron_r.get().unwrap(), KEY_LAST_INDEXED).unwrap();
        let last_indexed = last_indexed.unwrap_or(CASHTOKEN_ACTIVATION_HEIGHT.to_string());
        let mut last_indexed = BlockHash::from_hex(&last_indexed).unwrap();

        // Check if last indexed has been orphaned
        loop {
            if chain.lock().unwrap().contains(&last_indexed) {
                break;
            }
            info!(
                "Last indexed block ({}) has been orphaned",
                last_indexed.to_hex()
            );
            let header = db_get_header(&db.cauldron_r.get().unwrap(), &last_indexed)
                .context("Failed to get header for last indexed")
                .unwrap();
            last_indexed = header.prev_blockhash;

            info!(
                "Last indexed block rolled back to {}",
                last_indexed.to_hex()
            );
        }

        loop {
            if tip_header.block_hash() == last_indexed {
                if let Err(e) = block_send.send(None) {
                    warn!("Failed to end EOL to block reader: {}", e);
                }
                break;
            }

            let next_height = chain
                .lock()
                .unwrap()
                .get_block_height(&last_indexed)
                .expect("last_indexed height not found in main chain")
                + 1;

            let res = client
                .lock()
                .unwrap()
                .raw_call("blockchain.block.get", vec![Param::U32(next_height as u32)])
                .unwrap();

            let block_hex: String = serde_json::from_str(&res.to_string()).unwrap();
            let block: Block = deserialize(&hex::decode(&block_hex).unwrap()).unwrap();
            let block_hash = block.block_hash();

            if let Err(e) = block_send.send(Some((
                next_height,
                chain.lock().unwrap().get_mtp(next_height).unwrap(),
                block,
            ))) {
                warn!("Failed to send block to reader: {}", e);
                break;
            }

            last_indexed = block_hash;
        }
    });

    loop {
        let (block_height, mtp, block) = match block_recv.recv()? {
            Some(res) => res,
            None => break,
        };

        let mut db_conn = db.cauldron_w.get().unwrap();
        let db_tx = db_conn.transaction()?;

        let blockhash = block.block_hash();

        let mut total_cauldrons = 0;

        let mut all_cauldrons = vec![];

        for tx in &block.txdata {
            let cauldrons = parse_cauldrons(tx);
            if cauldrons.is_empty() {
                continue;
            }

            let txid = tx.txid();
            insert_block_tx(&db_tx, &txid, &blockhash, mtp as i64).context("inserting block tx")?;
            insert_utxo_funding(&db_tx, &cauldrons, &txid, true)
                .context("inserting funding utxos")?;
            insert_utxo_spending(&db_tx, &cauldrons, &txid, true)
                .context("inserting spending utxos")?;
            insert_user_action(&db_tx, &cauldrons, tx, true).context("inserting user actions")?;

            total_cauldrons += cauldrons.len();

            all_cauldrons.extend(cauldrons);
        }

        // Figuring out initial utxo needs to be done on all cauldrons in a block.
        db::cauldron::pool::update_pool_history(&db_tx, all_cauldrons)
            .context("update pool history")?;
        config_set(&db_tx, KEY_LAST_INDEXED, &blockhash.to_hex());

        let autheader_updates = if bcmr_enabled {
            let mut conn = db.bcmr_w.get().context("failed to get bcmr db")?;
            let bcmr_db_tx = conn.transaction()?;
            let updates = index_bcmr(&bcmr_db_tx, &blockhash, block.txdata)?;
            bcmr_db_tx.commit()?;
            updates as i64
        } else {
            -1
        };

        // cauldron commit needs to come at last; as it tracks what the last successful block indexed was
        db_tx.commit()?;

        info!(
            "Indexed {}; mtp: {}, height {}, {} trades, {} autheader updates.",
            blockhash.to_hex(),
            mtp,
            block_height,
            total_cauldrons,
            autheader_updates,
        );
    }
    Ok(tip_header.block_hash())
}