// 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 // db_utils.rs // Copyright (C) 2024 Riften Labs AS // AGPL-3.0-or-later use anyhow::Result; use rusqlite::{params, Connection}; pub enum CachedSort { ScoreDesc, ScoreAsc, NameAsc, NameDesc, SymbolAsc, SymbolDesc, TvlDesc, TvlAsc, VolumeDesc, VolumeAsc, Change24hDesc, Change24hAsc, Change7dDesc, Change7dAsc, PriceUsdDesc, PriceUsdAsc, Change24hUsdDesc, Change24hUsdAsc, Change7dUsdDesc, Change7dUsdAsc, Apy30dDesc, Apy30dAsc, } pub fn order_clause(sort: CachedSort) -> &'static str { match sort { // Best first: rank ascending if set; else score/volume fallback. CachedSort::ScoreDesc => { "ORDER BY (NULLIF(score_rank,0) IS NULL) ASC, \ NULLIF(score_rank,0) ASC, \ score DESC, trade_volume DESC, token_id ASC" } // Worst first: rank descending if set; else score/volume fallback. CachedSort::ScoreAsc => { "ORDER BY (NULLIF(score_rank,0) IS NULL) ASC, \ NULLIF(score_rank,0) DESC, \ score ASC, trade_volume ASC, token_id ASC" } CachedSort::TvlDesc => "ORDER BY tvl_sats DESC, token_id ASC", CachedSort::TvlAsc => "ORDER BY tvl_sats ASC, token_id ASC", CachedSort::VolumeDesc => "ORDER BY trade_volume DESC, token_id ASC", CachedSort::VolumeAsc => "ORDER BY trade_volume ASC, token_id ASC", // NULLS LAST via leading `IS NULL` (false { "ORDER BY display_name IS NULL, display_name COLLATE NOCASE ASC, token_id ASC" } CachedSort::NameDesc => { "ORDER BY display_name IS NULL, display_name COLLATE NOCASE DESC, token_id ASC" } CachedSort::SymbolAsc => { "ORDER BY display_symbol IS NULL, display_symbol COLLATE NOCASE ASC, token_id ASC" } CachedSort::SymbolDesc => { "ORDER BY display_symbol IS NULL, display_symbol COLLATE NOCASE DESC, token_id ASC" } CachedSort::Change24hDesc => { "ORDER BY change_24h_bp IS NULL, change_24h_bp DESC, token_id ASC" } CachedSort::Change24hAsc => { "ORDER BY change_24h_bp IS NULL, change_24h_bp ASC, token_id ASC" } CachedSort::Change7dDesc => { "ORDER BY change_7d_bp IS NULL, change_7d_bp DESC, token_id ASC" } CachedSort::Change7dAsc => { "ORDER BY change_7d_bp IS NULL, change_7d_bp ASC, token_id ASC" } CachedSort::PriceUsdDesc => "ORDER BY price_now_usd DESC, token_id ASC", CachedSort::PriceUsdAsc => "ORDER BY price_now_usd ASC, token_id ASC", CachedSort::Change24hUsdDesc => { "ORDER BY change_24h_usd_bp IS NULL, change_24h_usd_bp DESC, token_id ASC" } CachedSort::Change24hUsdAsc => { "ORDER BY change_24h_usd_bp IS NULL, change_24h_usd_bp ASC, token_id ASC" } CachedSort::Change7dUsdDesc => { "ORDER BY change_7d_usd_bp IS NULL, change_7d_usd_bp DESC, token_id ASC" } CachedSort::Change7dUsdAsc => { "ORDER BY change_7d_usd_bp IS NULL, change_7d_usd_bp ASC, token_id ASC" } CachedSort::Apy30dDesc => "ORDER BY apy_30d_bp IS NULL, apy_30d_bp DESC, token_id ASC", CachedSort::Apy30dAsc => "ORDER BY apy_30d_bp IS NULL, apy_30d_bp ASC, token_id ASC", } } /// Create the *current* index set only (no drops). pub fn create_cached_token_metrics_indexes(conn: &Connection) -> Result<()> { conn.execute_batch( r#" -- Consolidated sort indexes (ASC can be reverse-scanned for DESC; ties may sort in-memory on token_id) -- Score / Volume tie-break CREATE INDEX IF NOT EXISTS idx_ctm_score ON cached_token_metrics(score, trade_volume, token_id); -- TVL & Volume single-key sorts CREATE INDEX IF NOT EXISTS idx_ctm_tvl ON cached_token_metrics(tvl_sats, token_id); CREATE INDEX IF NOT EXISTS idx_ctm_volume ON cached_token_metrics(trade_volume, token_id); -- Price (USD) CREATE INDEX IF NOT EXISTS idx_ctm_price_usd ON cached_token_metrics(price_now_usd, token_id); -- APY 30d CREATE INDEX IF NOT EXISTS idx_ctm_apy ON cached_token_metrics(apy_30d_bp, token_id); -- NULLS LAST expression indexes for text sorts CREATE INDEX IF NOT EXISTS idx_ctm_name_ord ON cached_token_metrics((display_name IS NULL), display_name COLLATE NOCASE, token_id); CREATE INDEX IF NOT EXISTS idx_ctm_symbol_ord ON cached_token_metrics((display_symbol IS NULL), display_symbol COLLATE NOCASE, token_id); -- % change (BCH) CREATE INDEX IF NOT EXISTS idx_ctm_ch24_ord ON cached_token_metrics((change_24h_bp IS NULL), change_24h_bp, token_id); CREATE INDEX IF NOT EXISTS idx_ctm_ch7d_ord ON cached_token_metrics((change_7d_bp IS NULL), change_7d_bp, token_id); -- % change (USD) CREATE INDEX IF NOT EXISTS idx_ctm_ch24_usd_ord ON cached_token_metrics((change_24h_usd_bp IS NULL), change_24h_usd_bp, token_id); CREATE INDEX IF NOT EXISTS idx_ctm_ch7d_usd_ord ON cached_token_metrics((change_7d_usd_bp IS NULL), change_7d_usd_bp, token_id); CREATE INDEX IF NOT EXISTS idx_ctm_score_rank ON cached_token_metrics(score_rank, token_id); ANALYZE; PRAGMA optimize; "#, )?; Ok(()) } /// Indexes that speed up the aggregation path (tx → phe → pool). pub fn create_aggregation_path_indexes(conn: &Connection) -> Result<()> { conn.execute_batch( r#" -- Scan tx in a time window, then join to phe by txid CREATE INDEX IF NOT EXISTS idx_tx_effective_ts ON tx(effective_timestamp, txid); -- Join tx → phe on txid CREATE INDEX IF NOT EXISTS idx_phe_txid ON pool_history_entry(txid); -- Mixed access patterns (pool then txid) CREATE INDEX IF NOT EXISTS idx_phe_pool_txid ON pool_history_entry(pool, txid); -- Time-bounded scans per pool (snapshots/APY) CREATE INDEX IF NOT EXISTS idx_phe_pool_timestamp ON pool_history_entry(pool, effective_timestamp); -- Join phe.pool → pool.creation_utxo (and cover token_id) CREATE INDEX IF NOT EXISTS idx_pool_creation_utxo ON pool(creation_utxo, token_id); -- Convenience for token-centric lookups CREATE INDEX IF NOT EXISTS idx_pool_token_id ON pool(token_id, creation_utxo); ANALYZE; PRAGMA optimize; "#, )?; Ok(()) } 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, trade_volume INTEGER NOT NULL DEFAULT 0, tvl_sats INTEGER NOT NULL DEFAULT 0, tvl_tokens INTEGER NOT NULL DEFAULT 0, score INTEGER NOT NULL DEFAULT 0, score_rank INTEGER NOT NULL DEFAULT 0, display_name TEXT, display_symbol TEXT, -- BCH (sats-per-human) prices price_now REAL NOT NULL DEFAULT 0, price_24h REAL, price_7d REAL, change_24h_bp INTEGER, change_7d_bp INTEGER, -- USD per human token (from time-matched oracle) price_now_usd REAL NOT NULL DEFAULT 0, price_24h_usd REAL, price_7d_usd REAL, change_24h_usd_bp INTEGER, change_7d_usd_bp INTEGER, apy_30d_bp INTEGER, last_trade_ts INTEGER, updated_at INTEGER NOT NULL ); "#, )?; // This could be inserted directly instead but this makes migration easier to add first pool seen add_column_if_missing(conn, "cached_token_metrics", "first_pool_ts", "INTEGER")?; Ok(()) } fn add_column_if_missing(conn: &Connection, table: &str, column: &str, decl: &str) -> Result<()> { let mut exists = false; let mut stmt = conn.prepare(&format!("PRAGMA table_info({table});"))?; let mut rows = stmt.query([])?; while let Some(row) = rows.next()? { let name: String = row.get(1)?; // column name if name.eq_ignore_ascii_case(column) { exists = true; break; } } if !exists { conn.execute( &format!("ALTER TABLE {table} ADD COLUMN {column} {decl};"), [], )?; } Ok(()) } fn table_has_column(conn: &Connection, table: &str, column: &str) -> Result { let mut stmt = conn.prepare(&format!("PRAGMA table_info({table});"))?; let mut rows = stmt.query([])?; while let Some(row) = rows.next()? { let name: String = row.get(1)?; if name.eq_ignore_ascii_case(column) { return Ok(true); } } Ok(false) } /// First pool info: (creation_utxo, txid, first_ts, first_height) pub type FirstPoolInfo = (String, String, i64, Option); /// 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<()> { // Ensure a row exists so the update can succeed. conn.execute( r#" INSERT INTO cached_token_metrics (token_id, updated_at) VALUES (?1, strftime('%s','now')) ON CONFLICT(token_id) DO NOTHING; "#, params![token_id], )?; // Accept NULL *or* 0 as “empty/unset” conn.execute( r#" UPDATE cached_token_metrics SET first_pool_ts = ?2 WHERE token_id = ?1 AND (first_pool_ts IS NULL OR first_pool_ts = 0); "#, params![token_id, ts], )?; Ok(()) } /// Returns (creation_utxo, txid, first_ts, first_height) for the first-ever pool of `token_id`. pub fn db_first_pool_creation_row( conn: &Connection, token_id: &str, ) -> Result> { // Prefer height when available; otherwise fall back to timestamp-only ordering. let has_height = table_has_column(conn, "tx", "block_height")?; let sql = if has_height { r#" SELECT p.creation_utxo, uf.txid, t.effective_timestamp, t.block_height FROM pool AS p JOIN utxo_funding AS uf ON uf.new_utxo_hash = p.creation_utxo JOIN tx AS t ON t.txid = uf.txid WHERE p.token_id = ? ORDER BY (t.block_height IS NULL) ASC, t.block_height ASC, t.effective_timestamp ASC, uf.txid ASC, p.creation_utxo ASC LIMIT 1; "# } else { r#" SELECT p.creation_utxo, uf.txid, t.effective_timestamp, NULL AS block_height FROM pool AS p JOIN utxo_funding AS uf ON uf.new_utxo_hash = p.creation_utxo JOIN tx AS t ON t.txid = uf.txid WHERE p.token_id = ? ORDER BY t.effective_timestamp ASC, uf.txid ASC, p.creation_utxo ASC LIMIT 1; "# }; let mut stmt = conn.prepare(sql)?; let mut rows = stmt.query(params![token_id])?; if let Some(row) = rows.next()? { let creation_utxo: String = row.get(0)?; let txid: String = row.get(1)?; let ts: i64 = row.get(2)?; let height: Option = row.get(3)?; Ok(Some((creation_utxo, txid, ts, height))) } else { Ok(None) } }