riftenlabs-indexer/src/index.rs

// Copyright (C) 2024-2026 Whiterun LLC
//
// 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::{atomic::Ordering, Arc, Mutex},
    time::Duration,
};

use bitcoin_hashes::Hash;
use bitcoincash::{
    blockdata::block::Header as BlockHeader, consensus::deserialize, Block, BlockHash, Network,
    TokenID, Transaction, Txid,
};
use electrum_client_netagnostic::{Client, ElectrumApi, Param};
use log::{debug, info, warn};
use riftenlabs_defi::cauldron::{parse_cauldrons_from_tx, ParsedContract};
use riftenlabs_defi::moria::MoriaTokenIds;
use riftenlabs_defi::tokentoken::parse_tokentoken_from_tx;

use crate::{
    bcmr::index_bcmr,
    chain::{get_new_headers, Chain, StoreBlockUndoer},
    crc20::index_crc20,
    db::{
        self,
        blob::ToBlob,
        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,
        },
        oracle::index_oracle,
        DB,
    },
    electrum::{electrum_fetch_mempool, electrum_get_tip, electrum_get_tx},
    signal::shutdown_requested,
    timeutil::time_now,
    utiltx::ttor_sorted_kahn,
    IbdState, CASHTOKEN_ACTIVATION_HEIGHT, CHIPNET_START_BLOCK, KEY_LAST_INDEXED,
};
use anyhow::{bail, Context, Result};

/// Get the Moria token IDs for a given network
fn moria_token_ids(network: Option<Network>) -> MoriaTokenIds {
    match network {
        Some(Network::Chipnet) => MoriaTokenIds {
            moria: "29566f4884539dbcedfcb55cc7f5e66b5ae14975b70c0b6eb12de7e9707775ea"
                .parse::<TokenID>()
                .expect("valid chipnet moria token_id"),
            bp_oracle: "f3d6b85bfb0eaaf417ccabc8c8032464c5ec410e40b3b13f0c369a541bfb2a6a"
                .parse::<TokenID>()
                .expect("valid chipnet bp_oracle token_id"),
        },
        _ => MoriaTokenIds {
            moria: "b38a33f750f84c5c169a6f23cb873e6e79605021585d4f3408789689ed87f366"
                .parse::<TokenID>()
                .expect("valid mainnet moria token_id"),
            bp_oracle: "01711e39e7bf3b8ca0d9a6fc6ea32e340caa1d64dc7d1dc51fae20fd66755558"
                .parse::<TokenID>()
                .expect("valid mainnet bp_oracle token_id"),
        },
    }
}

