riftenlabs-indexer/src/main.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 anyhow::{bail, Result};
use bcmr::wellknowndowloader::WellKnownDownloader;
use bitcoincash::{consensus::deserialize, Block, BlockHash, Network};
use crc20::crc20fetcher::CRC20Fetcher;
use db::cauldron::tokenlist::db_utils::create_cached_token_metrics_table;
use db::DB;
use electrum::electrum_get_tip;
use electrum_client_netagnostic::{Client, ElectrumApi, Param};
use log::{error, info, warn};
use rocket::{launch, routes};
use rocket_cors::{AllowedHeaders, AllowedOrigins};
use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU64, Ordering};
use std::{
    backtrace::Backtrace,
    panic, process,
    sync::{Arc, Mutex},
    time::Duration,
};
use stderrlog::LogLevelNum;

/// Tracks how far the OHLCV pre-aggregation table has been populated.
/// `materialized_end` is the exclusive upper bound: `ohlcv_1h` contains
/// data for all complete 1-hour buckets whose `bucket_ts + 3600 ≤ materialized_end`.
/// Value 0 means nothing has been materialised yet.
pub struct OhlcvState {
    pub materialized_end: AtomicI64,
}

/// State tracking for Initial Block Download (IBD).
/// Used to return 503 errors while the indexer is catching up.
pub struct IbdState {
    /// Set to true once the initial sync completes
    pub initial_sync_complete: AtomicBool,
    /// Current block height being indexed
    pub current_height: AtomicU64,
    /// Target chain tip height
    pub target_height: AtomicU64,
    /// If true, serve requests even during IBD (for debugging)
    pub serve_during_ibd: bool,
}

use crate::bcmr::bcmrdownloader::BCMRDownloader;
use crate::db::cauldron::header::load_all_headers;
use crate::db::cauldron::tokenlist::metrics_cache::spawn_token_metrics_updater;
use crate::db::init::{initialize_databases, ReadSlots};
use crate::index::{index_blocks, update_mempool};

#[macro_use]
extern crate configure_me;

include_config!();

// The block where first cauldron contract was deployed. (Block 799870)
#[allow(dead_code)]
const RIFTEN_LABS_GENESIS_BLOCK: &str =
    "000000000000000000ed24c811077f7268a21ecf25cb437655aaba33d8ff4997";

// Start parsing for BCMR data from this height
const CASHTOKEN_ACTIVATION_HEIGHT: &str =
    "000000000000000002b678c471841c3e404ec7ae9ca9c32026fe27eb6e3a1ed1";

// Chipnet genesis
const CHIPNET_START_BLOCK: &str =
    "000000001dd410c49a788668ce26751718cc797474d3152a5fc073dd44fd9f7b";

// Last indexed block height.
const KEY_LAST_INDEXED: &str = "last_indexed";

mod bcmr;
mod cashaddr;
mod chain;
mod crc20;
mod db;
mod def;
mod electrum;
mod index;
mod rpc;
mod signal;
mod timeutil;
mod utiltest;
mod utiltoken;
mod utiltx;

fn set_panic_hook() {
    panic::set_hook(Box::new(|panic_info| {
        error!("A thread panicked, terminating the program.");
        if let Some(error) = panic_info.payload().downcast_ref::<anyhow::Error>() {
            error!("Panic occurred: {error:?}");
            error!("Anyhow backtrace:\n{}", error.backtrace());
            let mut source = error.source();
            while let Some(cause) = source {
                error!("Caused by: {cause:?}");
                source = cause.source();
            }
        } else if let Some(message) = panic_info.payload().downcast_ref::<&str>() {
            error!("Panic occurred: {message}");
        } else if let Some(message) = panic_info.payload().downcast_ref::<String>() {
            error!("Panic occurred: {message}");
        } else {
            error!("Panic info: {panic_info:?}");
        }

        let backtrace = Backtrace::capture();
        error!("Backtrace (if RUST_BACKTRACE=1):\n{backtrace}");
        process::exit(1);
    }));
}

