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

Convert hash columns from TEXT to BLOB storage

Browse source 
Migrate all 32-byte hash fields from TEXT (64-char hex) to BLOB
(32 bytes) for ~50% storage savings on hash columns.

Changes:
- Add src/db/blob.rs with ToBlob/FromBlob traits for hash types
- Update all table schemas to use BLOB for hash columns
- Replace .to_hex() with .to_blob() for DB inserts
- Replace ::from_hex() with ::from_blob() for DB reads
- Use SQL hex() function where API responses need hex strings
- Bump DB_VERSION from 4 to 5

Affected tables: tx, pool, pool_history_entry, utxo_spending,
utxo_funding, user_action, auth_chain_entry, bcmr_data,
bcmr_failure, bcmr_well_known, crc20, crc20_candidates,
delphi_entry, cached_token_metrics

Fixes #1
This commit is contained in:
Dagur Valberg Johannsson 2026-01-21 13:25:35 +01:00
parent 66097c1abb
commit 58fd9595d7
No known key found for this signature in database
GPG key ID: FD701804AEE88107
29 changed files with 641 additions and 310 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
.vscode/settings.json vendored Normal file
View file

@ -0,0 +1,3 @@
{
"rust-analyzer.showUnlinkedFileNotification": false
}

37
src/bcmr/mod.rs
View file

