469 lines
14 KiB
Rust
469 lines
14 KiB
Rust
// 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 bitcoin_hashes::{
|
|
hex::{FromHex, ToHex},
|
|
Hash,
|
|
};
|
|
use bitcoincash::{consensus::deserialize, Block, BlockHash, Transaction, Txid};
|
|
use chain::{get_new_headers, Chain, StoreBlockUndoer};
|
|
use db::{
|
|
db_load_mempool, get_token_tvl, historic_price, list_tokens_by_volume, load_all_headers,
|
|
store_headers,
|
|
};
|
|
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 rusqlite::Connection;
|
|
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_get, config_set, db_add_first_seen, db_delete_mempool_tx, insert_utxo_funding,
|
|
insert_utxo_spending, prepare_tables,
|
|
},
|
|
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;
|
|
|
|
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: Arc<Mutex<Connection>>, electrum: Arc<Mutex<Client>>) -> Result<()> {
|
|
let our_mempool_txs: HashSet<Txid> = db_load_mempool(&db.lock().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_lock = db.lock().unwrap();
|
|
let db_tx = db_lock.transaction()?;
|
|
|
|
for txid in txs_to_delete {
|
|
debug!("mempool remove {}", txid.to_hex());
|
|
db_delete_mempool_tx(&db_tx, txid)?;
|
|
}
|
|
|
|
for tx in txs_to_add {
|
|
let txid = tx.txid();
|
|
debug!("mempool add {}", txid.to_hex());
|
|
db_add_first_seen(&db_tx, &txid)?;
|
|
let cauldrons: Vec<ParsedContract> = tx
|
|
.input
|
|
.iter()
|
|
.enumerate()
|
|
.filter_map(|(i, _)| parse_cauldron(i, &tx))
|
|
.collect();
|
|
|
|
let mtp = 0;
|
|
let blockhash = BlockHash::all_zeros();
|
|
insert_utxo_funding(&db_tx, mtp, &cauldrons, &blockhash, &txid, false)?;
|
|
insert_utxo_spending(&db_tx, mtp, &cauldrons, &blockhash, &txid, false)?;
|
|
}
|
|
|
|
Ok(db_tx.commit()?)
|
|
}
|
|
|
|
fn index_blocks(
|
|
chain: Arc<Mutex<Chain>>,
|
|
conn: Arc<Mutex<Connection>>,
|
|
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_lock = conn.lock().unwrap();
|
|
for chunk in new_headers.chunks(100000) {
|
|
let db_tx = db_lock.transaction()?;
|
|
store_headers(&db_tx, chunk)?;
|
|
db_tx.commit()?;
|
|
}
|
|
}
|
|
let undoer = StoreBlockUndoer::new(conn.clone())?;
|
|
chain.update(undoer, new_headers, None)?;
|
|
debug!("Header update done");
|
|
}
|
|
}
|
|
|
|
let conn_cpy = conn.clone();
|
|
|
|
thread::spawn(move || {
|
|
let conn = conn_cpy;
|
|
|
|
let last_indexed = config_get(&conn.lock().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();
|
|
|
|
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)
|
|
.unwrap()
|
|
+ 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 lock = conn.lock().unwrap();
|
|
let db_tx = lock.transaction()?;
|
|
|
|
let blockhash = block.block_hash();
|
|
|
|
let mut total_cauldrons = 0;
|
|
|
|
for tx in block.txdata {
|
|
let cauldrons = parse_cauldrons(&tx);
|
|
let txid = tx.txid();
|
|
insert_utxo_funding(&db_tx, mtp as u32, &cauldrons, &blockhash, &txid, true)?;
|
|
insert_utxo_spending(&db_tx, mtp as u32, &cauldrons, &blockhash, &txid, true)?;
|
|
total_cauldrons += cauldrons.len();
|
|
}
|
|
|
|
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<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))
|
|
}
|
|
|
|
#[get("/price/<token>/history?<start>&<end>&<stepsize>")]
|
|
fn price_history(
|
|
token: &str,
|
|
start: Option<i64>,
|
|
end: Option<i64>,
|
|
stepsize: Option<i64>,
|
|
conn: &State<Arc<Mutex<Connection>>>,
|
|
) -> Result<Json<Value>, Custom<String>> {
|
|
let current_timestamp = SystemTime::now()
|
|
.duration_since(UNIX_EPOCH)
|
|
.unwrap()
|
|
.as_secs() as i64;
|
|
|
|
let history = historic_price(
|
|
&conn.lock().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)
|
|
})))
|
|
}
|
|
|
|
fn set_panic_hook() {
|
|
panic::set_hook(Box::new(|panic_info| {
|
|
let backtrace = Backtrace::capture();
|
|
error!(
|
|
"Backtrace (if RUST_BACKTRACE=1):\n{}",
|
|
backtrace.to_string()
|
|
);
|
|
|
|
if let Some(message) = panic_info.payload().downcast_ref::<&str>() {
|
|
error!("Panic occurred: {}", message);
|
|
} else {
|
|
error!("{:?} {}", panic_info, panic_info.to_string());
|
|
}
|
|
|
|
error!("A thread panicked, terminating the program.");
|
|
process::exit(1);
|
|
}));
|
|
}
|
|
|
|
fn start_program() -> Result<Arc<Mutex<Connection>>> {
|
|
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 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(&conn.lock().unwrap()).unwrap();
|
|
info!("Initializing {} headers...", all_headers.len());
|
|
chain.lock().unwrap().load(all_headers).unwrap();
|
|
info!("Headers loaded.");
|
|
|
|
let conn_cpy = Arc::clone(&conn);
|
|
let client_cpy = client.clone();
|
|
|
|
thread::spawn(move || {
|
|
let conn = conn_cpy;
|
|
let client = client_cpy;
|
|
|
|
let mut tip: BlockHash = match index_blocks(chain.clone(), conn.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 = match index_blocks(chain.clone(), conn.clone(), client.clone()) {
|
|
Ok(tip) => tip,
|
|
Err(e) => {
|
|
panic!("Indexing failed: {}", e);
|
|
}
|
|
}
|
|
}
|
|
if let Err(e) = update_mempool(conn.clone(), client.clone()) {
|
|
error!("Failed to update mempool: {}", e);
|
|
}
|
|
thread::sleep(Duration::from_secs(5));
|
|
}
|
|
});
|
|
|
|
Ok(conn)
|
|
}
|
|
|
|
#[launch]
|
|
fn launch() -> _ {
|
|
stderrlog::new()
|
|
.verbosity(LogLevelNum::Debug)
|
|
.init()
|
|
.unwrap();
|
|
|
|
set_panic_hook();
|
|
|
|
let conn = 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(conn)
|
|
.mount("/cauldron/", routes![tvl, list_by_volume, price_history])
|
|
.attach(cors)
|
|
}
|