riftenlabs-indexer/src/main.rs

use anyhow::Context;
use bitcoin_hashes::hex::ToHex;
use bitcoincash::{consensus::deserialize, Block};
use db::{get_token_tvl, list_tokens_by_volume};
use electrum_client::{Client, ElectrumApi, Param};
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 rusqlite::Connection;
use serde_json::Value;
use std::{
    backtrace::Backtrace,
    collections::VecDeque,
    panic,
    path::Path,
    process,
    sync::{mpsc::sync_channel, Arc, Mutex},
    thread,
    time::Duration,
};

use crate::{
    chainutil::calculate_mtp,
    db::{config_get, config_set, insert_utxo_funding, insert_utxo_spending, prepare_tables},
};

// The block where first cauldron contract was deployed.
const GENESIS_BLOCK: i64 = 799870;
// Trail behind the 10 block reorg-protection. We have not implemented logic for reorgs.
const TRAIL_BEHIND_OFFSET: i64 = 10;
// Last indexed block height.
const KEY_LAST_INDEXED: &str = "last_indexed";

mod chainutil;
mod db;

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

fn get_tip_height(client: &Client) -> anyhow::Result<i64> {
    let tip: Value =
        serde_json::from_str(&client.raw_call("blockchain.headers.tip", [])?.to_string())?;

    tip.get("height")
        .context("no height")?
        .as_i64()
        .context("no int")
}

fn index_blocks(conn: Arc<Mutex<Connection>>, client: Arc<Mutex<Client>>) -> anyhow::Result<i64> {
    let last_indexed =
        config_get(&conn.lock().unwrap(), KEY_LAST_INDEXED)?.unwrap_or(GENESIS_BLOCK - 1);

    let tip_height = get_tip_height(&client.lock().unwrap())?;
    let stop_height = tip_height - TRAIL_BEHIND_OFFSET;
    let mut block_height = last_indexed - 11; // 11 to get correct mtp

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

    thread::spawn(move || {
        let mut block_timestamps: VecDeque<u32> = VecDeque::with_capacity(11);

        loop {
            if block_height >= stop_height {
                block_send.send(None).unwrap();
                break;
            }
            let res = client
                .lock()
                .unwrap()
                .raw_call(
                    "blockchain.block.get",
                    vec![Param::U32(block_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();
            block_timestamps.push_back(block.header.time);
            // Ensure the buffer only keeps the last 11 timestamps
            if block_timestamps.len() > 11 {
                block_timestamps.pop_front();
            }
            let mtp = calculate_mtp(&block_timestamps);

            block_send.send(Some((block_height, mtp, block))).unwrap();

            block_height += 1;
        }
    });

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

        if block_height <= last_indexed {
            continue;
        }
        let mut lock = conn.lock().unwrap();
        let tx = lock.transaction()?;
        let cauldrons = parse_cauldrons(&block);
        insert_utxo_funding(&tx, mtp, &cauldrons)?;
        insert_utxo_spending(&tx, mtp, &cauldrons)?;
        config_set(&tx, KEY_LAST_INDEXED, block_height);
        tx.commit()?;

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

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

#[get("/tvl/<time>")]
fn tvl(
    time: usize,
    conn: &State<Arc<Mutex<Connection>>>,
) -> Result<Json<Vec<Value>>, Custom<String>> {
    let tvl: Vec<(String, u64, u64)> = get_token_tvl(&conn.lock().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<Arc<Mutex<Connection>>>,
) -> 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)> =
        list_tokens_by_volume(&conn.lock().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))
}

fn set_panic_hook() {
    panic::set_hook(Box::new(|panic_info| {
        let backtrace = Backtrace::capture();
        eprintln!("Backtrace:\n{:?}", backtrace);

        if let Some(message) = panic_info.payload().downcast_ref::<&str>() {
            eprintln!("Panic occurred: {}", message);
        }

        println!("A thread panicked, terminating the program.");
        process::exit(1);
    }));
}

#[launch]
fn launch() -> _ {
    set_panic_hook();

    let db_path = "cauldron.db";
    let db_exists = Path::new(db_path).exists();
    let conn = Connection::open(db_path).unwrap();
    if !db_exists {
        prepare_tables(&conn);
    }

    let conn = Arc::new(Mutex::new(conn));
    let conn_cpy = Arc::clone(&conn);

    thread::spawn(move || {
        let conn = conn_cpy;
        let client = Arc::new(Mutex::new(
            Client::new("tcp://rostrum.cauldron.quest:50001").unwrap(),
        ));
        let mut tip: i64 = 0;
        loop {
            let new_tip = get_tip_height(&client.lock().unwrap()).unwrap();
            if new_tip != tip {
                index_blocks(conn.clone(), client.clone()).unwrap();
                tip = new_tip;
            }
            thread::sleep(Duration::from_secs(60));
        }
    });

    rocket::build()
        .manage(conn)
        .mount("/cauldron/", routes![tvl, list_by_volume])
}