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};
use crc20::crc20fetcher::CRC20Fetcher;
use db::{DBPool, 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 rusqlite::OpenFlags;
use std::sync::atomic::AtomicBool;
use std::{
    backtrace::Backtrace,
    collections::HashMap,
    panic,
    path::Path,
    process,
    sync::{Arc, Mutex},
    thread::{self, sleep},
    time::{Duration, Instant},
};
use stderrlog::LogLevelNum;

use crate::bcmr::bcmrdownloader::BCMRDownloader;
use crate::db::cauldron::config::check_db_version;
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::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";

// 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 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 db_initialize(c: &mut rusqlite::Connection) -> Result<(), rusqlite::Error> {
    // Set busy timeout for normal queries.
    c.busy_timeout(Duration::from_secs(30))?;
    c.execute("PRAGMA foreign_keys=1;", [])?;

    // PRAGMA journal_mode may not honor busy_timeout, so retry manually.
    let start = Instant::now();
    let timeout = Duration::from_secs(30);
    loop {
        match c.execute_batch("PRAGMA journal_mode=WAL;") {
            Ok(_) => break,
            Err(rusqlite::Error::SqliteFailure(err, _))
                if err.code == rusqlite::ErrorCode::DatabaseBusy =>
            {
                if start.elapsed() >= timeout {
                    return Err(rusqlite::Error::SqliteFailure(
                        err,
                        Some("database locked after retries".into()),
                    ));
                }
                sleep(Duration::from_millis(100));
            }
            Err(e) => return Err(e),
        }
    }
    Ok(())
}

fn db_sanity_check(c: &rusqlite::Connection) -> rusqlite::Result<()> {
    // Verify that foreign_keys are enabled
    let mut stmt = c.prepare("PRAGMA foreign_keys;")?;
    let foreign_keys_enabled: i32 = stmt.query_row([], |row| row.get(0))?;
    if foreign_keys_enabled != 1 {
        panic!(
            "Foreign key enforcement is not enabled for the connection. (result: {foreign_keys_enabled})"
        );
    }
    Ok(())
}

fn start_program(
    config: Config,
) -> Result<(DB, BCMRDownloader, WellKnownDownloader, CRC20Fetcher)> {
    let create_db_pool = |db_path| -> (bool, DBPool, DBPool) {
        let db_exists = Path::new(db_path).exists();

        info!("Initializing connection to {db_path}");

        let write_manager = r2d2_sqlite::SqliteConnectionManager::file(db_path)
            .with_flags(OpenFlags::SQLITE_OPEN_READ_WRITE | OpenFlags::SQLITE_OPEN_CREATE)
            .with_init(db_initialize);

        let write_pool =
            Arc::new(r2d2::Pool::new(write_manager).expect("Failed to initialize database"));

        let read_manager = r2d2_sqlite::SqliteConnectionManager::file(db_path)
            .with_flags(OpenFlags::SQLITE_OPEN_READ_ONLY)
            .with_init(db_initialize);

        let read_pool =
            Arc::new(r2d2::Pool::new(read_manager).expect("Failed to initialize database"));

        db_sanity_check(&read_pool.get().unwrap()).expect("db sanity check failed");
        db_sanity_check(&write_pool.get().unwrap()).expect("db sanity check failed");

        (db_exists, write_pool, read_pool)
    };

    let (db_exists, cauldron_db_write, cauldron_db_read) = create_db_pool("cauldron.db");
    if !db_exists {
        db::cauldron::prepare_tables(
            &cauldron_db_write
                .get()
                .expect("failed to get sqlite connection"),
        );
    } else {
        // Check database version for existing database
        check_db_version(&cauldron_db_read.get().unwrap()).expect("Database version check failed");
    }
    let (db_exists, bcmr_db_write, bcmr_db_read) = create_db_pool("bcmr.db");
    if !db_exists {
        db::bcmr::prepare_tables(
            &bcmr_db_write
                .get()
                .expect("failed to create sql connection"),
        );
    }

    let (db_exists, crc20_db_write, crc20_db_read) = create_db_pool("crc20.db");
    if !db_exists {
        db::crc20::prepare_tables(
            &crc20_db_write
                .get()
                .expect("failed to create sqlite crc20 connection"),
        );
    }

    let (db_exists, oracle_db_write, oracle_db_read) = create_db_pool("oracle.db");
    if !db_exists {
        db::oracle::prepare_tables(
            &oracle_db_write
                .get()
                .expect("failed to create sqlite oracle connection"),
        );
    }

    // 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://{}", config.rostrum_addr)) {
            Ok(server) => server,
            Err(e) => {
                error!(
                    "Failed to connect to {}: {}. See --help for setting a different server.",
                    config.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(&cauldron_db_read.get().unwrap());

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

    let db = DB {
        cauldron_w: cauldron_db_write,
        cauldron_r: cauldron_db_read,
        bcmr_w: bcmr_db_write,
        bcmr_r: bcmr_db_read,
        crc20_w: crc20_db_write,
        crc20_r: crc20_db_read,
        oracle_w: oracle_db_write,
        oracle_r: oracle_db_read,
    };

    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();

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

        let mut tip: BlockHash = loop {
            indexing_in_progress_clone.store(true, std::sync::atomic::Ordering::Relaxed);

            break match index_blocks(chain.clone(), db.clone(), client.clone(), true) {
                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, std::sync::atomic::Ordering::Relaxed);

        loop {
            let new_tip = match electrum_get_tip(&client.lock().unwrap()) {
                Ok(t) => t.0.block_hash(),
                Err(e) => {
                    warn!("Failed to get block chain tip from electrum: {e}");
                    thread::sleep(Duration::from_secs(5));
                    continue;
                }
            };

            if new_tip != tip {
                indexing_in_progress_clone.store(true, std::sync::atomic::Ordering::Relaxed);

                tip = match index_blocks(chain.clone(), db.clone(), client.clone(), true) {
                    Ok(t) => t,
                    Err(e) => {
                        warn!("Indexing block failed: {} {}", e, e.backtrace());
                        tip
                    }
                };

                indexing_in_progress_clone.store(false, std::sync::atomic::Ordering::Relaxed);
            }
            if let Err(e) =
                update_mempool(db.cauldron_w.clone(), db.oracle_w.clone(), client.clone())
            {
                error!("Failed to update mempool: {e}");
            }
            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()?;

    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");
    }

    spawn_token_metrics_updater(dbpool.clone());

    rocket::build()
        .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::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,
            ],
        )
        .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)
}