pub async fn update_mempool(
    db: &DB,
    electrum: Arc<Mutex<Client>>,
    network: Option<Network>,
) -> Result<()> {
    let our_mempool_txs: HashSet<Txid> =
        db::cauldron::mempool::load_mempool(&db.cauldron_w).await?;

    let electrum_clone = electrum.clone();
    let (cauldron_txs, oracle_txs, ido_txs, bcmr_txs) = tokio::task::spawn_blocking(move || {
        electrum_fetch_mempool(&electrum_clone.lock().unwrap())
    })
    .await??;

    let txs_to_delete: Vec<Txid> = our_mempool_txs.difference(&cauldron_txs).cloned().collect();
    let txs_to_add: Vec<&Txid> = cauldron_txs.difference(&our_mempool_txs).collect();

    let electrum_for_fetch = electrum.clone();
    let txids_to_fetch: Vec<Txid> = txs_to_add.iter().map(|t| **t).collect();
    let txs_to_add: Vec<Transaction> = tokio::task::spawn_blocking(move || {
        txids_to_fetch
            .iter()
            .filter_map(
                |txid| match electrum_get_tx(&electrum_for_fetch.lock().unwrap(), txid) {
                    Ok(tx) => Some(tx),
                    Err(e) => {
                        info!("Failed to get mempool tx {txid}: {e}");
                        None
                    }
                },
            )
            .collect()
    })
    .await?;

    let txs_to_add = ttor_sorted_kahn(txs_to_add);

    {
        let mut db_tx = db.cauldron_w.begin().await?;

        db::cauldron::mempool::delete_mempool_txs(&mut db_tx, txs_to_delete.iter()).await?;

        let mut all_cauldrons = vec![];
        let mut all_tokentokens = vec![];
        let current_timestamp = time_now() as u64;

        for btx in txs_to_add {
            let cauldrons: Vec<ParsedContract> = parse_cauldrons_from_tx(&btx);
            let tokentokens = parse_tokentoken_from_tx(&btx);

            if cauldrons.is_empty() && tokentokens.is_empty() {
                info!("ignoring non-defi tx {}", btx.compute_txid());
                continue;
            }
            let txid = btx.compute_txid();
            db::cauldron::tx::insert_mempool_tx(&mut db_tx, &txid, current_timestamp).await?;
            info!(
                "mempool add {} with {} cauldrons, {} tokentoken states",
                txid,
                cauldrons.len(),
                tokentokens.len()
            );

            if !cauldrons.is_empty() {
                insert_utxo_funding(&mut db_tx, &cauldrons, &txid).await?;
                insert_utxo_spending(&mut db_tx, &cauldrons, &txid, false).await?;
                insert_user_action(&mut db_tx, &cauldrons, &btx, true).await?;

                all_cauldrons.extend(cauldrons);
            }
            all_tokentokens.extend(tokentokens);
        }

        db::cauldron::pool::update_pool_history(
            &mut db_tx,
            all_cauldrons,
            None,
            Some(current_timestamp),
        )
        .await?;
        db::cauldron::tokentoken::update_tokentoken_pool_history(
            &mut db_tx,
            all_tokentokens,
            None,
            Some(current_timestamp),
        )
        .await?;
        db_tx.commit().await?;
    }

    // oracle updates
    let oracle_electrum = electrum.clone();

    // filter entries we have
    let mut oracle_to_add = Vec::new();
    for txid in oracle_txs {
        if !db::oracle::has_entry(&db.oracle_w, &txid).await? {
            oracle_to_add.push(txid);
        }
    }

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

    index_oracle(&db.oracle_w, &txs_to_add, &BlockHash::all_zeros()).await?;

    // ido updates
    let ido_electrum = electrum.clone();

    // filter txs already recorded in an ido's state chain
    let mut ido_to_add = Vec::new();
    for txid in ido_txs {
        if !db::ido::has_indexed_tx(&db.ido_w, &txid.to_blob()).await? {
            ido_to_add.push(txid);
        }
    }

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

    // an ido chain can have several unconfirmed txs in flight; index parents first
    let txs_to_add = ttor_sorted_kahn(txs_to_add);
    db::ido::index_txs(network, &db.ido_w, &txs_to_add, None).await?;

    // bcmr updates: index newly registered BCMRs without waiting for a confirmation

    // Drop mempool-indexed entries whose tx left the mempool: either it
    // confirmed (index_blocks already re-stamped the entry with the real
    // blockhash, so it no longer matches the sentinel) or it was evicted or
    // replaced and must not linger as a stale auth head.
    for txid in db::bcmr::get_unconfirmed_txids(&db.bcmr_w).await? {
        if !bcmr_txs.contains(&txid) {
            db::bcmr::delete_unconfirmed_tx(&db.bcmr_w, &txid).await?;
        }
    }

    // filter txs already indexed into the auth chain
    let mut bcmr_to_add = Vec::new();
    for txid in bcmr_txs {
        if !db::bcmr::has_indexed_tx(&db.bcmr_w, &txid).await? {
            bcmr_to_add.push(txid);
        }
    }

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

    // an auth chain can have several unconfirmed txs in flight; index parents first
    let txs_to_add = ttor_sorted_kahn(txs_to_add);
    index_bcmr(&db.bcmr_w, &BlockHash::all_zeros(), txs_to_add).await?;

    Ok(())
}