async fn start_program(
    config: Config,
) -> Result<(
    DB,
    BCMRDownloader,
    WellKnownDownloader,
    CRC20Fetcher,
    Arc<IbdState>,
    Arc<AtomicBool>, // indexing_in_progress
)> {
    let network = match config.network.to_lowercase().as_str() {
        "mainnet" => Network::Bitcoin,
        "chipnet" => Network::Chipnet,
        _ => bail!(
            "Invalid network '{}'. Must be 'mainnet' or 'chipnet'",
            config.network
        ),
    };

    info!("Using network: {:?}", network);

    let rostrum_addr = if config.rostrum_addr.is_empty() {
        match network {
            Network::Chipnet => "127.0.0.1:64001".to_string(),
            _ => "127.0.0.1:50001".to_string(),
        }
    } else {
        config.rostrum_addr
    };

    // Normalize address: if no scheme, default to tcp:// for backwards compatibility.
    // Supported schemes: tcp://, ssl://, ws://, wss://
    let electrum_url = {
        let s = rostrum_addr.trim();
        if s.starts_with("tcp://")
            || s.starts_with("ssl://")
            || s.starts_with("ws://")
            || s.starts_with("wss://")
        {
            s.to_string()
        } else {
            format!("tcp://{}", s)
        }
    };

    let network_str = match network {
        Network::Bitcoin => "mainnet",
        Network::Chipnet => "chipnet",
        _ => "mainnet",
    };
    let db = initialize_databases(
        network_str,
        ReadSlots {
            cauldron: config.cauldron_read_slots,
            bcmr: config.bcmr_read_slots,
            crc20: config.crc20_read_slots,
            oracle: config.oracle_read_slots,
            moria: config.moria_read_slots,
            ido: config.ido_read_slots,
        },
    )
    .await?;

    let indexing_in_progress = Arc::new(AtomicBool::new(false));

    let ibd_state = Arc::new(IbdState {
        initial_sync_complete: AtomicBool::new(false),
        current_height: AtomicU64::new(0),
        target_height: AtomicU64::new(0),
        serve_during_ibd: config.serve_during_ibd,
    });

    let client = Arc::new(Mutex::new(match Client::new(&electrum_url) {
        Ok(server) => server,
        Err(e) => {
            error!(
                "Failed to connect to {}: {}. See --help for setting a different server.",
                electrum_url, e
            );
            bail!(e)
        }
    }));

    let genesis = match client
        .lock()
        .unwrap()
        .raw_call("blockchain.block.get", vec![Param::U32(0)])
    {
        Ok(g) => g,
        Err(e) => {
            error!(
                "Failed to fetch genesis block from {}: {}. See --help for setting a different server.",
                electrum_url, e
            );
            bail!(e)
        }
    };
    let genesis: Block = deserialize(&hex::decode(genesis.as_str().unwrap()).unwrap()).unwrap();
    let chain = Arc::new(Mutex::new(chain::Chain::new(genesis.header)));

    db::cauldron::pool::initialize_seq(&db.cauldron_r).await;
    db::cauldron::tokentoken::initialize_seq(&db.cauldron_r).await;
    db::cauldron::tokenbch::initialize_seq(&db.cauldron_r).await;

    info!("Loading block headers...");
    let all_headers = load_all_headers(&db.cauldron_r).await.unwrap();
    info!("Initializing {} headers...", all_headers.len());
    chain.lock().unwrap().load(all_headers).unwrap();
    info!("Headers loaded.");

    let db_cpy = db.clone();

    let mut crc20fetcher = CRC20Fetcher::new();
    crc20fetcher.start(
        db.crc20_w.clone(),
        db.bcmr_r.clone(),
        client.clone(),
        indexing_in_progress.clone(),
    )?;

    db::oracle::clear_mempool(&db.oracle_w).await.unwrap();
    db::moria_v11::clear_mempool(&db.moria_w).await.unwrap();
    db::bcmr::clear_mempool(&db.bcmr_w).await.unwrap();

    let indexing_in_progress_clone = indexing_in_progress.clone();
    let ibd_state_clone = ibd_state.clone();
    let start_height = config.start_height;

    tokio::spawn(async move {
        let db = db_cpy;

        {
            let client_lock = client.lock().unwrap();
            if let Ok(tip_info) = electrum_get_tip(&client_lock) {
                let tip_height = chain.lock().unwrap().height();
                ibd_state_clone
                    .target_height
                    .store(tip_height.max(tip_info.1), Ordering::Relaxed);
            }
        }

        // Initial full index
        let mut tip: BlockHash = loop {
            if signal::shutdown_requested() {
                info!("Shutdown requested, exiting indexing task during initial index");
                return;
            }
            indexing_in_progress_clone.store(true, Ordering::Relaxed);
            break match index_blocks(
                chain.clone(),
                db.clone(),
                client.clone(),
                true,
                Some(network),
                Some(ibd_state_clone.clone()),
                start_height,
            )
            .await
            {
                Ok(tip) => tip,
                Err(e) => {
                    if e.to_string().contains("database is locked") {
                        warn!("initial index error, trying again: {e}");
                        continue;
                    }
                    panic!("Initial index failed: {}\n {}", e, e.backtrace());
                }
            };
        };
        indexing_in_progress_clone.store(false, Ordering::Relaxed);
        info!("Initial block download complete");

        // Run ANALYZE before signalling initial_sync_complete so the ohlcv post-IBD
        // backfill (which waits for that flag) doesn't race with this write.
        info!("Running ANALYZE on cauldron database...");
        if let Err(e) = sqlx::query("ANALYZE;").execute(&db.cauldron_w).await {
            warn!("ANALYZE failed: {e}");
        } else {
            info!("ANALYZE complete");
        }

        ibd_state_clone
            .initial_sync_complete
            .store(true, Ordering::Relaxed);

        // Follow chain
        loop {
            if signal::shutdown_requested() {
                info!("Shutdown requested, exiting indexing task");
                return;
            }

            let new_tip_result = {
                let client_lock = client.lock().unwrap();
                electrum_get_tip(&client_lock).map(|t| t.0.block_hash())
            };
            let new_tip = match new_tip_result {
                Ok(t) => t,
                Err(e) => {
                    warn!("Failed to get chain tip from electrum: {e}");
                    tokio::time::sleep(Duration::from_secs(5)).await;
                    continue;
                }
            };

            if new_tip != tip {
                indexing_in_progress_clone.store(true, Ordering::Relaxed);
                tip = match index_blocks(
                    chain.clone(),
                    db.clone(),
                    client.clone(),
                    true,
                    Some(network),
                    None,
                    0, // start_height only matters for initial sync
                )
                .await
                {
                    Ok(t) => t,
                    Err(e) => {
                        warn!("Indexing block failed: {} {}", e, e.backtrace());
                        tip
                    }
                };
                indexing_in_progress_clone.store(false, Ordering::Relaxed);
            }

            // Avoid overlapping writer while indexer is on
            if !indexing_in_progress_clone.load(Ordering::Relaxed) {
                if let Err(e) = update_mempool(&db, client.clone(), Some(network)).await {
                    error!("Failed to update mempool: {e}");
                }
            }
            tokio::time::sleep(Duration::from_secs(5)).await;
        }
    });

    let mut bcmrdownloader =
        BCMRDownloader::new(db.bcmr_w.clone(), config.riften_ipfs_gateway.clone());
    bcmrdownloader.start()?;

    let mut wellknowndownloader = WellKnownDownloader::new(db.bcmr_w.clone());
    wellknowndownloader.start()?;

    spawn_token_metrics_updater(db.clone(), indexing_in_progress.clone());

    Ok((
        db,
        bcmrdownloader,
        wellknowndownloader,
        crc20fetcher,
        ibd_state,
        indexing_in_progress,
    ))
}

