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::{Context, Result};
use bitcoin_hashes::hex::{FromHex, ToHex};
use bitcoincash::{consensus::deserialize, Block, BlockHash, Transaction, Txid};
use chain::{get_new_headers, Chain, StoreBlockUndoer};
use db::{config::config_get, DBPool};
use electrum::{electrum_fetch_mempool, electrum_get_tip};
use electrum_client::{Client, ElectrumApi, Param};
use log::{debug, error, info};
use rayon::prelude::*;
use riftenlabs_defi::cauldron::{parse_cauldron, ParsedContract};
use rocket::{
    get,
    http::Status,
    launch,
    response::status::Custom,
    routes,
    serde::{
        json::{json, Json},
        Serialize,
    },
    State,
};
use rocket_cors::{AllowedHeaders, AllowedOrigins};
use serde_json::Value;
use std::{
    backtrace::Backtrace,
    collections::HashSet,
    panic,
    path::Path,
    process,
    sync::{mpsc::sync_channel, Arc, Mutex},
    thread,
    time::{Duration, SystemTime, UNIX_EPOCH},
};
use stderrlog::LogLevelNum;

use crate::{
    db::{
        config::config_set,
        header::{db_get_header, load_all_headers, store_headers},
        prepare_tables,
        tx::{self, insert_block_tx},
        utxo_funding::insert_utxo_funding,
        utxo_spending::insert_utxo_spending,
    },
    electrum::electrum_get_tx,
};

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

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

mod chain;
mod db;
mod electrum;
mod rpc;

fn parse_cauldrons(tx: &Transaction) -> Vec<ParsedContract> {
    tx.input
        .par_iter()
        .enumerate()
        .filter_map(move |(i, _)| parse_cauldron(i, tx))
        .collect()
}

fn update_mempool(db: DBPool, electrum: Arc<Mutex<Client>>) -> Result<()> {
    let our_mempool_txs: HashSet<Txid> = db::mempool::load_mempool(&db.get().unwrap())?;
    let node_mempool: HashSet<Txid> = electrum_fetch_mempool(&electrum.lock().unwrap())?;

    let txs_to_delete = our_mempool_txs.difference(&node_mempool);
    let txs_to_add = node_mempool.difference(&our_mempool_txs);

    let txs_to_add: Vec<Transaction> = txs_to_add
        .into_iter()
        .filter_map(
            |txid| match electrum_get_tx(&electrum.lock().unwrap(), txid) {
                Ok(tx) => Some(tx),
                Err(e) => {
                    info!("Failed to get mempool tx {}: {}", txid, e);
                    None
                }
            },
        )
        .collect();

    let mut db_conn = db.get().unwrap();
    let db_tx = db_conn.transaction()?;

    let mut all_cauldrons = vec![];

    for txid in txs_to_delete {
        debug!("mempool remove {}", txid.to_hex());
        db::mempool::delete_mempool_tx(&db_tx, txid)?;
    }

    for tx in txs_to_add {
        let txid = tx.txid();
        debug!("mempool add {}", txid.to_hex());
        tx::insert_mempool_tx(&db_tx, &txid)?;
        let cauldrons: Vec<ParsedContract> = tx
            .input
            .iter()
            .enumerate()
            .filter_map(|(i, _)| parse_cauldron(i, &tx))
            .collect();

        insert_utxo_funding(&db_tx, &cauldrons, &txid, false)?;
        insert_utxo_spending(&db_tx, &cauldrons, &txid, false)?;

        all_cauldrons.extend(cauldrons);
    }

    db::pool::update_pool_history(&db_tx, all_cauldrons).context("update pool history")?;

    Ok(db_tx.commit()?)
}

