From b0929a5dd4e5d104879bed483b4fd97dd1ed6b7e Mon Sep 17 00:00:00 2001 From: Jakob Notland Date: Thu, 19 Mar 2026 13:46:16 +0100 Subject: [PATCH] Checkpoint --- src/cache.rs | 24 ++ src/db/cauldron/ohlcv.rs | 465 +++++++++++++++++++++++++++++++++++++++ src/db/cauldron/tx.rs | 17 +- src/rpc/candlesticks.rs | 7 +- 4 files changed, 499 insertions(+), 14 deletions(-) create mode 100644 src/cache.rs create mode 100644 src/db/cauldron/ohlcv.rs diff --git a/src/cache.rs b/src/cache.rs new file mode 100644 index 0000000..f2adadc --- /dev/null +++ b/src/cache.rs @@ -0,0 +1,24 @@ +// 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 + +use crate::rpc::candlesticks::CandlestickData; +use moka::future::Cache; +use std::sync::Arc; + +/// Cache key: (token_display_hex, start_ts, end_ts, stepsize_secs) +type CandleCacheKey = (String, i64, i64, i64); + +pub type CandlestickCache = Cache>>; + +/// Max total candles stored across all entries. +/// Each CandlestickData is ~72 bytes; 500_000 candles ≈ 36 MB upper bound. +const MAX_CANDLE_CAPACITY: u64 = 500_000; + +pub fn new_candlestick_cache() -> CandlestickCache { + Cache::builder() + .max_capacity(MAX_CANDLE_CAPACITY) + .weigher(|_k, v: &Arc>| v.len().max(1) as u32) + .build() +} diff --git a/src/db/cauldron/ohlcv.rs b/src/db/cauldron/ohlcv.rs new file mode 100644 index 0000000..0848670 --- /dev/null +++ b/src/db/cauldron/ohlcv.rs @@ -0,0 +1,465 @@ +// 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 + +use anyhow::Result; +use sqlx::{Row, SqlitePool}; + +pub async fn create_table(pool: &SqlitePool) { + sqlx::query( + "CREATE TABLE IF NOT EXISTS ohlcv_1h ( + token_id BLOB NOT NULL, + bucket_ts INTEGER NOT NULL, + open REAL NOT NULL, + high REAL NOT NULL, + low REAL NOT NULL, + close REAL NOT NULL, + volume_sats INTEGER NOT NULL, + volume_tokens INTEGER NOT NULL, + tx_count INTEGER NOT NULL, + PRIMARY KEY (token_id, bucket_ts) + )", + ) + .execute(pool) + .await + .expect("failed to create ohlcv_1h table"); + + // Needed for fast joins in rebuild_range; use IF NOT EXISTS so this is safe on existing DBs. + sqlx::query( + "CREATE INDEX IF NOT EXISTS idx_phe_txid ON pool_history_entry(txid)", + ) + .execute(pool) + .await + .expect("failed to create idx_phe_txid index"); + + // Composite index for fast token+time range scans in the candlestick raw CTE. + sqlx::query( + "CREATE INDEX IF NOT EXISTS idx_phe_token_id_ts ON pool_history_entry(token_id, effective_timestamp)", + ) + .execute(pool) + .await + .expect("failed to create idx_phe_token_id_ts index"); + + // Index for tx_latest token-filtered query: JOIN utxo_funding WHERE token_id = ? → ORDER BY tx.effective_timestamp. + sqlx::query( + "CREATE INDEX IF NOT EXISTS idx_utxo_funding_token_txid ON utxo_funding(token_id, txid)", + ) + .execute(pool) + .await + .expect("failed to create idx_utxo_funding_token_txid index"); +} + +/// Returns the highest `bucket_ts` in `ohlcv_1h`, or `None` if the table is empty. +pub async fn get_max_bucket_ts(pool: &SqlitePool) -> Result> { + let row: Option<(Option,)> = sqlx::query_as("SELECT MAX(bucket_ts) FROM ohlcv_1h") + .fetch_optional(pool) + .await?; + Ok(row.and_then(|r| r.0)) +} + +/// Returns the earliest confirmed trade timestamp floored to the nearest 1-hour bucket, +/// or `None` if there are no confirmed trades. Used to seed the initial backfill start +/// so the background task doesn't scan from Unix epoch 0. +pub async fn get_min_trade_bucket_ts(pool: &SqlitePool) -> Result> { + let row: Option<(Option,)> = sqlx::query_as( + "SELECT (MIN(tx.effective_timestamp) / 3600) * 3600 + FROM pool_history_entry AS phe + JOIN tx ON tx.txid = phe.txid + WHERE tx.blockhash IS NOT NULL", + ) + .fetch_optional(pool) + .await?; + Ok(row.and_then(|r| r.0)) +} + +/// Materialise all 1-hour OHLCV buckets for confirmed trades whose effective timestamp falls +/// in `[since_ts, until_ts)`. +/// +/// Two-phase approach: the slow aggregation SELECT runs against `read_pool` (no write lock), +/// then the pre-computed rows are bulk-inserted via `write_pool` (write lock held briefly). +/// Uses INSERT OR IGNORE so existing rows are never overwritten. +/// Returns the number of rows inserted. +pub async fn rebuild_range( + read_pool: &SqlitePool, + write_pool: &SqlitePool, + since_ts: i64, + until_ts: i64, +) -> Result { + if since_ts >= until_ts { + return Ok(0); + } + + // Phase 1: aggregate using the read pool — no write lock held during the slow CTE. + let select_sql = r#" +WITH per_pool_tx_raw AS ( + SELECT + uf.token_id, + phe.txid, + tx.effective_timestamp AS ts, + phe.utxo, + phe.sats_delta, + phe.token_delta + FROM pool_history_entry AS phe + JOIN utxo_funding AS uf ON phe.utxo = uf.new_utxo_hash + JOIN tx ON tx.txid = phe.txid + WHERE tx.blockhash IS NOT NULL + AND tx.effective_timestamp >= ? + AND tx.effective_timestamp < ? +), +per_pool_tx AS ( + SELECT + token_id, + txid, + ts, + (ts / 3600) * 3600 AS bucket_ts, + utxo, + SUM(sats_delta) AS signed_sats, + SUM(token_delta) AS signed_tokens, + SUM(ABS(sats_delta)) AS vol_sats, + SUM(ABS(token_delta)) AS vol_tokens + FROM per_pool_tx_raw + GROUP BY token_id, txid, ts, utxo +), +tx_trades AS ( + SELECT + token_id, + txid, + ts, + bucket_ts, + SUM(signed_sats) AS signed_sats, + SUM(signed_tokens) AS signed_tokens, + SUM(vol_sats) AS vol_sats, + SUM(vol_tokens) AS vol_tokens + FROM per_pool_tx + GROUP BY token_id, txid, ts +), +priceable AS ( + SELECT + token_id, + bucket_ts, + ABS(CAST(signed_sats AS REAL) / CAST(signed_tokens AS REAL)) AS price, + ROW_NUMBER() OVER (PARTITION BY token_id, bucket_ts ORDER BY ts ASC, txid ASC) AS rn_asc, + ROW_NUMBER() OVER (PARTITION BY token_id, bucket_ts ORDER BY ts DESC, txid DESC) AS rn_desc + FROM tx_trades + WHERE signed_tokens != 0 +), +ohlc AS ( + SELECT + token_id, + bucket_ts, + MAX(CASE WHEN rn_asc = 1 THEN price END) AS open, + MAX(CASE WHEN rn_desc = 1 THEN price END) AS close, + MAX(price) AS high, + MIN(price) AS low + FROM priceable + GROUP BY token_id, bucket_ts +), +vol AS ( + SELECT + token_id, + bucket_ts, + SUM(vol_sats) AS volume_sats, + SUM(vol_tokens) AS volume_tokens, + COUNT(*) AS tx_count + FROM tx_trades + GROUP BY token_id, bucket_ts +) +SELECT + ohlc.token_id, + ohlc.bucket_ts, + ohlc.open, + ohlc.high, + ohlc.low, + ohlc.close, + vol.volume_sats, + vol.volume_tokens, + vol.tx_count +FROM ohlc +JOIN vol ON ohlc.token_id = vol.token_id AND ohlc.bucket_ts = vol.bucket_ts +"#; + + let rows = sqlx::query(select_sql) + .bind(since_ts) + .bind(until_ts) + .fetch_all(read_pool) + .await?; + + if rows.is_empty() { + return Ok(0); + } + + // Phase 2: insert pre-computed rows inside a single transaction. + // The write lock is held only for these fast INSERTs, not during aggregation. + let mut tx = write_pool.begin().await?; + let mut inserted = 0u64; + for row in &rows { + let token_id: Vec = row.get(0); + let bucket_ts: i64 = row.get(1); + let open: f64 = row.get(2); + let high: f64 = row.get(3); + let low: f64 = row.get(4); + let close: f64 = row.get(5); + let volume_sats: i64 = row.get(6); + let volume_tokens: i64 = row.get(7); + let tx_count: i64 = row.get(8); + + inserted += sqlx::query( + "INSERT OR IGNORE INTO ohlcv_1h + (token_id, bucket_ts, open, high, low, close, volume_sats, volume_tokens, tx_count) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", + ) + .bind(token_id) + .bind(bucket_ts) + .bind(open) + .bind(high) + .bind(low) + .bind(close) + .bind(volume_sats) + .bind(volume_tokens) + .bind(tx_count) + .execute(&mut *tx) + .await? + .rows_affected(); + } + tx.commit().await?; + + Ok(inserted) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::db::cauldron::{pool as cauldron_pool, tx, utxo_funding}; + use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions}; + use std::sync::atomic::{AtomicU64, Ordering}; + + static OHLCV_TEST_COUNTER: AtomicU64 = AtomicU64::new(0); + + async fn test_pool() -> SqlitePool { + let id = OHLCV_TEST_COUNTER.fetch_add(1, Ordering::SeqCst); + let uri = format!("file:ohlcv_test_{}?mode=memory&cache=shared", id); + let opts = SqliteConnectOptions::new() + .filename(&uri) + .foreign_keys(false); + SqlitePoolOptions::new().connect_with(opts).await.unwrap() + } + + async fn setup_db(pool: &SqlitePool) { + tx::create_table(pool).await; + utxo_funding::create_table(pool).await; + cauldron_pool::create_table(pool).await; + create_table(pool).await; // ohlcv_1h + idx_phe_txid + } + + /// Insert a single confirmed trade via raw SQL (FK disabled in tests). + async fn insert_confirmed_trade( + pool: &SqlitePool, + txid: [u8; 32], + utxo: [u8; 32], + token_id: [u8; 32], + mtp_ts: i64, + sats_delta: i64, + token_delta: i64, + ) { + let blockhash = [0xAA_u8; 32]; + sqlx::query("INSERT INTO tx (txid, blockhash, mtp_timestamp) VALUES (?, ?, ?)") + .bind(txid.as_slice()) + .bind(blockhash.as_slice()) + .bind(mtp_ts) + .execute(pool) + .await + .unwrap(); + sqlx::query( + "INSERT INTO utxo_funding (new_utxo_hash, txid, spent_utxo_hash, new_utxo_txid, new_utxo_n, sats, token_amount, token_id) + VALUES (?, ?, ?, ?, 0, 1000, 1000, ?)", + ) + .bind(utxo.as_slice()) + .bind(txid.as_slice()) + .bind([0u8; 32].as_slice()) + .bind(txid.as_slice()) + .bind(token_id.as_slice()) + .execute(pool) + .await + .unwrap(); + let seq: i64 = + sqlx::query_scalar("SELECT IFNULL(MAX(sequence), 0) + 1 FROM pool_history_entry") + .fetch_one(pool) + .await + .unwrap(); + sqlx::query( + "INSERT INTO pool_history_entry + (utxo, pool, token_id, txid, tx_pos, mtp_timestamp, sequence, sats, token_amount, sats_delta, token_delta) + VALUES (?, ?, ?, ?, 0, ?, ?, 1000, 1000, ?, ?)", + ) + .bind(utxo.as_slice()) + .bind([0xBB_u8; 32].as_slice()) // dummy pool hash (FK disabled) + .bind(token_id.as_slice()) + .bind(txid.as_slice()) + .bind(mtp_ts) + .bind(seq) + .bind(sats_delta) + .bind(token_delta) + .execute(pool) + .await + .unwrap(); + } + + /// Insert a mempool-only trade (no blockhash on the tx row). + async fn insert_mempool_trade( + pool: &SqlitePool, + txid: [u8; 32], + utxo: [u8; 32], + token_id: [u8; 32], + first_seen_ts: i64, + ) { + sqlx::query("INSERT INTO tx (txid, first_seen_timestamp) VALUES (?, ?)") + .bind(txid.as_slice()) + .bind(first_seen_ts) + .execute(pool) + .await + .unwrap(); + sqlx::query( + "INSERT INTO utxo_funding (new_utxo_hash, txid, spent_utxo_hash, new_utxo_txid, new_utxo_n, sats, token_amount, token_id) + VALUES (?, ?, ?, ?, 0, 1000, 1000, ?)", + ) + .bind(utxo.as_slice()) + .bind(txid.as_slice()) + .bind([0u8; 32].as_slice()) + .bind(txid.as_slice()) + .bind(token_id.as_slice()) + .execute(pool) + .await + .unwrap(); + let seq: i64 = + sqlx::query_scalar("SELECT IFNULL(MAX(sequence), 0) + 1 FROM pool_history_entry") + .fetch_one(pool) + .await + .unwrap(); + sqlx::query( + "INSERT INTO pool_history_entry + (utxo, pool, token_id, txid, tx_pos, first_seen_timestamp, sequence, sats, token_amount, sats_delta, token_delta) + VALUES (?, ?, ?, ?, 0, ?, ?, 1000, 1000, -1000, 25)", + ) + .bind(utxo.as_slice()) + .bind([0xBB_u8; 32].as_slice()) + .bind(token_id.as_slice()) + .bind(txid.as_slice()) + .bind(first_seen_ts) + .bind(seq) + .execute(pool) + .await + .unwrap(); + } + + /// `get_min_trade_bucket_ts` should floor a mid-hour timestamp to the hour boundary. + #[tokio::test] + async fn test_get_min_trade_bucket_ts_floors_to_hour() { + let pool = test_pool().await; + setup_db(&pool).await; + + // Trade at 1727963400 — not hour-aligned; floor to 1727960400 + insert_confirmed_trade(&pool, [0x01; 32], [0x02; 32], [0x03; 32], 1727963400, -1000, 25) + .await; + + let result = get_min_trade_bucket_ts(&pool).await.unwrap(); + assert_eq!( + result, + Some(1727960400), + "1727963400 should floor to 1727960400" + ); + } + + /// `get_min_trade_bucket_ts` returns `None` when only unconfirmed (mempool) trades exist. + #[tokio::test] + async fn test_get_min_trade_bucket_ts_no_confirmed_trades() { + let pool = test_pool().await; + setup_db(&pool).await; + + insert_mempool_trade(&pool, [0x01; 32], [0x02; 32], [0x03; 32], 1727963400).await; + + let result = get_min_trade_bucket_ts(&pool).await.unwrap(); + assert_eq!(result, None, "mempool-only trades must not be returned"); + } + + /// Running `rebuild_range` twice on the same range inserts nothing on the second call + /// because INSERT OR IGNORE skips rows that already exist. + #[tokio::test] + async fn test_rebuild_range_idempotent() { + let pool = test_pool().await; + setup_db(&pool).await; + + insert_confirmed_trade(&pool, [0x01; 32], [0x02; 32], [0x03; 32], 1727963400, -1000, 25) + .await; + + let n1 = rebuild_range(&pool, &pool, 1727960400, 1727964000) + .await + .unwrap(); + assert!(n1 > 0, "first rebuild should insert at least one bucket"); + + let n2 = rebuild_range(&pool, &pool, 1727960400, 1727964000) + .await + .unwrap(); + assert_eq!(n2, 0, "second rebuild must insert nothing (INSERT OR IGNORE)"); + } + + /// Unconfirmed trades (tx.blockhash IS NULL) must not appear in `ohlcv_1h`. + #[tokio::test] + async fn test_rebuild_range_excludes_mempool() { + let pool = test_pool().await; + setup_db(&pool).await; + + insert_mempool_trade(&pool, [0x01; 32], [0x02; 32], [0x03; 32], 1727963400).await; + + let n = rebuild_range(&pool, &pool, 1727960400, 1727967600) + .await + .unwrap(); + assert_eq!(n, 0, "mempool trades must not be materialised"); + } +} + +pub struct OhlcvRow { + pub bucket_ts: i64, + pub open: f64, + pub high: f64, + pub low: f64, + pub close: f64, + pub volume_sats: i64, + pub volume_tokens: i64, + pub tx_count: i64, +} + +/// Return the materialised 1-hour candles for a single token in `[start, end)`. +/// Only buckets that had at least one trade are returned (gaps must be filled by the caller). +pub async fn get_active_candles( + pool: &SqlitePool, + token_blob: &[u8], + start: i64, + end: i64, +) -> Result> { + let rows = sqlx::query( + "SELECT bucket_ts, open, high, low, close, volume_sats, volume_tokens, tx_count + FROM ohlcv_1h + WHERE token_id = ? AND bucket_ts >= ? AND bucket_ts < ? + ORDER BY bucket_ts ASC", + ) + .bind(token_blob) + .bind(start) + .bind(end) + .fetch_all(pool) + .await?; + + Ok(rows + .into_iter() + .map(|r| OhlcvRow { + bucket_ts: r.get(0), + open: r.get(1), + high: r.get(2), + low: r.get(3), + close: r.get(4), + volume_sats: r.get(5), + volume_tokens: r.get(6), + tx_count: r.get(7), + }) + .collect()) +} diff --git a/src/db/cauldron/tx.rs b/src/db/cauldron/tx.rs index e4afda7..d2bba6c 100644 --- a/src/db/cauldron/tx.rs +++ b/src/db/cauldron/tx.rs @@ -75,17 +75,12 @@ pub async fn latest( ) -> Result, u64)>> { let (sql, token_blob) = match &token_id { Some(tid) => ( - "SELECT tx.txid, tx.blockhash, tx.mtp_timestamp, tx.first_seen_timestamp - FROM ( - SELECT txid, MAX(effective_timestamp) AS ts - FROM pool_history_entry - WHERE token_id = ?1 - GROUP BY txid - ORDER BY ts DESC - LIMIT ?2 OFFSET ?3 - ) AS recent - JOIN tx ON tx.txid = recent.txid - ORDER BY recent.ts DESC", + "SELECT DISTINCT tx.txid, tx.blockhash, tx.mtp_timestamp, tx.first_seen_timestamp + FROM tx + JOIN utxo_funding ON tx.txid = utxo_funding.txid + WHERE utxo_funding.token_id = ?1 + ORDER BY tx.effective_timestamp DESC + LIMIT ?2 OFFSET ?3", Some(tid.to_blob()), ), None => ( diff --git a/src/rpc/candlesticks.rs b/src/rpc/candlesticks.rs index 8243305..16b8b05 100644 --- a/src/rpc/candlesticks.rs +++ b/src/rpc/candlesticks.rs @@ -234,14 +234,15 @@ async fn fetch_raw_trades( WITH per_pool_tx_raw AS ( SELECT phe.txid, - phe.effective_timestamp, + tx.effective_timestamp, phe.utxo, phe.sats_delta, phe.token_delta FROM pool_history_entry AS phe + JOIN tx ON tx.txid = phe.txid WHERE phe.token_id = ? - AND phe.effective_timestamp >= ? - AND phe.effective_timestamp < ? + AND tx.effective_timestamp >= ? + AND tx.effective_timestamp < ? ), per_pool_tx AS ( SELECT