BitcoinCash/riftenlabs-indexer
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add BCMR indexer

Browse source 
This adds BCMR indexer, including downloading and parsing the actual
BCMR files.
This commit is contained in:
Dagur Valberg Johannsson 2024-05-09 11:25:27 +02:00
parent 49730f27ca
commit e38eaaa26c
No known key found for this signature in database
GPG key ID: FD701804AEE88107
26 changed files with 1994 additions and 451 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

108
Cargo.lock generated
View file

@ -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"

2
Cargo.toml
View file

@ -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

16
contrib/volume.py Normal file
View file

@ -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}")

449
src/bcmr/bcmrdownloader.rs Normal file
View file

@ -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<AtomicBool>,
download_thread: Option<JoinHandle<()>>,
}
const LOOP_SLEEP_TIME: Duration = Duration::from_secs(10);
fn get_download_candidates(db: &DBPool) -> Vec<AuthChainEntry> {
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<String> = 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<ParsedBCMR> {
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::<i64>()
.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);
}
}

345
src/bcmr/mod.rs Normal file
View file

@ -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<S>(bytes: &[u8; 32], serializer: S) -> Result<S::Ok, S::Error>
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<String>,
#[serde(skip)]
pub op_return: Vec<u8>,
}
pub fn parse_bcmr_from_opreturn(bcmr_op_return: &Script) -> Option<BCMR> {
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<String> = 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<BCMR> {
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<TokenID> {
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<Transaction>) -> Vec<Transaction> {
let txs = {
let mut queue: VecDeque<Transaction> = txs.into_iter().collect();
let mut queue_txids: HashSet<Txid> = queue.par_iter().map(|tx| tx.txid()).collect();
let mut txs: Vec<Transaction> = 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<Transaction>) -> Result<usize> {
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<Transaction>,
) -> Result<usize> {
let sorted = ttor_sorted(txs);
let mut new_tokens: Vec<(Transaction, TokenID)> = Vec::default();
let sorted: Vec<Transaction> = 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<OutPointHash, usize> = HashMap::default();
// Look for authchain updates. We collect all candidates first.
let mut candidates: HashMap<OutPointHash, Transaction> = 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<AuthChainEntry> = 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(&not_genesis_tx).is_some());
assert!(is_genesis_tx(&genesis_tx).is_some());
}
}

104
src/bcmr/parsedbcmr.rs Normal file
View file

@ -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<String>,
pub web: Option<String>,
}
#[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,
},
}
}
}

19
src/chain.rs
View file

@ -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<Self> {
pub fn new(db: DB) -> Result<Self> {
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(),
);

337
src/db/bcmr/mod.rs Normal file
View file

@ -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<Vec<u8>>,
}
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<bool> {
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<Vec<u8>>,
) -> 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<Option<AuthChainEntry>> {
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<String> = 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<Item = &'a OutPointHash>,
) -> Result<Vec<AuthChainEntry>> {
// To not exceed SQL query length; we chunk the queries into entries of this size.
let chunk_size = 10000;
let keys: Vec<String> = keys.map(|utxo| utxo.to_hex()).collect();
let mut matches: Vec<AuthChainEntry> = 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<String> = 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<Vec<AuthChainEntry>> {
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<AuthChainEntry> = 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<String> = 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<Option<ParsedBCMR>> {
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)
}
}

6
src/db/config.rs → src/db/cauldron/config.rs
View file

@ -19,5 +19,9 @@ pub fn config_get(conn: &Connection, key: &str) -> Result<Option<String>> {
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)
}
}

0
src/db/header.rs → src/db/cauldron/header.rs
View file

0
src/db/mempool.rs → src/db/cauldron/mempool.rs
View file

76
src/db/cauldron/mod.rs Normal file
View file

@ -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<bool> {
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)
}

0
src/db/pool.rs → src/db/cauldron/pool.rs
View file

0
src/db/tx.rs → src/db/cauldron/tx.rs
View file

0
src/db/utxo_funding.rs → src/db/cauldron/utxo_funding.rs
View file

0
src/db/utxo_spending.rs → src/db/cauldron/utxo_spending.rs
View file

81
src/db/mod.rs
View file

@ -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<r2d2::Pool<r2d2_sqlite::SqliteConnectionManager>>;
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<bool> {
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,
}

262
src/index.rs Normal file
View file