pub async fn index_blocks(
    chain: Arc<Mutex<Chain>>,
    db: DB,
    client: Arc<Mutex<Client>>,
    bcmr_enabled: bool,
    network: Option<Network>,
    ibd_state: Option<Arc<IbdState>>,
    start_height: u64,
) -> Result<BlockHash> {
    let electrum_clone = client.clone();
    let (tip_header, tip_height) =
        tokio::task::spawn_blocking(move || electrum_get_tip(&electrum_clone.lock().unwrap()))
            .await??;

    // Update target height for IBD progress tracking
    if let Some(ref state) = ibd_state {
        state.target_height.store(tip_height, Ordering::Relaxed);
    }

    // Validate network before proceeding
    let start_block_hash = match network {
        Some(Network::Chipnet) => CHIPNET_START_BLOCK
            .parse::<BlockHash>()
            .map_err(|e| anyhow::anyhow!("Invalid CHIPNET_START_BLOCK: {}", e))?,
        Some(Network::Bitcoin) => CASHTOKEN_ACTIVATION_HEIGHT
            .parse::<BlockHash>()
            .map_err(|e| anyhow::anyhow!("Invalid CASHTOKEN_ACTIVATION_HEIGHT: {}", e))?,
        None => CASHTOKEN_ACTIVATION_HEIGHT
            .parse::<BlockHash>()
            .map_err(|e| anyhow::anyhow!("Invalid CASHTOKEN_ACTIVATION_HEIGHT: {}", e))?,
        Some(net) => bail!("Unknown network: {:?}", net),
    };

    let (block_send, mut block_recv) = tokio::sync::mpsc::channel::<Option<(u64, u64, Block)>>(10);

    // Update header chain (and undo any blocks that may have reorged away)
    {
        let needs_update = {
            let chain_guard = chain.lock().unwrap();
            tip_header.block_hash() != chain_guard.tip_hash()
        };
        if needs_update {
            {
                let chain_guard = chain.lock().unwrap();
                debug!(
                    "Updating header chain from {} to {}",
                    chain_guard.tip_hash(),
                    tip_header.block_hash()
                );
            }
            let client_for_headers = client.clone();
            let chain_for_headers = chain.clone();
            let tip_hash = tip_header.block_hash();
            let new_headers = tokio::task::spawn_blocking(move || {
                let chain_guard = chain_for_headers.lock().unwrap();
                get_new_headers(&client_for_headers.lock().unwrap(), &chain_guard, &tip_hash)
            })
            .await??;
            debug!("Storing headers");
            for chunk in new_headers.chunks(100000) {
                let mut db_tx = db.cauldron_w.begin().await?;
                store_headers(&mut db_tx, chunk).await?;
                db_tx.commit().await?;
            }
            // Run chain update in spawn_blocking because StoreBlockUndoer::undo_block
            // uses Handle::current().block_on() which panics on Tokio worker threads.
            let chain_for_update = chain.clone();
            let undoer_db = db.clone();
            tokio::task::spawn_blocking(move || {
                let chain_guard = chain_for_update.lock().unwrap();
                let undoer = StoreBlockUndoer::new(undoer_db)?;
                chain_guard.update(undoer, new_headers, None)
            })
            .await??;
            debug!("Header update done");
        }
    }

    // Before applying confirmed blocks, drop any mempool-indexed IDO state
    // (blockhash IS NULL). A mempool tx that never confirmed — e.g. one replaced
    // by a different on-chain tx — must not leave a stale snapshot that wins
    // MAX(seq) or resurfaces after the reorg undo above. Confirmed blocks
    // re-create rows for the txs they contain, and the next mempool pass
    // (update_mempool) rebuilds the rest.
    db::ido::delete_entries(&db.ido_w, None).await?;

    let db_cpy = db.clone();
    let chain_cpy = chain.clone();
    let client_cpy = client.clone();

    tokio::task::spawn_blocking(move || {
        let db = db_cpy;
        let handle = tokio::runtime::Handle::current();

        let last_indexed = handle
            .block_on(config_get(&db.cauldron_r, KEY_LAST_INDEXED))
            .unwrap();
        let mut last_indexed = if let Some(last) = last_indexed {
            if start_height > 0 {
                warn!("--start-height is ignored because database already has indexed blocks. Delete the database to re-index from a different height.");
            }
            last.parse::<BlockHash>().unwrap()
        } else if start_height > 0 {
            // Resolve configured start height to a block hash via electrum
            let resp = client_cpy
                .lock()
                .unwrap()
                .raw_call(
                    "blockchain.block.header",
                    vec![Param::U32(start_height as u32)],
                )
                .unwrap_or_else(|e| panic!("Failed to fetch header at height {start_height}: {e}"));
            let header_hex: String = serde_json::from_str(&resp.to_string()).unwrap();
            let header: BlockHeader = deserialize(&hex::decode(&header_hex).unwrap()).unwrap();
            info!(
                "Starting indexing from configured height {} ({})",
                start_height,
                header.block_hash()
            );
            header.block_hash()
        } else {
            start_block_hash
        };

        // Check if last indexed has been orphaned
        loop {
            if chain_cpy.lock().unwrap().contains(&last_indexed) {
                break;
            }
            info!("Last indexed block ({}) has been orphaned", last_indexed);
            let header = handle
                .block_on(db_get_header(&db.cauldron_r, &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);
        }

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

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

            let res = client_cpy
                .lock()
                .unwrap()
                .raw_call("blockchain.block.get", vec![Param::U32(next_height as u32)])
                .unwrap_or_else(|e| {
                    panic!("Failed to fetch block at height {next_height} from electrum: {e}")
                });

            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.blocking_send(Some((
                next_height,
                chain_cpy.lock().unwrap().get_mtp(next_height).unwrap(),
                block,
            ))) {
                warn!("Failed to send block to reader: {e}");
                break;
            }

            last_indexed = block_hash;
        }
    });

    loop {
        let recv_result = tokio::time::timeout(Duration::from_secs(1), block_recv.recv()).await;
        let (block_height, mtp, block) = match recv_result {
            Ok(Some(Some(data))) => data,
            Ok(Some(None)) | Ok(None) => break, // End of blocks signal or channel closed
            Err(_) => {
                // Timeout
                if shutdown_requested() {
                    info!("Shutdown requested, exiting block indexing");
                    break;
                }
                continue;
            }
        };

        let mut db_tx = db.cauldron_w.begin().await?;

        let blockhash = block.block_hash();

        let mut total_cauldrons = 0;

        let mut all_cauldrons = vec![];
        let mut all_tokentokens = vec![];

        let sorted_txs = ttor_sorted_kahn(block.txdata);

        for tx in &sorted_txs {
            let cauldrons = parse_cauldrons_from_tx(tx);
            // Note: parse_tokentoken_from_tx skips the pool-creation check when
            // any input spends an existing pool, so a single tx that both swaps
            // pool P and creates pool Q drops Q (upstream riftenlabs-defi
            // limitation, shared with parse_cauldrons_from_tx).
            let tokentokens = parse_tokentoken_from_tx(tx);
            if cauldrons.is_empty() && tokentokens.is_empty() {
                continue;
            }

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

                total_cauldrons += cauldrons.len();
                all_cauldrons.extend(cauldrons);
            }

            all_tokentokens.extend(tokentokens);
        }

        // Figuring out initial utxo needs to be done on all cauldrons in a block.
        db::cauldron::pool::update_pool_history(&mut db_tx, all_cauldrons, Some(mtp), None).await?;
        db::cauldron::tokentoken::update_tokentoken_pool_history(
            &mut db_tx,
            all_tokentokens,
            Some(mtp),
            None,
        )
        .await?;
        // config_set must be inside the cauldron transaction, as it tracks what the
        // last successful block indexed was. It must commit atomically with block data.
        config_set(&mut *db_tx, KEY_LAST_INDEXED, &blockhash.to_string()).await;

        // crc20
        index_crc20(&db.crc20_w, &sorted_txs).await?;

        // oracle updates
        index_oracle(&db.oracle_w, &sorted_txs, &blockhash).await?;

        // moria lending
        let moria_actions = db::moria::index_moria(
            &db.moria_w,
            &sorted_txs,
            &blockhash,
            mtp as i64,
            &moria_token_ids(network),
        )
        .await?;

        db::ido::index_txs(network, &db.ido_w, &sorted_txs, Some(&blockhash)).await?;

        let autheader_updates = if bcmr_enabled {
            let updates = index_bcmr(&db.bcmr_w, &blockhash, sorted_txs).await?;
            updates as i64
        } else {
            -1
        };

        db_tx.commit().await?;

        // Update IBD progress
        if let Some(ref state) = ibd_state {
            state.current_height.store(block_height, Ordering::Relaxed);
        }

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