@ -7,7 +7,9 @@ use rayon::prelude::*;
use std::convert::TryInto;
use anyhow::Result;
use bitcoin_hashes::{hex::FromHex, hex::ToHex, Hash};
#[cfg(test)]
use bitcoin_hashes::hex::FromHex;
use bitcoin_hashes::{hex::ToHex, Hash};
use bitcoincash::{BlockHash, Script, TokenID, Transaction, Txid};
use log::debug;
use riftenlabs_defi::chainutil::{compute_outpoint_hash, read_push_from_script, OutPointHash};
@ -52,6 +54,8 @@ pub struct BCMR {
}
fn find_parent_auth_entries(conn: &Connection, tx: &Transaction) -> Result<Vec<AuthChainEntry>> {
use crate::db::blob::{FromBlob, ToBlob};
let mut parents = Vec::new();
for vin in &tx.input {
@ -66,23 +70,18 @@ fn find_parent_auth_entries(conn: &Connection, tx: &Transaction) -> Result<Vec<A
FROM auth_chain_entry
WHERE utxo = ?1",
)?;
let mut rows = stmt.query(params![prev_utxo.to_hex()])?;
let mut rows = stmt.query(params![prev_utxo.to_blob()])?;
while let Some(row) = rows.next()? {
let token_hex: String = row.get(0)?;
let txid_hex: String = row.get(1)?;
let token_blob: Vec<u8> = row.get(0)?;
let txid_blob: Vec<u8> = row.get(1)?;
let height: usize = row.get(2)?;
let bcmr_data_hex: Option<String> = row.get(3)?;
let utxo_hex: String = row.get(4)?;
let bcmr_data: Option<Vec<u8>> = row.get(3)?;
let utxo_blob: Vec<u8> = row.get(4)?;
let token_id = TokenID::from_hex(&token_hex)?;
let txid = Txid::from_hex(&txid_hex)?;
let utxo = OutPointHash::from_hex(&utxo_hex)?;
let bcmr_data = match bcmr_data_hex {
Some(h) => Some(hex::decode(&h)?),
None => None,
};
let token_id = TokenID::from_blob(&token_blob)?;
let txid = Txid::from_blob(&txid_blob)?;
let utxo = OutPointHash::from_blob(&utxo_blob)?;
parents.push(AuthChainEntry {
utxo,
@ -312,11 +311,12 @@ fn multi_parent_auth_update_creates_entries_for_all_tokens() {
// Expect 2 auth_chain_entry rows for this tx (one per token)
{
use crate::db::blob::ToBlob;
let mut stmt = db_tx
.prepare("SELECT COUNT(*) FROM auth_chain_entry WHERE txid = ?")
.unwrap();
let count: i64 = stmt
.query_row([update_txid.to_hex()], |row| row.get(0))
.query_row([update_txid.to_blob()], |row| row.get(0))
.unwrap();
assert_eq!(
@ -327,19 +327,20 @@ fn multi_parent_auth_update_creates_entries_for_all_tokens() {
// Expect each token to have height 1 in this tx
{
use crate::db::blob::ToBlob;
let mut stmt = db_tx
.prepare(
"SELECT token_id, height
"SELECT hex(token_id) as token_id, height
FROM auth_chain_entry
WHERE txid = ?
ORDER BY token_id",
)
.unwrap();
let mut rows = stmt.query([update_txid.to_hex()]).unwrap();
let mut rows = stmt.query([update_txid.to_blob()]).unwrap();
let mut seen: Vec<(String, i64)> = Vec::new();
while let Some(row) = rows.next().unwrap() {
let t: String = row.get(0).unwrap();
let t: String = row.get::<_, String>(0).unwrap().to_lowercase();
let h: i64 = row.get(1).unwrap();
seen.push((t, h));
}

75
src/db/bcmr/mod.rs
View file

@ -4,8 +4,9 @@
// 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, SOURCE_ON_CHAIN};
use crate::db::blob::{FromBlob, ToBlob};
use anyhow::*;
use bitcoin_hashes::hex::{FromHex, ToHex};
use bitcoin_hashes::hex::ToHex;
use bitcoincash::{BlockHash, TokenID, Txid};
use log::info;
use riftenlabs_defi::chainutil::OutPointHash;
@ -27,10 +28,10 @@ pub struct AuthChainEntry {
pub fn prepare_tables(conn: &Connection) {
conn.execute(
"CREATE TABLE auth_chain_entry (
token_id TEXT NOT NULL,
utxo TEXT NOT NULL,
blockhash TEXT NOT NULL,
txid TEXT NOT NULL,
token_id BLOB NOT NULL,
utxo BLOB NOT NULL,
blockhash BLOB NOT NULL,
txid BLOB NOT NULL,
height INT NOT NULL,
bcmr_data TEXT,
PRIMARY KEY (token_id, utxo)
@ -41,8 +42,8 @@ pub fn prepare_tables(conn: &Connection) {
conn.execute(
"CREATE TABLE bcmr_data (
token_id TEXT NOT NULL,
utxo TEXT NOT NULL,
token_id BLOB NOT NULL,
utxo BLOB NOT NULL,
symbol TEXT NOT NULL,
decimals INT NOT NULL,
name TEXT NOT NULL,
@ -63,9 +64,9 @@ pub fn prepare_tables(conn: &Connection) {
conn.execute(
"CREATE TABLE bcmr_failure (
token_id TEXT NOT NULL,
utxo TEXT NOT NULL,
txid TEXT NOT NULL,
token_id BLOB NOT NULL,
utxo BLOB NOT NULL,
txid BLOB NOT NULL,
last_attempt INT NOT NULL,
attempts INT NOT NULL,
error_message TEXT,
@ -87,7 +88,7 @@ pub fn prepare_tables(conn: &Connection) {
CREATE TABLE bcmr_well_known (
source TEXT NOT NULL,
symbol TEXT NOT NULL,
token_id TEXT NOT NULL,
token_id BLOB NOT NULL,
decimals INT NOT NULL,
name TEXT NOT NULL,
description TEXT NOT NULL,
@ -148,7 +149,7 @@ pub fn delete_entries_for_well_known(tx: &Connection, source: &str) -> Result<()
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()])?;
let rows_deleted = stmt.execute(params![blockhash.to_blob()])?;
Ok(rows_deleted != 0)
}
@ -168,10 +169,10 @@ pub fn insert_authheader(
VALUES (?, ?, ?, ?, ?, ?)",
)?;
stmt.execute(params![
utxo.to_hex(),
blockhash.to_hex(),
txid.to_hex(),
token_id.to_hex(),
utxo.to_blob(),
blockhash.to_blob(),
txid.to_blob(),
token_id.to_blob(),
height,
bcmr_data.map(|bcmr| bcmr.to_hex())
])?;
@ -185,12 +186,12 @@ pub fn get_authheader(conn: &Connection, utxo: &OutPointHash) -> Result<Option<A
"SELECT token_id, txid, height, bcmr_data
FROM auth_chain_entry WHERE utxo = ?",
)?;
let mut row = stmt.query([utxo.to_hex()])?;
let mut row = stmt.query([utxo.to_blob()])?;
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 token_blob: Vec<u8> = header.get(0)?;
let txid_blob: Vec<u8> = 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 {
@ -201,8 +202,8 @@ pub fn get_authheader(conn: &Connection, utxo: &OutPointHash) -> Result<Option<A
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")?,
token_id: TokenID::from_blob(&token_blob).context("failed to decode token blob")?,
txid: Txid::from_blob(&txid_blob).context("failed to decode txid blob")?,
height,
bcmr_data,
}))
@ -247,8 +248,8 @@ pub fn get_entries_missing_bcmr_download(conn: &Connection) -> Result<Vec<AuthCh
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 token_blob: Vec<u8> = header.get(0)?;
let txid_blob: Vec<u8> = header.get(1)?;
let height: usize = header.get(2)?;
let bcmr_data_hex: Option<String> = header.get(3)?;
let bcmr_data = if let Some(data) = bcmr_data_hex {
@ -256,13 +257,13 @@ pub fn get_entries_missing_bcmr_download(conn: &Connection) -> Result<Vec<AuthCh
} else {
None
};
let utxo_hex: String = header.get(4)?;
let utxo = OutPointHash::from_hex(&utxo_hex)?;
let utxo_blob: Vec<u8> = header.get(4)?;
let utxo = OutPointHash::from_blob(&utxo_blob)?;
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")?,
token_id: TokenID::from_blob(&token_blob).context("failed to decode token blob")?,
txid: Txid::from_blob(&txid_blob).context("failed to decode txid blob")?,
height,
bcmr_data,
});
@ -308,8 +309,8 @@ pub fn insert_bcmr_data(
conn.execute(
sql,
rusqlite::params![
&token_id.to_hex(),
&utxo.to_hex(),
&token_id.to_blob(),
&utxo.to_blob(),
&bcmr.token.symbol,
bcmr.token.decimals,
&bcmr.name,
@ -333,13 +334,15 @@ pub fn insert_well_known_bcmr(
VALUES (?, ?, ?, ?, ?, ?, ?, ?)";
let empty_string: String = "".to_owned();
let token_blob =
hex::decode(&bcmr.token.category).context("failed to decode token category hex")?;
conn.execute(
sql,
rusqlite::params![
source,
bcmr.token.symbol,
bcmr.token.category,
token_blob,
bcmr.token.decimals,
&bcmr.name,
&bcmr.description,
@ -377,9 +380,9 @@ pub fn update_bcmr_failure(
conn.execute(
&sql,
params![
&token_id.to_hex(),
&utxo.to_hex(),
&txid.to_hex(),
&token_id.to_blob(),
&utxo.to_blob(),
&txid.to_blob(),
error_message,
give_up
],
@ -396,9 +399,10 @@ pub fn get_token_bcmr(conn: &Connection, token_hex: &str) -> Result<Option<Parse
ORDER BY ROWID DESC
LIMIT 1;
"#;
let token_blob = hex::decode(token_hex).context("invalid token hex")?;
let mut stmt = conn.prepare(sql)?;
let mut row = stmt.query([token_hex])?;
let mut row = stmt.query([token_blob])?;
if let Some(r) = row.next()? {
let symbol: String = r.get(0)?;
@ -438,9 +442,10 @@ pub fn get_well_known_bcmr(conn: &Connection, token_hex: &str) -> Result<Vec<Par
source, symbol, decimals, name, description, icon, web
FROM bcmr_well_known
WHERE token_id = ?";
let token_blob = hex::decode(token_hex).context("invalid token hex")?;
let mut stmt = conn.prepare(sql)?;
let mut row = stmt.query([token_hex])?;
let mut row = stmt.query([token_blob])?;
let mut entries: Vec<ParsedBCMR> = Vec::default();

179
src/db/blob.rs Normal file
View file

@ -0,0 +1,179 @@
// Copyright (C) 2024-2026 Whiterun LLC
//
// 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
//! BLOB storage helpers for efficient database storage of hash types.
//!
//! Converts 32-byte hashes from TEXT (64-char hex) to BLOB (32 bytes) for ~50% storage savings.
use anyhow::{Context, Result};
use bitcoin_hashes::Hash;
use bitcoincash::{BlockHash, PubkeyHash, TokenID, Txid};
use riftenlabs_defi::chainutil::OutPointHash;
use crate::def::PoolID;
/// Trait for converting hash types to BLOB bytes for database storage.
pub trait ToBlob {
fn to_blob(&self) -> Vec<u8>;
}
/// Trait for converting BLOB bytes back to hash types.
pub trait FromBlob: Sized {
fn from_blob(bytes: &[u8]) -> Result<Self>;
}
// Implement ToBlob for 32-byte hash types
impl ToBlob for Txid {
fn to_blob(&self) -> Vec<u8> {
self.into_inner().to_vec()
}
}
impl ToBlob for BlockHash {
fn to_blob(&self) -> Vec<u8> {
self.into_inner().to_vec()
}
}
impl ToBlob for TokenID {
fn to_blob(&self) -> Vec<u8> {
self.into_inner().to_vec()
}
}
impl ToBlob for OutPointHash {
fn to_blob(&self) -> Vec<u8> {
self.into_inner().to_vec()
}
}
impl ToBlob for PoolID {
fn to_blob(&self) -> Vec<u8> {
self.into_inner().to_vec()
}
}
// Implement ToBlob for 20-byte PubkeyHash
impl ToBlob for PubkeyHash {
fn to_blob(&self) -> Vec<u8> {
self.into_inner().to_vec()
}
}
// Implement FromBlob for 32-byte hash types
impl FromBlob for Txid {
fn from_blob(bytes: &[u8]) -> Result<Self> {
let arr: [u8; 32] = bytes.try_into().context("Txid blob must be 32 bytes")?;
Ok(Txid::from_inner(arr))
}
}
impl FromBlob for BlockHash {
fn from_blob(bytes: &[u8]) -> Result<Self> {
let arr: [u8; 32] = bytes
.try_into()
.context("BlockHash blob must be 32 bytes")?;
Ok(BlockHash::from_inner(arr))
}
}
impl FromBlob for TokenID {
fn from_blob(bytes: &[u8]) -> Result<Self> {
let arr: [u8; 32] = bytes.try_into().context("TokenID blob must be 32 bytes")?;
Ok(TokenID::from_inner(arr))
}
}
impl FromBlob for OutPointHash {
fn from_blob(bytes: &[u8]) -> Result<Self> {
let arr: [u8; 32] = bytes
.try_into()
.context("OutPointHash blob must be 32 bytes")?;
Ok(OutPointHash::from_inner(arr))
}
}
impl FromBlob for PoolID {
fn from_blob(bytes: &[u8]) -> Result<Self> {
let arr: [u8; 32] = bytes.try_into().context("PoolID blob must be 32 bytes")?;
Ok(PoolID::from_inner(arr))
}
}
// Implement FromBlob for 20-byte PubkeyHash
impl FromBlob for PubkeyHash {
fn from_blob(bytes: &[u8]) -> Result<Self> {
let arr: [u8; 20] = bytes
.try_into()
.context("PubkeyHash blob must be 20 bytes")?;
Ok(PubkeyHash::from_inner(arr))
}
}
#[cfg(test)]
mod tests {
use super::*;
use bitcoin_hashes::hex::{FromHex, ToHex};
#[test]
fn test_txid_roundtrip() {
let hex = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";
let txid = Txid::from_hex(hex).unwrap();
let blob = txid.to_blob();
assert_eq!(blob.len(), 32);
let recovered = Txid::from_blob(&blob).unwrap();
assert_eq!(txid, recovered);
assert_eq!(recovered.to_hex(), hex);
}
#[test]
fn test_blockhash_roundtrip() {
let hex = "fedcba9876543210fedcba9876543210fedcba9876543210fedcba9876543210";
let hash = BlockHash::from_hex(hex).unwrap();
let blob = hash.to_blob();
assert_eq!(blob.len(), 32);
let recovered = BlockHash::from_blob(&blob).unwrap();
assert_eq!(hash, recovered);
}
#[test]
fn test_token_id_roundtrip() {
let hex = "abcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789";
let token = TokenID::from_hex(hex).unwrap();
let blob = token.to_blob();
assert_eq!(blob.len(), 32);
let recovered = TokenID::from_blob(&blob).unwrap();
assert_eq!(token, recovered);
}
#[test]
fn test_outpointhash_roundtrip() {
let hex = "1111111111111111111111111111111111111111111111111111111111111111";
let hash = OutPointHash::from_hex(hex).unwrap();
let blob = hash.to_blob();
assert_eq!(blob.len(), 32);
let recovered = OutPointHash::from_blob(&blob).unwrap();
assert_eq!(hash, recovered);
}
#[test]
fn test_pubkeyhash_roundtrip() {
let hex = "0123456789abcdef01230123456789abcdef0123";
let pkh = PubkeyHash::from_hex(hex).unwrap();
let blob = pkh.to_blob();
assert_eq!(blob.len(), 20);
let recovered = PubkeyHash::from_blob(&blob).unwrap();
assert_eq!(pkh, recovered);
}
#[test]
fn test_invalid_blob_size() {
let short_blob = vec![0u8; 16];
assert!(Txid::from_blob(&short_blob).is_err());
assert!(BlockHash::from_blob(&short_blob).is_err());
assert!(TokenID::from_blob(&short_blob).is_err());
assert!(PubkeyHash::from_blob(&short_blob).is_err());
}
}

2
src/db/cauldron/config.rs
View file

@ -6,7 +6,7 @@
use anyhow::{bail, Result};
use rusqlite::{params, Connection};
pub const DB_VERSION: u32 = 4;
pub const DB_VERSION: u32 = 5;
const DB_VERSION_KEY: &str = "db_version";
/// Create the config table

13
src/db/cauldron/mempool.rs
View file

@ -6,18 +6,19 @@
use std::collections::HashSet;
use anyhow::Result;
use bitcoin_hashes::hex::{FromHex, ToHex};
use bitcoincash::Txid;
use rusqlite::Connection;
use crate::db::blob::{FromBlob, ToBlob};
pub fn load_mempool(conn: &Connection) -> Result<HashSet<Txid>> {
let mut stmt = conn.prepare("SELECT txid FROM tx WHERE blockhash is NULL")?;
let txid_iter = stmt.query_map([], |row| row.get(0))?;
let mut txids: HashSet<Txid> = HashSet::new();
for txid_res in txid_iter {
let txid_hex: String = txid_res?;
let txid = Txid::from_hex(&txid_hex).expect("invalid txid in db");
let txid_blob: Vec<u8> = txid_res?;
let txid = Txid::from_blob(&txid_blob).expect("invalid txid in db");
txids.insert(txid);
}
Ok(txids)
@ -27,15 +28,15 @@ pub fn delete_mempool_txs<'a, I>(db_tx: &Connection, txids: I) -> Result<bool>
where
I: IntoIterator<Item = &'a Txid>,
{
let txid_hexes: Vec<String> = txids.into_iter().map(|txid| txid.to_hex()).collect();
let placeholders = txid_hexes
let txid_blobs: Vec<Vec<u8>> = txids.into_iter().map(|txid| txid.to_blob()).collect();
let placeholders = txid_blobs
.iter()
.map(|_| "?")
.collect::<Vec<_>>()
.join(", ");
let query = format!("DELETE FROM tx WHERE txid IN ({placeholders}) AND blockhash is NULL");
let params: Vec<&dyn rusqlite::ToSql> = txid_hexes
let params: Vec<&dyn rusqlite::ToSql> = txid_blobs
.iter()
.map(|s| s as &dyn rusqlite::ToSql)
.collect();

6
src/db/cauldron/mod.rs
View file

@ -4,10 +4,10 @@
// 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};
use crate::db::blob::ToBlob;
use crate::db::cauldron::tokenlist::db_utils::{
create_aggregation_path_indexes, create_cached_token_metrics_indexes,
create_cached_token_metrics_table,
@ -69,7 +69,7 @@ pub fn delete_entries_for_block(tx: &Connection, blockhash: &BlockHash) -> Resul
SELECT txid FROM tx WHERE blockhash = ?
)",
)?;
let mut rows_deleted = stmt.execute(params![blockhash.to_hex()])?;
let mut rows_deleted = stmt.execute(params![blockhash.to_blob()])?;
let mut stmt = tx.prepare(
"DELETE FROM utxo_spending
@ -79,7 +79,7 @@ pub fn delete_entries_for_block(tx: &Connection, blockhash: &BlockHash) -> Resul
)
)",
)?;
rows_deleted += stmt.execute(params![blockhash.to_hex()])?;
rows_deleted += stmt.execute(params![blockhash.to_blob()])?;
// remaining tables should clear themselves due to foregn keys usage

67
src/db/cauldron/pool.rs
View file

@ -8,9 +8,10 @@ use std::{
sync::atomic::AtomicI64,
};
use crate::db::blob::{FromBlob, ToBlob};
use crate::def::PoolID;
use anyhow::{Context, Result};
use bitcoin_hashes::hex::{FromHex, ToHex};
use bitcoin_hashes::hex::ToHex;
use log::{debug, info, warn};
use malachite::Integer;
use riftenlabs_defi::{cauldron::ParsedContract, chainutil::OutPointHash};
@ -32,10 +33,10 @@ pub fn create_table(conn: &Connection) {
conn.execute(
"
CREATE TABLE pool (
creation_utxo TEXT PRIMARY KEY REFERENCES utxo_funding(new_utxo_hash) ON DELETE CASCADE,
owner_pkh TEXT NOT NULL,
token_id TEXT NOT NULL,
withdrawn_in_utxo TEXT REFERENCES utxo_spending(spent_utxo_hash) ON DELETE SET NULL
creation_utxo BLOB PRIMARY KEY REFERENCES utxo_funding(new_utxo_hash) ON DELETE CASCADE,
owner_pkh BLOB NOT NULL,
token_id BLOB NOT NULL,
withdrawn_in_utxo BLOB REFERENCES utxo_spending(spent_utxo_hash) ON DELETE SET NULL
)",
[],
)
@ -43,10 +44,10 @@ pub fn create_table(conn: &Connection) {
conn.execute(
"CREATE TABLE pool_history_entry (
utxo TEXT PRIMARY KEY REFERENCES utxo_funding(new_utxo_hash) ON DELETE CASCADE,
pool TEXT REFERENCES pool(creation_utxo) ON DELETE CASCADE,
token_id TEXT NOT NULL,
txid TEXT REFERENCES tx(txid) ON DELETE CASCADE,
utxo BLOB PRIMARY KEY REFERENCES utxo_funding(new_utxo_hash) ON DELETE CASCADE,
pool BLOB REFERENCES pool(creation_utxo) ON DELETE CASCADE,
token_id BLOB NOT NULL,
txid BLOB REFERENCES tx(txid) ON DELETE CASCADE,
tx_pos INT NOT NULL,
mtp_timestamp BIGINT,
first_seen_timestamp BIGINT,
@ -101,12 +102,12 @@ pub fn create_table(conn: &Connection) {
fn get_pool_by_utxo(conn: &Connection, utxo_hash: &OutPointHash) -> Result<Option<OutPointHash>> {
let mut stmt = conn.prepare("SELECT pool FROM pool_history_entry WHERE utxo = ?")?;
let mut row = stmt.query([utxo_hash.to_hex()])?;
let utxo_hex: Option<String> = row.next()?.map(|r| r.get(0).unwrap());
let mut row = stmt.query([utxo_hash.to_blob()])?;
let utxo_blob: Option<Vec<u8>> = row.next()?.map(|r| r.get(0).unwrap());
match utxo_hex {
Some(utxo) => Ok(Some(
OutPointHash::from_hex(&utxo).expect("invalid original_utxo utxo in db"),
match utxo_blob {
Some(blob) => Ok(Some(
OutPointHash::from_blob(&blob).expect("invalid original_utxo utxo in db"),
)),
None => Ok(None),
}
@ -119,7 +120,7 @@ pub fn flag_as_withdrawn(
) -> Result<()> {
conn.execute(
"UPDATE pool SET withdrawn_in_utxo = ? WHERE creation_utxo = ?",
params![cauldron.spent_utxo_hash.to_hex(), pool_utxo.to_hex()],
params![cauldron.spent_utxo_hash.to_blob(), pool_utxo.to_blob()],
)
.map_err(|e| anyhow::anyhow!("failed flag pool as withdrawn. Original error: {:?}", e))?;
@ -131,10 +132,10 @@ pub fn insert_new_pool(conn: &Connection, cauldron: &ParsedContract) -> Result<(
conn.execute(
"INSERT OR REPLACE INTO pool (creation_utxo, owner_pkh, token_id, withdrawn_in_utxo) VALUES (?, ?, ?, ?)",
params![
cauldron.new_utxo_hash.expect("outpoint hash for new pool missing").to_hex(),
cauldron.pkh.to_hex(),
cauldron.token_id.expect("token id for new pool missing").to_hex(),
None::<String>,
cauldron.new_utxo_hash.expect("outpoint hash for new pool missing").to_blob(),
cauldron.pkh.to_blob(),
cauldron.token_id.expect("token id for new pool missing").to_blob(),
None::<Vec<u8>>,
]
).map_err(|e| {
anyhow::anyhow!(
@ -192,13 +193,13 @@ pub fn insert_pool_history_entry(
cauldron
.new_utxo_hash
.expect("utxo hash on new pool history entry")
.to_hex(),
pool.to_hex(),
cauldron.token_id.expect("token id on new pool history entry").to_hex(),
.to_blob(),
pool.to_blob(),
cauldron.token_id.expect("token id on new pool history entry").to_blob(),
cauldron
.new_utxo_txid
.expect("txid of new pool history entry")
.to_hex(),
.to_blob(),
cauldron
.new_utxo_n
.expect("utxo index of new pool history entry"),
@ -459,8 +460,8 @@ pub struct PoolHistoryEntry {
}
fn get_pool_history_entry(conn: &Connection, utxo_hash: OutPointHash) -> Result<PoolHistoryEntry> {
let mut stmt = conn.prepare("SELECT txid, sats, token_amount, effective_timestamp as timestamp FROM pool_history_entry WHERE utxo = ?")?;
let mut rows = stmt.query(params![utxo_hash.to_hex()])?;
let mut stmt = conn.prepare("SELECT hex(txid), sats, token_amount, effective_timestamp as timestamp FROM pool_history_entry WHERE utxo = ?")?;
let mut rows = stmt.query(params![utxo_hash.to_blob()])?;
let row = rows
.next()?
.context("no pool history entry found for UTXO hash")?;
@ -481,7 +482,7 @@ pub fn db_pool_history(
start_time: u64,
) -> Result<Vec<PoolHistoryEntry>> {
let query = "SELECT
phe.txid,
hex(phe.txid),
phe.sats,
phe.token_amount,
phe.effective_timestamp as timestamp
@ -495,7 +496,7 @@ pub fn db_pool_history(
";
let mut stmt = conn.prepare(query)?;
let mut rows = stmt.query(params![pool.to_hex(), start_time])?;
let mut rows = stmt.query(params![pool.to_blob(), start_time])?;
let from_row = |row: &Row<'_>| -> Result<PoolHistoryEntry> {
let sats = row.get(1)?;
@ -523,16 +524,16 @@ pub fn db_pool_history(
pub fn db_pool_get_details(db: &Connection, pool: &PoolID) -> Result<(String, String)> {
let res = db.query_row(
"SELECT token_id, owner_pkh FROM pool WHERE creation_utxo = ?1",
[pool.to_hex()],
"SELECT hex(token_id), hex(owner_pkh) FROM pool WHERE creation_utxo = ?1",
[pool.to_blob()],
|row| Ok((row.get(0)?, row.get(1)?)),
)?;
Ok(res)
}
pub fn db_pool_id_from_utxo(conn: &Connection, utxo_hash: &OutPointHash) -> Result<Option<String>> {
let mut stmt = conn.prepare("SELECT pool FROM pool_history_entry WHERE utxo = ?")?;
let mut rows = stmt.query(params![utxo_hash.to_hex()])?;
let mut stmt = conn.prepare("SELECT hex(pool) FROM pool_history_entry WHERE utxo = ?")?;
let mut rows = stmt.query(params![utxo_hash.to_blob()])?;
if let Some(row) = rows.next()? {
let pool_id: String = row.get(0)?;
@ -581,10 +582,10 @@ pub fn get_token_volume_sats(
JOIN pool p ON phe.pool = p.creation_utxo
WHERE tx.effective_timestamp BETWEEN ? AND ?
AND p.token_id = ?";
let token_blob = hex::decode(token_id)?;
let (sats_volume, token_volume): (i64, i64) = db.query_row(
sql,
params![start_timestamp, end_timestamp, token_id],
params![start_timestamp, end_timestamp, token_blob],
|row| Ok((row.get(0)?, row.get(1)?)),
)?;

16
src/db/cauldron/poolvisitor.rs
View file

@ -137,8 +137,9 @@ pub(crate) fn db_visit_pool_entries<T: PoolVisitor>(
let mut param_index = 2 + params.len();
let phe_tokenid_filter = if let Some(token_id) = &filters.token_id {
let token_blob = hex::decode(token_id).expect("valid token hex");
params.push(rusqlite::types::ToSqlOutput::Owned(
token_id.to_owned().into(),
rusqlite::types::Value::Blob(token_blob),
));
let this_param_index = param_index;
param_index += 1;
@ -149,7 +150,10 @@ pub(crate) fn db_visit_pool_entries<T: PoolVisitor>(
if let Some(owner) = &filters.owner {
sql_filters.push(format!("p.owner_pkh = ?{}", param_index));
params.push(rusqlite::types::ToSqlOutput::Owned(owner.to_owned().into()));
let owner_blob = hex::decode(owner).expect("valid owner hex");
params.push(rusqlite::types::ToSqlOutput::Owned(
rusqlite::types::Value::Blob(owner_blob),
));
}
let sql_filters_str = if sql_filters.is_empty() {
"".to_string()
@ -159,13 +163,13 @@ pub(crate) fn db_visit_pool_entries<T: PoolVisitor>(
let query: String = "
SELECT
p.owner_pkh,
hex(p.owner_pkh),
phe.sats,
phe.token_amount,
phe.txid,
hex(phe.txid),
phe.tx_pos,
p.token_id,
p.creation_utxo,
hex(p.token_id),
hex(p.creation_utxo),
phe.effective_timestamp
FROM
pool p

19
src/db/cauldron/tokenlist/db_utils.rs
View file

@ -7,7 +7,7 @@
// Copyright (C) 2024-2026 Whiterun LLC
// AGPL-3.0-or-later
use anyhow::Result;
use anyhow::{Context, Result};
use rusqlite::{params, Connection};
pub enum CachedSort {
@ -207,7 +207,7 @@ pub fn create_cached_token_metrics_table(conn: &Connection) -> Result<()> {
conn.execute_batch(
r#"
CREATE TABLE IF NOT EXISTS cached_token_metrics (
token_id TEXT PRIMARY KEY,
token_id BLOB PRIMARY KEY,
trade_volume INTEGER NOT NULL DEFAULT 0,
tvl_sats INTEGER NOT NULL DEFAULT 0,
tvl_tokens INTEGER NOT NULL DEFAULT 0,
@ -279,6 +279,8 @@ pub type FirstPoolInfo = (String, String, i64, Option<i64>);
/// Only set `first_pool_ts` when it is currently NULL or 0.
pub fn cache_first_pool_ts_if_empty(conn: &Connection, token_id: &str, ts: i64) -> Result<()> {
let token_blob = hex::decode(token_id).context("invalid token hex")?;
// Ensure a row exists so the update can succeed.
conn.execute(
r#"
@ -286,7 +288,7 @@ pub fn cache_first_pool_ts_if_empty(conn: &Connection, token_id: &str, ts: i64)
VALUES (?1, strftime('%s','now'))
ON CONFLICT(token_id) DO NOTHING;
"#,
params![token_id],
params![&token_blob],
)?;
// Accept NULL *or* 0 as “empty/unset”
@ -296,7 +298,7 @@ pub fn cache_first_pool_ts_if_empty(conn: &Connection, token_id: &str, ts: i64)
SET first_pool_ts = ?2
WHERE token_id = ?1 AND (first_pool_ts IS NULL OR first_pool_ts = 0);
"#,
params![token_id, ts],
params![&token_blob, ts],
)?;
Ok(())
}
@ -360,12 +362,15 @@ pub fn db_first_pool_creation_row(
LIMIT 1;
"#
};
let token_blob = hex::decode(token_id).context("invalid token hex")?;
let mut stmt = conn.prepare(sql)?;
let mut rows = stmt.query(params![token_id])?;
let mut rows = stmt.query(params![&token_blob])?;
if let Some(row) = rows.next()? {
let creation_utxo: String = row.get(0)?;
let txid: String = row.get(1)?;
let creation_utxo_blob: Vec<u8> = row.get(0)?;
let txid_blob: Vec<u8> = row.get(1)?;
let creation_utxo: String = hex::encode(&creation_utxo_blob).to_lowercase();
let txid: String = hex::encode(&txid_blob).to_lowercase();
let ts: i64 = row.get(2)?;
let height: Option<i64> = row.get(3)?;
Ok(Some((creation_utxo, txid, ts, height)))

12
src/db/cauldron/tokenlist/list_cached.rs
View file

@ -37,7 +37,7 @@ pub struct TokenListItemCached {
}
const TOKEN_METRICS_COLUMNS: &str = r#"
token_id,
hex(token_id) as token_id,
trade_volume,
tvl_sats,
tvl_tokens,
@ -84,7 +84,7 @@ pub fn db_list_tokens_cached(
let mut tokens = Vec::with_capacity(limit);
while let Some(row) = rows.next()? {
let token_id: String = row.get("token_id")?;
let token_id: String = row.get::<_, String>("token_id")?.to_lowercase();
let trade_volume: u64 = row.get("trade_volume")?;
let tvl_sats: u64 = row.get::<_, i64>("tvl_sats")? as u64;
let tvl_tokens: u64 = row.get::<_, i64>("tvl_tokens")? as u64;
@ -161,13 +161,17 @@ pub fn db_list_tokens_cached_by_ids(
ph = placeholders,
order_sql = order_sql
);
let token_blobs: Vec<Vec<u8>> = token_ids
.iter()
.filter_map(|s| hex::decode(s).ok())
.collect();
let mut stmt = cauldron_conn.prepare(&sql)?;
let mut rows = stmt.query(params_from_iter(token_ids.iter().map(|s| s.as_str())))?;
let mut rows = stmt.query(params_from_iter(token_blobs.iter()))?;
let mut tokens = Vec::with_capacity(token_ids.len());
while let Some(row) = rows.next()? {
let token_id: String = row.get("token_id")?;
let token_id: String = row.get::<_, String>("token_id")?.to_lowercase();
let trade_volume: u64 = row.get("trade_volume")?;
let tvl_sats: u64 = row.get::<_, i64>("tvl_sats")? as u64;
let tvl_tokens: u64 = row.get::<_, i64>("tvl_tokens")? as u64;

4
src/db/cauldron/tokenlist/list_tokens_volume.rs
View file

@ -55,7 +55,7 @@ pub fn db_list_tokens_by_volume(
GROUP BY all_tokens.token_id
)
SELECT
token_id,
hex(token_id) as token_id,
total_trade_volume
FROM AggregateTradeData
ORDER BY total_trade_volume DESC
@ -68,7 +68,7 @@ pub fn db_list_tokens_by_volume(
let mut tokens: Vec<TokenListItem> = Vec::with_capacity(limit);
while let Some(row) = query.next()? {
let token_id: String = row.get(0)?;
let token_id: String = row.get::<_, String>(0)?.to_lowercase();
let trade_volume: u64 = row.get(1)?;
let (tvl_sats, tvl_tokens) = get_token_tvl(cauldron_conn, None, &token_id)?;

53
src/db/cauldron/tokenlist/metrics_cache.rs
View file

@ -207,9 +207,10 @@ fn flush_fast_batch(conn: &Connection, rows: &[(&str, i64, i64, f64, f64)]) -> a
"#,
)?;
for (tid, sats, toks, p_now, p_now_usd) in rows {
upsert.execute(params![*tid, *sats, *toks, *p_now, *p_now_usd])?;
let tid_blob = hex::decode(tid).unwrap_or_default();
upsert.execute(params![tid_blob, *sats, *toks, *p_now, *p_now_usd])?;
}
} // drop(upsert)
}
tx.commit()?;
Ok(())
},
@ -221,12 +222,12 @@ fn flush_delete_absent<'a>(
conn: &Connection,
present_ids: impl Iterator<Item = &'a str>,
) -> anyhow::Result<()> {
let ids: Vec<String> = present_ids.map(|s| s.to_owned()).collect();
let ids: Vec<Vec<u8>> = present_ids.filter_map(|s| hex::decode(s).ok()).collect();
with_busy_retry(
|| {
let tx = conn.unchecked_transaction()?;
tx.execute_batch(
"CREATE TEMP TABLE IF NOT EXISTS __present (token_id TEXT PRIMARY KEY);",
"CREATE TEMP TABLE IF NOT EXISTS __present (token_id BLOB PRIMARY KEY);",
)?;
tx.execute_batch("DELETE FROM __present;")?;
{
@ -234,7 +235,7 @@ fn flush_delete_absent<'a>(
for tid in &ids {
ins.execute(params![tid])?;
}
} // drop(ins)
}
tx.execute(
r#"
DELETE FROM cached_token_metrics
@ -293,7 +294,7 @@ pub fn update_changes_score_volume_and_ranking(
// volume per token
let mut vol_stmt = cauldron_conn.prepare(
r#"
SELECT p.token_id, COALESCE(SUM(ABS(phe.sats_delta)), 0) AS vol
SELECT hex(p.token_id) as token_id, COALESCE(SUM(ABS(phe.sats_delta)), 0) AS vol
FROM pool_history_entry phe
JOIN pool p ON p.creation_utxo = phe.pool
JOIN tx ON tx.txid = phe.txid
@ -304,7 +305,7 @@ pub fn update_changes_score_volume_and_ranking(
let mut vol_rows = vol_stmt.query([since])?;
let mut vol_by_token: HashMap<String, u64> = HashMap::new();
while let Some(row) = vol_rows.next()? {
let token_id: String = row.get(0)?;
let token_id: String = row.get::<_, String>(0)?.to_lowercase();
let vol: i64 = row.get(1)?;
vol_by_token.insert(token_id, vol.max(0) as u64);
}
@ -352,14 +353,19 @@ pub fn update_changes_score_volume_and_ranking(
let mut cur_stmt = tx.prepare(
"SELECT tvl_sats, tvl_tokens FROM cached_token_metrics WHERE token_id = ?1",
)?;
let mut id_stmt =
tx.prepare("SELECT token_id FROM cached_token_metrics WHERE tvl_sats > 0")?;
let mut id_stmt = tx.prepare(
"SELECT hex(token_id) as token_id FROM cached_token_metrics WHERE tvl_sats > 0",
)?;
let mut id_rows = id_stmt.query([])?;
let mut dec_cache: HashMap<String, u32> = HashMap::new();
while let Some(row) = id_rows.next()? {
let token_id: String = row.get(0)?;
let token_id: String = row.get::<_, String>(0)?.to_lowercase();
let token_blob = match hex::decode(&token_id) {
Ok(b) => b,
Err(_) => continue,
};
// decimals → factor
let decimals_u32 = *dec_cache
@ -369,7 +375,7 @@ pub fn update_changes_score_volume_and_ranking(
// tvl
let (tvl_sats_i64, tvl_tokens_i64): (i64, i64) =
cur_stmt.query_row([&token_id], |r| Ok((r.get(0)?, r.get(1)?)))?;
cur_stmt.query_row([&token_blob], |r| Ok((r.get(0)?, r.get(1)?)))?;
let tvl_sats = tvl_sats_i64.max(0) as u64;
let tvl_tokens = tvl_tokens_i64.max(0) as u64;
@ -406,7 +412,7 @@ pub fn update_changes_score_volume_and_ranking(
.unwrap_or((None, None));
upd.execute(params![
token_id,
&token_blob,
vol as i64,
score,
Option::<f64>::None, // price_24h
@ -448,7 +454,7 @@ pub fn update_changes_score_volume_and_ranking(
Err(_) => {
// still update the "now" fields and basics
upd.execute(params![
token_id,
&token_blob,
vol as i64,
score,
Option::<f64>::None,
@ -522,7 +528,7 @@ pub fn update_changes_score_volume_and_ranking(
};
upd.execute(params![
token_id,
&token_blob,
vol as i64,
score,
price_24h_human_f,
@ -589,7 +595,7 @@ pub fn update_changes_score_volume_and_ranking(
};
upd.execute(params![
token_id,
&token_blob,
vol as i64,
score,
price_24h_human_f,
@ -647,16 +653,19 @@ pub fn update_apy_only(cauldron_conn: &Connection) -> anyhow::Result<()> {
let mut ids = Vec::<(String, i64)>::new();
{
let mut s = cauldron_conn.prepare(
"SELECT token_id, trade_volume FROM cached_token_metrics WHERE tvl_sats > 0",
"SELECT hex(token_id) as token_id, trade_volume FROM cached_token_metrics WHERE tvl_sats > 0",
)?;
let mut r = s.query([])?;
while let Some(row) = r.next()? {
ids.push((row.get::<_, String>(0)?, row.get::<_, i64>(1)?));
ids.push((
row.get::<_, String>(0)?.to_lowercase(),
row.get::<_, i64>(1)?,
));
}
}
// Compute/set APY; when vol==0, write 0 directly
let mut apys = Vec::<(String, Option<i64>)>::with_capacity(ids.len());
let mut apys = Vec::<(Vec<u8>, Option<i64>)>::with_capacity(ids.len());
for (token_id, vol30) in ids {
let apy_opt = if vol30 == 0 {
Some(0)
@ -669,7 +678,9 @@ pub fn update_apy_only(cauldron_conn: &Connection) -> anyhow::Result<()> {
}
}
};
apys.push((token_id, apy_opt));
if let Ok(blob) = hex::decode(&token_id) {
apys.push((blob, apy_opt));
}
}
with_busy_retry(
@ -823,7 +834,7 @@ pub fn backfill_first_pool_ts_batch(conn: &Connection, limit: i64) -> anyhow::Re
let token_ids: Vec<String> = {
let mut stmt = conn.prepare(
r#"
SELECT token_id
SELECT hex(token_id) as token_id
FROM cached_token_metrics
WHERE first_pool_ts IS NULL OR first_pool_ts = 0
ORDER BY token_id
@ -833,7 +844,7 @@ pub fn backfill_first_pool_ts_batch(conn: &Connection, limit: i64) -> anyhow::Re
let mut rows = stmt.query([limit])?;
let mut v = Vec::new();
while let Some(row) = rows.next()? {
v.push(row.get::<_, String>(0)?);
v.push(row.get::<_, String>(0)?.to_lowercase());
}
v
};

103
src/db/cauldron/tokenlist/test.rs
View file

@ -15,6 +15,7 @@ mod tests {
use crate::db::bcmr::{
insert_authheader, insert_bcmr_data, prepare_tables as bcmr_prepare_tables,
};
use crate::db::blob::ToBlob;
use crate::db::cauldron::pool::{dummy_init_seq, insert_new_pool, insert_pool_history_entry};
use crate::db::cauldron::prepare_tables as cauldron_prepare_tables;
use crate::db::cauldron::tokenlist::db_utils::{create_cached_token_metrics_table, CachedSort};
@ -215,7 +216,7 @@ mod tests {
"INSERT INTO bcmr_well_known (token_id, name, symbol, decimals, description, icon, web, source)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
params![
token_b.to_hex(),
token_b.to_blob(),
"WKName",
"WKS",
2i64,
@ -230,7 +231,7 @@ mod tests {
let token_c = TokenID::from_inner([0x03; 32]);
rw.execute(
"INSERT INTO crc20 (token_id, name, symbol, decimals) VALUES (?1, ?2, ?3, ?4)",
params![token_c.to_hex(), "CRCName", "CRC", 6i64],
params![token_c.to_blob(), "CRCName", "CRC", 6i64],
)
.unwrap();
@ -276,6 +277,7 @@ mod tests {
ch7d_usd: i64,
apy_bp: i64,
) {
let token_blob = hex::decode(token_hex).expect("valid hex");
conn.execute(
r#"
INSERT INTO cached_token_metrics
@ -291,7 +293,7 @@ mod tests {
?18, strftime('%s','now'))
"#,
params![
token_hex,
token_blob,
trade_volume,
tvl_sats,
tvl_tokens,
@ -566,13 +568,15 @@ mod tests {
// insert a minimal row so tvl_sats>0 for one token before calling the fast updater.
// Minimal fake: create a row in cached_token_metrics by hand (simulating the fast path)
cw.execute_batch(
let tok_a = TokenID::from_inner([0x11; 32]);
cw.execute(
r#"
INSERT INTO cached_token_metrics(token_id, trade_volume, tvl_sats, tvl_tokens, score,
display_name, display_symbol,
price_now, price_now_usd, updated_at)
VALUES('tokA', 0, 1, 1, 0, NULL, NULL, 1.0, 0.5, strftime('%s','now'));
VALUES(?1, 0, 1, 1, 0, NULL, NULL, 1.0, 0.5, strftime('%s','now'));
"#,
params![tok_a.to_blob()],
)
.unwrap();
@ -580,8 +584,8 @@ mod tests {
let (c24, c7d, c24u, c7du, apy, dn, ds): (Option<i64>, Option<i64>, Option<i64>, Option<i64>, Option<i64>, Option<String>, Option<String>) =
cw.query_row(
"SELECT change_24h_bp, change_7d_bp, change_24h_usd_bp, change_7d_usd_bp, apy_30d_bp, display_name, display_symbol
FROM cached_token_metrics WHERE token_id='tokA'",
[],
FROM cached_token_metrics WHERE token_id=?1",
params![tok_a.to_blob()],
|r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?, r.get(5)?, r.get(6)?))
).unwrap();
@ -648,7 +652,7 @@ mod tests {
"INSERT INTO cached_token_metrics(token_id, trade_volume, tvl_sats, tvl_tokens, score,
price_now, price_now_usd, updated_at)
VALUES (?1, 0, 1, 1, 0, 1.0, 0.5, strftime('%s','now'))",
rusqlite::params![token.to_hex()],
rusqlite::params![token.to_blob()],
)
.unwrap();
@ -659,7 +663,7 @@ mod tests {
.query_row(
"SELECT change_24h_bp, change_7d_bp, change_24h_usd_bp, change_7d_usd_bp
FROM cached_token_metrics WHERE token_id=?1",
rusqlite::params![token.to_hex()],
rusqlite::params![token.to_blob()],
|r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?)),
)
.unwrap();
@ -679,23 +683,25 @@ mod tests {
let orc = mock.oracle_r.get().unwrap();
// Seed a cached row with some tvl so it exists initially
cw.execute_batch(
let tok_remove = TokenID::from_inner([0x33; 32]);
cw.execute(
r#"
INSERT INTO cached_token_metrics
(token_id, trade_volume, tvl_sats, tvl_tokens, score,
display_name, display_symbol,
price_now, price_now_usd, updated_at)
VALUES
('tok_to_remove', 0, 123, 456, 0, 'Tmp', 'TMP', 1.0, 0.5, strftime('%s','now'));
(?1, 0, 123, 456, 0, 'Tmp', 'TMP', 1.0, 0.5, strftime('%s','now'));
"#,
params![tok_remove.to_blob()],
)
.unwrap();
// Sanity: row exists
let count_before: i64 = cw
.query_row(
"SELECT COUNT(*) FROM cached_token_metrics WHERE token_id='tok_to_remove'",
[],
"SELECT COUNT(*) FROM cached_token_metrics WHERE token_id=?1",
params![tok_remove.to_blob()],
|r| r.get(0),
)
.unwrap();
@ -706,8 +712,8 @@ mod tests {
let count_after: i64 = cw
.query_row(
"SELECT COUNT(*) FROM cached_token_metrics WHERE token_id='tok_to_remove'",
[],
"SELECT COUNT(*) FROM cached_token_metrics WHERE token_id=?1",
params![tok_remove.to_blob()],
|r| r.get(0),
)
.unwrap();
@ -726,17 +732,35 @@ mod tests {
let orc = mock.oracle_r.get().unwrap();
// Two tokens in cache with initial TVL and scores so they get ranks 1 & 2
let tok_keep = TokenID::from_inner([0xAA; 32]).to_hex();
let tok_drop = TokenID::from_inner([0xAB; 32]).to_hex();
let tok_keep_id = TokenID::from_inner([0xAA; 32]);
let tok_drop_id = TokenID::from_inner([0xAB; 32]);
let tok_keep = tok_keep_id.to_hex();
let _tok_drop = tok_drop_id.to_hex();
cw.execute_batch(&format!(
r#"
cw.execute(
r#"
INSERT INTO cached_token_metrics
(token_id, trade_volume, tvl_sats, tvl_tokens, score,
price_now, price_now_usd, updated_at)
VALUES
('{tok_keep}', 100, 10, 10, 1000, 1.0, 0.5, strftime('%s','now')),
('{tok_drop}', 50, 10, 10, 900, 1.0, 0.5, strftime('%s','now'));
(?1, 100, 10, 10, 1000, 1.0, 0.5, strftime('%s','now'));
"#,
params![tok_keep_id.to_blob()],
)
.unwrap();
cw.execute(
r#"
INSERT INTO cached_token_metrics
(token_id, trade_volume, tvl_sats, tvl_tokens, score,
price_now, price_now_usd, updated_at)
VALUES
(?1, 50, 10, 10, 900, 1.0, 0.5, strftime('%s','now'));
"#,
params![tok_drop_id.to_blob()],
)
.unwrap();
cw.execute_batch(
r#"
-- initial ranks like core would do
WITH ranked AS (
SELECT token_id,
@ -746,8 +770,9 @@ mod tests {
UPDATE cached_token_metrics
SET score_rank = (SELECT rnk FROM ranked WHERE ranked.token_id = cached_token_metrics.token_id)
WHERE token_id IN (SELECT token_id FROM ranked);
"#
)).unwrap();
"#,
)
.unwrap();
// Create *real* TVL only for tok_keep so the fast updater will keep it and drop tok_drop.
// We do this by seeding minimal pool history for tok_keep so TvlByTokenVisitor sees liquidity.
@ -777,11 +802,12 @@ mod tests {
let rows: Vec<(String, i64)> = {
let mut v = Vec::new();
let mut stmt = cw
.prepare("SELECT token_id, score_rank FROM cached_token_metrics ORDER BY token_id")
.prepare("SELECT hex(token_id) as token_id, score_rank FROM cached_token_metrics ORDER BY token_id")
.unwrap();
let mut r = stmt.query([]).unwrap();
while let Some(row) = r.next().unwrap() {
v.push((row.get(0).unwrap(), row.get(1).unwrap()));
let tid: String = row.get(0).unwrap();
v.push((tid.to_lowercase(), row.get(1).unwrap()));
}
v
};
@ -878,7 +904,7 @@ mod tests {
"INSERT INTO cached_token_metrics(token_id, updated_at)
VALUES(?1, strftime('%s','now'))
ON CONFLICT(token_id) DO NOTHING",
params![token.to_hex()],
params![token.to_blob()],
)
.unwrap();
@ -890,7 +916,7 @@ mod tests {
let cached_ts: Option<i64> = rw
.query_row(
"SELECT first_pool_ts FROM cached_token_metrics WHERE token_id=?1",
params![token.to_hex()],
params![token.to_blob()],
|r| r.get(0),
)
.unwrap();
@ -913,7 +939,7 @@ mod tests {
seed_minimal_token_history(&rw, token_b, t0b, t0b + 100, 1000, 10, 2000, 20);
// Create bare cache rows so the backfill selector sees them
for tid in [token_a.to_hex(), token_b.to_hex()] {
for tid in [token_a.to_blob(), token_b.to_blob()] {
rw.execute(
"INSERT INTO cached_token_metrics(token_id, updated_at)
VALUES(?1, strftime('%s','now'))",
@ -930,14 +956,14 @@ mod tests {
let a_ts: Option<i64> = rw
.query_row(
"SELECT first_pool_ts FROM cached_token_metrics WHERE token_id=?1",
params![token_a.to_hex()],
params![token_a.to_blob()],
|r| r.get(0),
)
.unwrap();
let b_ts: Option<i64> = rw
.query_row(
"SELECT first_pool_ts FROM cached_token_metrics WHERE token_id=?1",
params![token_b.to_hex()],
params![token_b.to_blob()],
|r| r.get(0),
)
.unwrap();
@ -954,14 +980,14 @@ mod tests {
let a_ts2: Option<i64> = rw
.query_row(
"SELECT first_pool_ts FROM cached_token_metrics WHERE token_id=?1",
params![token_a.to_hex()],
params![token_a.to_blob()],
|r| r.get(0),
)
.unwrap();
let b_ts2: Option<i64> = rw
.query_row(
"SELECT first_pool_ts FROM cached_token_metrics WHERE token_id=?1",
params![token_b.to_hex()],
params![token_b.to_blob()],
|r| r.get(0),
)
.unwrap();
@ -974,17 +1000,18 @@ mod tests {
let mock = mock_db_pool(|conn| setup_basic_schemas(conn));
let rw = mock.cauldron_w.get().unwrap();
let token = TokenID::from_inner([0xFE; 32]).to_hex();
let token = TokenID::from_inner([0xFE; 32]);
let token_hex = token.to_hex();
// No pools for this token.
let row = db_first_pool_creation_row(&rw, &token).unwrap();
let row = db_first_pool_creation_row(&rw, &token_hex).unwrap();
assert!(row.is_none(), "no pools → no first row");
// Put a cache row so backfill inspects it; should remain NULL
rw.execute(
"INSERT INTO cached_token_metrics(token_id, updated_at)
VALUES(?1, strftime('%s','now'))",
params![token],
params![token.to_blob()],
)
.unwrap();
@ -994,7 +1021,7 @@ mod tests {
let cached_ts: Option<i64> = rw
.query_row(
"SELECT first_pool_ts FROM cached_token_metrics WHERE token_id=?1",
params![token],
params![token.to_blob()],
|r| r.get(0),
)
.unwrap();
@ -1150,7 +1177,7 @@ mod tests {
// Create presence in cache
cw.execute("INSERT INTO cached_token_metrics(token_id, trade_volume, tvl_sats, tvl_tokens, score, updated_at)
VALUES(?1,0,1,1,0,strftime('%s','now'))
ON CONFLICT(token_id) DO NOTHING", rusqlite::params![token.to_hex()]).unwrap();
ON CONFLICT(token_id) DO NOTHING", rusqlite::params![token.to_blob()]).unwrap();
// Should not panic
update_changes_score_volume_and_ranking(&cw, &bcmr_r, &crc_r, &orc_r).unwrap();
@ -1158,7 +1185,7 @@ mod tests {
// Prices likely NULL after overflow guard
let (p_now_usd, p_24h, p_7d): (Option<f64>, Option<f64>, Option<f64>) =
cw.query_row("SELECT price_now_usd, price_24h, price_7d FROM cached_token_metrics WHERE token_id=?1",
rusqlite::params![token.to_hex()],
rusqlite::params![token.to_blob()],
|r| Ok((r.get(0)?, r.get(1)?, r.get(2)?))).unwrap();
assert!(p_now_usd.is_none() || p_now_usd.unwrap().is_finite());
// allow None here; the point is: no panic

21
src/db/cauldron/tokenlist/token_utils.rs
View file

@ -3,7 +3,7 @@
// 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 anyhow::{Context, Result};
use log::warn;
use rusqlite::Connection;
@ -121,6 +121,11 @@ pub fn compute_score(tvl_sats: u64, vol_30d: u64) -> i64 {
score_big.to_string().parse::<i64>().unwrap_or(i64::MAX)
}
pub fn resolve_decimals(bcmr_conn: &Connection, crc20_conn: &Connection, token_id: &str) -> u32 {
let token_blob = match hex::decode(token_id) {
Ok(b) => b,
Err(_) => return 0, // Invalid hex, return fallback
};
// On-chain BCMR (latest by height)
if let Ok(Some(v)) = bcmr_conn.query_row(
r#"
@ -132,7 +137,7 @@ pub fn resolve_decimals(bcmr_conn: &Connection, crc20_conn: &Connection, token_i
WHERE a.bcmr_data IS NOT NULL AND a.token_id = ?1
) s WHERE rn = 1
"#,
[token_id],
[&token_blob],
|r| r.get::<_, Option<i64>>(0),
) {
let dd = v.max(0) as u32;
@ -144,7 +149,7 @@ pub fn resolve_decimals(bcmr_conn: &Connection, crc20_conn: &Connection, token_i
// Well-known BCMR
if let Ok(Some(v)) = bcmr_conn.query_row(
r#"SELECT decimals FROM bcmr_well_known WHERE token_id = ?1 ORDER BY source LIMIT 1"#,
[token_id],
[&token_blob],
|r| r.get::<_, Option<i64>>(0),
) {
let dd = v.max(0) as u32;
@ -156,7 +161,7 @@ pub fn resolve_decimals(bcmr_conn: &Connection, crc20_conn: &Connection, token_i
// CRC20 fallback
if let Ok(Some(v)) = crc20_conn.query_row(
r#"SELECT decimals FROM crc20 WHERE token_id = ?1 LIMIT 1"#,
[token_id],
[&token_blob],
|r| r.get::<_, Option<i64>>(0),
) {
let dd = v.max(0) as u32;
@ -254,6 +259,8 @@ pub fn resolve_display_labels(
crc20_conn: &Connection,
token_id: &str,
) -> Result<(String, String)> {
let token_blob = hex::decode(token_id).context("invalid token hex")?;
// On-chain BCMR (latest by height)
let mut onchain = bcmr_conn.prepare(
r#"
@ -266,7 +273,7 @@ pub fn resolve_display_labels(
) s WHERE rn = 1
"#,
)?;
if let Ok(row) = onchain.query_row([token_id], |r| {
if let Ok(row) = onchain.query_row([&token_blob], |r| {
Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?))
}) {
let (name, sym) = row;
@ -284,7 +291,7 @@ pub fn resolve_display_labels(
let mut wk = bcmr_conn.prepare(
r#"SELECT name, symbol FROM bcmr_well_known WHERE token_id = ?1 ORDER BY source LIMIT 1"#,
)?;
if let Ok(row) = wk.query_row([token_id], |r| {
if let Ok(row) = wk.query_row([&token_blob], |r| {
Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?))
}) {
let (name, sym) = row;
@ -301,7 +308,7 @@ pub fn resolve_display_labels(
// CRC20 fallback
let mut crc =
crc20_conn.prepare(r#"SELECT name, symbol FROM crc20 WHERE token_id = ?1 LIMIT 1"#)?;
if let Ok(row) = crc.query_row([token_id], |r| {
if let Ok(row) = crc.query_row([&token_blob], |r| {
Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?))
}) {
let (name, sym) = row;

21
src/db/cauldron/tx.rs
View file

@ -4,14 +4,15 @@
// A copy of the license can be found in the LICENSE file or at https://www.gnu.org/licenses/agpl-3.0.html
use anyhow::Result;
use bitcoin_hashes::hex::{FromHex, ToHex};
use bitcoincash::{BlockHash, TokenID, Txid};
use rusqlite::{params, Connection};
use crate::db::blob::{FromBlob, ToBlob};
pub fn create_table(conn: &Connection) {
let tbl = "CREATE TABLE tx (
txid TEXT PRIMARY KEY,
blockhash TEXT,
txid BLOB PRIMARY KEY,
blockhash BLOB,
mtp_timestamp BIGINT,
first_seen_timestamp BIGINT,
effective_timestamp BIGINT GENERATED ALWAYS AS (COALESCE(first_seen_timestamp, mtp_timestamp))
@ -38,7 +39,7 @@ pub fn insert_block_tx(
ON CONFLICT(txid) DO UPDATE SET
blockhash = excluded.blockhash,
mtp_timestamp = excluded.mtp_timestamp";
conn.execute(sql, params![&txid.to_hex(), &blockhash.to_hex(), &mtp])?;
conn.execute(sql, params![&txid.to_blob(), &blockhash.to_blob(), &mtp])?;
Ok(())
}
@ -52,7 +53,7 @@ pub fn insert_mempool_tx(
ON CONFLICT(txid) DO UPDATE SET
first_seen_timestamp = excluded.first_seen_timestamp";
conn.execute(sql, params![&txid.to_hex(), first_seen_timestamp])?;
conn.execute(sql, params![&txid.to_blob(), first_seen_timestamp])?;
Ok(())
}
@ -70,7 +71,7 @@ pub fn latest(
WHERE utxo_funding.token_id = ?
ORDER BY tx.effective_timestamp DESC
LIMIT ? OFFSET ?",
params![tid.to_hex(), limit, offset],
params![tid.to_blob(), limit, offset],
),
None => (
"SELECT tx.txid, tx.blockhash, tx.mtp_timestamp, tx.first_seen_timestamp
@ -83,8 +84,8 @@ pub fn latest(
let mut stmt = conn.prepare(sql)?;
let tx_iter = stmt.query_map(params, |row| {
let txid_hex: String = row.get(0)?;
let blockhash_hex: Option<String> = row.get(1)?;
let txid_blob: Vec<u8> = row.get(0)?;
let blockhash_blob: Option<Vec<u8>> = row.get(1)?;
let mtp_timestamp: Option<u64> = row.get(2).ok(); // Handle potential NULL
let first_seen_timestamp: Option<u64> = row.get(3).ok(); // Handle potential NULL
@ -98,8 +99,8 @@ pub fn latest(
})?;
Ok((
Txid::from_hex(&txid_hex).expect("Invalid Txid hex"),
blockhash_hex.map(|hex| BlockHash::from_hex(&hex).expect("Invalid BlockHash hex")),
Txid::from_blob(&txid_blob).expect("Invalid Txid blob"),
blockhash_blob.map(|blob| BlockHash::from_blob(&blob).expect("Invalid BlockHash blob")),
timestamp,
))
})?;

14
src/db/cauldron/user.rs
View file

@ -8,16 +8,18 @@
use std::collections::HashSet;
use anyhow::Result;
use bitcoin_hashes::{hex::ToHex, Hash};
use bitcoin_hashes::Hash;
use bitcoincash::{PubkeyHash, Transaction};
use riftenlabs_defi::cauldron::ParsedContract;
use rusqlite::{params, Connection};
use crate::db::blob::ToBlob;
pub fn create_table(conn: &Connection) {
conn.execute(
"CREATE TABLE user_action (
spent_utxo_hash TEXT NOT NULL,
user TEXT NOT NULL,
spent_utxo_hash BLOB NOT NULL,
user BLOB NOT NULL,
address_type TEXT NOT NULL,
FOREIGN KEY(spent_utxo_hash) REFERENCES utxo_spending(spent_utxo_hash) ON DELETE CASCADE,
PRIMARY KEY (spent_utxo_hash, user)
@ -59,7 +61,11 @@ pub fn insert_user_action(
for c in cauldrons {
for user in &p2pkh_output_hashes {
statement.execute(params![c.spent_utxo_hash.to_hex(), user.to_hex(), "p2pkh"])?;
statement.execute(params![
c.spent_utxo_hash.to_blob(),
user.to_blob(),
"p2pkh"
])?;
}
}
Ok(())

23
src/db/cauldron/utxo_funding.rs
View file

@ -4,22 +4,23 @@
// 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::Txid;
use riftenlabs_defi::cauldron::ParsedContract;
use rusqlite::{params, Connection};
use crate::db::blob::ToBlob;
pub fn create_table(conn: &Connection) {
conn.execute(
"CREATE TABLE utxo_funding (
new_utxo_hash TEXT PRIMARY KEY,
txid TEXT,
spent_utxo_hash TEXT,
new_utxo_txid TEXT,
new_utxo_hash BLOB PRIMARY KEY,
txid BLOB,
spent_utxo_hash BLOB,
new_utxo_txid BLOB,
new_utxo_n INT,
sats BIGINT,
token_amount BIGINT,
token_id TEXT,
token_id BLOB,
FOREIGN KEY(txid) REFERENCES tx(txid) ON DELETE CASCADE
)",
[],
@ -46,14 +47,14 @@ pub fn insert_utxo_funding(
continue;
}
statement.execute(params![
c.new_utxo_hash.unwrap().to_hex(),
txid.to_hex(),
c.spent_utxo_hash.to_hex(),
c.new_utxo_txid.unwrap().to_hex(),
c.new_utxo_hash.unwrap().to_blob(),
txid.to_blob(),
c.spent_utxo_hash.to_blob(),
c.new_utxo_txid.unwrap().to_blob(),
c.new_utxo_n.unwrap(),
c.sats.unwrap(),
c.token_amount.unwrap(),
c.token_id.unwrap().to_hex(),
c.token_id.unwrap().to_blob(),
])?;
}
Ok(())

9
src/db/cauldron/utxo_spending.rs
View file

@ -4,16 +4,17 @@
// 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::Txid;
use riftenlabs_defi::cauldron::ParsedContract;
use rusqlite::{params, Connection};
use crate::db::blob::ToBlob;
pub fn create_table(conn: &Connection) {
conn.execute(
"CREATE TABLE utxo_spending (
spent_utxo_hash TEXT PRIMARY KEY,
txid TEXT,
spent_utxo_hash BLOB PRIMARY KEY,
txid BLOB,
FOREIGN KEY(txid) REFERENCES tx(txid) ON DELETE CASCADE
)",
[],
@ -33,7 +34,7 @@ pub fn insert_utxo_spending(
))?;
for c in cauldrons {
statement.execute(params![c.spent_utxo_hash.to_hex(), txid.to_hex(),])?;
statement.execute(params![c.spent_utxo_hash.to_blob(), txid.to_blob(),])?;
}
Ok(())
}

41
src/db/crc20/mod.rs
View file

@ -8,6 +8,8 @@ use bitcoin_hashes::hex::ToHex;
use bitcoincash::TokenID;
use rusqlite::{params, Connection};
use crate::db::blob::ToBlob;
const STATE_NOT_INDEXED: i32 = -1;
const STATE_NOT_CRC20: i32 = 0;
const STATE_IS_CRC20: i32 = 1;
@ -17,7 +19,7 @@ const MAX_FAILED_ATTEMPTS: i32 = 20;
pub fn prepare_tables(conn: &Connection) {
conn.execute(
"CREATE TABLE crc20 (
token_id TEXT PRIMARY KEY,
token_id BLOB PRIMARY KEY,
symbol TEXT NOT NULL,
name TEXT NOT NULL,
decimals INT NOT NULL
@ -28,7 +30,7 @@ pub fn prepare_tables(conn: &Connection) {
conn.execute(
"CREATE TABLE crc20_candidates (
token_id TEXT PRIMARY KEY,
token_id BLOB PRIMARY KEY,
is_crc20 INT NOT NULL,
failed_attempts INT NOT NULL
)",
@ -41,7 +43,7 @@ pub fn insert_crc20_candidate(conn: &Connection, token_id: &TokenID) -> Result<(
conn.execute(
"INSERT OR IGNORE INTO crc20_candidates (token_id, is_crc20, failed_attempts)
VALUES (?1, ?2, ?3)",
params![token_id.to_hex(), STATE_NOT_INDEXED, 0],
params![token_id.to_blob(), STATE_NOT_INDEXED, 0],
)
.map_err(|e| {
anyhow::anyhow!(
@ -56,7 +58,7 @@ pub fn insert_crc20_candidate(conn: &Connection, token_id: &TokenID) -> Result<(
pub fn get_not_indexed_tokens(conn: &Connection) -> Result<Vec<String>> {
let mut stmt = conn.prepare(
"SELECT token_id FROM crc20_candidates WHERE is_crc20 = ?1 AND failed_attempts <= ?2",
"SELECT hex(token_id) FROM crc20_candidates WHERE is_crc20 = ?1 AND failed_attempts <= ?2",
)?;
let token_ids = stmt
.query_map(params![STATE_NOT_INDEXED, MAX_FAILED_ATTEMPTS], |row| {
@ -67,15 +69,18 @@ pub fn get_not_indexed_tokens(conn: &Connection) -> Result<Vec<String>> {
Ok(token_ids)
}
pub fn update_to_not_crc20(conn: &Connection, token: &str) -> Result<()> {
pub fn update_to_not_crc20(conn: &Connection, token_hex: &str) -> Result<()> {
let token_blob = hex::decode(token_hex)
.map_err(|e| anyhow::anyhow!("invalid token hex {}: {:?}", token_hex, e))?;
conn.execute(
"UPDATE crc20_candidates SET is_crc20 = ?1 WHERE token_id = ?2",
params![STATE_NOT_CRC20, token],
params![STATE_NOT_CRC20, token_blob],
)
.map_err(|e| {
anyhow::anyhow!(
"failed to update token_id = {} to STATE_NOT_CRC20. Original error: {:?}",
token,
token_hex,
e
)
})?;
@ -85,20 +90,23 @@ pub fn update_to_not_crc20(conn: &Connection, token: &str) -> Result<()> {
pub fn update_to_crc20(
conn: &Connection,
token_id: &str,
token_hex: &str,
symbol: &str,
name: &str,
decimals: i32,
) -> Result<()> {
let token_blob = hex::decode(token_hex)
.map_err(|e| anyhow::anyhow!("invalid token hex {}: {:?}", token_hex, e))?;
// Update is_crc20 in crc20_candidates table
conn.execute(
"UPDATE crc20_candidates SET is_crc20 = ?1 WHERE token_id = ?2",
params![STATE_IS_CRC20, token_id],
params![STATE_IS_CRC20, &token_blob],
)
.map_err(|e| {
anyhow::anyhow!(
"failed to update token_id = {} to STATE_IS_CRC20 in crc20_candidates. Original error: {:?}",
token_id,
token_hex,
e
)
})?;
@ -107,12 +115,12 @@ pub fn update_to_crc20(
conn.execute(
"INSERT OR REPLACE INTO crc20 (token_id, symbol, name, decimals)
VALUES (?1, ?2, ?3, ?4)",
params![token_id, symbol, name, decimals],
params![&token_blob, symbol, name, decimals],
)
.map_err(|e| {
anyhow::anyhow!(
"failed to insert or replace token_id = {} in crc20. Original error: {:?}",
token_id,
token_hex,
e
)
})?;
@ -120,15 +128,18 @@ pub fn update_to_crc20(
Ok(())
}
pub fn bump_failed_attempts(conn: &Connection, token_id: &str) -> Result<()> {
pub fn bump_failed_attempts(conn: &Connection, token_hex: &str) -> Result<()> {
let token_blob = hex::decode(token_hex)
.map_err(|e| anyhow::anyhow!("invalid token hex {}: {:?}", token_hex, e))?;
conn.execute(
"UPDATE crc20_candidates SET failed_attempts = failed_attempts + 1 WHERE token_id = ?1",
params![token_id],
params![token_blob],
)
.map_err(|e| {
anyhow::anyhow!(
"Failed to bump failed_attempts for token_id = {}. Original error: {:?}",
token_id,
token_hex,
e
)
})?;

1
src/db/mod.rs
View file

@ -6,6 +6,7 @@
use std::sync::Arc;
pub mod bcmr;
pub mod blob;
pub mod cauldron;
pub mod crc20;
pub mod init;

96
src/db/oracle/mod.rs
View file

@ -10,6 +10,8 @@ use log::debug;
use riftenlabs_defi::delphi::parse_delphi_update;
use rusqlite::{params, Connection, Row};
use crate::db::blob::ToBlob;
#[derive(Debug, Clone, serde::Serialize)]
pub struct DelphiEntry {
pub txid: String,
@ -22,10 +24,14 @@ pub struct DelphiEntry {
impl DelphiEntry {
fn from_row(row: &Row) -> Result<Self> {
// hex() returns uppercase, convert to lowercase for consistency
let txid: String = row.get(0)?;
let token_id: String = row.get(1)?;
let blockhash: String = row.get(2)?;
Ok(Self {
txid: row.get(0)?,
token_id: row.get(1)?,
blockhash: row.get(2)?,
txid: txid.to_lowercase(),
token_id: token_id.to_lowercase(),
blockhash: blockhash.to_lowercase(),
oracle_timestamp: row.get(3)?,
oracle_price: row.get(4)?,
oracle_sequence: row.get(5)?,
@ -36,9 +42,9 @@ impl DelphiEntry {
pub fn prepare_tables(conn: &Connection) {
conn.execute(
"CREATE TABLE delphi_entry (
txid TEXT PRIMARY KEY,
token_id TEXT NOT NULL,
blockhash TEXT NOT NULL,
txid BLOB PRIMARY KEY,
token_id BLOB NOT NULL,
blockhash BLOB NOT NULL,
oracle_timestamp BIGINT NOT NULL,
oracle_price BIGINT NOT NULL,
oracle_sequence BIGINT NOT NULL
@ -56,13 +62,20 @@ pub fn prepare_tables(conn: &Connection) {
}
pub fn insert_delphi_entry(conn: &Connection, entry: &DelphiEntry) -> Result<()> {
let txid_blob =
hex::decode(&entry.txid).map_err(|e| anyhow::anyhow!("invalid txid hex: {}", e))?;
let token_id_blob =
hex::decode(&entry.token_id).map_err(|e| anyhow::anyhow!("invalid token_id hex: {}", e))?;
let blockhash_blob = hex::decode(&entry.blockhash)
.map_err(|e| anyhow::anyhow!("invalid blockhash hex: {}", e))?;
conn.execute(
"INSERT OR REPLACE INTO delphi_entry (txid, token_id, blockhash, oracle_timestamp, oracle_price, oracle_sequence)
VALUES (?, ?, ?, ?, ?, ?)",
params![
entry.txid,
entry.token_id,
entry.blockhash,
txid_blob,
token_id_blob,
blockhash_blob,
entry.oracle_timestamp,
entry.oracle_price,
entry.oracle_sequence
@ -77,7 +90,7 @@ pub fn delete_entries_for_block(conn: &Connection, blockhash: &BlockHash) -> Res
let rows_deleted = conn
.execute(
"DELETE FROM delphi_entry WHERE blockhash = ?",
params![blockhash.to_hex()],
params![blockhash.to_blob()],
)
.map_err(|e| {
anyhow::anyhow!(
@ -114,7 +127,7 @@ pub fn get_closest(
timestamp: i64,
) -> Result<Option<DelphiEntry>> {
let mut stmt = conn.prepare(
"SELECT txid, token_id,blockhash, oracle_timestamp, oracle_price, oracle_sequence
"SELECT hex(txid), hex(token_id), hex(blockhash), oracle_timestamp, oracle_price, oracle_sequence
FROM delphi_entry
WHERE oracle_timestamp <= ?
AND (? IS NULL OR token_id = ?)
@ -124,8 +137,8 @@ pub fn get_closest(
let mut rows = stmt.query(params![
timestamp,
token_id.as_ref().map(|t| t.to_string()),
token_id.as_ref().map(|t| t.to_string())
token_id.as_ref().map(|t| t.to_blob()),
token_id.as_ref().map(|t| t.to_blob())
])?;
if let Some(row) = rows.next()? {
Ok(Some(DelphiEntry::from_row(row)?))
@ -141,7 +154,7 @@ pub fn get_range(
end_timestamp: i64,
) -> Result<Vec<DelphiEntry>> {
let mut stmt = conn.prepare(
"SELECT txid, token_id, blockhash, oracle_timestamp, oracle_price, oracle_sequence
"SELECT hex(txid), hex(token_id), hex(blockhash), oracle_timestamp, oracle_price, oracle_sequence
FROM delphi_entry
WHERE oracle_timestamp >= ? AND oracle_timestamp <= ?
AND (? IS NULL OR token_id = ?)
@ -151,8 +164,8 @@ pub fn get_range(
let mut rows = stmt.query(params![
start_timestamp,
end_timestamp,
token_id.as_ref().map(|t| t.to_string()),
token_id.as_ref().map(|t| t.to_string())
token_id.as_ref().map(|t| t.to_blob()),
token_id.as_ref().map(|t| t.to_blob())
])?;
let mut entries = Vec::new();
@ -165,7 +178,7 @@ pub fn get_range(
pub fn has_entry(conn: &rusqlite::Connection, txid: &Txid) -> Result<bool> {
let mut stmt = conn.prepare("SELECT 1 FROM delphi_entry WHERE txid = ?")?;
let exists = stmt.query_row(params![txid.to_hex()], |_| Ok(())) == Ok(());
let exists = stmt.query_row(params![txid.to_blob()], |_| Ok(())) == Ok(());
Ok(exists)
}
@ -173,7 +186,7 @@ pub fn clear_mempool(conn: &Connection) -> Result<usize> {
let rows_deleted = conn
.execute(
"DELETE FROM delphi_entry WHERE blockhash = ?",
params![BlockHash::all_zeros().to_hex()],
params![BlockHash::all_zeros().to_blob()],
)
.map_err(|e| anyhow::anyhow!("failed to clear mempool entries: {}", e))?;
@ -199,7 +212,7 @@ pub fn get_range_with_step(
}
let mut stmt = conn.prepare(
"SELECT txid, token_id, blockhash, oracle_timestamp, oracle_price, oracle_sequence
"SELECT hex(txid), hex(token_id), hex(blockhash), oracle_timestamp, oracle_price, oracle_sequence
FROM delphi_entry
WHERE oracle_timestamp BETWEEN ? AND ?
AND (? IS NULL OR token_id = ?)
@ -209,8 +222,8 @@ pub fn get_range_with_step(
let mut rows = stmt.query(params![
timestamp_start,
timestamp_end,
token_id.as_ref().map(|t| t.to_string()),
token_id.as_ref().map(|t| t.to_string()),
token_id.as_ref().map(|t| t.to_blob()),
token_id.as_ref().map(|t| t.to_blob()),
])?;
let mut all_entries = Vec::new();
@ -262,39 +275,49 @@ mod tests {
}
fn insert_test_entries(conn: &Connection, token_id: &TokenID, base_ts: i64) {
use bitcoin_hashes::hex::ToHex;
use bitcoin_hashes::Hash;
use bitcoincash::Txid;
let blockhash = BlockHash::hash(b"block");
// Generate proper 32-byte txid hashes
let txid1 = Txid::hash(b"tx1").to_hex();
let txid2 = Txid::hash(b"tx2").to_hex();
let txid3 = Txid::hash(b"tx3").to_hex();
let txid4 = Txid::hash(b"tx4").to_hex();
let entries = vec![
DelphiEntry {
txid: "tx1".to_string(),
txid: txid1.clone(),
blockhash: blockhash.to_hex(),
oracle_timestamp: base_ts - 200,
oracle_price: 1000,
oracle_sequence: 1,
token_id: token_id.to_string(),
token_id: token_id.to_hex(),
},
DelphiEntry {
txid: "tx2".to_string(),
txid: txid2.clone(),
blockhash: blockhash.to_hex(),
oracle_timestamp: base_ts - 150,
oracle_price: 1100,
oracle_sequence: 2,
token_id: token_id.to_string(),
token_id: token_id.to_hex(),
},
DelphiEntry {
txid: "tx3".to_string(),
txid: txid3.clone(),
blockhash: blockhash.to_hex(),
oracle_timestamp: base_ts - 100,
oracle_price: 1200,
oracle_sequence: 3,
token_id: token_id.to_string(),
token_id: token_id.to_hex(),
},
DelphiEntry {
txid: "tx4".to_string(),
txid: txid4.clone(),
blockhash: blockhash.to_hex(),
oracle_timestamp: base_ts - 50,
oracle_price: 1300,
oracle_sequence: 4,
token_id: token_id.to_string(),
token_id: token_id.to_hex(),
},
];
@ -305,25 +328,34 @@ mod tests {
#[test]
fn test_get_range_with_step_downsamples_correctly() {
use bitcoin_hashes::hex::ToHex;
use bitcoin_hashes::Hash;
use bitcoincash::Txid;
let conn = setup_test_db();
let token_id = TokenID::all_zeros();
let now = 1_720_000_000;
insert_test_entries(&conn, &token_id, now);
// Expected txids (matching insert_test_entries)
let txid1 = Txid::hash(b"tx1").to_hex();
let txid3 = Txid::hash(b"tx3").to_hex();
let txid4 = Txid::hash(b"tx4").to_hex();
let result = get_range_with_step(&conn, &Some(token_id), now - 250, now, 100)
.expect("query should succeed");
assert_eq!(result.len(), 3, "should return 3 buckets");
assert_eq!(
result[0].txid, "tx1",
result[0].txid, txid1,
"bucket 0 should be latest in its range"
);
assert_eq!(
result[1].txid, "tx3",
result[1].txid, txid3,
"bucket 1 should be latest in its range (tx2 overwritten)"
);
assert_eq!(result[2].txid, "tx4", "bucket 2 should contain final entry");
assert_eq!(result[2].txid, txid4, "bucket 2 should contain final entry");
}
#[test]

89
src/db/search/mod.rs
View file

@ -24,7 +24,7 @@ fn search_bcmr(bcmr_conn: &Connection, search_query: &str) -> Result<Vec<TokenBa
let is_full_hex_token_id = TokenID::from_hex(search_query).is_ok();
// Fetch the latest (highest) record that has BCMR data.
let sql = if is_full_hex_token_id {
"SELECT token_id, name, symbol
"SELECT hex(token_id), name, symbol
FROM (
SELECT a.token_id, b.name, b.symbol, a.height,
ROW_NUMBER() OVER (PARTITION BY a.token_id ORDER BY a.height DESC) AS rn
@ -34,7 +34,7 @@ fn search_bcmr(bcmr_conn: &Connection, search_query: &str) -> Result<Vec<TokenBa
WHERE rn = 1
AND (token_id = ?1);"
} else {
"SELECT token_id, name, symbol
"SELECT hex(token_id), name, symbol
FROM (
SELECT a.token_id, b.name, b.symbol, a.height,
ROW_NUMBER() OVER (PARTITION BY a.token_id ORDER BY a.height DESC) AS rn
@ -47,20 +47,26 @@ fn search_bcmr(bcmr_conn: &Connection, search_query: &str) -> Result<Vec<TokenBa
};
let mut statement = bcmr_conn.prepare(sql)?;
let search_pattern = if is_full_hex_token_id {
search_query.to_string()
} else {
format!("%{search_query}%")
};
let mut bcmr_data: Vec<TokenBasicInfo> = Vec::new();
let mut query = statement.query([search_pattern.as_str()])?;
while let Some(result) = query.next()? {
let token_id: String = result.get(0)?;
let name: Option<String> = result.get(1)?;
let ticker: Option<String> = result.get(2)?;
bcmr_data.push((token_id, name, ticker));
if is_full_hex_token_id {
let token_blob = hex::decode(search_query).context("invalid token hex")?;
let mut query = statement.query([&token_blob as &dyn rusqlite::ToSql])?;
while let Some(result) = query.next()? {
let token_id: String = result.get(0)?;
let name: Option<String> = result.get(1)?;
let ticker: Option<String> = result.get(2)?;
bcmr_data.push((token_id.to_lowercase(), name, ticker));
}
} else {
let search_pattern = format!("%{search_query}%");
let mut query = statement.query([&search_pattern as &dyn rusqlite::ToSql])?;
while let Some(result) = query.next()? {
let token_id: String = result.get(0)?;
let name: Option<String> = result.get(1)?;
let ticker: Option<String> = result.get(2)?;
bcmr_data.push((token_id.to_lowercase(), name, ticker));
}
}
Ok(bcmr_data)
@ -72,33 +78,39 @@ fn search_crc20(crc20_conn: &Connection, search_query: &str) -> Result<Vec<Token
let sql = if is_full_hex_token_id {
// Search by token_id if it's a full hex token ID
"
SELECT token_id, name, symbol
SELECT hex(token_id), name, symbol
FROM crc20
WHERE token_id = ?1;
"
} else {
// Only search by name or symbol otherwise
"
SELECT token_id, name, symbol
SELECT hex(token_id), name, symbol
FROM crc20
WHERE name LIKE ?1 OR symbol LIKE ?1;
"
};
let mut statement = crc20_conn.prepare(sql)?;
let search_pattern = if is_full_hex_token_id {
search_query.to_string()
} else {
format!("%{search_query}%")
};
let mut crc20_data: Vec<TokenBasicInfo> = Vec::new();
let mut query = statement.query([&search_pattern])?;
while let Some(result) = query.next()? {
let token_id: String = result.get(0)?;
let name: Option<String> = result.get(1)?;
let ticker: Option<String> = result.get(2)?;
crc20_data.push((token_id, name, ticker));
if is_full_hex_token_id {
let token_blob = hex::decode(search_query).context("invalid token hex")?;
let mut query = statement.query([&token_blob as &dyn rusqlite::ToSql])?;
while let Some(result) = query.next()? {
let token_id: String = result.get(0)?;
let name: Option<String> = result.get(1)?;
let ticker: Option<String> = result.get(2)?;
crc20_data.push((token_id.to_lowercase(), name, ticker));
}
} else {
let search_pattern = format!("%{search_query}%");
let mut query = statement.query([&search_pattern as &dyn rusqlite::ToSql])?;
while let Some(result) = query.next()? {
let token_id: String = result.get(0)?;
let name: Option<String> = result.get(1)?;
let ticker: Option<String> = result.get(2)?;
crc20_data.push((token_id.to_lowercase(), name, ticker));
}
}
Ok(crc20_data)
@ -129,15 +141,15 @@ fn token_volume(
AND p.token_id = ?
)
SELECT
token_id,
hex(token_id),
COALESCE(SUM(trade_volume), 0) as total_trade_volume
FROM TradeData;
",
)
.context("Failed to prepare statement")?;
let token_blob = hex::decode(&token_id).context("invalid token hex")?;
let trade_volume: u64 = statement
.query_row(params![&token_id], |row| row.get(1))
.query_row(params![&token_blob], |row| row.get(1))
.unwrap_or(0);
// Return result as Ok tuple for successful case
@ -231,7 +243,7 @@ pub fn db_search_tokens_cached(
let sql = format!(
"SELECT
token_id,
hex(token_id) as token_id,
trade_volume,
tvl_sats,
tvl_tokens,
@ -260,10 +272,19 @@ pub fn db_search_tokens_cached(
let limit_i64 = limit as i64;
let offset_i64 = offset as i64;
let id_blob: Option<Vec<u8>> = match &filter {
Filter::Id(s) => Some(hex::decode(s).context("invalid token hex")?),
_ => None,
};
let mut params_vec: Vec<(&str, &dyn ToSql)> = Vec::with_capacity(3);
match &filter {
Filter::Id(s) => params_vec.push((":id", s as &dyn ToSql)),
Filter::Id(_) => {
if let Some(ref blob) = id_blob {
params_vec.push((":id", blob as &dyn ToSql));
}
}
Filter::Pat(s) => params_vec.push((":pat", s as &dyn ToSql)),
Filter::None => {}
}
@ -274,7 +295,7 @@ pub fn db_search_tokens_cached(
let mut out = Vec::with_capacity(limit);
while let Some(row) = rows.next()? {
let token_id: String = row.get("token_id")?;
let token_id: String = row.get::<_, String>("token_id")?.to_lowercase();
let trade_volume: u64 = row.get("trade_volume")?;
let tvl_sats: u64 = row.get::<_, i64>("tvl_sats")? as u64;
let tvl_tokens: u64 = row.get::<_, i64>("tvl_tokens")? as u64;

18
src/db/search/test.rs
View file

@ -9,6 +9,7 @@ mod tests {
use crate::db::bcmr::{
insert_authheader, insert_bcmr_data, prepare_tables as bcmr_prepare_tables,
};
use crate::db::blob::ToBlob;
use crate::db::cauldron::pool::{dummy_init_seq, insert_new_pool, insert_pool_history_entry};
use crate::db::cauldron::prepare_tables as cauldron_prepare_tables;
use crate::db::cauldron::tokenlist::db_utils::{create_cached_token_metrics_table, CachedSort};
@ -24,6 +25,9 @@ mod tests {
use riftenlabs_defi::cauldron::ParsedContract;
use riftenlabs_defi::chainutil::OutPointHash;
use rusqlite::{params, Connection};
fn hex_to_blob(hex: &str) -> Vec<u8> {
hex::decode(hex).expect("valid hex string")
}
fn setup_mock_db(conn: &Connection) {
cauldron_prepare_tables(conn);
@ -72,7 +76,7 @@ mod tests {
?18, strftime('%s','now'))
"#,
params![
token_hex,
hex_to_blob(token_hex),
trade_volume,
tvl_sats,
tvl_tokens,
@ -667,7 +671,7 @@ mod tests {
conn.execute(
"INSERT INTO crc20 (token_id, name, symbol, decimals) VALUES (?, ?, ?, ?)",
params![&token_id4.to_hex(), "TokenFour", "TFOUR", 18],
params![token_id4.to_blob(), "TokenFour", "TFOUR", 18],
)?;
let txid4_init = Txid::from_inner([0xf4; 32]);
@ -742,8 +746,10 @@ mod tests {
let w = mock.cauldron_w.get().unwrap();
let bcmr = mock.bcmr_r.get().unwrap();
let t_named = TokenID::from_inner([0x62; 32]).to_hex();
let t_null = TokenID::from_inner([0x63; 32]).to_hex();
let tok_named = TokenID::from_inner([0x62; 32]);
let tok_null = TokenID::from_inner([0x63; 32]);
let t_named = tok_named.to_hex();
let t_null = tok_null.to_hex();
// Named row
w.execute(
@ -754,7 +760,7 @@ mod tests {
VALUES (?1, 0, 1000, 500, 0,
?2, ?3,
1.0, 0.5, strftime('%s','now'))"#,
rusqlite::params![t_named, "Named", "NMD"],
rusqlite::params![tok_named.to_blob(), "Named", "NMD"],
)
.unwrap();
@ -767,7 +773,7 @@ mod tests {
VALUES (?1, 0, 1000, 500, 0,
NULL, NULL,
1.0, 0.5, strftime('%s','now'))"#,
rusqlite::params![t_null],
rusqlite::params![tok_null.to_blob()],
)
.unwrap();

6
src/rpc/candlesticks.rs
View file

@ -7,7 +7,7 @@ use crate::db::DB;
use crate::rpc::err::{bad_request, db_error, ApiErrorCode, CachedApiResult};
use crate::rpc::response::{cached_ok, CACHE_IMMUTABLE, CACHE_NONE};
use crate::timeutil::time_now;
use anyhow::{bail, Result};
use anyhow::{bail, Context, Result};
use rocket::{get, State};
use rusqlite::{params, Connection};
use serde::Serialize;
@ -139,9 +139,9 @@ SELECT
FROM tx_trades
ORDER BY effective_timestamp ASC;
"#;
let token_blob = hex::decode(token_id).context("invalid token hex")?;
let mut statement = connection.prepare(sql)?;
let mut rows = statement.query(params![token_id, timestamp_start, timestamp_end])?;
let mut rows = statement.query(params![token_blob, timestamp_start, timestamp_end])?;
let mut all_trades = Vec::new();
while let Some(row) = rows.next()? {

10
src/rpc/contract.rs
View file

@ -3,7 +3,7 @@
// 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 anyhow::{Context, Result};
use rayon::prelude::*;
use rocket::{get, State};
use rusqlite::Connection;
@ -49,21 +49,23 @@ fn db_contract_count_all(db: &Connection) -> Result<ContractCount> {
}
fn db_contract_count_by_token(db: &Connection, token_id: &str) -> Result<ContractCount> {
let token_blob = hex::decode(token_id).context("invalid token hex")?;
let active: u64 = db.query_row(
"SELECT COUNT(*) FROM pool WHERE withdrawn_in_utxo IS NULL AND token_id = ?",
[token_id],
[&token_blob],
|row| row.get(0),
)?;
let ended: u64 = db.query_row(
"SELECT COUNT(*) FROM pool WHERE withdrawn_in_utxo IS NOT NULL AND token_id = ?",
[token_id],
[&token_blob],
|row| row.get(0),
)?;
let interactions: u64 = db.query_row(
"SELECT COUNT(*) FROM pool_history_entry phe JOIN pool p ON phe.pool = p.creation_utxo WHERE p.token_id = ?",
[token_id],
[&token_blob],
|row| row.get(0),
)?;

8
src/rpc/pool.rs
View file

@ -193,18 +193,18 @@ impl PoolVisitor for ActivePoolList {
}
fn visit(&mut self, sats: u64, tokens: u64, optional_fields: OptionalFields) -> Result<bool> {
let owner_pkh = optional_fields.owner.unwrap();
let owner_pkh = optional_fields.owner.unwrap().to_lowercase();
let owner_p2pkh_addr = p2pkh_hex_to_addr(&owner_pkh)?;
self.active.push(ActivePool {
owner_pkh,
owner_p2pkh_addr,
token_id: optional_fields.token_id.unwrap(),
token_id: optional_fields.token_id.unwrap().to_lowercase(),
sats,
tokens,
txid: optional_fields.txid.unwrap(),
txid: optional_fields.txid.unwrap().to_lowercase(),
tx_pos: optional_fields.tx_pos.unwrap(),
pool_id: optional_fields.pool_id.unwrap(),
pool_id: optional_fields.pool_id.unwrap().to_lowercase(),
});
Ok(true)

3
src/rpc/price.rs
View file

@ -160,9 +160,10 @@ fn historic_price(
ORDER BY
effective_timestamp ASC
";
let token_blob = hex::decode(token_id).context("invalid token hex")?;
let mut statement = connection.prepare(sql)?;
let mut rows = statement.query(params![token_id, timestamp_start, timestamp_end])?;
let mut rows = statement.query(params![token_blob, timestamp_start, timestamp_end])?;
let mut result: Vec<(i64, f64, f64, f64)> = Vec::with_capacity(total_intervals as usize);

2
src/rpc/tvl.rs
View file

@ -37,7 +37,7 @@ impl PoolVisitor for TvlByTokenVisitor {
}
fn visit(&mut self, sats: u64, tokens: u64, optional_fields: OptionalFields) -> Result<bool> {
let token_id = optional_fields.token_id.unwrap();
let token_id = optional_fields.token_id.unwrap().to_lowercase();
let entry = self.tvl.entry(token_id).or_insert((0u64, 0u64));
entry.0 += sats;