fn index_blocks(
    chain: Arc<Mutex<Chain>>,
    pool: DBPool,
    client: Arc<Mutex<Client>>,
) -> Result<BlockHash> {
    let (tip_header, _) = electrum_get_tip(&client.lock().unwrap())?;

    let (block_send, block_recv) = sync_channel::<Option<(u64, u64, Block)>>(10);

    // Update header chain (and undo any blocks that may have reorged away)
    {
        let chain = chain.lock().unwrap();
        if tip_header.block_hash() != chain.tip_hash() {
            debug!(
                "Updating header chain from {} to {}",
                chain.tip_hash().to_hex(),
                tip_header.block_hash().to_hex()
            );
            let new_headers =
                get_new_headers(&client.lock().unwrap(), &chain, &tip_header.block_hash())?;
            debug!("Storing headers");
            {
                let mut db_conn = pool.get().unwrap();
                for chunk in new_headers.chunks(100000) {
                    let db_tx = db_conn.transaction()?;
                    store_headers(&db_tx, chunk)?;
                    db_tx.commit()?;
                }
            }
            let undoer = StoreBlockUndoer::new(pool.clone())?;
            chain.update(undoer, new_headers, None)?;
            debug!("Header update done");
        }
    }

    let pool_cpy = pool.clone();

    thread::spawn(move || {
        let pool = pool_cpy;

        let last_indexed = config_get(&pool.get().unwrap(), KEY_LAST_INDEXED).unwrap();
        let last_indexed = last_indexed.unwrap_or(RIFTEN_LABS_GENESIS_BLOCK.to_string());
        let mut last_indexed = BlockHash::from_hex(&last_indexed).unwrap();

        // Check if last indexed has been orphaned
        loop {
            if chain.lock().unwrap().contains(&last_indexed) {
                break;
            }
            info!(
                "Last indexed block ({}) has been orphaned",
                last_indexed.to_hex()
            );
            let header = db_get_header(&pool.get().unwrap(), &last_indexed)
                .context("Failed to get header for last indexed")
                .unwrap();
            last_indexed = header.prev_blockhash;

            info!(
                "Last indexed block rolled back to {}",
                last_indexed.to_hex()
            );
        }

        loop {
            if tip_header.block_hash() == last_indexed {
                block_send.send(None).unwrap();
                break;
            }

            let next_height = chain
                .lock()
                .unwrap()
                .get_block_height(&last_indexed)
                .expect("last_indexed height not found in main chain")
                + 1;

            let res = client
                .lock()
                .unwrap()
                .raw_call("blockchain.block.get", vec![Param::U32(next_height as u32)])
                .unwrap();

            let block_hex: String = serde_json::from_str(&res.to_string()).unwrap();
            let block: Block = deserialize(&hex::decode(&block_hex).unwrap()).unwrap();
            let block_hash = block.block_hash();

            block_send
                .send(Some((
                    next_height,
                    chain.lock().unwrap().get_mtp(next_height).unwrap(),
                    block,
                )))
                .unwrap();

            last_indexed = block_hash;
        }
    });

    loop {
        let (block_height, mtp, block) = match block_recv.recv()? {
            Some(res) => res,
            None => break,
        };

        let mut db_conn = pool.get().unwrap();
        let db_tx = db_conn.transaction()?;

        let blockhash = block.block_hash();

        let mut total_cauldrons = 0;

        let mut all_cauldrons = vec![];

        for tx in block.txdata {
            let cauldrons = parse_cauldrons(&tx);
            if cauldrons.is_empty() {
                continue;
            }

            let txid = tx.txid();
            insert_block_tx(&db_tx, &txid, &blockhash, mtp as i64).context("inserting block tx")?;
            insert_utxo_funding(&db_tx, &cauldrons, &txid, true)
                .context("inserting funding utxos")?;
            insert_utxo_spending(&db_tx, &cauldrons, &txid, true)
                .context("inserting spending utxos")?;
            total_cauldrons += cauldrons.len();

            all_cauldrons.extend(cauldrons);
        }

        // Figuring out initial utxo needs to be done on all cauldrons in a block.
        db::pool::update_pool_history(&db_tx, all_cauldrons).context("update pool history")?;

        config_set(&db_tx, KEY_LAST_INDEXED, &blockhash.to_hex());
        db_tx.commit()?;

        println!(
            "Indexed {}; mtp: {}, height {}, {} trades.",
            block.header.block_hash().to_hex(),
            mtp,
            block_height,
            total_cauldrons
        );
    }
    Ok(tip_header.block_hash())
}

#[derive(Serialize)]
#[serde(crate = "rocket::serde")]
struct TVLResponse {}

#[get("/tvl/<time>")]
fn tvl(time: usize, conn: &State<DBPool>) -> Result<Json<Vec<Value>>, Custom<String>> {
    let tvl: Vec<(String, u64, u64)> = rpc::get_token_tvl(&conn.get().unwrap(), time)
        .map_err(|e| Custom(Status::InternalServerError, format!("Error: {}", e)))?;

    let result: Vec<Value> = tvl
        .into_par_iter()
        .map(|(token, token_amount, sats)| {
            json!({
                "token_id": token,
                "token_amount": token_amount,
                "satoshis": sats
            })
        })
        .collect();

    Ok(Json(result))
}

