riftenlabs-indexer/src/main.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 anyhow::{bail, Result};
use bcmr::wellknowndowloader::WellKnownDownloader;
use bitcoincash::{consensus::deserialize, Block, BlockHash, Network};
use crc20::crc20fetcher::CRC20Fetcher;
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 rpc::ResponseCache;
use std::sync::atomic::{AtomicBool, Ordering};
use std::{
    backtrace::Backtrace,
    collections::HashMap,
    panic, process,
    sync::{Arc, Mutex},
    time::Duration,
};
use stderrlog::LogLevelNum;

use crate::bcmr::bcmrdownloader::BCMRDownloader;
use crate::db::cauldron::header::load_all_headers;
use crate::db::cauldron::tokenlist::db_utils::create_cached_token_metrics_table;
use crate::db::cauldron::tokenlist::metrics_cache::spawn_token_metrics_updater;
use crate::db::init::initialize_databases;
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);
    }));
}

fn start_program(
    config: Config,
) -> Result<(DB, BCMRDownloader, WellKnownDownloader, CRC20Fetcher)> {
    // Parse network parameter
    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);

    // Set default rostrum address based on network if not explicitly provided
    let rostrum_addr = if config.rostrum_addr.is_empty() {
        // User didn't provide --rostrum-addr, use network-specific default
        match network {
            Network::Chipnet => "127.0.0.1:64001".to_string(),
            _ => "127.0.0.1:50001".to_string(),
        }
    } else {
        // User explicitly provided a value, use it as-is
        config.rostrum_addr
    };

    // Initialize all databases
    let network_str = match network {
        Network::Bitcoin => "mainnet",
        Network::Chipnet => "chipnet",
        _ => "mainnet",
    };
    let db = initialize_databases(network_str)?;

    // Create a shared flag for indexing status
    let indexing_in_progress = Arc::new(AtomicBool::new(false));

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

    let genesis = client
        .lock()
        .unwrap()
        .raw_call("blockchain.block.get", vec![Param::U32(0)])
        .unwrap();
    let genesis: Block = deserialize(&hex::decode(genesis.as_str().unwrap()).unwrap()).unwrap();
    let chain = Arc::new(Mutex::new(chain::Chain::new(genesis.header)));

    // initialize insert sequence for pool history
    db::cauldron::pool::initialize_seq(&db.cauldron_r.get().unwrap());

    info!("Loading block headers...");
    let all_headers = load_all_headers(&db.cauldron_r.get().unwrap()).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(),
        client.clone(),
        indexing_in_progress.clone(),
    )?;

    {
        // clear oracle mempool
        let conn = db.oracle_w.get().unwrap();
        db::oracle::clear_mempool(&conn).unwrap();
    }

    let indexing_in_progress_clone = indexing_in_progress.clone();

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

        // Initial full index
        let mut tip: BlockHash = loop {
            if signal::shutdown_requested() {
                info!("Shutdown requested, exiting indexing thread 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),
            ) {
                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);

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

            let new_tip = match electrum_get_tip(&client.lock().unwrap()) {
                Ok(t) => t.0.block_hash(),
                Err(e) => {
                    warn!("Failed to get chain tip from electrum: {e}");
                    std::thread::sleep(Duration::from_secs(5));
                    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),
                ) {
                    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.cauldron_w.clone(), db.oracle_w.clone(), client.clone())
                {
                    error!("Failed to update mempool: {e}");
                }
            }
            std::thread::sleep(Duration::from_secs(5));
        }
    });

    let mut bcmrdownloader = BCMRDownloader::new(db.bcmr_w.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))
}

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

    set_panic_hook();

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

    let (dbpool, bcmrdownloader, wellknowndownloader, crc20fetcher) = match start_program(config) {
        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();

    let response_cache: ResponseCache = Arc::new(Mutex::new(HashMap::default()));

    {
        let conn = dbpool.cauldron_w.get().expect("get write conn");
        // Ensure the table exists on both fresh and existing DBs
        create_cached_token_metrics_table(&conn).expect("ensure cached_token_metrics exists");
    }

    rocket::build()
        .attach(signal::ShutdownFairing)
        .manage(dbpool)
        .manage(response_cache)
        // give rocket ownership of downloader to ensure thread isn'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::list_by_volume,
                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_pools_by_apy,
                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],
        )
        .mount(
            "/oracle",
            routes![
                rpc::oracle::oracle_get_closest,
                rpc::oracle::oracle_get_range,
                rpc::oracle::oracle_get_history
            ],
        )
        .attach(cors)
}