From e38eaaa26c076881ae4cc8d96163328ffbc4ebc8 Mon Sep 17 00:00:00 2001 From: Dagur Valberg Johannsson Date: Thu, 9 May 2024 11:25:27 +0200 Subject: [PATCH] Add BCMR indexer This adds BCMR indexer, including downloading and parsing the actual BCMR files. --- Cargo.lock | 108 +++++- Cargo.toml | 2 + contrib/volume.py | 16 + src/bcmr/bcmrdownloader.rs | 449 ++++++++++++++++++++++++ src/bcmr/mod.rs | 345 +++++++++++++++++++ src/bcmr/parsedbcmr.rs | 104 ++++++ src/chain.rs | 19 +- src/db/bcmr/mod.rs | 337 ++++++++++++++++++ src/db/{ => cauldron}/config.rs | 6 +- src/db/{ => cauldron}/header.rs | 0 src/db/{ => cauldron}/mempool.rs | 0 src/db/cauldron/mod.rs | 76 +++++ src/db/{ => cauldron}/pool.rs | 0 src/db/{ => cauldron}/tx.rs | 0 src/db/{ => cauldron}/utxo_funding.rs | 0 src/db/{ => cauldron}/utxo_spending.rs | 0 src/db/mod.rs | 81 +---- src/index.rs | 262 ++++++++++++++ src/main.rs | 450 ++++++------------------- src/rpc/bcmr.rs | 32 ++ src/rpc/contract.rs | 44 ++- src/rpc/mod.rs | 17 +- src/rpc/pool.rs | 8 +- src/rpc/price.rs | 8 +- src/rpc/tokens.rs | 65 ++++ src/rpc/tvl.rs | 16 +- 26 files changed, 1994 insertions(+), 451 deletions(-) create mode 100644 contrib/volume.py create mode 100644 src/bcmr/bcmrdownloader.rs create mode 100644 src/bcmr/mod.rs create mode 100644 src/bcmr/parsedbcmr.rs create mode 100644 src/db/bcmr/mod.rs rename src/db/{ => cauldron}/config.rs (89%) rename src/db/{ => cauldron}/header.rs (100%) rename src/db/{ => cauldron}/mempool.rs (100%) create mode 100644 src/db/cauldron/mod.rs rename src/db/{ => cauldron}/pool.rs (100%) rename src/db/{ => cauldron}/tx.rs (100%) rename src/db/{ => cauldron}/utxo_funding.rs (100%) rename src/db/{ => cauldron}/utxo_spending.rs (100%) create mode 100644 src/index.rs create mode 100644 src/rpc/bcmr.rs create mode 100644 src/rpc/tokens.rs diff --git a/Cargo.lock b/Cargo.lock index 7a6ee10..88bcf80 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -154,6 +154,12 @@ dependencies = [ "rustc-demangle", ] +[[package]] +name = "base64" +version = "0.22.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" + [[package]] name = "bech32" version = "0.9.1" @@ -329,6 +335,15 @@ version = "0.8.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "06ea2b9bc92be3c2baa9334a323ebca2d6f074ff852cd1d7b11064035cd3868f" +[[package]] +name = "crc32fast" +version = "1.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a97769d94ddab943e4510d138150169a2758b5ef3eb191a9ee688de3e23ef7b3" +dependencies = [ + "cfg-if", +] + [[package]] name = "crossbeam-deque" version = "0.8.3" @@ -418,10 +433,10 @@ dependencies = [ "byteorder", "libc", "log", - "rustls", + "rustls 0.21.9", "serde", "serde_json", - "webpki-roots", + "webpki-roots 0.25.4", "winapi", ] @@ -482,6 +497,16 @@ dependencies = [ "version_check", ] +[[package]] +name = "flate2" +version = "1.0.30" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5f54427cfd1c7829e2a139fcefea601bf088ebca651d2bf53ebc600eac295dae" +dependencies = [ + "crc32fast", + "miniz_oxide", +] + [[package]] name = "fnv" version = "1.0.7" @@ -1324,6 +1349,7 @@ dependencies = [ "log", "r2d2", "r2d2_sqlite", + "rand", "rayon", "riftenlabs-defi", "rocket", @@ -1334,6 +1360,7 @@ dependencies = [ "serde", "serde_json", "stderrlog", + "ureq", ] [[package]] @@ -1545,10 +1572,30 @@ checksum = "629648aced5775d558af50b2b4c7b02983a04b312126d45eeead26e7caa498b9" dependencies = [ "log", "ring", - "rustls-webpki", + "rustls-webpki 0.101.7", "sct", ] +[[package]] +name = "rustls" +version = "0.22.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bf4ef73721ac7bcd79b2b315da7779d8fc09718c6b3d2d1b2d94850eb8c18432" +dependencies = [ + "log", + "ring", + "rustls-pki-types", + "rustls-webpki 0.102.4", + "subtle", + "zeroize", +] + +[[package]] +name = "rustls-pki-types" +version = "1.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "976295e77ce332211c0d24d92c0e83e50f5c5f046d11082cea19f3df13a3562d" + [[package]] name = "rustls-webpki" version = "0.101.7" @@ -1559,6 +1606,17 @@ dependencies = [ "untrusted", ] +[[package]] +name = "rustls-webpki" +version = "0.102.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff448f7e92e913c4b7d4c6d8e4540a1724b319b4152b8aef6d4cf8339712b33e" +dependencies = [ + "ring", + "rustls-pki-types", + "untrusted", +] + [[package]] name = "rustversion" version = "1.0.14" @@ -1702,9 +1760,9 @@ dependencies = [ [[package]] name = "smallvec" -version = "1.11.2" +version = "1.13.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4dccd0940a2dcdf68d092b8cbab7dc0ad8fa938bf95787e1b916b0e3d0e8e970" +checksum = "3c5e1a9a646d36c3599cd173a41282daf47c44583ad367b8e6837255952e5c67" [[package]] name = "socket2" @@ -1763,6 +1821,12 @@ dependencies = [ "thread_local", ] +[[package]] +name = "subtle" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "81cdd64d312baedb58e21336b31bc043b77e01cc99033ce76ef539f78e965ebc" + [[package]] name = "syn" version = "1.0.109" @@ -2111,6 +2175,25 @@ version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" +[[package]] +name = "ureq" +version = "2.9.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d11a831e3c0b56e438a28308e7c810799e3c118417f342d30ecec080105395cd" +dependencies = [ + "base64", + "flate2", + "log", + "once_cell", + "rustls 0.22.4", + "rustls-pki-types", + "rustls-webpki 0.102.4", + "serde", + "serde_json", + "url", + "webpki-roots 0.26.1", +] + [[package]] name = "url" version = "2.5.0" @@ -2225,6 +2308,15 @@ version = "0.25.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5f20c57d8d7db6d3b86154206ae5d8fba62dd39573114de97c2cb0578251f8e1" +[[package]] +name = "webpki-roots" +version = "0.26.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b3de34ae270483955a94f4b21bdaaeb83d508bb84a01435f393818edb0012009" +dependencies = [ + "rustls-pki-types", +] + [[package]] name = "winapi" version = "0.3.9" @@ -2452,3 +2544,9 @@ dependencies = [ "quote", "syn 2.0.39", ] + +[[package]] +name = "zeroize" +version = "1.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "525b4ec142c6b68a2d10f01f7bbf6755599ca3f81ea53b8431b7dd348f5fdb2d" diff --git a/Cargo.toml b/Cargo.toml index 9bb35f2..f783cb6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -29,6 +29,8 @@ r2d2 = "0.8.10" r2d2_sqlite = "0.24.0" rust_decimal = "1.34" rust_decimal_macros = "1.34" +ureq = { version = "2.9", features = ["json"] } +rand = "0.8.5" [profile.release] debug = true diff --git a/contrib/volume.py b/contrib/volume.py new file mode 100644 index 0000000..550f570 --- /dev/null +++ b/contrib/volume.py @@ -0,0 +1,16 @@ +#!/usr/bin/env python3 +# 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 + +from urllib.request import urlopen +import json + +url = "http://indexer.cauldron.quest/cauldron/contract/volume?end=9999999999" + +with urlopen(url) as response: + data = json.load(response) + total_satoshis = sum(item["total_sats"] for item in data) + print(f"Total BCH: {total_satoshis / 100000000}") + + diff --git a/src/bcmr/bcmrdownloader.rs b/src/bcmr/bcmrdownloader.rs new file mode 100644 index 0000000..bb633f3 --- /dev/null +++ b/src/bcmr/bcmrdownloader.rs @@ -0,0 +1,449 @@ +// 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 std::{ + sync::{atomic::AtomicBool, Arc}, + thread::{self, JoinHandle}, + time::Duration, +}; + +use crate::bcmr::parsedbcmr::ParsedBCMR; +use crate::db::bcmr::insert_bcmr_data; +use crate::db::bcmr::update_bcmr_failure; +use crate::{ + bcmr::parse_bcmr_from_opreturn, + db::{ + bcmr::{get_entries_missing_bcmr_download, AuthChainEntry}, + DBPool, + }, +}; +use anyhow::*; +use bitcoin_hashes::{hex::ToHex, sha256, Hash}; +use bitcoincash::{Script, TokenID}; +use log::{info, warn}; +use rand::thread_rng; +use serde_json::Value; +use ureq::Agent; + +use serde_json::json; + +use rand::seq::SliceRandom; +use rayon::prelude::*; +use std::result::Result::Ok; + +// We are generous on timeout to allow for slow ipfs gateway +const DOWNLOAD_TIMEOUT: Duration = Duration::from_secs(60); +const IPFS_GATEWAY: &str = "https://ipfs.io/ipfs/"; +const MAX_BCMR_SIZE: usize = 1024 * 1024 * 100; // 100 MB + +const MAX_PARALLEL_DOWNLOADS: usize = 5; + +pub struct BCMRDownloader { + db: DBPool, + keep_running: Arc, + download_thread: Option>, +} + +const LOOP_SLEEP_TIME: Duration = Duration::from_secs(10); + +fn get_download_candidates(db: &DBPool) -> Vec { + let conn = match db.get() { + Ok(c) => c, + Err(e) => { + warn!("bcmr: Failed to get a db connection: {}", e); + return Vec::default(); + } + }; + match get_entries_missing_bcmr_download(&conn) { + Ok(c) => c, + Err(e) => { + warn!("bcmr: Failed to fetch bcmr download candidates: {}", e); + Vec::default() + } + } +} + +fn get_url(url: &str, max_download_size: usize) -> Result<(String, String)> { + let agent = Agent::new(); + let response = agent.get(url).timeout(DOWNLOAD_TIMEOUT).call()?; + + if response.status() != 200 { + bail!("Failed to fetch {}: HTTP {}", url, response.status()); + } + + let mut reader = response.into_reader(); + let mut content = Vec::new(); + let mut buffer = [0; 1024]; // Read in chunks of 1KB. + + while let Ok(count) = reader.read(&mut buffer) { + if count == 0 { + break; + } + if content.len() + count > max_download_size { + bail!("Download exceeded the maximum allowed size"); + } + content.extend_from_slice(&buffer[..count]); + } + + let content_str = String::from_utf8(content)?; + let hash = sha256::Hash::hash(content_str.as_bytes()); + Ok((content_str, hash.to_hex())) +} + +fn fetch_bcmr( + urls: &[String], + expected_hash: &str, +) -> ( + Option<(String, String)>, + String, /* error */ + bool, /* fatal error (don't try again) */ +) { + let mut candidate: Option<(String, String)> = None; + let mut errors: Vec = Vec::new(); + + for url in urls { + let url = if let Some(stripped) = url.strip_prefix("ipfs://") { + format!("{}{}", IPFS_GATEWAY, stripped) + } else if !url.starts_with("https://") { + format!("https://{}", url) + } else { + url.to_string() + }; + + let (contents, actual_hash) = match get_url(&url, MAX_BCMR_SIZE) { + Ok(c) => c, + Err(e) => { + errors.push(e.to_string()); + continue; + } + }; + + if actual_hash == expected_hash { + return (Some((contents, actual_hash)), String::default(), false); + } else { + candidate = Some((contents, actual_hash)) + } + } + + // No URL's with matching exepected hash + if candidate.is_some() { + (candidate, String::default(), false) + } else if errors.is_empty() { + (None, "No URLs to fetch BCMR from".to_string(), true) + } else { + ( + None, + format!("Errors fetching BCMR: {}", errors.join("; ")), + false, + ) + } +} + +pub fn parse_bcmr_json( + bcmr: &Value, + token_id: &TokenID, + actual_hash: String, + expected_hash: String, +) -> Result { + let identities = bcmr + .get("identities") + .context("BCMR missing 'identities'")?; + let token_history = identities + .get(token_id.to_hex()) + .context("'identities' does not contain token ID")?; + let first_entry = token_history + .as_object() + .context("token history is not an object")? + .values() + .next() + .context("no entries in token history")?; + + let token = first_entry + .get("token") + .context("'token' missing in token history entry")?; + + let symbol = token + .get("symbol") + .context("'symbol' missing in 'token' entry")?; + let symbol = symbol.as_str().context("'symbol' field is not a string")?; + + // decimals is optional and defaults to 0 + let decimals_default = json!(0); + let decimals = token + .get("decimals") + .unwrap_or(&decimals_default) + .to_owned(); + + let decimals = { + let d_int = if let Some(d) = decimals.as_i64() { + d + } else if let Some(d_str) = decimals.as_str() { + d_str + .parse::() + .context("Invalid number in 'decimal' field for 'token")? + } else { + bail!("BCMR contains a 'decimal' field for 'token', but it's not a number") + }; + + if d_int < 0 { + bail!("'decimal' field for 'token' cannot be negative") + } + d_int as usize + }; + + let empty_string = json!(""); + let name = first_entry + .get("name") + .unwrap_or(&empty_string) + .as_str() + .unwrap_or(""); + let description = first_entry + .get("description") + .unwrap_or(&empty_string) + .as_str() + .unwrap_or(""); + + let (icon, web) = if let Some(uris) = first_entry.get("uris") { + let icon = uris + .get("icon") + .unwrap_or(&empty_string) + .as_str() + .unwrap_or(""); + let web = uris + .get("web") + .unwrap_or(&empty_string) + .as_str() + .unwrap_or(""); + (icon, web) + } else { + ("", "") + }; + + Ok(ParsedBCMR::new( + &token_id.to_hex(), + symbol, + decimals, + name, + description, + icon, + web, + expected_hash, + actual_hash, + )) +} + +impl BCMRDownloader { + pub fn new(db: DBPool) -> Self { + Self { + db, + keep_running: Arc::new(AtomicBool::new(true)), + download_thread: None, + } + } + + pub fn start(&mut self) -> Result<()> { + let db_cpy = self.db.clone(); + let keep_running_cpy = self.keep_running.clone(); + self.download_thread = Some( + thread::Builder::new() + .name("bcmr downloader".to_string()) + .spawn(move || loop { + if !keep_running_cpy.load(std::sync::atomic::Ordering::Relaxed) { + info!("Exiting bcmr download thread"); + return; + } + + let mut queue = get_download_candidates(&db_cpy); + + let mut rng = thread_rng(); + queue.shuffle(&mut rng); + + if queue.is_empty() { + thread::sleep(LOOP_SLEEP_TIME); + continue; + } + + info!("bcmr: {} tokens need BCMR download", queue.len()); + + let pool = rayon::ThreadPoolBuilder::new() + .num_threads(MAX_PARALLEL_DOWNLOADS) + .build() + .unwrap(); + + pool.install(|| { + queue.par_iter().for_each(|entry| { + info!( + "bcmr: Dowloading BCMR for token {}", + entry.token_id.to_hex() + ); + + let bcmr = parse_bcmr_from_opreturn(&Script::from( + entry.bcmr_data.as_ref().expect("bcmr data missing").clone(), + )) + .expect("invalid bcmr entry in db"); + + let (json_str, error, is_fatal) = + fetch_bcmr(&bcmr.uris, &bcmr.hash.to_hex()); + + let conn = match db_cpy.get() { + Ok(c) => c, + Err(e) => { + warn!("bcmr: Failed to get BCMR db connection: {}", e); + return; + } + }; + + let (json_str, actual_hash) = match json_str { + Some(j) => j, + None => { + if let Err(e) = update_bcmr_failure( + &conn, + &entry.utxo, + &format!("Failed to fetch BCMR: {}", error), + is_fatal, + ) { + warn!("bcmr: Failed to set BCMR error: {}", e) + } + return; + } + }; + + let json: Value = match serde_json::from_str(&json_str) { + Ok(b) => b, + Err(e) => { + if let Err(e) = update_bcmr_failure( + &conn, + &entry.utxo, + &format!("BCMR invalid JSON error: {}", e), + true, + ) { + warn!("bcmr: Failed to set BCMR error: {}", e) + } + return; + } + }; + + let bcmr_parsed = match parse_bcmr_json( + &json, + &entry.token_id, + actual_hash, + bcmr.hash.to_hex(), + ) { + Ok(b) => b, + Err(err) => { + if let Err(e) = update_bcmr_failure( + &conn, + &entry.utxo, + &format!("BCMR contents error: {}", err), + true, + ) { + warn!("bcmr: Failed to set BCMR error: {}", e) + } + return; + } + }; + + if let Err(err) = insert_bcmr_data(&conn, &entry.utxo, &bcmr_parsed) { + warn!("bcmr: Failed to insert BCMR data {}", err) + } + }); + }); + + thread::sleep(LOOP_SLEEP_TIME) + }) + .expect("failed to start bcmr download thread"), + ); + + Ok(()) + } +} + +impl Drop for BCMRDownloader { + fn drop(&mut self) { + self.keep_running + .store(false, std::sync::atomic::Ordering::SeqCst); + if let Some(thread) = self.download_thread.take() { + // Wake the thread in case it is sleeping + thread.thread().unpark(); + + match thread.join() { + Ok(_) => info!("bcmr download thread done"), + Err(e) => warn!("Failed to join bcmr download thread: {:?}", e), + } + } + } +} + +#[cfg(test)] +mod tests { + + use super::*; + use bitcoin_hashes::hex::FromHex; + + #[test] + fn test_bcmr_parse() { + let bcmr = r#" + { + "$schema": "https://cashtokens.org/bcmr-v2.schema.json", + "version": { + "major": 0, + "minor": 1, + "patch": 0 + }, + "latestRevision": "2023-05-15T12:18:32.912Z", + "registryIdentity": { + "name": "Fallout Coin", + "description": "One of the earliest coins ever created on BCH", + "uris": { + "icon": "https://c3-soft.com/tokens/icon.ico", + "web": "https://c3-soft.com/tokens/", + "registry": "https://c3-soft.com/tokens/registry.json" + } + }, + "identities": { + "83e12eea20b19a9a0906bb0521ff18520db69a4a8136293bafbfca0acb2c2313": { + "2023-05-12T12:00:00.000Z": { + "name": "Fallout Coin", + "description": "Commemorative Coin for the computer game Fallout", + "token": { + "category": "83e12eea20b19a9a0906bb0521ff18520db69a4a8136293bafbfca0acb2c2313", + "symbol": "FC", + "decimals": 2 + }, + "uris": { + "icon": "https://c3-soft.com/tokens/fallout.ico" + } + } + } + }, + "license": "CC0-1.0" + } + "#; + + let dummy_hash = TokenID::all_zeros().to_hex(); + + let parsed = parse_bcmr_json( + &serde_json::from_str(bcmr).unwrap(), + &TokenID::from_hex("83e12eea20b19a9a0906bb0521ff18520db69a4a8136293bafbfca0acb2c2313") + .unwrap(), + dummy_hash.clone(), + dummy_hash, + ) + .unwrap(); + + assert_eq!("FC", parsed.token.symbol); + assert_eq!(2, parsed.token.decimals); + assert_eq!("Fallout Coin", parsed.name); + assert_eq!( + "Commemorative Coin for the computer game Fallout", + parsed.description + ); + assert_eq!( + "https://c3-soft.com/tokens/fallout.ico", + parsed.uris.icon.unwrap() + ); + assert_eq!(None, parsed.uris.web); + } +} diff --git a/src/bcmr/mod.rs b/src/bcmr/mod.rs new file mode 100644 index 0000000..4a2d92f --- /dev/null +++ b/src/bcmr/mod.rs @@ -0,0 +1,345 @@ +// 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 rayon::prelude::*; +use std::{ + cmp, + collections::{HashMap, HashSet, VecDeque}, + convert::TryInto, +}; + +use anyhow::Result; +use bitcoin_hashes::{hex::ToHex, Hash}; +use bitcoincash::BlockHash; +use bitcoincash::{Script, TokenID, Transaction, Txid}; +use log::debug; +use riftenlabs_defi::chainutil::{compute_outpoint_hash, read_push_from_script, OutPointHash}; +use rusqlite::Connection; +use serde::{Deserialize, Serialize, Serializer}; + +use crate::db::bcmr::{get_matching_autheaders, insert_authheader, AuthChainEntry}; + +pub mod bcmrdownloader; +pub mod parsedbcmr; + +pub const BCMR_PREFIX: &[u8] = &[ + 0x6a, // OP_RETURN + 0x04, // PUSH 4 + 0x42, // B + 0x43, // C + 0x4d, // M + 0x52, // R +]; + +// Custom function to serialize [u8; 32] as a hex string +fn as_hex(bytes: &[u8; 32], serializer: S) -> Result +where + S: Serializer, +{ + serializer.serialize_str(&hex::encode(bytes)) +} + +#[derive(Debug, Serialize, Deserialize)] +#[allow(clippy::upper_case_acronyms)] +pub struct BCMR { + #[serde(serialize_with = "as_hex")] + pub hash: [u8; 32], + pub uris: Vec, + #[serde(skip)] + pub op_return: Vec, +} + +pub fn parse_bcmr_from_opreturn(bcmr_op_return: &Script) -> Option { + let iter = bcmr_op_return[6..].iter(); + let (iter, hash) = read_push_from_script(iter).ok()?; + + let hash: [u8; 32] = hash.and_then(|h| h.try_into().ok())?; + let mut uris: Vec = vec![]; + let mut uri_element; + let mut uri_iter = iter; + loop { + (uri_iter, uri_element) = match read_push_from_script(uri_iter) { + Ok(r) => r, + Err(_) => break, + }; + match uri_element { + Some(url) => { + if let Ok(uri) = String::from_utf8(url) { + uris.push(uri) + } + } + None => break, + } + } + + Some(BCMR { + hash, + uris, + op_return: bcmr_op_return.to_bytes(), + }) +} + +pub fn parse_bcmr(tx: &Transaction) -> Option { + let bcmr_op_return = tx + .output + .iter() + .find(|o| o.script_pubkey.as_bytes().starts_with(BCMR_PREFIX))?; + + parse_bcmr_from_opreturn(&bcmr_op_return.script_pubkey) +} + +fn is_genesis_tx(tx: &Transaction) -> Option { + let potential_token_ids: HashSet<[u8; 32]> = tx + .input + .iter() + .filter(|i| i.previous_output.vout == 0) + .map(|i| i.previous_output.txid.into_inner()) + .collect(); + + for o in &tx.output { + match &o.token { + Some(token) => { + if potential_token_ids.contains(token.id.as_inner()) { + return Some(token.id); + } + } + None => continue, + } + } + None +} + +pub fn ttor_sorted(txs: Vec) -> Vec { + let txs = { + let mut queue: VecDeque = txs.into_iter().collect(); + + let mut queue_txids: HashSet = queue.par_iter().map(|tx| tx.txid()).collect(); + + let mut txs: Vec = Vec::with_capacity(queue.len()); + + while let Some(tx) = queue.pop_front() { + let mut has_parent = false; + + for i in &tx.input { + if queue_txids.contains(&i.previous_output.txid) { + // depends on parent + has_parent = true; + break; + }; + } + + if has_parent { + queue.push_back(tx); + } else { + queue_txids.remove(&tx.txid()); + txs.push(tx) + } + } + txs + }; + txs +} + +// TODO: Use when we get scriptpubkey filter for electrum.mempoo.get +#[allow(dead_code)] +pub fn mempool_index_genesis(conn: &Connection, txs: &Vec) -> Result { + let genesis: Vec<(Transaction, TokenID)> = txs + .par_iter() + .filter_map(|tx| is_genesis_tx(tx).map(|token| (tx.clone(), token))) + .collect(); + + let mut inserts = 0; + for (tx, token) in genesis { + let txid = tx.txid(); + let utxo = compute_outpoint_hash(&txid, 0); + let bcmr_data = parse_bcmr(&tx); + insert_authheader( + conn, + &utxo, + &BlockHash::all_zeros(), + &txid, + &token, + 0, + bcmr_data.map(|b| b.op_return), + )?; + inserts += 1; + } + + Ok(inserts) +} + +pub fn index_bcmr( + conn: &Connection, + blockhash: &BlockHash, + txs: Vec, +) -> Result { + let sorted = ttor_sorted(txs); + + let mut new_tokens: Vec<(Transaction, TokenID)> = Vec::default(); + + let sorted: Vec = sorted + .into_iter() + .filter_map(|tx| { + if let Some(token_id) = is_genesis_tx(&tx) { + // filter out new; we know it to be start of authchain + new_tokens.push((tx, token_id)); + None + } else { + Some(tx) + } + }) + .collect(); + + let mut position_index: HashMap = HashMap::default(); + + // Look for authchain updates. We collect all candidates first. + let mut candidates: HashMap = sorted + .into_iter() + .enumerate() + .filter_map(|(pos, tx)| { + let prevout = tx.input.first().expect("tx with no inputs").previous_output; + if prevout.vout != 0 { + // authchain needs to spend first output of a previous tx + None + } else { + let utxohash = compute_outpoint_hash(&prevout.txid, prevout.vout); + position_index.insert(utxohash, pos); + Some((utxohash, tx)) + } + }) + .collect(); + + let mut inserts = 0; + + // Start by inserting new auth chains + for (tx, token_id) in new_tokens { + let bcmr = parse_bcmr(&tx); + debug!( + "Found token gensis for {} in {}; has bcmr: {}", + token_id.to_hex(), + tx.txid().to_hex(), + bcmr.is_some() + ); + let txid = tx.txid(); + let utxohash = compute_outpoint_hash(&tx.txid(), 0); + insert_authheader( + conn, + &utxohash, + blockhash, + &txid, + &token_id, + 0, + bcmr.map(|bcmr| bcmr.op_return), + )?; + inserts += 1; + } + + // Finally; we need to loop for remaining updates; as each update has the potential to add more candidates. + loop { + let matches: Vec = get_matching_autheaders(conn, candidates.keys())?; + + if matches.is_empty() { + break; + } + + let mut smallest_index = usize::MAX; + + for prev_autheader in matches { + let tx = candidates + .remove(&prev_autheader.utxo) + .expect("match not in candidate list"); + + let bcmr = parse_bcmr(&tx); + + let txid = tx.txid(); + let utxohash = compute_outpoint_hash(&txid, 0); + debug!( + "Found authheader update {} in {}; has bcmr: {}; new height: {}, utxo: {}", + prev_autheader.token_id.to_hex(), + txid.to_hex(), + bcmr.is_some(), + prev_autheader.height + 1, + utxohash.to_hex() + ); + + insert_authheader( + conn, + &utxohash, + blockhash, + &txid, + &prev_autheader.token_id, + prev_autheader.height + 1, + bcmr.map(|bcmr| bcmr.op_return), + )?; + + smallest_index = cmp::min( + smallest_index, + *position_index + .get(&prev_autheader.utxo) + .expect("entry not in position index"), + ); + inserts += 1; + } + + // since it's ttor sorted; there can be no chained updates above the first candidate; + // and we can remove all candidates above smallest_index + position_index.retain(|hash, pos| { + if pos > &mut smallest_index { + true + } else { + candidates.remove(hash); + false + } + }); + } + + Ok(inserts) +} + +#[cfg(test)] +mod tests { + use bitcoincash::consensus::deserialize; + + use super::*; + + #[test] + fn test_parse_bcmr() { + let tx_hex = "0200000001925bd92e424c0f0bc290a794f491abf19e61b6dcdec7e70747fcb54682fc9bb700000000644181707613f45069c8df981b2ef0cbd05158d5623602fac3ae2e13f69d05fa356c278f2f3a4e34f33a79e0f8375abd35051150ec024b3625807cea185528a5de9e412103b4680dffa1e34b3bfdb024fd0a8498eb24f59ba5ead34acfb74f9d8549db6f3a0000000003e8030000000000003eef925bd92e424c0f0bc290a794f491abf19e61b6dcdec7e70747fcb54682fc9bb710fdf40176a914bd4c90f2c64743fc0d3ea6a14973a5d628260b7388ac0000000000000000456a0442434d5220c705cc90a56ac7ef9a15ef90ebbc8ba7e60e4c622e5464d52d8baf7887949fcc1d736f636b2e6361756c64726f6e2e71756573742f62636d722e6a736f6ed9210000000000001976a914bd4c90f2c64743fc0d3ea6a14973a5d628260b7388ac00000000"; + let tx: Transaction = deserialize(&hex::decode(&tx_hex).unwrap()).unwrap(); + + let result = parse_bcmr(&tx); + assert!(result.is_some()); + let bcmr = result.unwrap(); + + assert_eq!( + hex::encode(bcmr.hash), + "c705cc90a56ac7ef9a15ef90ebbc8ba7e60e4c622e5464d52d8baf7887949fcc" + ); + assert_eq!(bcmr.uris.len(), 1); + assert_eq!(bcmr.uris[0], "sock.cauldron.quest/bcmr.json"); + } + + #[test] + fn invalid_bcmr_hash() { + // txid 4e44c45cf1fb956fe4752c6dfa9637f29cf18e9251e20454dd65dbcc298617b6 + let tx_hex = "0200000002020056e81cda3b55f6b912acf05131e4f50594dc76702e8022d5522582f74bd40000000064414048069ab85d72e0f195e6587a4ef30360d5304a7d24995e21f8bb3d9c9c52eb5d1408a3a4754fad8da6593dc65fd248e507fe1620004bcb87f3237fcea9679e4121028bb6fdf7137233695237a6fa538827afb4f7490bad82215098b40dc3d9c645fc00000000020056e81cda3b55f6b912acf05131e4f50594dc76702e8022d5522582f74bd4010000006441d501d321b457b2622388efc9290245e7021f082d2914bf1883184ea29f546a2b08d74537f848fba17d41e44ad6f9d6949b44670b379635b00fe69caa5236e47f4121028bb6fdf7137233695237a6fa538827afb4f7490bad82215098b40dc3d9c645fc0000000003e80300000000000042ef020056e81cda3b55f6b912acf05131e4f50594dc76702e8022d5522582f74bd431ff00b402869d7e0100a9147ca730ccf0ba8c552bae1287d01a9e7e5590f0fa870000000000000000d56a0442434d5240646331303939346363323530363731643830383737626639356131613136616164356431343033623365623263373136346366623231383532323363353162664c506e667473746f726167652e6c696e6b2f697066732f6261666b726569673463636d757a7173716d346f796262333337666e627566766b32786975616f7a36776c64726d746833656763736570637278343b6261666b726569673463636d757a7173716d346f796262333337666e627566766b32786975616f7a36776c64726d74683365676373657063727834bb0f0000000000001976a914d948abf743a76f472703b6a91437854dc1d7ced588ac00000000"; + let tx: Transaction = deserialize(&hex::decode(&tx_hex).unwrap()).unwrap(); + let result = parse_bcmr(&tx); + // The hash value in this BCMR is 64 bytes. Not a valid sha256 hash. + assert!(result.is_none()); + } + + #[test] + fn test_is_genesis_tx() { + let tx_hex = "01000000032b70d6279de034fea76d4dd329c44a3e162b17a0cebc32eeeebbd14f5e8a97250100000064411b42a9dcb22c59ff0e800b3c63d6b8ac045bd18eb1effc8d2f23ec49ed0abdc316215a469a206d45153eac999d14eb8abc3d33b91c5ac4b61e7868d162d5d5184121021735db91ac477c1007da74c8c117c383eebc664d21f12871ca6e69401f8072f8feffffff2b70d6279de034fea76d4dd329c44a3e162b17a0cebc32eeeebbd14f5e8a972502000000644193e4d5bb533279fbcbebe15ba62f226e53e5c64a1624c755b255b13f0b1011e382182896026575370b5e30d03390d31faabf854a9f41c215b8a5bb57bae221df4121021735db91ac477c1007da74c8c117c383eebc664d21f12871ca6e69401f8072f8feffffff8d80b5567b6a3cac6e19a6934fd7544a02616fd6868a304c93bff6aed3eb12280100000064418b576b5defc8942182d01654cc2a0706b3e86e9031f7a13a694ffd830944ca00ecf5ad3271a242993580cbe78c13a41e35b06dfd68e72c4242dd7ff9e7e039a6412103b666cf8bc6ef2e3c3f100fc883013040f01f7085d75e5ab0811ae00b6a0a6d24feffffff03200300000000000040ef8d80b5567b6a3cac6e19a6934fd7544a02616fd6868a304c93bff6aed3eb1228600401db010076a914bfe9ed4ea9c7830c2ee96c17a3bd5ff563c12f2b88ac200300000000000040ef8d80b5567b6a3cac6e19a6934fd7544a02616fd6868a304c93bff6aed3eb1228600402db010076a914bfe9ed4ea9c7830c2ee96c17a3bd5ff563c12f2b88ac9f117901000000001976a91402a67e8d884367e7a22d6cda96a6df21e56a2f3288ac851b0c00"; + let not_genesis_tx: Transaction = deserialize(&hex::decode(&tx_hex).unwrap()).unwrap(); + + let tx_hex = "0200000001925bd92e424c0f0bc290a794f491abf19e61b6dcdec7e70747fcb54682fc9bb700000000644181707613f45069c8df981b2ef0cbd05158d5623602fac3ae2e13f69d05fa356c278f2f3a4e34f33a79e0f8375abd35051150ec024b3625807cea185528a5de9e412103b4680dffa1e34b3bfdb024fd0a8498eb24f59ba5ead34acfb74f9d8549db6f3a0000000003e8030000000000003eef925bd92e424c0f0bc290a794f491abf19e61b6dcdec7e70747fcb54682fc9bb710fdf40176a914bd4c90f2c64743fc0d3ea6a14973a5d628260b7388ac0000000000000000456a0442434d5220c705cc90a56ac7ef9a15ef90ebbc8ba7e60e4c622e5464d52d8baf7887949fcc1d736f636b2e6361756c64726f6e2e71756573742f62636d722e6a736f6ed9210000000000001976a914bd4c90f2c64743fc0d3ea6a14973a5d628260b7388ac00000000"; + let genesis_tx: Transaction = deserialize(&hex::decode(&tx_hex).unwrap()).unwrap(); + + assert!(!is_genesis_tx(¬_genesis_tx).is_some()); + assert!(is_genesis_tx(&genesis_tx).is_some()); + } +} diff --git a/src/bcmr/parsedbcmr.rs b/src/bcmr/parsedbcmr.rs new file mode 100644 index 0000000..2b2535f --- /dev/null +++ b/src/bcmr/parsedbcmr.rs @@ -0,0 +1,104 @@ +// 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 + +// The ParsedBCMR struct mirrors identity struct in the BCMR definition + +use serde::Serialize; + +const MAX_SYMBOL_LEN: usize = 100; +const MAX_NAME_LEN: usize = 500; +const MAX_DESCRIPTION_LEN: usize = 10000; +const MAX_URL_LEN: usize = 2000; + +#[derive(Serialize)] +pub struct Token { + pub category: String, + pub symbol: String, + pub decimals: usize, +} + +#[derive(Serialize)] +pub struct Uris { + pub icon: Option, + pub web: Option, +} + +#[derive(Serialize)] +pub struct FileMeta { + pub expected_hash: String, + pub actual_hash: String, +} + +#[derive(Serialize)] +pub struct ParsedBCMR { + pub name: String, + pub description: String, + pub token: Token, + pub uris: Uris, + pub filemeta: FileMeta, +} + +fn trim_str(str: &str, new_size: usize) -> String { + let mut iter = str.char_indices(); + + let (start, _) = match iter.next() { + Some(i) => i, + None => { + // not sure why this should fail (empty string?) + return str.to_owned(); + } + }; + let (end, _) = match iter.nth(new_size) { + Some(i) => i, + None => { + return str.to_owned(); + } + }; + + str[start..end].to_owned() + "..." +} + +impl ParsedBCMR { + #[allow(clippy::too_many_arguments)] + pub fn new( + category: &str, + symbol: &str, + decimals: usize, + name: &str, + description: &str, + icon: &str, + web: &str, + expected_hash: String, + actual_hash: String, + ) -> Self { + // Trim inputs so avoid bogus BCMR files populating our db with garbage + + Self { + name: trim_str(name, MAX_NAME_LEN), + description: trim_str(description, MAX_DESCRIPTION_LEN), + token: Token { + category: category.to_owned(), + symbol: trim_str(symbol, MAX_SYMBOL_LEN), + decimals, + }, + uris: Uris { + icon: if icon.is_empty() { + None + } else { + Some(trim_str(icon, MAX_URL_LEN)) + }, + web: if web.is_empty() { + None + } else { + Some(trim_str(web, MAX_URL_LEN)) + }, + }, + filemeta: FileMeta { + expected_hash, + actual_hash, + }, + } + } +} diff --git a/src/chain.rs b/src/chain.rs index a683840..6c4d50e 100644 --- a/src/chain.rs +++ b/src/chain.rs @@ -16,7 +16,7 @@ use serde_json::Value; use std::{collections::HashMap, sync::RwLock}; use crate::{ - db::{config::config_set, delete_entries_for_block, DBPool}, + db::{self, cauldron::config::config_set, DB}, KEY_LAST_INDEXED, }; @@ -43,22 +43,29 @@ pub trait BlockUndoer { } pub struct StoreBlockUndoer { - db: DBPool, + db: DB, } impl StoreBlockUndoer { - pub fn new(db: DBPool) -> Result { + pub fn new(db: DB) -> Result { Ok(Self { db }) } } impl BlockUndoer for StoreBlockUndoer { fn undo_block(&self, blockheader: &BlockHeader) -> Result<()> { - let conn = self.db.get().unwrap(); - if delete_entries_for_block(&conn, &blockheader.block_hash())? { + let cauldron_conn = self.db.cauldron_w.get().unwrap(); + let mut was_indexed = + db::cauldron::delete_entries_for_block(&cauldron_conn, &blockheader.block_hash())?; + let bcmr_conn = self.db.bcmr_w.get().unwrap(); + if db::bcmr::delete_entries_for_block(&bcmr_conn, &blockheader.block_hash())? { + was_indexed = true; + } + + if was_indexed { // This block was indexed. Need to update indexing flag. config_set( - &conn, + &cauldron_conn, KEY_LAST_INDEXED, &blockheader.prev_blockhash.to_hex(), ); diff --git a/src/db/bcmr/mod.rs b/src/db/bcmr/mod.rs new file mode 100644 index 0000000..3d1c470 --- /dev/null +++ b/src/db/bcmr/mod.rs @@ -0,0 +1,337 @@ +// 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 crate::bcmr::parsedbcmr::{FileMeta, ParsedBCMR, Token, Uris}; +use anyhow::*; +use bitcoin_hashes::hex::{FromHex, ToHex}; +use bitcoincash::{BlockHash, TokenID, Txid}; +use log::info; +use riftenlabs_defi::chainutil::OutPointHash; +use rusqlite::{params, Connection}; + +// Flag BCMR download as give up after this many attempts. +const MAX_DOWNLOAD_ATTEMPTS: usize = 100; + +// BCMR auth chain entry, matches auth_chain_entry table +pub struct AuthChainEntry { + pub utxo: OutPointHash, + pub txid: Txid, + pub token_id: TokenID, + pub height: usize, + pub bcmr_data: Option>, +} + +pub fn prepare_tables(conn: &Connection) { + conn.execute( + "CREATE TABLE auth_chain_entry ( + utxo TEXT PRIMARY KEY, + blockhash TEXT NOT NULL, + txid TEXT NOT NULL, + token_id TEXT NOT NULL, + height INT NOT NULL, + bcmr_data TEXT + )", + [], + ) + .expect("failed to create auth_chain_entry table"); + + conn.execute( + "CREATE TABLE bcmr_data ( + utxo TEXT PRIMARY KEY, + symbol TEXT NOT NULL, + decimals INT NOT NULL, + name TEXT NOT NULL, + description TEXT NOT NULL, + icon TEXT NOT NULL, + web TEXT NOT NULL, + expected_hash TEXT NOT NULL, + actual_hash TEXT NOT NULL, + FOREIGN KEY (utxo) REFERENCES auth_chain_entry(utxo) + )", + [], + ) + .expect("failed to create bcmr_data table"); + + conn.execute( + "CREATE TABLE bcmr_failure ( + utxo TEXT PRIMARY KEY, + last_attempt INT NOT NULL, + attempts INT NOT NULL, + error_message TEXT, + give_up BOOLEAN, + FOREIGN KEY (utxo) REFERENCES auth_chain_entry(utxo) + )", + [], + ) + .expect("failed to create bcmr_failure table"); +} + +pub fn delete_entries_for_block(tx: &Connection, blockhash: &BlockHash) -> Result { + let mut stmt = tx.prepare("DELETE FROM auth_chain_entry WHERE blockhash = ?")?; + let rows_deleted = stmt.execute(params![blockhash.to_hex()])?; + + Ok(rows_deleted != 0) +} + +/// Add or update config entry +pub fn insert_authheader( + conn: &Connection, + utxo: &OutPointHash, + blockhash: &BlockHash, + txid: &Txid, + token_id: &TokenID, + height: usize, + bcmr_data: Option>, +) -> Result<()> { + let mut stmt = conn + .prepare("INSERT OR REPLACE INTO auth_chain_entry (utxo, blockhash, txid, token_id, height, bcmr_data) VALUES (?, ?, ?, ?, ?, ?)")?; + stmt.execute(params![ + utxo.to_hex(), + blockhash.to_hex(), + txid.to_hex(), + token_id.to_hex(), + height, + bcmr_data.map(|bcmr| bcmr.to_hex()) + ])?; + Ok(()) +} + +/// Get config entry +#[allow(dead_code)] +pub fn get_authheader(conn: &Connection, utxo: &OutPointHash) -> Result> { + let mut stmt = conn + .prepare("SELECT token_id, txid, height, bcmr_data FROM auth_chain_entry WHERE utxo = ?")?; + let mut row = stmt.query([utxo.to_hex()])?; + + let auth_header = row.next()?; + + if let Some(header) = auth_header { + let token_hex: String = header.get(0)?; + let txid_hex: String = header.get(1)?; + let height = header.get(2)?; + let bcmr_data_hex: Option = header.get(3)?; + let bcmr_data = if let Some(data) = bcmr_data_hex { + Some(hex::decode(data).context("failed to decode bcmr data")?) + } else { + None + }; + + Ok(Some(AuthChainEntry { + utxo: *utxo, + token_id: TokenID::from_hex(&token_hex).context("failed to decode token hex")?, + txid: Txid::from_hex(&txid_hex).context("failed to decode txid")?, + height, + bcmr_data, + })) + } else { + Ok(None) + } +} + +pub fn get_matching_autheaders<'a>( + conn: &Connection, + keys: impl Iterator, +) -> Result> { + // To not exceed SQL query length; we chunk the queries into entries of this size. + let chunk_size = 10000; + + let keys: Vec = keys.map(|utxo| utxo.to_hex()).collect(); + let mut matches: Vec = Vec::new(); + + // Process in chunks + for chunk in keys.chunks(chunk_size) { + let in_clause = chunk.join("','"); + let sql = format!( + "SELECT token_id, txid, height, bcmr_data, utxo FROM auth_chain_entry WHERE utxo IN ('{}')", + in_clause + ); + + let mut stmt = conn.prepare(&sql)?; + + let mut rows = stmt.query([])?; + + while let Some(header) = rows.next()? { + let token_hex: String = header.get(0)?; + let txid_hex: String = header.get(1)?; + let height = header.get(2)?; + let bcmr_data_hex: Option = header.get(3)?; + let bcmr_data = if let Some(data) = bcmr_data_hex { + Some(hex::decode(&data).context("failed to decode bcmr data")?) + } else { + None + }; + + let utxo_hex: String = header.get(4)?; + let utxo = OutPointHash::from_hex(&utxo_hex)?; + + matches.push(AuthChainEntry { + utxo, + token_id: TokenID::from_hex(&token_hex).context("failed to decode token hex")?, + txid: Txid::from_hex(&txid_hex).context("failed to decode txid")?, + height, + bcmr_data, + }) + } + } + + Ok(matches) +} + +pub fn get_entries_missing_bcmr_download(conn: &Connection) -> Result> { + let sql = "WITH MaxHeight AS ( + -- CTE to find the highest height for each token_id where bcmr_data is not null initially + SELECT + token_id, + MAX(height) AS max_height + FROM + auth_chain_entry + WHERE + bcmr_data IS NOT NULL + GROUP BY + token_id + ) + SELECT + ace.token_id, ace.txid, ace.height, ace.bcmr_data, ace.utxo + FROM + auth_chain_entry AS ace + -- Join with MaxHeight to get the maximum height entries + JOIN MaxHeight mh ON ace.token_id = mh.token_id AND ace.height = mh.max_height + -- Exclude entries that have a corresponding entry in bcmr_data + LEFT JOIN bcmr_data bd ON ace.utxo = bd.utxo + -- Exclude entries that have a corresponding failure entry where give_up = true or last attempt was less than 1800 seconds ago + LEFT JOIN bcmr_failure bf ON ace.utxo = bf.utxo AND (bf.give_up = 1 OR (strftime('%s', 'now') - bf.last_attempt) < 1800) + WHERE + -- Ensure no matching entries in bcmr_data and only include bcmr_failure when filtered conditions aren't met + bd.utxo IS NULL AND + bf.utxo IS NULL; + "; + + let mut stmt = conn.prepare(sql)?; + + let mut rows = stmt.query([])?; + + let mut matches: Vec = Vec::new(); + + while let Some(header) = rows.next()? { + let token_hex: String = header.get(0)?; + let txid_hex: String = header.get(1)?; + let height = header.get(2)?; + let bcmr_data_hex: Option = header.get(3)?; + let bcmr_data = if let Some(data) = bcmr_data_hex { + Some(hex::decode(&data).context("failed to decode bcmr data")?) + } else { + None + }; + + let utxo_hex: String = header.get(4)?; + let utxo = OutPointHash::from_hex(&utxo_hex)?; + + matches.push(AuthChainEntry { + utxo, + token_id: TokenID::from_hex(&token_hex).context("failed to decode token hex")?, + txid: Txid::from_hex(&txid_hex).context("failed to decode txid")?, + height, + bcmr_data, + }) + } + + Ok(matches) +} + +pub fn insert_bcmr_data( + conn: &rusqlite::Connection, + utxo: &OutPointHash, + bcmr: &ParsedBCMR, +) -> Result<()> { + let sql = "INSERT OR REPLACE INTO bcmr_data (utxo, symbol, decimals, name, description, icon, web, expected_hash, actual_hash) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)"; + + info!("Trying to insert {}", utxo.to_hex()); + let empty_string: String = "".to_owned(); + conn.execute( + sql, + rusqlite::params![ + &utxo.to_hex(), + &bcmr.token.symbol, + bcmr.token.decimals, + &bcmr.name, + &bcmr.description, + &bcmr.uris.icon.as_ref().unwrap_or(&empty_string), + &bcmr.uris.web.as_ref().unwrap_or(&empty_string), + &bcmr.filemeta.expected_hash, + &bcmr.filemeta.actual_hash + ], + )?; + Ok(()) +} + +pub fn update_bcmr_failure( + conn: &Connection, + utxo: &OutPointHash, + error_message: &str, + give_up: bool, +) -> Result<()> { + let sql = format!( + " + INSERT INTO bcmr_failure (utxo, last_attempt, attempts, error_message, give_up) + VALUES (?1, strftime('%s', 'now'), 1, ?2, ?3) + ON CONFLICT(utxo) DO UPDATE SET + last_attempt = excluded.last_attempt, + attempts = bcmr_failure.attempts + 1, + error_message = excluded.error_message, + give_up = CASE + WHEN bcmr_failure.attempts + 1 > {} THEN true + ELSE excluded.give_up + END", + MAX_DOWNLOAD_ATTEMPTS + ); + conn.execute(&sql, params![&utxo.to_hex(), error_message, give_up])?; + Ok(()) +} + +pub fn get_token_bcmr(conn: &Connection, token_hex: &str) -> Result> { + let sql = "SELECT + b.symbol, b.decimals, b.name, b.description, b.icon, b.web, b.actual_hash, b.expected_hash + FROM auth_chain_entry a + JOIN bcmr_data b ON a.utxo = b.utxo + WHERE a.token_id = ? + AND a.bcmr_data IS NOT NULL + ORDER BY a.height DESC + LIMIT 1"; + + let mut stmt = conn.prepare(sql)?; + let mut row = stmt.query([token_hex])?; + + if let Some(r) = row.next()? { + let symbol: String = r.get(0)?; + let decimals: usize = r.get(1)?; + let name = r.get(2)?; + let description = r.get(3)?; + let icon: String = r.get(4)?; + let web: String = r.get(5)?; + let actual_hash: String = r.get(6)?; + let expected_hash: String = r.get(7)?; + + Ok(Some(ParsedBCMR { + name, + description, + token: Token { + category: token_hex.to_owned(), + symbol, + decimals, + }, + uris: Uris { + web: if web.is_empty() { None } else { Some(web) }, + icon: if icon.is_empty() { None } else { Some(icon) }, + }, + filemeta: FileMeta { + expected_hash, + actual_hash, + }, + })) + } else { + Ok(None) + } +} diff --git a/src/db/config.rs b/src/db/cauldron/config.rs similarity index 89% rename from src/db/config.rs rename to src/db/cauldron/config.rs index cfe50ae..e212825 100644 --- a/src/db/config.rs +++ b/src/db/cauldron/config.rs @@ -19,5 +19,9 @@ pub fn config_get(conn: &Connection, key: &str) -> Result> { let mut stmt = conn.prepare("SELECT value FROM config WHERE key = ?")?; let mut row = stmt.query([key])?; - Ok(row.next()?.map(|r| r.get(0).unwrap())) + if let Some(r) = row.next()? { + Ok(r.get(0)?) + } else { + Ok(None) + } } diff --git a/src/db/header.rs b/src/db/cauldron/header.rs similarity index 100% rename from src/db/header.rs rename to src/db/cauldron/header.rs diff --git a/src/db/mempool.rs b/src/db/cauldron/mempool.rs similarity index 100% rename from src/db/mempool.rs rename to src/db/cauldron/mempool.rs diff --git a/src/db/cauldron/mod.rs b/src/db/cauldron/mod.rs new file mode 100644 index 0000000..8b54357 --- /dev/null +++ b/src/db/cauldron/mod.rs @@ -0,0 +1,76 @@ +// 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::ToHex; +use bitcoincash::BlockHash; +use rusqlite::{params, Connection}; + +pub mod config; +pub mod header; +pub mod mempool; +pub mod pool; +pub mod tx; +pub mod utxo_funding; +pub mod utxo_spending; + +pub fn prepare_tables(conn: &Connection) { + tx::create_table(conn); + utxo_funding::create_table(conn); + utxo_spending::create_table(conn); + + conn.execute( + "CREATE TABLE config ( + key TEXT PRIMARY KEY, + value TEXT + )", + [], + ) + .unwrap(); + + conn.execute( + "CREATE TABLE headers ( + key TEXT PRIMARY KEY, + height INT, + header BLOB + )", + [], + ) + .unwrap(); + + // Keep track of what was the initial deployment of a Cauldron + pool::create_table(conn); + // Indexes for list_by_volume + conn.execute( + "CREATE INDEX idx_utxo_funding_join ON utxo_funding(new_utxo_hash, sats, token_id);", + [], + ) + .unwrap(); + conn.execute("CREATE INDEX idx_utxo_funding_tvl_highest ON utxo_funding(token_id, new_utxo_hash, sats, token_amount);", []).unwrap(); +} + +pub fn delete_entries_for_block(tx: &Connection, blockhash: &BlockHash) -> Result { + let mut stmt = tx.prepare( + "DELETE FROM utxo_funding + WHERE txid IN ( + SELECT txid FROM tx WHERE blockhash = ? + )", + )?; + let mut rows_deleted = stmt.execute(params![blockhash.to_hex()])?; + + let mut stmt = tx.prepare( + "DELETE FROM utxo_spending + WHERE txid IN ( + SELECT txid FROM utxo_funding WHERE txid IN ( + SELECT txid FROM tx WHERE blockhash = ? + ) + )", + )?; + rows_deleted += stmt.execute(params![blockhash.to_hex()])?; + + // remaining tables should clear themselves due to foregn keys usage + + Ok(rows_deleted != 0) +} diff --git a/src/db/pool.rs b/src/db/cauldron/pool.rs similarity index 100% rename from src/db/pool.rs rename to src/db/cauldron/pool.rs diff --git a/src/db/tx.rs b/src/db/cauldron/tx.rs similarity index 100% rename from src/db/tx.rs rename to src/db/cauldron/tx.rs diff --git a/src/db/utxo_funding.rs b/src/db/cauldron/utxo_funding.rs similarity index 100% rename from src/db/utxo_funding.rs rename to src/db/cauldron/utxo_funding.rs diff --git a/src/db/utxo_spending.rs b/src/db/cauldron/utxo_spending.rs similarity index 100% rename from src/db/utxo_spending.rs rename to src/db/cauldron/utxo_spending.rs diff --git a/src/db/mod.rs b/src/db/mod.rs index 4f64bb0..92fee2d 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -5,76 +5,19 @@ use std::sync::Arc; -use anyhow::Result; -use bitcoin_hashes::hex::ToHex; -use bitcoincash::BlockHash; -use rusqlite::{params, Connection}; - -pub mod config; -pub mod header; -pub mod mempool; -pub mod pool; -pub mod tx; -pub mod utxo_funding; -pub mod utxo_spending; +pub mod bcmr; +pub mod cauldron; pub type DBPool = Arc>; -pub fn prepare_tables(conn: &Connection) { - tx::create_table(conn); - utxo_funding::create_table(conn); - utxo_spending::create_table(conn); - - conn.execute( - "CREATE TABLE config ( - key TEXT PRIMARY KEY, - value TEXT - )", - [], - ) - .unwrap(); - - conn.execute( - "CREATE TABLE headers ( - key TEXT PRIMARY KEY, - height INT, - header BLOB - )", - [], - ) - .unwrap(); - - // Keep track of what was the initial deployment of a Cauldron - pool::create_table(conn); - // Indexes for list_by_volume - conn.execute( - "CREATE INDEX idx_utxo_funding_join ON utxo_funding(new_utxo_hash, sats, token_id);", - [], - ) - .unwrap(); - conn.execute("CREATE INDEX idx_utxo_funding_tvl_highest ON utxo_funding(token_id, new_utxo_hash, sats, token_amount);", []).unwrap(); -} - -pub fn delete_entries_for_block(tx: &Connection, blockhash: &BlockHash) -> Result { - let mut stmt = tx.prepare( - "DELETE FROM utxo_funding - WHERE txid IN ( - SELECT txid FROM tx WHERE blockhash = ? - )", - )?; - let mut rows_deleted = stmt.execute(params![blockhash.to_hex()])?; - - let mut stmt = tx.prepare( - "DELETE FROM utxo_spending - WHERE txid IN ( - SELECT txid FROM utxo_funding WHERE txid IN ( - SELECT txid FROM tx WHERE blockhash = ? - ) - )", - )?; - rows_deleted += stmt.execute(params![blockhash.to_hex()])?; - - // remaining tables should clear themselves due to foregn keys usage - - Ok(rows_deleted != 0) +#[derive(Clone)] +pub struct DB { + // readwrite + pub cauldron_w: DBPool, + // readonly + pub cauldron_r: DBPool, + // readonly + pub bcmr_r: DBPool, + // readwrite + pub bcmr_w: DBPool, } diff --git a/src/index.rs b/src/index.rs new file mode 100644 index 0000000..25a09c6 --- /dev/null +++ b/src/index.rs @@ -0,0 +1,262 @@ +// 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 std::{ + collections::HashSet, + sync::{mpsc::sync_channel, Arc, Mutex}, +}; + +use bitcoin_hashes::hex::{FromHex, ToHex}; +use bitcoincash::{consensus::deserialize, Block, BlockHash, Transaction, Txid}; +use electrum_client::{Client, Param}; +use log::{debug, info, warn}; +use rayon::prelude::*; +use riftenlabs_defi::cauldron::{parse_cauldron, ParsedContract}; + +use crate::{ + bcmr::index_bcmr, + chain::{get_new_headers, Chain, StoreBlockUndoer}, + db::{ + self, + cauldron::{ + config::{config_get, config_set}, + header::{db_get_header, store_headers}, + tx::insert_block_tx, + utxo_funding::insert_utxo_funding, + utxo_spending::insert_utxo_spending, + }, + DBPool, DB, + }, + electrum::{electrum_fetch_mempool, electrum_get_tip, electrum_get_tx}, + CASHTOKEN_ACTIVATION_HEIGHT, KEY_LAST_INDEXED, +}; +use anyhow::{Context, Result}; +use electrum_client::ElectrumApi; + +fn parse_cauldrons(tx: &Transaction) -> Vec { + tx.input + .par_iter() + .enumerate() + .filter_map(move |(i, _)| parse_cauldron(i, tx)) + .collect() +} + +pub fn update_mempool(db: DBPool, electrum: Arc>) -> Result<()> { + let our_mempool_txs: HashSet = db::cauldron::mempool::load_mempool(&db.get().unwrap())?; + let node_mempool: HashSet = 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 = 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::cauldron::mempool::delete_mempool_tx(&db_tx, txid)?; + } + + for tx in txs_to_add { + let txid = tx.txid(); + debug!("mempool add {}", txid.to_hex()); + db::cauldron::tx::insert_mempool_tx(&db_tx, &txid)?; + let cauldrons: Vec = 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::cauldron::pool::update_pool_history(&db_tx, all_cauldrons) + .context("update pool history")?; + + Ok(db_tx.commit()?) +} + +pub fn index_blocks( + chain: Arc>, + db: DB, + client: Arc>, + bcmr_enabled: bool, +) -> Result { + let (tip_header, _) = electrum_get_tip(&client.lock().unwrap())?; + + let (block_send, block_recv) = sync_channel::>(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 = db.cauldron_w.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(db.clone())?; + chain.update(undoer, new_headers, None)?; + debug!("Header update done"); + } + } + + let db_cpy = db.clone(); + + std::thread::spawn(move || { + let db = db_cpy; + + let last_indexed = config_get(&db.cauldron_r.get().unwrap(), KEY_LAST_INDEXED).unwrap(); + let last_indexed = last_indexed.unwrap_or(CASHTOKEN_ACTIVATION_HEIGHT.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(&db.cauldron_r.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 { + if let Err(e) = block_send.send(None) { + warn!("Failed to end EOL to block reader: {}", e); + } + 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(); + + if let Err(e) = block_send.send(Some(( + next_height, + chain.lock().unwrap().get_mtp(next_height).unwrap(), + block, + ))) { + warn!("Failed to send block to reader: {}", e); + break; + } + + last_indexed = block_hash; + } + }); + + loop { + let (block_height, mtp, block) = match block_recv.recv()? { + Some(res) => res, + None => break, + }; + + let mut db_conn = db.cauldron_w.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::cauldron::pool::update_pool_history(&db_tx, all_cauldrons) + .context("update pool history")?; + config_set(&db_tx, KEY_LAST_INDEXED, &blockhash.to_hex()); + + let autheader_updates = if bcmr_enabled { + let mut conn = db.bcmr_w.get().context("failed to get bcmr db")?; + let bcmr_db_tx = conn.transaction()?; + let updates = index_bcmr(&bcmr_db_tx, &blockhash, block.txdata)?; + bcmr_db_tx.commit()?; + updates as i64 + } else { + -1 + }; + + // cauldron commit needs to come at last; as it tracks what the last successful block indexed was + db_tx.commit()?; + + info!( + "Indexed {}; mtp: {}, height {}, {} trades, {} autheader updates.", + blockhash.to_hex(), + mtp, + block_height, + total_cauldrons, + autheader_updates, + ); + } + Ok(tip_header.block_hash()) +} diff --git a/src/main.rs b/src/main.rs index 90af59a..abf6eaf 100644 --- a/src/main.rs +++ b/src/main.rs @@ -3,348 +3,51 @@ // 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 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::{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}, - State, -}; +use log::{error, info, warn}; +use rocket::{launch, routes}; use rocket_cors::{AllowedHeaders, AllowedOrigins}; -use serde_json::Value; +use rusqlite::OpenFlags; use std::{ backtrace::Backtrace, - collections::HashSet, panic, path::Path, process, - sync::{mpsc::sync_channel, Arc, Mutex}, + sync::{Arc, Mutex}, thread, time::Duration, }; use stderrlog::LogLevelNum; -use timeutil::time_now; -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, -}; +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 parse_cauldrons(tx: &Transaction) -> Vec { - tx.input - .par_iter() - .enumerate() - .filter_map(move |(i, _)| parse_cauldron(i, tx)) - .collect() -} - -fn update_mempool(db: DBPool, electrum: Arc>) -> Result<()> { - let our_mempool_txs: HashSet = db::mempool::load_mempool(&db.get().unwrap())?; - let node_mempool: HashSet = 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 = 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 = 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>, - pool: DBPool, - client: Arc>, -) -> Result { - let (tip_header, _) = electrum_get_tip(&client.lock().unwrap())?; - - let (block_send, block_recv) = sync_channel::>(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()) -} - -#[get("/tokens/list_by_volume?&")] -fn list_by_volume( - duration: Option, - limit: Option, - conn: &State, -) -> Result>, Custom> { - 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 = 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("/contract/volume?")] -fn contract_volume( - end: Option, - conn: &State, -) -> Result>, Custom> { - let end_timestamp = match end { - Some(e) => e, - None => time_now(), - }; - - let db = conn - .get() - .map_err(|e| Custom(Status::InternalServerError, format!("Error: {}", e)))?; - let volume = rpc::contract_volume(&db, end_timestamp as u64) - .map_err(|e| Custom(Status::BadRequest, format!("Error: {}", e)))?; - - let result: Vec = volume - .into_par_iter() - .map(|(token, (total, thirty_days, one_day))| { - json!({ - "token_id": token, - "total_sats": total, - "thirty_days_sats": thirty_days, - "one_day_sats": one_day - }) - }) - .collect(); - - Ok(Json(result)) -} - fn set_panic_hook() { panic::set_hook(Box::new(|panic_info| { error!("A thread panicked, terminating the program."); @@ -368,16 +71,46 @@ fn set_panic_hook() { })); } -fn start_program() -> Result { - let db_path = "cauldron.db"; - let db_exists = Path::new(db_path).exists(); +fn start_program() -> Result<(DB, BCMRDownloader)> { + let create_db_pool = |db_path| -> (bool, DBPool, DBPool) { + let db_exists = Path::new(db_path).exists(); - let manager = r2d2_sqlite::SqliteConnectionManager::file(db_path) - .with_init(|c| c.execute_batch("PRAGMA foreign_keys=1;")); + 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 pool = Arc::new(r2d2::Pool::new(manager).expect("Failed to initialize database")); + 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 { - prepare_tables(&pool.get().expect("failed to get sqlite connection")); + 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( @@ -393,53 +126,79 @@ fn start_program() -> Result { 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(); + 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 pool_cpy = Arc::clone(&pool); - let client_cpy = client.clone(); + 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 pool = pool_cpy; - let client = client_cpy; + let db = db_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); - } + 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 = electrum_get_tip(&client.lock().unwrap()) - .unwrap() - .0 - .block_hash(); + 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 = index_blocks(chain.clone(), pool.clone(), client.clone()) - .expect("Indexing failed"); + 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(pool.clone(), client.clone()) { + if let Err(e) = update_mempool(db.cauldron_w.clone(), client.clone()) { error!("Failed to update mempool: {}", e); } thread::sleep(Duration::from_secs(5)); } }); - Ok(pool) + let mut bcmrdownloader = BCMRDownloader::new(db.bcmr_w.clone()); + bcmrdownloader.start()?; + + Ok((db, bcmrdownloader)) } #[launch] fn launch() -> _ { stderrlog::new() - .verbosity(LogLevelNum::Debug) + .verbosity(LogLevelNum::Info) .init() .unwrap(); set_panic_hook(); - let dbpool = match start_program() { + let (dbpool, bcmrdownloader) = match start_program() { Ok(db) => db, Err(e) => { let backtrace = Backtrace::capture(); @@ -468,21 +227,24 @@ fn launch() -> _ { rocket::build() .manage(dbpool) + // 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, - list_by_volume, + 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, - contract_volume + rpc::contract::contract_volume ], ) + .mount("/bcmr", routes![rpc::bcmr::token_bcmr,]) .attach(cors) } diff --git a/src/rpc/bcmr.rs b/src/rpc/bcmr.rs new file mode 100644 index 0000000..26cb25d --- /dev/null +++ b/src/rpc/bcmr.rs @@ -0,0 +1,32 @@ +// 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; +use bitcoin_hashes::hex::{FromHex, ToHex}; +use bitcoincash::TokenID; +use rocket::{get, http::Status, response::status::Custom, serde::json::Json, State}; +use serde_json::json; +use serde_json::Value; + +use crate::db::{bcmr::get_token_bcmr, DB}; + +#[get("/token/")] +pub fn token_bcmr(category: Option<&str>, db: &State) -> Result, Custom> { + let token_id_hex = category + .context("category missing") + .map_err(|e| Custom(Status::BadRequest, format!("Error: {}", e)))?; + + // validate input by parsing it into TokenID + let token_id = TokenID::from_hex(token_id_hex) + .map_err(|e| Custom(Status::BadRequest, format!("Error: {}", e)))?; + let conn = db + .bcmr_r + .get() + .map_err(|e| Custom(Status::InternalServerError, format!("Error: {}", e)))?; + let bcmr = get_token_bcmr(&conn, &token_id.to_hex()) + .map_err(|e| Custom(Status::InternalServerError, format!("Error: {}", e)))?; + + Ok(Json(json!(bcmr))) +} diff --git a/src/rpc/contract.rs b/src/rpc/contract.rs index 783cf4a..49999a8 100644 --- a/src/rpc/contract.rs +++ b/src/rpc/contract.rs @@ -4,12 +4,13 @@ // 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 rayon::prelude::*; use rocket::{get, http::Status, response::status::Custom, serde::json::Json, State}; use rusqlite::Connection; use serde::Serialize; use serde_json::{json, Value}; -use crate::db::DBPool; +use crate::{db::DB, timeutil::time_now}; #[derive(Serialize)] pub struct ContractCount { @@ -50,8 +51,9 @@ fn db_contract_count_by_token(db: &Connection, token_id: &str) -> Result) -> Result, Custom> { +pub fn contract_count_all(conn: &State) -> Result, Custom> { let db = conn + .cauldron_r .get() .map_err(|e| Custom(Status::InternalServerError, format!("Error: {}", e)))?; let count = db_contract_count_all(&db) @@ -60,14 +62,44 @@ pub fn contract_count_all(conn: &State) -> Result, Custom")] -pub fn contract_count_token( - token: &str, - conn: &State, -) -> Result, Custom> { +pub fn contract_count_token(token: &str, conn: &State) -> Result, Custom> { let db = conn + .cauldron_r .get() .map_err(|e| Custom(Status::InternalServerError, format!("Error: {}", e)))?; let count = db_contract_count_by_token(&db, token) .map_err(|e| Custom(Status::BadRequest, format!("Error: {}", e)))?; Ok(Json(json!(count))) } + +#[get("/contract/volume?")] +pub fn contract_volume( + end: Option, + conn: &State, +) -> Result>, Custom> { + let end_timestamp = match end { + Some(e) => e, + None => time_now(), + }; + + let db = conn + .cauldron_r + .get() + .map_err(|e| Custom(Status::InternalServerError, format!("Error: {}", e)))?; + let volume = super::contract_volume(&db, end_timestamp as u64) + .map_err(|e| Custom(Status::BadRequest, format!("Error: {}", e)))?; + + let result: Vec = volume + .into_par_iter() + .map(|(token, (total, thirty_days, one_day))| { + json!({ + "token_id": token, + "total_sats": total, + "thirty_days_sats": thirty_days, + "one_day_sats": one_day + }) + }) + .collect(); + + Ok(Json(result)) +} diff --git a/src/rpc/mod.rs b/src/rpc/mod.rs index bf57126..388c4ba 100644 --- a/src/rpc/mod.rs +++ b/src/rpc/mod.rs @@ -8,19 +8,24 @@ use std::collections::HashMap; use anyhow::{Context, Result}; use rusqlite::{params, Connection}; +use crate::{bcmr::parsedbcmr::ParsedBCMR, db::bcmr::get_token_bcmr}; + +pub mod bcmr; pub mod contract; pub mod pool; pub mod price; +pub mod tokens; pub mod tvl; #[allow(clippy::type_complexity)] pub fn list_tokens_by_volume( - connection: &Connection, + cauldron_conn: &Connection, + bcmr_conn: &Connection, seconds: usize, limit: usize, -) -> Result> { +) -> Result)>> { // List token ID's by volume last n seconds - let mut statement = connection + let mut statement = cauldron_conn .prepare( " WITH TradeData AS ( @@ -78,7 +83,7 @@ pub fn list_tokens_by_volume( ", )?; - let tokens: Vec<(String, u64, u64, u64, u64, u64, u64)> = statement + let tokens: Vec<(String, u64, u64, u64, u64, u64, u64, Option)> = statement .query_and_then([seconds, limit], |row| { let token_id: String = row.get(0)?; let trade_volume: u64 = row.get(1)?; @@ -87,6 +92,7 @@ pub fn list_tokens_by_volume( let tvl_token: u64 = row.get(4)?; let best_contract_sats: u64 = row.get(5)?; let best_contracts_token: u64 = row.get(6)?; + let bcmr = get_token_bcmr(bcmr_conn, &token_id)?; Ok(( token_id, trade_volume, @@ -95,10 +101,11 @@ pub fn list_tokens_by_volume( tvl_token, best_contract_sats, best_contracts_token, + bcmr, )) }) .unwrap() - .map(|row: Result<(String, u64, u64, u64, u64, u64, u64)>| row.unwrap()) + .map(|row: Result<(String, u64, u64, u64, u64, u64, u64, Option)>| row.unwrap()) .collect(); Ok(tokens) diff --git a/src/rpc/pool.rs b/src/rpc/pool.rs index 20968bd..03d142d 100644 --- a/src/rpc/pool.rs +++ b/src/rpc/pool.rs @@ -5,7 +5,7 @@ use crate::{ cashaddr::{self, version_byte_flags}, - db::DBPool, + db::DB, timeutil::time_now, }; use anyhow::{Context, Result}; @@ -139,8 +139,9 @@ fn pools_by_apy(connection: &Connection) -> Result> { } #[get("/pool/list_by_apy")] -pub fn list_pools_by_apy(conn: &State) -> Result, Custom> { +pub fn list_pools_by_apy(conn: &State) -> Result, Custom> { let db = conn + .cauldron_r .get() .map_err(|e| Custom(Status::InternalServerError, format!("Error: {}", e)))?; @@ -237,9 +238,10 @@ fn db_list_active_pools( pub fn list_active_pools( token: Option<&str>, pkh: Option<&str>, - conn: &State, + conn: &State, ) -> Result, Custom> { let db = conn + .cauldron_r .get() .map_err(|e| Custom(Status::InternalServerError, format!("Error: {}", e)))?; diff --git a/src/rpc/price.rs b/src/rpc/price.rs index c6fdce8..ded2e3f 100644 --- a/src/rpc/price.rs +++ b/src/rpc/price.rs @@ -9,7 +9,7 @@ use rusqlite::{params, Connection}; use rust_decimal::prelude::*; use serde_json::{json, Value}; -use crate::{db::DBPool, timeutil::time_now}; +use crate::{db::DB, timeutil::time_now}; struct PriceInterval { start: i64, @@ -181,8 +181,9 @@ fn historic_price( } #[get("/price//current")] -pub fn price_current(token: &str, conn: &State) -> Result, Custom> { +pub fn price_current(token: &str, conn: &State) -> Result, Custom> { let db = conn + .cauldron_r .get() .map_err(|e| Custom(Status::InternalServerError, format!("Error: {}", e)))?; @@ -200,11 +201,12 @@ pub fn price_history( start: Option, end: Option, stepsize: Option, - conn: &State, + conn: &State, ) -> Result, Custom> { let current_timestamp = time_now(); let db = conn + .cauldron_r .get() .map_err(|e| Custom(Status::InternalServerError, format!("Error: {}", e)))?; diff --git a/src/rpc/tokens.rs b/src/rpc/tokens.rs new file mode 100644 index 0000000..7fde915 --- /dev/null +++ b/src/rpc/tokens.rs @@ -0,0 +1,65 @@ +// 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 rocket::http::Status; +use rocket::{get, response::status::Custom, serde::json::Json, State}; +use serde_json::json; +use serde_json::Value; + +use crate::bcmr::parsedbcmr::ParsedBCMR; +use crate::db::DB; +use crate::rpc; +use rayon::prelude::*; + +#[get("/tokens/list_by_volume?&")] +pub fn list_by_volume( + duration: Option, + limit: Option, + db: &State, +) -> Result>, Custom> { + 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); + + #[allow(clippy::type_complexity)] + let list: Vec<(String, u64, u64, u64, u64, u64, u64, Option)> = + rpc::list_tokens_by_volume( + &db.cauldron_r.get().unwrap(), + &db.bcmr_r.get().unwrap(), + duration, + limit, + ) + .map_err(|e| Custom(Status::InternalServerError, format!("Error: {}", e)))?; + + let result: Vec = list + .into_par_iter() + .map( + |( + token_id, + trade_volume, + trade_count, + tvl_sats, + tvl_token, + best_contract_sats, + best_contract_tokens, + bcmr, + )| { + 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, + "bcmr": bcmr + }) + }, + ) + .collect(); + + Ok(Json(result)) +} diff --git a/src/rpc/tvl.rs b/src/rpc/tvl.rs index a3bc222..d827bca 100644 --- a/src/rpc/tvl.rs +++ b/src/rpc/tvl.rs @@ -9,7 +9,7 @@ use rocket::{get, http::Status, response::status::Custom, serde::json::Json, Sta use rusqlite::{params, Connection}; use serde_json::{json, Value}; -use crate::{db::DBPool, timeutil::time_now}; +use crate::{db::DB, timeutil::time_now}; /// Fetch TVL for all tokens pub fn get_all_token_tvl( @@ -92,11 +92,9 @@ pub fn get_token_tvl( // Deprecated; use valuelocked with optional parameters #[get("/tvl/