#[get("/tokens/list_by_volume?<duration>&<limit>")]
fn list_by_volume(
    duration: Option<usize>,
    limit: Option<usize>,
    conn: &State<DBPool>,
) -> Result<Json<Vec<Value>>, Custom<String>> {
    let thirty_days = 24 * 60 * 60 * 30;
    let duration = duration.unwrap_or(thirty_days);
    let limit = limit.unwrap_or(50);
    let limit = std::cmp::max(limit, 1000);

    let list: Vec<(String, u64, u64, u64, u64, u64, u64)> =
        rpc::list_tokens_by_volume(&conn.get().unwrap(), duration, limit)
            .map_err(|e| Custom(Status::InternalServerError, format!("Error: {}", e)))?;

    let result: Vec<Value> = list
        .into_par_iter()
        .map(
            |(
                token_id,
                trade_volume,
                trade_count,
                tvl_sats,
                tvl_token,
                best_contract_sats,
                best_contract_tokens,
            )| {
                json!({
                    "token_id": token_id,
                    "trade_volume": trade_volume,
                    "trade_count": trade_count,
                    "tvl_sats": tvl_sats,
                    "tvl_tokens": tvl_token,
                    "best_contract_sats": best_contract_sats,
                    "best_contract_tokens": best_contract_tokens
                })
            },
        )
        .collect();

    Ok(Json(result))
}

#[get("/price/<token>/history?<start>&<end>&<stepsize>")]
fn price_history(
    token: &str,
    start: Option<i64>,
    end: Option<i64>,
    stepsize: Option<i64>,
    conn: &State<DBPool>,
) -> Result<Json<Value>, Custom<String>> {
    let current_timestamp = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap()
        .as_secs() as i64;

    let history = rpc::historic_price(
        &conn.get().unwrap(),
        start.unwrap_or(current_timestamp - 30 * 24 * 3600 /* 30 days */),
        end.unwrap_or(current_timestamp),
        stepsize.unwrap_or(3600 /* 1 hour */),
        token,
    )
    .map_err(|e| Custom(Status::BadRequest, format!("Error: {}", e)))?;

    let history_json: Vec<Value> = history
        .iter()
        .map(|(time, avg, max, min)| {
            json!({
                "time": time,
                "avg": avg,
                "max": max,
                "min": min,
            })
        })
        .collect();

    Ok(Json(json!({
        "history": json!(history_json)
    })))
}

#[get("/pool/list_by_apy")]
fn list_pools_by_apy(conn: &State<DBPool>) -> Result<Json<Value>, Custom<String>> {
    let pools: Vec<rpc::PoolYield> = rpc::pools_by_apy(&conn.get().unwrap())
        .map_err(|e| Custom(Status::BadRequest, format!("Error: {}", e)))?;

    Ok(Json(json!({
        "pools": json!(pools)
    })))
}

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<DBPool> {
    let db_path = "cauldron.db";
    let db_exists = Path::new(db_path).exists();

    let manager = r2d2_sqlite::SqliteConnectionManager::file(db_path);
    let pool = Arc::new(r2d2::Pool::new(manager).expect("Failed to initialize database"));
    if !db_exists {
        prepare_tables(&pool.get().expect("failed to get sqlite 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(&pool.get().unwrap()).unwrap();
    info!("Initializing {} headers...", all_headers.len());
    chain.lock().unwrap().load(all_headers).unwrap();
    info!("Headers loaded.");

    let pool_cpy = Arc::clone(&pool);
    let client_cpy = client.clone();

    thread::spawn(move || {
        let pool = pool_cpy;
        let client = client_cpy;

        let mut tip: BlockHash = match index_blocks(chain.clone(), pool.clone(), client.clone()) {
            Ok(tip) => tip,
            Err(e) => {
                panic!("Initial index failed: {}", e);
            }
        };
        loop {
            let new_tip = electrum_get_tip(&client.lock().unwrap())
                .unwrap()
                .0
                .block_hash();
            if new_tip != tip {
                tip = index_blocks(chain.clone(), pool.clone(), client.clone())
                    .expect("Indexing failed");
            }
            if let Err(e) = update_mempool(pool.clone(), client.clone()) {
                error!("Failed to update mempool: {}", e);
            }
            thread::sleep(Duration::from_secs(5));
        }
    });

    Ok(pool)
}

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

    set_panic_hook();

    let dbpool = 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();

    rocket::build()
        .manage(dbpool)
        .mount(
            "/cauldron/",
            routes![tvl, list_by_volume, price_history, list_pools_by_apy],
        )
        .attach(cors)
}