// 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, Ordering}; use std::{ backtrace::Backtrace, collections::HashMap, panic, path::Path, process, sync::{Arc, Mutex}, thread::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::() { 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::() { 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(); std::thread::spawn(move || { let db = db_cpy; // Initial full index let mut tip: BlockHash = loop { indexing_in_progress_clone.store(true, 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, Ordering::Relaxed); // Follow chain loop { 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) { 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::()) .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() .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, 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) }