@ -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<ParsedContract> {
tx.input
.par_iter()
.enumerate()
.filter_map(move |(i, _)| parse_cauldron(i, tx))
.collect()
}
pub fn update_mempool(db: DBPool, electrum: Arc<Mutex<Client>>) -> Result<()> {
let our_mempool_txs: HashSet<Txid> = db::cauldron::mempool::load_mempool(&db.get().unwrap())?;
let node_mempool: HashSet<Txid> = electrum_fetch_mempool(&electrum.lock().unwrap())?;
let txs_to_delete = our_mempool_txs.difference(&node_mempool);
let txs_to_add = node_mempool.difference(&our_mempool_txs);
let txs_to_add: Vec<Transaction> = txs_to_add
.into_iter()
.filter_map(
|txid| match electrum_get_tx(&electrum.lock().unwrap(), txid) {
Ok(tx) => Some(tx),
Err(e) => {
info!("Failed to get mempool tx {}: {}", txid, e);
None
}
},
)
.collect();
let mut db_conn = db.get().unwrap();
let db_tx = db_conn.transaction()?;
let mut all_cauldrons = vec![];
for txid in txs_to_delete {
debug!("mempool remove {}", txid.to_hex());
db::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<ParsedContract> = tx
.input
.iter()
.enumerate()
.filter_map(|(i, _)| parse_cauldron(i, &tx))
.collect();
insert_utxo_funding(&db_tx, &cauldrons, &txid, false)?;
insert_utxo_spending(&db_tx, &cauldrons, &txid, false)?;
all_cauldrons.extend(cauldrons);
}
db::cauldron::pool::update_pool_history(&db_tx, all_cauldrons)
.context("update pool history")?;
Ok(db_tx.commit()?)
}
pub fn index_blocks(
chain: Arc<Mutex<Chain>>,
db: DB,
client: Arc<Mutex<Client>>,
bcmr_enabled: bool,
) -> Result<BlockHash> {
let (tip_header, _) = electrum_get_tip(&client.lock().unwrap())?;
let (block_send, block_recv) = sync_channel::<Option<(u64, u64, Block)>>(10);
// Update header chain (and undo any blocks that may have reorged away)
{
let chain = chain.lock().unwrap();
if tip_header.block_hash() != chain.tip_hash() {
debug!(
"Updating header chain from {} to {}",
chain.tip_hash().to_hex(),
tip_header.block_hash().to_hex()
);
let new_headers =
get_new_headers(&client.lock().unwrap(), &chain, &tip_header.block_hash())?;
debug!("Storing headers");
{
let mut db_conn = 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())
}

450
src/main.rs
View file

@ -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<ParsedContract> {
tx.input
.par_iter()
.enumerate()
.filter_map(move |(i, _)| parse_cauldron(i, tx))
.collect()
}
fn update_mempool(db: DBPool, electrum: Arc<Mutex<Client>>) -> Result<()> {
let our_mempool_txs: HashSet<Txid> = db::mempool::load_mempool(&db.get().unwrap())?;
let node_mempool: HashSet<Txid> = electrum_fetch_mempool(&electrum.lock().unwrap())?;
let txs_to_delete = our_mempool_txs.difference(&node_mempool);
let txs_to_add = node_mempool.difference(&our_mempool_txs);
let txs_to_add: Vec<Transaction> = txs_to_add
.into_iter()
.filter_map(
|txid| match electrum_get_tx(&electrum.lock().unwrap(), txid) {
Ok(tx) => Some(tx),
Err(e) => {
info!("Failed to get mempool tx {}: {}", txid, e);
None
}
},
)
.collect();
let mut db_conn = db.get().unwrap();
let db_tx = db_conn.transaction()?;
let mut all_cauldrons = vec![];
for txid in txs_to_delete {
debug!("mempool remove {}", txid.to_hex());
db::mempool::delete_mempool_tx(&db_tx, txid)?;
}
for tx in txs_to_add {
let txid = tx.txid();
debug!("mempool add {}", txid.to_hex());
tx::insert_mempool_tx(&db_tx, &txid)?;
let cauldrons: Vec<ParsedContract> = tx
.input
.iter()
.enumerate()
.filter_map(|(i, _)| parse_cauldron(i, &tx))
.collect();
insert_utxo_funding(&db_tx, &cauldrons, &txid, false)?;
insert_utxo_spending(&db_tx, &cauldrons, &txid, false)?;
all_cauldrons.extend(cauldrons);
}
db::pool::update_pool_history(&db_tx, all_cauldrons).context("update pool history")?;
Ok(db_tx.commit()?)
}
fn index_blocks(
chain: Arc<Mutex<Chain>>,
pool: DBPool,
client: Arc<Mutex<Client>>,
) -> Result<BlockHash> {
let (tip_header, _) = electrum_get_tip(&client.lock().unwrap())?;
let (block_send, block_recv) = sync_channel::<Option<(u64, u64, Block)>>(10);
// Update header chain (and undo any blocks that may have reorged away)
{
let chain = chain.lock().unwrap();
if tip_header.block_hash() != chain.tip_hash() {
debug!(
"Updating header chain from {} to {}",
chain.tip_hash().to_hex(),
tip_header.block_hash().to_hex()
);
let new_headers =
get_new_headers(&client.lock().unwrap(), &chain, &tip_header.block_hash())?;
debug!("Storing headers");
{
let mut db_conn = pool.get().unwrap();
for chunk in new_headers.chunks(100000) {
let db_tx = db_conn.transaction()?;
store_headers(&db_tx, chunk)?;
db_tx.commit()?;
}
}
let undoer = StoreBlockUndoer::new(pool.clone())?;
chain.update(undoer, new_headers, None)?;
debug!("Header update done");
}
}
let pool_cpy = pool.clone();
thread::spawn(move || {
let pool = pool_cpy;
let last_indexed = config_get(&pool.get().unwrap(), KEY_LAST_INDEXED).unwrap();
let last_indexed = last_indexed.unwrap_or(RIFTEN_LABS_GENESIS_BLOCK.to_string());
let mut last_indexed = BlockHash::from_hex(&last_indexed).unwrap();
// Check if last indexed has been orphaned
loop {
if chain.lock().unwrap().contains(&last_indexed) {
break;
}
info!(
"Last indexed block ({}) has been orphaned",
last_indexed.to_hex()
);
let header = db_get_header(&pool.get().unwrap(), &last_indexed)
.context("Failed to get header for last indexed")
.unwrap();
last_indexed = header.prev_blockhash;
info!(
"Last indexed block rolled back to {}",
last_indexed.to_hex()
);
}
loop {
if tip_header.block_hash() == last_indexed {
block_send.send(None).unwrap();
break;
}
let next_height = chain
.lock()
.unwrap()
.get_block_height(&last_indexed)
.expect("last_indexed height not found in main chain")
+ 1;
let res = client
.lock()
.unwrap()
.raw_call("blockchain.block.get", vec![Param::U32(next_height as u32)])
.unwrap();
let block_hex: String = serde_json::from_str(&res.to_string()).unwrap();
let block: Block = deserialize(&hex::decode(&block_hex).unwrap()).unwrap();
let block_hash = block.block_hash();
block_send
.send(Some((
next_height,
chain.lock().unwrap().get_mtp(next_height).unwrap(),
block,
)))
.unwrap();
last_indexed = block_hash;
}
});
loop {
let (block_height, mtp, block) = match block_recv.recv()? {
Some(res) => res,
None => break,
};
let mut db_conn = pool.get().unwrap();
let db_tx = db_conn.transaction()?;
let blockhash = block.block_hash();
let mut total_cauldrons = 0;
let mut all_cauldrons = vec![];
for tx in block.txdata {
let cauldrons = parse_cauldrons(&tx);
if cauldrons.is_empty() {
continue;
}
let txid = tx.txid();
insert_block_tx(&db_tx, &txid, &blockhash, mtp as i64).context("inserting block tx")?;
insert_utxo_funding(&db_tx, &cauldrons, &txid, true)
.context("inserting funding utxos")?;
insert_utxo_spending(&db_tx, &cauldrons, &txid, true)
.context("inserting spending utxos")?;
total_cauldrons += cauldrons.len();
all_cauldrons.extend(cauldrons);
}
// Figuring out initial utxo needs to be done on all cauldrons in a block.
db::pool::update_pool_history(&db_tx, all_cauldrons).context("update pool history")?;
config_set(&db_tx, KEY_LAST_INDEXED, &blockhash.to_hex());
db_tx.commit()?;
println!(
"Indexed {}; mtp: {}, height {}, {} trades.",
block.header.block_hash().to_hex(),
mtp,
block_height,
total_cauldrons
);
}
Ok(tip_header.block_hash())
}
#[get("/tokens/list_by_volume?<duration>&<limit>")]
fn list_by_volume(
duration: Option<usize>,
limit: Option<usize>,
conn: &State<DBPool>,
) -> Result<Json<Vec<Value>>, Custom<String>> {
let thirty_days = 24 * 60 * 60 * 30;
let duration = duration.unwrap_or(thirty_days);
let limit = limit.unwrap_or(50);
let limit = std::cmp::max(limit, 1000);
let list: Vec<(String, u64, u64, u64, u64, u64, u64)> =
rpc::list_tokens_by_volume(&conn.get().unwrap(), duration, limit)
.map_err(|e| Custom(Status::InternalServerError, format!("Error: {}", e)))?;
let result: Vec<Value> = list
.into_par_iter()
.map(
|(
token_id,
trade_volume,
trade_count,
tvl_sats,
tvl_token,
best_contract_sats,
best_contract_tokens,
)| {
json!({
"token_id": token_id,
"trade_volume": trade_volume,
"trade_count": trade_count,
"tvl_sats": tvl_sats,
"tvl_tokens": tvl_token,
"best_contract_sats": best_contract_sats,
"best_contract_tokens": best_contract_tokens
})
},
)
.collect();
Ok(Json(result))
}
#[get("/contract/volume?<end>")]
fn contract_volume(
end: Option<i64>,
conn: &State<DBPool>,
) -> Result<Json<Vec<Value>>, Custom<String>> {
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<Value> = 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<DBPool> {
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<DBPool> {
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)
}

32
src/rpc/bcmr.rs Normal file
View file

@ -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/<category>")]
pub fn token_bcmr(category: Option<&str>, db: &State<DB>) -> Result<Json<Value>, Custom<String>> {
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)))
}

