From ec49bf765b2c44018818eb50685e1fb27b0ae18f Mon Sep 17 00:00:00 2001 From: Dagur Valberg Johannsson Date: Mon, 18 Aug 2025 22:30:10 +0200 Subject: [PATCH] Improve pool visitor queries; make price a visitor This adds a big performance boost to pool visitor queries, also moving token price queries to use the visitor model. --- linters/timestamp_coalesce_check.py | 72 +++++++ src/db/cauldron/config.rs | 2 +- src/db/cauldron/mod.rs | 5 +- src/db/cauldron/pool.rs | 36 +++- src/db/cauldron/poolvisitor.rs | 143 +++++++++++--- src/db/cauldron/tokenlist.rs | 9 +- src/db/cauldron/tx.rs | 13 +- src/db/cauldron/utxo_spending.rs | 4 +- src/db/search/mod.rs | 35 ++-- src/rpc/candlesticks.rs | 4 +- src/rpc/mod.rs | 4 +- src/rpc/pool.rs | 11 +- src/rpc/price.rs | 289 ++++++++++++++++------------ src/rpc/tvl.rs | 15 +- src/utiltest.rs | 25 --- 15 files changed, 435 insertions(+), 232 deletions(-) create mode 100755 linters/timestamp_coalesce_check.py diff --git a/linters/timestamp_coalesce_check.py b/linters/timestamp_coalesce_check.py new file mode 100755 index 0000000..1d08262 --- /dev/null +++ b/linters/timestamp_coalesce_check.py @@ -0,0 +1,72 @@ +#!/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 + +import os +import sys +import re + +""" +Disallow COALESCE with first_seen_timestamp and mtp_timestamp on the same line in SQL queries. +These combinations should use effective_timestamp instead. +Note: Database schema definitions using GENERATED ALWAYS AS are allowed. +Note: COALESCE aliased as effective_timestamp is allowed (this is the correct pattern). +""" + +OUR_PATH = os.path.dirname(os.path.realpath(__file__)) + +def check_file_for_timestamp_coalesce(file_path): + """Check if the file contains disallowed timestamp combinations.""" + with open(file_path, 'r', encoding='utf-8') as file: + for line_num, line in enumerate(file, 1): + # Check if line contains COALESCE and both timestamp fields + if (re.search(r'\bCOALESCE\b', line, re.IGNORECASE) and + re.search(r'\bfirst_seen_timestamp\b', line) and + re.search(r'\bmtp_timestamp\b', line)): + + # Skip database schema definitions (GENERATED ALWAYS AS) + if re.search(r'GENERATED\s+ALWAYS\s+AS', line, re.IGNORECASE): + continue + + # Skip comments that are just documenting the pattern + if line.strip().startswith('//') and 'COALESCE' in line: + continue + + # Allow COALESCE aliased as effective_timestamp (this is the correct pattern) + if re.search(r'AS\s+effective_timestamp', line, re.IGNORECASE): + continue + + print(f"{file_path}:{line_num}: Disallowed COALESCE with first_seen_timestamp and mtp_timestamp") + print(f" Line: {line.strip()}") + print(f" Suggestion: Use effective_timestamp instead") + return False + + return True + +def check_files(directories): + """ + Traverse the specified directories and check each .rs file. + """ + found_forbidden_combinations = False + + for directory in directories: + for root, dirs, files in os.walk(os.path.join(OUR_PATH, "..", directory)): + for file in files: + if file.endswith('.rs'): + file_path = os.path.join(root, file) + if not check_file_for_timestamp_coalesce(file_path): + found_forbidden_combinations = True + + if found_forbidden_combinations: + print("\nForbidden timestamp combinations found.") + print("Please use effective_timestamp instead of COALESCE with first_seen_timestamp and mtp_timestamp.") + print("Note: Database schema definitions using GENERATED ALWAYS AS are allowed.") + print("Note: COALESCE aliased as effective_timestamp is allowed (this is the correct pattern).") + sys.exit(1) + else: + print("OK") + sys.exit(0) + +if __name__ == "__main__": + check_files(['src', 'contrib']) diff --git a/src/db/cauldron/config.rs b/src/db/cauldron/config.rs index 4664533..14784b3 100644 --- a/src/db/cauldron/config.rs +++ b/src/db/cauldron/config.rs @@ -6,7 +6,7 @@ use anyhow::{bail, Result}; use rusqlite::{params, Connection}; -pub const DB_VERSION: u32 = 3; +pub const DB_VERSION: u32 = 4; const DB_VERSION_KEY: &str = "db_version"; /// Create the config table diff --git a/src/db/cauldron/mod.rs b/src/db/cauldron/mod.rs index b10bf60..91cf268 100644 --- a/src/db/cauldron/mod.rs +++ b/src/db/cauldron/mod.rs @@ -46,10 +46,7 @@ pub fn prepare_tables(conn: &Connection) { ) .unwrap(); conn.execute("CREATE INDEX idx_utxo_funding_tvl_highest ON utxo_funding(token_id, new_utxo_hash, sats, token_amount);", []).unwrap(); - // Create the new index for (pool, COALESCE(first_seen_timestamp, mtp_timestamp)) - conn.execute("CREATE INDEX idx_pool_history_entry_pool_timestamp ON pool_history_entry (pool, COALESCE(first_seen_timestamp, mtp_timestamp));", -[], -).unwrap(); + conn.execute("CREATE INDEX idx_pool_history_entry_pool_timestamp ON pool_history_entry (pool, effective_timestamp);", []).unwrap(); // Set the database version config::set_db_version(conn); diff --git a/src/db/cauldron/pool.rs b/src/db/cauldron/pool.rs index fec7967..f4f65b3 100644 --- a/src/db/cauldron/pool.rs +++ b/src/db/cauldron/pool.rs @@ -45,10 +45,12 @@ pub fn create_table(conn: &Connection) { "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, tx_pos INT NOT NULL, mtp_timestamp BIGINT, first_seen_timestamp BIGINT, + effective_timestamp BIGINT GENERATED ALWAYS AS (COALESCE(first_seen_timestamp, mtp_timestamp)), sequence BIGINT NOT NULL, sats BIGINT NOT NULL, token_amount BIGINT NOT NULL, @@ -77,7 +79,23 @@ pub fn create_table(conn: &Connection) { params![], ) .unwrap(); - conn.execute("CREATE INDEX idx_pool_history_entry_timestamp_sequence ON pool_history_entry(mtp_timestamp, sequence)", params![]).unwrap(); + + // speeds up pool visitor query + conn.execute( + "CREATE INDEX idx_pool_history_entry_pool_timestamp_sequence ON pool_history_entry ( + pool, + effective_timestamp, + sequence DESC)", + params![], + ) + .unwrap(); + + // for pool visitor; filtering on token + conn.execute( + "CREATE INDEX idx_pool_history_entry_token_id ON pool_history_entry(token_id)", + params![], + ) + .unwrap(); } fn get_pool_by_utxo(conn: &Connection, utxo_hash: &OutPointHash) -> Result> { @@ -160,10 +178,11 @@ pub fn insert_pool_history_entry( assert!(next_seq >= 0); conn.execute( - "INSERT INTO pool_history_entry (utxo, pool, txid, tx_pos, mtp_timestamp, first_seen_timestamp, sequence, sats, token_amount, sats_delta, token_delta) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + "INSERT INTO pool_history_entry (utxo, pool, token_id, txid, tx_pos, mtp_timestamp, first_seen_timestamp, sequence, sats, token_amount, sats_delta, token_delta) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(utxo) DO UPDATE SET pool = excluded.pool, + token_id = excluded.token_id, txid = excluded.txid, tx_pos = excluded.tx_pos, mtp_timestamp = COALESCE(excluded.mtp_timestamp, pool_history_entry.mtp_timestamp), @@ -175,6 +194,7 @@ pub fn insert_pool_history_entry( .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(), cauldron .new_utxo_txid .expect("txid of new pool history entry") @@ -323,7 +343,7 @@ fn get_nearest_entries( (false, false) => params![timestamp], }; - let ts = "COALESCE(first_seen_timestamp, mtp_timestamp)"; + let ts = "effective_timestamp"; let preferred = match resolution { SnapshotSelection::UseBefore => "ts ASC", // Prefer the lower timestamp first @@ -439,7 +459,7 @@ pub struct PoolHistoryEntry { } fn get_pool_history_entry(conn: &Connection, utxo_hash: OutPointHash) -> Result { - let mut stmt = conn.prepare("SELECT txid, sats, token_amount, COALESCE(first_seen_timestamp, mtp_timestamp) as timestamp FROM pool_history_entry WHERE utxo = ?")?; + 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 row = rows .next()? @@ -464,7 +484,7 @@ pub fn db_pool_history( phe.txid, phe.sats, phe.token_amount, - COALESCE(phe.first_seen_timestamp, phe.mtp_timestamp) as timestamp + phe.effective_timestamp as timestamp FROM pool_history_entry phe WHERE @@ -522,7 +542,7 @@ pub fn get_total_volume_sats( FROM pool_history_entry phe JOIN pool p ON phe.pool = p.creation_utxo JOIN tx ON phe.txid = tx.txid - WHERE COALESCE(tx.first_seen_timestamp, tx.mtp_timestamp) BETWEEN ? AND ?"; + WHERE tx.effective_timestamp BETWEEN ? AND ?"; let total_volume: i64 = db.query_row(sql, params![start_timestamp, end_timestamp], |row| { row.get(0) @@ -545,7 +565,7 @@ pub fn get_token_volume_sats( FROM pool_history_entry phe JOIN pool p ON phe.pool = p.creation_utxo JOIN tx ON phe.txid = tx.txid - WHERE COALESCE(tx.first_seen_timestamp, tx.mtp_timestamp) BETWEEN ? AND ? + WHERE tx.effective_timestamp BETWEEN ? AND ? AND p.token_id = ?"; let (sats_volume, token_volume): (i64, i64) = db.query_row( diff --git a/src/db/cauldron/poolvisitor.rs b/src/db/cauldron/poolvisitor.rs index f3a525a..7cd24cb 100644 --- a/src/db/cauldron/poolvisitor.rs +++ b/src/db/cauldron/poolvisitor.rs @@ -17,14 +17,59 @@ pub(crate) enum OptionalPoolFields { TxPos = 1 << 3, TokenId = 1 << 4, PoolId = 1 << 5, + Timestamp = 1 << 6, } +#[derive(Default)] pub(crate) struct PoolFilters { - pub timestamp_less_than: Option, + pub timestamp_lt: Option, + pub timestamp_lte: Option, + pub timestamp_gt: Option, + pub timestamp_gte: Option, pub token_id: Option, pub owner: Option, } +impl PoolFilters { + pub fn new() -> Self { + Self::default() + } + + #[allow(dead_code)] + pub fn lt(mut self, timestamp: u64) -> Self { + self.timestamp_lt = Some(timestamp); + self + } + + pub fn lte(mut self, timestamp: u64) -> Self { + self.timestamp_lte = Some(timestamp); + self + } + + #[allow(dead_code)] + pub fn gt(mut self, timestamp: u64) -> Self { + self.timestamp_gt = Some(timestamp); + self + } + + #[allow(dead_code)] + pub fn gte(mut self, timestamp: u64) -> Self { + self.timestamp_gte = Some(timestamp); + self + } + + pub fn token_id(mut self, token_id: &str) -> Self { + self.token_id = Some(token_id.to_string()); + self + } + + #[allow(dead_code)] + pub fn owner(mut self, owner: &str) -> Self { + self.owner = Some(owner.to_string()); + self + } +} + #[derive(Default)] pub(crate) struct OptionalFields { pub owner: Option, @@ -32,6 +77,7 @@ pub(crate) struct OptionalFields { pub tx_pos: Option, pub token_id: Option, pub pool_id: Option, + pub timestamp: Option, } pub(crate) trait PoolVisitor { @@ -40,46 +86,77 @@ pub(crate) trait PoolVisitor { fn visit(&mut self, sats: u64, tokens: u64, optional_fields: OptionalFields) -> Result; } +fn time_filters( + filters: &PoolFilters, + param_index: usize, +) -> (String, Vec) { + let now = time_now() as u64; + + let mut clauses: Vec = vec!["1=1".to_string()]; + let mut params: Vec = Vec::new(); + let mut next_idx: usize = param_index; + + if filters.timestamp_lt.map(|t| t <= now).unwrap_or(false) { + clauses.push(format!("effective_timestamp < ?{}", next_idx)); + params.push(rusqlite::types::ToSqlOutput::Owned( + (filters.timestamp_lt.unwrap() as i64).into(), + )); + next_idx += 1; + } + if filters.timestamp_lte.map(|t| t <= now).unwrap_or(false) { + clauses.push(format!("effective_timestamp <= ?{}", next_idx)); + params.push(rusqlite::types::ToSqlOutput::Owned( + (filters.timestamp_lte.unwrap() as i64).into(), + )); + next_idx += 1; + } + if let Some(gt) = filters.timestamp_gt { + clauses.push(format!("effective_timestamp > ?{}", next_idx)); + params.push(rusqlite::types::ToSqlOutput::Owned((gt as i64).into())); + next_idx += 1; + } + if let Some(gte) = filters.timestamp_gte { + clauses.push(format!("effective_timestamp >= ?{}", next_idx)); + params.push(rusqlite::types::ToSqlOutput::Owned((gte as i64).into())); + } + + let phe_sql = format!(" WHERE {} ", clauses.join(" AND ")); + + (phe_sql, params) +} + pub(crate) fn db_visit_pool_entries( conn: &Connection, visitor: &mut T, filters: PoolFilters, ) -> Result<()> { - let mut sql_filters = Vec::new(); - let mut params: Vec = Vec::new(); - let mut param_index = 2; // Start from 2 since ?1 is used by time filter + let (phe_timestamp_filter, time_params) = time_filters(&filters, 2); // Start at ?2, ?1 for withdrawn - if let Some(token_id) = &filters.token_id { - sql_filters.push(format!("p.token_id = ?{}", param_index)); + let mut sql_filters = Vec::new(); + let mut params: Vec = time_params; + let mut param_index = 2 + params.len(); + + let phe_tokenid_filter = if let Some(token_id) = &filters.token_id { params.push(rusqlite::types::ToSqlOutput::Owned( token_id.to_owned().into(), )); + let this_param_index = param_index; param_index += 1; - } + format!("AND token_id = ?{}", this_param_index) + } else { + "".to_string() + }; 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 sql_filters_str = if sql_filters.is_empty() { "".to_string() } else { format!("AND {}", sql_filters.join(" AND ")) }; - // don't need this filter if its in the future. - // faster without filter. - let time_filter = filters - .timestamp_less_than - .filter(|&t| t <= time_now() as u64); - - let phe_timestamp_filter = if time_filter.is_some() { - " WHERE mtp_timestamp < ?1 ".to_string() - } else { - "".to_string() - }; - let query: String = " SELECT p.owner_pkh, @@ -88,7 +165,8 @@ pub(crate) fn db_visit_pool_entries( phe.txid, phe.tx_pos, p.token_id, - p.creation_utxo + p.creation_utxo, + phe.effective_timestamp FROM pool p JOIN pool_history_entry phe ON p.creation_utxo = phe.pool @@ -96,23 +174,31 @@ pub(crate) fn db_visit_pool_entries( SELECT pool, MAX(sequence) AS max_sequence FROM pool_history_entry {phe_timestamp_filter} + {phe_tokenid_filter} + GROUP BY pool ) max_phe ON phe.pool = max_phe.pool AND phe.sequence = max_phe.max_sequence WHERE (p.withdrawn_in_utxo IS NULL OR ( - SELECT t.mtp_timestamp + SELECT effective_timestamp FROM utxo_spending us JOIN tx t ON us.txid = t.txid WHERE us.spent_utxo_hash = p.withdrawn_in_utxo ) >= ?1) {filters} - GROUP BY - p.creation_utxo" - .replace("{filters}", &sql_filters_str) - .replace("{phe_timestamp_filter}", &phe_timestamp_filter); + " + .replace("{filters}", &sql_filters_str) + .replace("{phe_timestamp_filter}", &phe_timestamp_filter) + .replace("{phe_tokenid_filter}", &phe_tokenid_filter); + + // allow to include pools that were withdrawn AFTER a time filter. + let withdraw_time_filter = filters + .timestamp_lt + .min(filters.timestamp_lte) + .unwrap_or(i64::MAX as u64); let mut final_params = vec![rusqlite::types::ToSqlOutput::Owned( - (time_filter.unwrap_or(i64::MAX as u64) as i64).into(), + (withdraw_time_filter as i64).into(), )]; final_params.extend(params); @@ -142,6 +228,9 @@ pub(crate) fn db_visit_pool_entries( if (wanted & OptionalPoolFields::PoolId as u64) != 0 { optional.pool_id = Some(row.get(6)?); } + if (wanted & OptionalPoolFields::Timestamp as u64) != 0 { + optional.timestamp = Some(row.get(7)?); + } if !visitor.visit(sats, tokens, optional)? { break; diff --git a/src/db/cauldron/tokenlist.rs b/src/db/cauldron/tokenlist.rs index 0245b22..133d94d 100644 --- a/src/db/cauldron/tokenlist.rs +++ b/src/db/cauldron/tokenlist.rs @@ -31,9 +31,8 @@ pub fn db_list_tokens_by_volume( seconds: usize, limit: usize, ) -> Result> { - let mut statement = cauldron_conn - .prepare( - " + let mut statement = cauldron_conn.prepare( + " WITH TradeData AS ( SELECT p.token_id, @@ -41,7 +40,7 @@ pub fn db_list_tokens_by_volume( FROM pool_history_entry phe JOIN pool p ON phe.pool = p.creation_utxo JOIN tx ON phe.txid = tx.txid - WHERE COALESCE(tx.first_seen_timestamp, tx.mtp_timestamp) >= (strftime('%s', 'now') - ?1) + WHERE tx.effective_timestamp >= (strftime('%s', 'now') - ?1) ), AllTokenIDs AS ( SELECT DISTINCT token_id @@ -62,7 +61,7 @@ pub fn db_list_tokens_by_volume( ORDER BY total_trade_volume DESC LIMIT ?2 ", - )?; + )?; let mut query = statement.query([seconds, limit])?; diff --git a/src/db/cauldron/tx.rs b/src/db/cauldron/tx.rs index caffd65..015a514 100644 --- a/src/db/cauldron/tx.rs +++ b/src/db/cauldron/tx.rs @@ -13,13 +13,18 @@ pub fn create_table(conn: &Connection) { txid TEXT PRIMARY KEY, blockhash TEXT, mtp_timestamp BIGINT, - first_seen_timestamp BIGINT + first_seen_timestamp BIGINT, + effective_timestamp BIGINT GENERATED ALWAYS AS (COALESCE(first_seen_timestamp, mtp_timestamp)) )"; conn.execute(tbl, []).expect("failed to create tx table"); // Create index for volume queries that filter by COALESCE(timestamp) - conn.execute("CREATE INDEX idx_tx_coalesce_timestamp ON tx(COALESCE(first_seen_timestamp, mtp_timestamp))", []).expect("failed to create tx timestamp index"); + conn.execute( + "CREATE INDEX idx_tx_effective_timestamp ON tx(effective_timestamp)", + [], + ) + .expect("failed to create tx timestamp index"); } pub fn insert_block_tx( @@ -63,14 +68,14 @@ pub fn latest( FROM tx JOIN utxo_funding ON tx.txid = utxo_funding.txid WHERE utxo_funding.token_id = ? - ORDER BY COALESCE(tx.first_seen_timestamp, tx.mtp_timestamp) DESC + ORDER BY tx.effective_timestamp DESC LIMIT ? OFFSET ?", params![tid.to_hex(), limit, offset], ), None => ( "SELECT tx.txid, tx.blockhash, tx.mtp_timestamp, tx.first_seen_timestamp FROM tx - ORDER BY COALESCE(tx.first_seen_timestamp, tx.mtp_timestamp) DESC + ORDER BY tx.effective_timestamp DESC LIMIT ? OFFSET ?", params![limit, offset], ), diff --git a/src/db/cauldron/utxo_spending.rs b/src/db/cauldron/utxo_spending.rs index 9a274d5..c3ef3c3 100644 --- a/src/db/cauldron/utxo_spending.rs +++ b/src/db/cauldron/utxo_spending.rs @@ -7,7 +7,7 @@ use anyhow::Result; use bitcoin_hashes::hex::ToHex; use bitcoincash::Txid; use riftenlabs_defi::cauldron::ParsedContract; -use rusqlite::{params, Connection, Transaction}; +use rusqlite::{params, Connection}; pub fn create_table(conn: &Connection) { conn.execute( @@ -22,7 +22,7 @@ pub fn create_table(conn: &Connection) { } pub fn insert_utxo_spending( - tx: &Transaction, + tx: &Connection, cauldrons: &Vec, txid: &Txid, replace: bool, diff --git a/src/db/search/mod.rs b/src/db/search/mod.rs index d05d23a..526ed3d 100644 --- a/src/db/search/mod.rs +++ b/src/db/search/mod.rs @@ -104,15 +104,15 @@ fn token_volume( tokens: Vec, ) -> Result> { let result: Vec = tokens - .into_par_iter() - .map(|(token_id, name, ticker)| { + .into_par_iter() + .map(|(token_id, name, ticker)| { + let conn = cauldron_pool + .get() + .context("Failed to get connection from pool")?; - let conn = cauldron_pool - .get() - .context("Failed to get connection from pool")?; - - let mut statement = conn.prepare( - " + let mut statement = conn + .prepare( + " WITH TradeData AS ( SELECT p.token_id, @@ -120,7 +120,7 @@ fn token_volume( FROM pool_history_entry phe JOIN pool p ON phe.pool = p.creation_utxo JOIN tx ON phe.txid = tx.txid - WHERE COALESCE(tx.first_seen_timestamp, tx.mtp_timestamp) >= (strftime('%s', 'now') - 2592000) + WHERE tx.effective_timestamp >= (strftime('%s', 'now') - 2592000) AND p.token_id = ? ) SELECT @@ -128,16 +128,17 @@ fn token_volume( COALESCE(SUM(trade_volume), 0) as total_trade_volume FROM TradeData; ", - ).context("Failed to prepare statement")?; + ) + .context("Failed to prepare statement")?; - let trade_volume: u64 = statement - .query_row(params![&token_id], |row| row.get(1)) - .unwrap_or(0); + let trade_volume: u64 = statement + .query_row(params![&token_id], |row| row.get(1)) + .unwrap_or(0); - // Return result as Ok tuple for successful case - Ok((token_id, name, ticker, trade_volume)) - }) - .collect::>>()?; // Collect results into a Vec, propagating errors + // Return result as Ok tuple for successful case + Ok((token_id, name, ticker, trade_volume)) + }) + .collect::>>()?; // Collect results into a Vec, propagating errors Ok(result) } diff --git a/src/rpc/candlesticks.rs b/src/rpc/candlesticks.rs index 2060862..4148c1a 100644 --- a/src/rpc/candlesticks.rs +++ b/src/rpc/candlesticks.rs @@ -135,8 +135,8 @@ pub fn candlesticks( JOIN tx ON tx.txid = phe.txid WHERE uf.token_id = ? - AND COALESCE(tx.first_seen_timestamp, tx.mtp_timestamp) >= ? - AND COALESCE(tx.first_seen_timestamp, tx.mtp_timestamp) < ? + AND tx.effective_timestamp >= ? + AND tx.effective_timestamp < ? GROUP BY tx.txid ) SELECT diff --git a/src/rpc/mod.rs b/src/rpc/mod.rs index e8546fb..ce7193d 100644 --- a/src/rpc/mod.rs +++ b/src/rpc/mod.rs @@ -41,7 +41,7 @@ fn all_time_volume(db: &Connection, end_timestamp: u64) -> Result Result> { uf.sats AS original_sats, uf.token_amount AS original_token_amount, uf.token_id as token_id, - COALESCE(tx.first_seen_timestamp, tx.mtp_timestamp) AS original_timestamp + tx.effective_timestamp AS original_timestamp FROM pool p JOIN utxo_funding uf ON p.creation_utxo = uf.new_utxo_hash @@ -57,7 +57,7 @@ fn pools_by_apy(connection: &Connection) -> Result> { phe.pool, uf.sats AS latest_sats, uf.token_amount AS latest_token_amount, - COALESCE(tx.first_seen_timestamp, tx.mtp_timestamp) AS latest_timestamp, + tx.effective_timestamp AS latest_timestamp, uf.new_utxo_txid, uf.new_utxo_n FROM @@ -70,7 +70,7 @@ fn pools_by_apy(connection: &Connection) -> Result> { AND us.spent_utxo_hash IS NULL ORDER BY - COALESCE(tx.first_seen_timestamp, tx.mtp_timestamp) DESC + tx.effective_timestamp DESC ) SELECT od.original_sats, @@ -257,7 +257,10 @@ pub fn list_active_pools( PoolFilters { token_id: token.map(|s| s.to_string()), owner: pkh.map(|s| s.to_string()), - timestamp_less_than: None, + timestamp_lt: None, + timestamp_lte: None, + timestamp_gt: None, + timestamp_gte: None, }, ) .map_err(|e| Custom(Status::InternalServerError, format!("Error: {e}")))?; diff --git a/src/rpc/price.rs b/src/rpc/price.rs index 07a948b..ae8e4bd 100644 --- a/src/rpc/price.rs +++ b/src/rpc/price.rs @@ -11,7 +11,16 @@ use rusqlite::{params, Connection}; use rust_decimal::prelude::*; use serde_json::{json, Value}; -use crate::{db::DB, rpc::tvl::get_token_tvl, timeutil::time_now}; +use crate::{ + db::{ + cauldron::poolvisitor::{ + db_visit_pool_entries, OptionalFields, OptionalPoolFields, PoolFilters, PoolVisitor, + }, + DB, + }, + rpc::tvl::get_token_tvl, + timeutil::time_now, +}; struct PriceInterval { start: i64, @@ -166,75 +175,84 @@ fn historic_price( Ok(result) } -fn price_at_or_before( +/// Works by getting the total sats/tokens for each unique pool close to the requested timestamp. +/// In case of multiple price points for the same pool, the one with the greater K (sats * tokens) is used, since K grows with usage. +#[derive(Default)] +struct PriceVisitor { + // key is pool id + max_timestamp: i64, + sum_sats: u64, + sum_tokens: u64, +} + +impl PoolVisitor for PriceVisitor { + fn optional_fields_wanted(&self) -> u64 { + OptionalPoolFields::Timestamp as u64 + } + + fn visit(&mut self, sats: u64, tokens: u64, optional_fields: OptionalFields) -> Result { + let timestamp = optional_fields.timestamp.unwrap(); + + self.max_timestamp = self.max_timestamp.max(timestamp); + self.sum_sats += sats; + self.sum_tokens += tokens; + + Ok(true) + } +} + +impl PriceVisitor { + fn price(&self) -> Result { + let sum_sats = Decimal::from_u64(self.sum_sats).unwrap_or_default(); + let sum_tokens = Decimal::from_u64(self.sum_tokens).unwrap_or_default(); + + sum_sats + .checked_div(sum_tokens) + .context("Division by zero or overflow") + } + + fn has_data(&self) -> bool { + self.sum_sats != 0 + } + + fn max_timestamp(&self) -> i64 { + self.max_timestamp + } +} + +fn price_at_or_before_2( connection: &Connection, timestamp: i64, token_id: &TokenID, ) -> Result<(i64, f64)> { - let sql = " - WITH max_timestamps AS ( - SELECT - phe.pool, - MAX(COALESCE(phe.first_seen_timestamp, phe.mtp_timestamp)) AS max_effective_timestamp - FROM - pool_history_entry phe - JOIN - utxo_funding uf ON phe.utxo = uf.new_utxo_hash - WHERE - COALESCE(phe.first_seen_timestamp, phe.mtp_timestamp) <= ? - AND uf.token_id = ? - GROUP BY - phe.pool - ) - SELECT - uf.token_amount, - uf.sats, - COALESCE(phe.first_seen_timestamp, phe.mtp_timestamp) AS effective_timestamp - FROM - pool_history_entry phe - JOIN - utxo_funding uf ON phe.utxo = uf.new_utxo_hash - JOIN - pool p ON p.creation_utxo = phe.pool - JOIN - max_timestamps mt ON phe.pool = mt.pool - AND COALESCE(phe.first_seen_timestamp, phe.mtp_timestamp) = mt.max_effective_timestamp - WHERE - p.withdrawn_in_utxo IS NULL - "; + // Create a visitor to collect price data + let mut visitor = PriceVisitor::default(); - let mut statement = connection.prepare(sql)?; - let mut rows = statement.query(params![timestamp, token_id.to_hex()])?; + // note that we may return old timestamp; but it's still the last known trade at that time - let mut sum_sats: u64 = 0; - let mut sum_tokens: u64 = 0; - let mut latest_timestamp = 0; + // Use db_visit_pool_entries with filters for the specific token and timestamp + db_visit_pool_entries( + connection, + &mut visitor, + PoolFilters::new() + .lte(timestamp as u64) + .token_id(&token_id.to_hex()), + )?; - while let Some(row) = rows.next()? { - let token_amount: i64 = row.get(0)?; - let sats: i64 = row.get(1)?; - let row_timestamp: i64 = row.get(2)?; - - if sats >= 0 && token_amount >= 0 { - sum_sats += sats as u64; - sum_tokens += token_amount as u64; - } - - if row_timestamp > latest_timestamp { - latest_timestamp = row_timestamp; - } + // Check if we found any data + if !visitor.has_data() { + return Err(anyhow::anyhow!( + "No price data found for token {} at timestamp {}", + token_id.to_hex(), + timestamp + )); } - let sum_sats_decimal = Decimal::from_u64(sum_sats).unwrap_or_default(); - let sum_tokens_decimal = Decimal::from_u64(sum_tokens).unwrap_or_default(); + // Calculate the price + let price = visitor.price()?; + let price_f64 = price.to_f64().context("Failed to convert price to f64")?; - let overall_price = sum_sats_decimal - .checked_div(sum_tokens_decimal) - .context("Division failed: sum_tokens is zero or invalid")?; - - let overall_price_f64 = overall_price.to_f64().context("Conversion to f64 failed")?; - - Ok((latest_timestamp, overall_price_f64)) + Ok((visitor.max_timestamp(), price_f64)) } /// Get the price of a given token at a specific timestamp. @@ -277,7 +295,7 @@ pub fn price_at( let token = TokenID::from_hex(token).map_err(|e| Custom(Status::BadRequest, format!("Error: {e}")))?; - match price_at_or_before(&db, timestamp, &token) { + match price_at_or_before_2(&db, timestamp, &token) { Ok((latest_timestamp, price)) => Ok(Json(json!({ "timestamp": latest_timestamp, "price": price @@ -394,17 +412,16 @@ pub fn price_history( #[cfg(test)] mod tests { - use crate::utiltest::mock_db_pool; - use crate::{ - db::cauldron::{ - pool::{ - self, dummy_init_seq, flag_as_withdrawn, insert_new_pool, insert_pool_history_entry, - }, - tx::{self, insert_block_tx, insert_mempool_tx}, - utxo_funding::{self, insert_utxo_funding}, + use crate::db::cauldron::pool::update_pool_history; + use crate::db::cauldron::utxo_spending::{self, insert_utxo_spending}; + use crate::db::cauldron::{ + pool::{ + self, dummy_init_seq, flag_as_withdrawn, insert_new_pool, insert_pool_history_entry, }, - utiltest::{calc_sats_delta, calc_token_delta}, + tx::{self, insert_block_tx, insert_mempool_tx}, + utxo_funding::{self, insert_utxo_funding}, }; + use crate::utiltest::mock_db_pool; use super::*; use bitcoin_hashes::Hash; @@ -426,11 +443,12 @@ mod tests { sats: u64, tokens: i64, pkh: &PubkeyHash, + spent_utxo_hash: &OutPointHash, ) -> ParsedContract { ParsedContract { pkh: *pkh, is_withdrawn: false, - spent_utxo_hash: OutPointHash::all_zeros(), + spent_utxo_hash: *spent_utxo_hash, new_utxo_hash: Some(*utxo), new_utxo_txid: Some(*txid), new_utxo_n: Some(0), @@ -442,6 +460,7 @@ mod tests { fn setup_mock_db(conn: &Connection) { utxo_funding::create_table(conn); + utxo_spending::create_table(&conn); tx::create_table(conn); pool::create_table(conn); dummy_init_seq(); @@ -452,14 +471,38 @@ mod tests { let txid1 = Txid::from_inner([0xf0; 32]); let txid2 = Txid::from_inner([0xf1; 32]); let txid3 = Txid::from_inner([0xf2; 32]); - let utxo1 = OutPointHash::from_inner([0xe0; 32]); - let utxo2 = OutPointHash::from_inner([0xe1; 32]); - let utxo3 = OutPointHash::from_inner([0xe2; 32]); + let utxo1 = OutPointHash::from_inner([0xe0; 32]); // pool 1 + let utxo2 = OutPointHash::from_inner([0xe1; 32]); // pool 2 + let utxo3 = OutPointHash::from_inner([0xe2; 32]); // pool 3 // Insert mock data into utxo_funding - let cauldron1 = dummy_cauldron(&txid1, &utxo1, &token_zero, 50000, 1000, &pkh_zero); - let cauldron2 = dummy_cauldron(&txid2, &utxo2, &token_zero, 60000, 2000, &pkh_zero); - let cauldron3 = dummy_cauldron(&txid2, &utxo3, &token_zero, 90000, 3000, &pkh_zero); + let cauldron1 = dummy_cauldron( + &txid1, + &utxo1, + &token_zero, + 50000, + 1000, + &pkh_zero, + &OutPointHash::all_zeros(), + ); + let cauldron2 = dummy_cauldron( + &txid2, + &utxo2, + &token_zero, + 60000, + 2000, + &pkh_zero, + &OutPointHash::all_zeros(), + ); + let cauldron3 = dummy_cauldron( + &txid2, + &utxo3, + &token_zero, + 90000, + 3000, + &pkh_zero, + &OutPointHash::all_zeros(), + ); insert_utxo_funding(conn, &vec![cauldron1.clone()], &txid1, true).unwrap(); insert_utxo_funding(conn, &vec![cauldron2.clone()], &txid2, true).unwrap(); insert_utxo_funding(conn, &vec![cauldron3.clone()], &txid3, true).unwrap(); @@ -474,17 +517,12 @@ mod tests { insert_mempool_tx(conn, &txid3, TIME_3).unwrap(); // Insert mock data into `pool_history_entry` - let pool1 = OutPointHash::from_inner([0x0a; 32]); - let pool2 = OutPointHash::from_inner([0x0b; 32]); - let pool3 = OutPointHash::from_inner([0x0c; 32]); + update_pool_history(conn, vec![cauldron1.clone()], Some(TIME_1), Some(TIME_1)).unwrap(); + update_pool_history(conn, vec![cauldron2], Some(TIME_2), Some(TIME_2)).unwrap(); + update_pool_history(conn, vec![cauldron3], Some(TIME_3), Some(TIME_3)).unwrap(); + let txid1_newer = Txid::from_inner([0xf3; 32]); let utxo1_newer = OutPointHash::from_inner([0xe3; 32]); - insert_pool_history_entry(conn, &pool1, &cauldron1, Some(TIME_1), Some(TIME_1), 0, 0) - .unwrap(); - insert_pool_history_entry(conn, &pool2, &cauldron2, Some(TIME_2), Some(TIME_2), 0, 0) - .unwrap(); - insert_pool_history_entry(conn, &pool3, &cauldron3, Some(TIME_3), Some(TIME_3), 0, 0) - .unwrap(); // Insert newer entry for pool1 let cauldron1_newer = dummy_cauldron( @@ -494,39 +532,14 @@ mod tests { 70000, 1500, &pkh_zero, + &utxo1, ); - insert_pool_history_entry( + insert_utxo_spending(conn, &vec![cauldron1], &txid1_newer, true).unwrap(); + update_pool_history( conn, - &pool1, - &cauldron1_newer, + vec![cauldron1_newer], Some(1727963500), Some(1727963500), - calc_sats_delta(&cauldron1, &cauldron1_newer), - calc_token_delta(&cauldron1, &cauldron1_newer), - ) - .unwrap(); - insert_utxo_funding(conn, &vec![cauldron1_newer], &txid1_newer, true).unwrap(); - - // Insert active pools with required columns (owner_pkh and token_id) - let token1 = TokenID::from_inner([0xda; 32]); - let token2 = TokenID::from_inner([0xdb; 32]); - let token3 = TokenID::from_inner([0xdc; 32]); - let pkh1 = PubkeyHash::from_inner([0xca; 20]); - let pkh2 = PubkeyHash::from_inner([0xcb; 20]); - let pkh3 = PubkeyHash::from_inner([0xcc; 20]); - insert_new_pool( - conn, - &dummy_cauldron(&Txid::all_zeros(), &pool1, &token1, 0, 0, &pkh1), - ) - .unwrap(); - insert_new_pool( - conn, - &dummy_cauldron(&Txid::all_zeros(), &pool2, &token2, 0, 0, &pkh2), - ) - .unwrap(); - insert_new_pool( - conn, - &dummy_cauldron(&Txid::all_zeros(), &pool3, &token3, 0, 0, &pkh3), ) .unwrap(); @@ -543,6 +556,7 @@ mod tests { 80000, 2500, &inactive_pkh, + &OutPointHash::all_zeros(), ); let inactive_cauldron_withdraw = ParsedContract { pkh: inactive_pkh, @@ -588,7 +602,12 @@ mod tests { let response = client .get(format!("/cauldron/price/{token_id}/at/{timestamp}")) .dispatch(); - assert_eq!(response.status(), Status::Ok); + assert_eq!( + response.status(), + Status::Ok, + "response: {}", + response.into_string().unwrap() + ); let json_value: serde_json::Value = serde_json::from_str(response.into_string().unwrap().as_str()).unwrap(); @@ -653,7 +672,12 @@ mod tests { let response = client .get(format!("/cauldron/price/{token_id}/at/{timestamp}")) .dispatch(); - assert_eq!(response.status(), Status::Ok); + assert_eq!( + response.status(), + Status::Ok, + "response: {}", + response.into_string().unwrap() + ); let json_value: serde_json::Value = serde_json::from_str(response.into_string().unwrap().as_str()).unwrap(); @@ -671,6 +695,9 @@ mod tests { #[test] fn test_price_with_multiple_entries_at_same_timestamp() { + // when encountering dupicate timestamp for same pool; it should + // use the latest entry + let mock_db = mock_db_pool(setup_mock_db); let rocket = rocket::build() @@ -685,27 +712,32 @@ mod tests { let conn = mock_db.cauldron_w.get().expect("Failed to get connection."); // These entries should have the same timestamp as previous mockdata and be counted in the price calculation - let pool1 = OutPointHash::from_inner([0x0a; 32]); + let pool1 = OutPointHash::from_inner([0xe0; 32]); let txid = Txid::hash("txid_extra".as_bytes()); let cauldron = dummy_cauldron( &txid, &OutPointHash::hash("utxo_extra".as_bytes()), &TokenID::all_zeros(), - 40000, - 500, + 1, + 1, &PubkeyHash::all_zeros(), + &pool1, ); - insert_pool_history_entry(&conn, &pool1, &cauldron, Some(TIME_1), Some(TIME_1), 0, 0) - .unwrap(); - insert_utxo_funding(&conn, &vec![cauldron], &txid, true).unwrap(); + update_pool_history(&conn, vec![cauldron], Some(TIME_1), Some(TIME_1)).unwrap(); + // insert_utxo_funding(&conn, &vec![cauldron], &txid, true).unwrap(); - // Test: Query at the timestamp matching test_txid1 (1727963300) and check the price + // Test: Query at the timestamp matching test_txid1 (TIME_1) and check the price let timestamp = TIME_1; let response = client .get(format!("/cauldron/price/{token_id}/at/{timestamp}")) .dispatch(); - assert_eq!(response.status(), Status::Ok); + assert_eq!( + response.status(), + Status::Ok, + "response: {}", + response.into_string().unwrap() + ); let json_value: serde_json::Value = serde_json::from_str(response.into_string().unwrap().as_str()).unwrap(); @@ -713,9 +745,9 @@ mod tests { .as_f64() .expect("Price field is not a valid f64"); - // Expected price based on (50,000 + 40,000) sats and (1,000 + 500) tokens = 90,000 / 1,500 = 60.0 - let expected_price = 60.0; - assert!((actual_price - expected_price).abs() < 0.01); + // Expected price based on 1 sats and 1 tokens + let expected_price = 1.0; + assert_eq!(actual_price, expected_price); } #[test] @@ -807,6 +839,7 @@ mod tests { 1, /* low sats */ 9999999999999999, /* many tokens */ &PubkeyHash::all_zeros(), + &OutPointHash::all_zeros(), ); insert_utxo_funding(&conn, &vec![cauldron.clone()], &txid, true).unwrap(); insert_pool_history_entry(&conn, &pool, &cauldron, Some(TIME_1), Some(TIME_1), 0, 0) diff --git a/src/rpc/tvl.rs b/src/rpc/tvl.rs index eca6d5c..7529286 100644 --- a/src/rpc/tvl.rs +++ b/src/rpc/tvl.rs @@ -85,7 +85,10 @@ pub fn deprecated_get_all_token_tvl( connection, &mut visitor, PoolFilters { - timestamp_less_than: Some(max_timestamp as u64), + timestamp_lt: Some(max_timestamp as u64), + timestamp_lte: None, + timestamp_gt: None, + timestamp_gte: None, token_id: None, owner: None, }, @@ -101,7 +104,10 @@ pub fn get_total_sats_tvl(connection: &Connection, max_timestamp: Option) connection, &mut visitor, PoolFilters { - timestamp_less_than: max_timestamp.map(|t| t as u64), + timestamp_lt: max_timestamp.map(|t| t as u64), + timestamp_lte: None, + timestamp_gt: None, + timestamp_gte: None, token_id: None, owner: None, }, @@ -121,7 +127,10 @@ pub fn get_token_tvl( connection, &mut visitor, PoolFilters { - timestamp_less_than: max_timestamp.map(|t| t as u64), + timestamp_lt: max_timestamp.map(|t| t as u64), + timestamp_lte: None, + timestamp_gt: None, + timestamp_gte: None, token_id: Some(token_id.to_string()), owner: None, }, diff --git a/src/utiltest.rs b/src/utiltest.rs index 6eac76d..b1a5b20 100644 --- a/src/utiltest.rs +++ b/src/utiltest.rs @@ -9,30 +9,9 @@ mod test_utils { use crate::db::DB; use r2d2::Pool; use r2d2_sqlite::SqliteConnectionManager; - use riftenlabs_defi::cauldron::ParsedContract; use rusqlite::Connection; use std::sync::Arc; - pub fn calc_sats_delta(a: &ParsedContract, b: &ParsedContract) -> i64 { - assert!( - a.token_id == b.token_id, - "Different token ID; so definetly not the same pool" - ); - let a_sats = a.sats.unwrap_or(0); - let b_sats = b.sats.unwrap_or(0); - b_sats as i64 - a_sats as i64 - } - - pub fn calc_token_delta(a: &ParsedContract, b: &ParsedContract) -> i64 { - assert!( - a.token_id == b.token_id, - "Different token ID; so definetly not the same pool" - ); - let a_tokens = a.token_amount.unwrap_or(0); - let b_tokens = b.token_amount.unwrap_or(0); - b_tokens as i64 - a_tokens as i64 - } - pub fn mock_db_pool(setup_fn: F) -> DB where F: Fn(&Connection), @@ -56,9 +35,5 @@ mod test_utils { } } -#[cfg(test)] -pub use test_utils::calc_sats_delta; -#[cfg(test)] -pub use test_utils::calc_token_delta; #[cfg(test)] pub use test_utils::mock_db_pool;