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::Result;
use bitcoincash::{consensus::deserialize, Block, BlockHash};
use db::{DBPool, DB};
use electrum::electrum_get_tip;
use electrum_client::{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::{
    backtrace::Backtrace,
    collections::HashMap,
    panic,
    path::Path,
    process,
    sync::{Arc, Mutex},
    thread,
    time::Duration,
};
use stderrlog::LogLevelNum;

use crate::bcmr::bcmrdownloader::BCMRDownloader;
use crate::db::cauldron::header::load_all_headers;
use crate::index::{index_blocks, update_mempool};

// 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 db;
mod electrum;
mod index;
mod rpc;
mod timeutil;

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 {
            error!("Panic info: {:?}", panic_info);
        }

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

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

        let write_manager = r2d2_sqlite::SqliteConnectionManager::file(db_path)
            .with_flags(OpenFlags::SQLITE_OPEN_READ_WRITE | OpenFlags::SQLITE_OPEN_CREATE)
            .with_init(|c| c.execute_batch("PRAGMA foreign_keys=1;"))
            .with_init(|c| c.busy_timeout(Duration::from_secs(30)))
            .with_init(|c| c.execute_batch("PRAGMA journal_mode=WAL;"));

        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(|c| c.execute_batch("PRAGMA foreign_keys=1;"))
            .with_init(|c| c.busy_timeout(Duration::from_secs(30)))
            .with_init(|c| c.execute_batch("PRAGMA journal_mode=WAL;"));

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

        (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"),
        );
    }
    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 client = Arc::new(Mutex::new(
        Client::new("tcp://rostrum.cauldron.quest:50001").unwrap(),
    ));

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

    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,
    };

    let db_cpy = db.clone();
    // let client_cpy = client.clone();

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

        let mut tip: BlockHash = loop {
            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());
                }
            };
        };
        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 {
                tip = match index_blocks(chain.clone(), db.clone(), client.clone(), true) {
                    Ok(t) => t,
                    Err(e) => {
                        warn!("Indexing block failed: {} {}", e, e.backtrace());
                        tip
                    }
                }
            }
            if let Err(e) = update_mempool(db.cauldron_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()?;

    Ok((db, bcmrdownloader))
}

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

    set_panic_hook();

    let (dbpool, bcmrdownloader) = match start_program() {
        Ok(db) => db,
        Err(e) => {
            let backtrace = Backtrace::capture();
            error!(
                "Backtrace (if RUST_BACKTRACE=1):\n{}",
                backtrace.to_string()
            );
            error!("Error: {}", e.to_string());
            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()));

    rocket::build()
        .manage(dbpool)
        .manage(response_cache)
        // give rocket ownership of downloader to ensure thread isn't dropped
        .manage(bcmrdownloader)
        .mount(
            "/cauldron/",
            routes![
                rpc::tvl::deprecated_tvl,
                rpc::tvl::valuelocked_token,
                rpc::tvl::valuelocked_all,
                rpc::tokens::list_by_volume,
                rpc::price::price_history,
                rpc::price::price_current,
                rpc::pool::list_pools_by_apy,
                rpc::pool::list_active_pools,
                rpc::contract::contract_count_token,
                rpc::contract::contract_count_all,
                rpc::contract::contract_volume
            ],
        )
        .mount("/bcmr", routes![rpc::bcmr::token_bcmr,])
        .attach(cors)
}