Checkpoint

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<CandleCacheKey, Arc<Vec<CandlestickData>>>;
+
+/// 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<Vec<CandlestickData>>| v.len().max(1) as u32)
+        .build()
+}

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<Option<i64>> {
+    let row: Option<(Option<i64>,)> = 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<Option<i64>> {
+    let row: Option<(Option<i64>,)> = 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<u64> {
+    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<u8> = 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<Vec<OhlcvRow>> {
+    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())
+}

src/db/cauldron/tx.rs

@@ -75,17 +75,12 @@ pub async fn latest(
 ) -> Result<Vec<(Txid, Option<BlockHash>, u64)>> {
     let (sql, token_blob) = match &token_id {
         Some(tid) => (
-            "SELECT tx.txid, tx.blockhash, tx.mtp_timestamp, tx.first_seen_timestamp
+            "SELECT DISTINCT tx.txid, tx.blockhash, tx.mtp_timestamp, tx.first_seen_timestamp
-            FROM (
+            FROM tx
-                SELECT txid, MAX(effective_timestamp) AS ts
+            JOIN utxo_funding ON tx.txid = utxo_funding.txid
-                FROM pool_history_entry
+            WHERE utxo_funding.token_id = ?1
-                WHERE token_id = ?1
+            ORDER BY tx.effective_timestamp DESC
-                GROUP BY txid
+            LIMIT ?2 OFFSET ?3",
-                ORDER BY ts DESC
-                LIMIT ?2 OFFSET ?3
-            ) AS recent
-            JOIN tx ON tx.txid = recent.txid
-            ORDER BY recent.ts DESC",
             Some(tid.to_blob()),
         ),
         None => (

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 tx.effective_timestamp >= ?
-    AND phe.effective_timestamp < ?
+    AND tx.effective_timestamp < ?
 ),
 per_pool_tx AS (
   SELECT