#[launch]
async fn launch() -> _ {
    stderrlog::new()
        .verbosity(LogLevelNum::Info)
        .init()
        .unwrap();

    set_panic_hook();

    let config = {
        let (config, _extra) =
            Config::including_optional_config_files(std::iter::empty::<std::ffi::OsString>())
                .unwrap_or_exit();
        config
    };

    let (
        dbpool,
        bcmrdownloader,
        wellknowndownloader,
        crc20fetcher,
        ibd_state,
        indexing_in_progress,
    ) = match start_program(config).await {
        Ok(db) => db,
        Err(e) => {
            let backtrace = Backtrace::capture();
            error!("Backtrace (if RUST_BACKTRACE=1):\n{backtrace}");
            error!("Error: {e}");
            panic!("Failed at program startup")
        }
    };
    let allowed_origins = AllowedOrigins::all();

    let cors = rocket_cors::CorsOptions {
        allowed_origins,
        allowed_methods: vec![rocket::http::Method::Get]
            .into_iter()
            .map(From::from)
            .collect(),
        allowed_headers: AllowedHeaders::some(&["Authorization", "Accept"]),
        allow_credentials: true,
        ..Default::default()
    }
    .to_cors()
    .unwrap();

    create_cached_token_metrics_table(&dbpool.cauldron_w)
        .await
        .expect("ensure cached_token_metrics exists");

    // Ensure the OHLCV pre-aggregation table exists (safe on both new and existing DBs).
    db::cauldron::ohlcv::create_table(&dbpool.cauldron_w).await;

    // Discard buckets materialised under a superseded pricing rule.
    let ohlcv_wiped =
        match db::cauldron::ohlcv::migrate_if_stale(&dbpool.cauldron_r, &dbpool.cauldron_w).await {
            Ok(wiped) => {
                if wiped {
                    info!(
                        "ohlcv: cleared for rebuild at version {}; \
                         serving the raw path until the background task catches up",
                        db::cauldron::ohlcv::OHLCV_VERSION
                    );
                }
                wiped
            }
            Err(e) => {
                warn!("ohlcv: version check failed, leaving table as-is: {e}");
                false
            }
        };

    // Bootstrap OhlcvState from whatever is already in the table (survives restarts).
    let max_bucket_ts = db::cauldron::ohlcv::get_max_bucket_ts(&dbpool.cauldron_r)
        .await
        .unwrap_or(None);
    let initial_ohlcv_end = max_bucket_ts.map(|ts| ts + 3600).unwrap_or(0);
    let ohlcv_state = Arc::new(OhlcvState {
        materialized_end: AtomicI64::new(initial_ohlcv_end),
    });

    // Post-IBD backfill: run the full ohlcv_1h materialisation before allowing
    // metrics_cache and other background writers to start.  We reuse the
    // indexing_in_progress flag so metrics_cache backs off during this window.
    //
    // Runs in a task: it must WAIT for IBD, and main() sits before
    // rocket::build(), so awaiting it inline kept the API from binding for the
    // entire initial sync — on a fresh database the server refused connections
    // for hours.  The gate it holds preserves the old writer ordering.
    //
    // Skipped after a version wipe: re-materialising all of history would
    // contend with the rebuild rather than degrading to the (correct, slower)
    // raw path.
    if !ohlcv_wiped {
        let dbpool = dbpool.clone();
        let ibd_state = ibd_state.clone();
        let indexing_in_progress = indexing_in_progress.clone();
        let ohlcv_state = ohlcv_state.clone();
        tokio::spawn(async move {
            const BACKFILL_BATCH_SECS: i64 = 24 * 3600;
            const BACKFILL_SAFETY_SECS: i64 = 3 * 3600;

            // Wait for IBD to finish — ohlcv_1h data is only useful for confirmed blocks.
            while !ibd_state.initial_sync_complete.load(Ordering::Relaxed) {
                tokio::time::sleep(Duration::from_secs(1)).await;
            }

            // Gate metrics_cache so it doesn't compete for cauldron_w during backfill.
            indexing_in_progress.store(true, Ordering::Relaxed);
            info!("ohlcv: starting post-IBD full backfill");

            let now = crate::timeutil::time_now();
            let cutoff = (now - BACKFILL_SAFETY_SECS) / 3600 * 3600;
            let since_opt = match max_bucket_ts {
                Some(ts) => Some(ts + 3600),
                None => {
                    match db::cauldron::ohlcv::get_min_trade_bucket_ts(&dbpool.cauldron_r).await {
                        Ok(v) => v,
                        Err(e) => {
                            warn!("ohlcv backfill: could not read min trade ts: {e}");
                            None
                        }
                    }
                }
            };
            if since_opt.is_none() {
                info!("ohlcv backfill: no confirmed trades found, skipping");
            }
            if let Some(mut batch_start) = since_opt {
                while batch_start < cutoff {
                    let batch_end = (batch_start + BACKFILL_BATCH_SECS).min(cutoff);
                    match db::cauldron::ohlcv::rebuild_range(
                        &dbpool.cauldron_r,
                        &dbpool.cauldron_w,
                        batch_start,
                        batch_end,
                    )
                    .await
                    {
                        Ok(n) => {
                            info!("ohlcv backfill: {n} buckets [{batch_start}, {batch_end})");
                            ohlcv_state
                                .materialized_end
                                .store(batch_end, Ordering::Relaxed);
                        }
                        Err(e) => {
                            warn!("ohlcv backfill failed at [{batch_start}, {batch_end}): {e}");
                            break;
                        }
                    }
                    batch_start = batch_end;
                    // Brief yield so new block writes are not starved.
                    tokio::time::sleep(Duration::from_millis(200)).await;
                }
            }

            info!("ohlcv: post-IBD backfill complete");
            indexing_in_progress.store(false, Ordering::Relaxed);
        });
    }

    // Background task: incrementally materialise new 1-hour OHLCV buckets as blocks arrive.
    // After a version wipe this is also what repopulates history, since the synchronous
    // backfill above is skipped in that case.
    // Only processes buckets older than 3 hours (well beyond BCH reorg depth).
    {
        let ohlcv_write = dbpool.cauldron_w.clone();
        let ohlcv_read = dbpool.cauldron_r.clone();
        let ohlcv_state_bg = ohlcv_state.clone();
        let ohlcv_ibd = ibd_state.clone();
        tokio::spawn(async move {
            // Batch size: 1 day per SQL call to keep each write short.
            const BATCH_SECS: i64 = 24 * 3600;
            // Safety margin: only materialise buckets older than this many seconds.
            const SAFETY_SECS: i64 = 3 * 3600;

            // Wait for IBD before materialising anything.
            //
            // The synchronous backfill above waits too, but it is skipped whenever
            // `ohlcv_wiped` is set — which includes every fresh database, since a missing
            // version key reads as stale.  Without this the task would sweep from the
            // first trade all the way to `now - 3h` while indexing is still years behind,
            // writing nothing, contending with block writes for the cauldron write lock,
            // and advancing `materialized_end` to roughly now against an empty table — at
            // which point `candlesticks()` would take the fast path over nothing.
            while !ohlcv_ibd.initial_sync_complete.load(Ordering::Relaxed) {
                tokio::time::sleep(Duration::from_secs(1)).await;
            }

            loop {
                let now = crate::timeutil::time_now();
                // Floor to 1-hour boundary, 3 hours ago.
                let cutoff = (now - SAFETY_SECS) / 3600 * 3600;

                let since = match db::cauldron::ohlcv::get_max_bucket_ts(&ohlcv_read).await {
                    Ok(Some(max_ts)) => max_ts + 3600,
                    Ok(None) => {
                        // Table is empty: start from the first confirmed trade rather than
                        // scanning from Unix epoch 0 through thousands of empty batches.
                        match db::cauldron::ohlcv::get_min_trade_bucket_ts(&ohlcv_read).await {
                            Ok(Some(min_ts)) => min_ts,
                            Ok(None) => {
                                // No confirmed trades yet; wait before retrying.
                                tokio::time::sleep(Duration::from_secs(60)).await;
                                continue;
                            }
                            Err(e) => {
                                error!("ohlcv rebuild (min trade ts): {e}");
                                tokio::time::sleep(Duration::from_secs(60)).await;
                                continue;
                            }
                        }
                    }
                    Err(e) => {
                        error!("ohlcv rebuild: {e}");
                        tokio::time::sleep(Duration::from_secs(60)).await;
                        continue;
                    }
                };

                let mut batch_start = since;
                while batch_start < cutoff {
                    let batch_end = (batch_start + BATCH_SECS).min(cutoff);
                    match db::cauldron::ohlcv::rebuild_range(
                        &ohlcv_read,
                        &ohlcv_write,
                        batch_start,
                        batch_end,
                    )
                    .await
                    {
                        Ok(n) => {
                            info!("ohlcv: materialised {n} buckets [{batch_start}, {batch_end})");
                            ohlcv_state_bg
                                .materialized_end
                                .store(batch_end, Ordering::Relaxed);
                        }
                        Err(e) => {
                            error!("ohlcv rebuild failed: {e}");
                            break;
                        }
                    }
                    batch_start = batch_end;
                    // Yield between batches so block indexing writes can proceed.
                    tokio::time::sleep(Duration::from_millis(200)).await;
                }

                tokio::time::sleep(Duration::from_secs(600)).await;
            }
        });
    }

    rocket::build()
        .attach(signal::ShutdownFairing)
        .attach(signal::IbdCheckFairing)
        .manage(dbpool)
        .manage(ibd_state)
        .manage(ohlcv_state)
        // give rocket ownership of downloaders/fetchers to ensure threads aren't dropped
        .manage(bcmrdownloader)
        .manage(wellknowndownloader)
        .manage(crc20fetcher)
        .mount(
            "/cauldron/",
            routes![
                rpc::tvl::deprecated_tvl,
                rpc::tvl::valuelocked_token,
                rpc::tvl::valuelocked_all,
                rpc::volume::volume_all,
                rpc::volume::volume_token,
                rpc::tokens::search_by_volume,
                rpc::tokens::search_cached,
                rpc::tokens::list_cached,
                rpc::tokens::list_cached_by_ids,
                rpc::price::price_history,
                rpc::candlesticks::price_candlesticks,
                rpc::price::price_current,
                rpc::price::price_at,
                rpc::pool::list_active_pools,
                rpc::pool::pool_history,
                rpc::pool::pool_id_from_utxo,
                rpc::apy::aggregate_apy,
                rpc::contract::contract_count_token,
                rpc::contract::contract_count_all,
                rpc::contract::contract_volume,
                rpc::user::unique_addresses,
                rpc::tx::tx_latest,
                rpc::tokens::first_pool_creation,
            ],
        )
        .mount(
            "/bcmr",
            routes![
                rpc::bcmr::token_bcmr,
                rpc::bcmr::token_bcmr_all,
                rpc::bcmr::token_authhead
            ],
        )
        .mount(
            "/oracle",
            routes![
                rpc::oracle::oracle_get_closest,
                rpc::oracle::oracle_get_range,
                rpc::oracle::oracle_get_history,
                rpc::oracle::oracle_cash_closest,
                rpc::oracle::oracle_cash_history,
            ],
        )
        .mount(
            "/moria/v11",
            routes![
                rpc::moria_v11::loan_history,
                rpc::moria_v11::global_history,
                rpc::moria_v11::active_loans,
                rpc::moria_v11::allowances,
                rpc::moria_v11::crankable,
                rpc::moria_v11::moria_stats,
                rpc::moria_v11::bar_history,
                rpc::moria_v11::pool_history,
                rpc::moria_v11::staking_apr,
            ],
        )
        .mount(
            "/tokentoken",
            routes![
                rpc::tokentoken::list_active_pools,
                rpc::tokentoken::list_all_pools,
                rpc::tokentoken::list_tokens,
            ],
        )
        .mount(
            "/tokenbch",
            routes![
                rpc::tokenbch::list_active_pools,
                rpc::tokenbch::list_all_pools,
                rpc::tokenbch::list_tokens,
            ],
        )
        .mount(
            "/ido",
            routes![
                rpc::ido::list_idos,
                rpc::ido::get_ido_by_id,
                rpc::ido::get_ido_by_offering_token,
                rpc::ido::list_ido_entries,
            ],
        )
        .mount("/", routes![rpc::health::health])
        .attach(cors)
}