44
src/rpc/contract.rs
View file

@ -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<Contrac
}
#[get("/contract/count")]
pub fn contract_count_all(conn: &State<DBPool>) -> Result<Json<Value>, Custom<String>> {
pub fn contract_count_all(conn: &State<DB>) -> Result<Json<Value>, Custom<String>> {
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<DBPool>) -> Result<Json<Value>, Custom<St
}
#[get("/contract/count/<token>")]
pub fn contract_count_token(
token: &str,
conn: &State<DBPool>,
) -> Result<Json<Value>, Custom<String>> {
pub fn contract_count_token(token: &str, conn: &State<DB>) -> Result<Json<Value>, Custom<String>> {
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?<end>")]
pub fn contract_volume(
end: Option<i64>,
conn: &State<DB>,
) -> Result<Json<Vec<Value>>, Custom<String>> {
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<Value> = 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))
}

17
src/rpc/mod.rs
View file

@ -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<Vec<(String, u64, u64, u64, u64, u64, u64)>> {
) -> Result<Vec<(String, u64, u64, u64, u64, u64, u64, Option<ParsedBCMR>)>> {
// 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<ParsedBCMR>)> = 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<ParsedBCMR>)>| row.unwrap())
.collect();
Ok(tokens)

8
src/rpc/pool.rs
View file

@ -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<Vec<PoolYield>> {
}
#[get("/pool/list_by_apy")]
pub fn list_pools_by_apy(conn: &State<DBPool>) -> Result<Json<Value>, Custom<String>> {
pub fn list_pools_by_apy(conn: &State<DB>) -> Result<Json<Value>, Custom<String>> {
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<DBPool>,
conn: &State<DB>,
) -> Result<Json<Value>, Custom<String>> {
let db = conn
.cauldron_r
.get()
.map_err(|e| Custom(Status::InternalServerError, format!("Error: {}", e)))?;

8
src/rpc/price.rs
View file

@ -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/<token>/current")]
pub fn price_current(token: &str, conn: &State<DBPool>) -> Result<Json<Value>, Custom<String>> {
pub fn price_current(token: &str, conn: &State<DB>) -> Result<Json<Value>, Custom<String>> {
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<i64>,
end: Option<i64>,
stepsize: Option<i64>,
conn: &State<DBPool>,
conn: &State<DB>,
) -> Result<Json<Value>, Custom<String>> {
let current_timestamp = time_now();
let db = conn
.cauldron_r
.get()
.map_err(|e| Custom(Status::InternalServerError, format!("Error: {}", e)))?;

65
src/rpc/tokens.rs Normal file
View file

@ -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?<duration>&<limit>")]
pub fn list_by_volume(
duration: Option<usize>,
limit: Option<usize>,
db: &State<DB>,
) -> 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);
#[allow(clippy::type_complexity)]
let list: Vec<(String, u64, u64, u64, u64, u64, u64, Option<ParsedBCMR>)> =
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<Value> = 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))
}

16
src/rpc/tvl.rs
View file

@ -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/<time>")]
pub fn deprecated_tvl(
time: usize,
conn: &State<DBPool>,
) -> Result<Json<Vec<Value>>, Custom<String>> {
pub fn deprecated_tvl(time: usize, conn: &State<DB>) -> Result<Json<Vec<Value>>, Custom<String>> {
let db = conn
.cauldron_r
.get()
.map_err(|e| Custom(Status::InternalServerError, format!("Error: {}", e)))?;
@ -120,9 +118,9 @@ pub fn deprecated_tvl(
#[get("/valuelocked?<time>")]
pub fn valuelocked_all(
time: Option<usize>,
conn: &State<DBPool>,
conn: &State<DB>,
) -> Result<Json<Vec<Value>>, Custom<String>> {
let db = conn.get().map_err(|_| {
let db = conn.cauldron_r.get().map_err(|_| {
Custom(
Status::InternalServerError,
"Failed to get DB connection".into(),
@ -152,9 +150,9 @@ pub fn valuelocked_all(
pub fn valuelocked_token(
token: &str,
time: Option<usize>,
conn: &State<DBPool>,
conn: &State<DB>,
) -> Result<Json<Value>, Custom<String>> {
let db = conn.get().map_err(|_| {
let db = conn.cauldron_r.get().map_err(|_| {
Custom(
Status::InternalServerError,
"Failed to get DB connection".into(),