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

@@ -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'])

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

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, effective_timestamp);", []).unwrap();
-    conn.execute("CREATE INDEX idx_pool_history_entry_pool_timestamp ON pool_history_entry (pool, COALESCE(first_seen_timestamp, mtp_timestamp));",
-[],
-).unwrap();
 
     // Set the database version
     config::set_db_version(conn);

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<Option<OutPointHash>> {
@@ -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)
+        "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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+         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<PoolHistoryEntry> {
-    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(

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<u64>,
+    pub timestamp_lt: Option<u64>,
+    pub timestamp_lte: Option<u64>,
+    pub timestamp_gt: Option<u64>,
+    pub timestamp_gte: Option<u64>,
     pub token_id: Option<String>,
     pub owner: Option<String>,
 }
 
+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<String>,
@@ -32,6 +77,7 @@ pub(crate) struct OptionalFields {
     pub tx_pos: Option<u32>,
     pub token_id: Option<String>,
     pub pool_id: Option<String>,
+    pub timestamp: Option<i64>,
 }
 
 pub(crate) trait PoolVisitor {
@@ -40,46 +86,77 @@ pub(crate) trait PoolVisitor {
     fn visit(&mut self, sats: u64, tokens: u64, optional_fields: OptionalFields) -> Result<bool>;
 }
 
+fn time_filters(
+    filters: &PoolFilters,
+    param_index: usize,
+) -> (String, Vec<rusqlite::types::ToSqlOutput>) {
+    let now = time_now() as u64;
+
+    let mut clauses: Vec<String> = vec!["1=1".to_string()];
+    let mut params: Vec<rusqlite::types::ToSqlOutput> = 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<T: PoolVisitor>(
     conn: &Connection,
     visitor: &mut T,
     filters: PoolFilters,
 ) -> Result<()> {
-    let mut sql_filters = Vec::new();
+    let (phe_timestamp_filter, time_params) = time_filters(&filters, 2); // Start at ?2, ?1 for withdrawn
-    let mut params: Vec<rusqlite::types::ToSqlOutput> = Vec::new();
-    let mut param_index = 2; // Start from 2 since ?1 is used by time filter
 
-    if let Some(token_id) = &filters.token_id {
+    let mut sql_filters = Vec::new();
-        sql_filters.push(format!("p.token_id = ?{}", param_index));
+    let mut params: Vec<rusqlite::types::ToSqlOutput> = 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<T: PoolVisitor>(
             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<T: PoolVisitor>(
             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("{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<T: PoolVisitor>(
         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;

src/db/cauldron/tokenlist.rs

@@ -31,8 +31,7 @@ pub fn db_list_tokens_by_volume(
     seconds: usize,
     limit: usize,
 ) -> Result<Vec<TokenListItem>> {
-    let mut statement = cauldron_conn
+    let mut statement = cauldron_conn.prepare(
-        .prepare(
         "
             WITH TradeData AS (
                 SELECT
@@ -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

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],
         ),

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<ParsedContract>,
     txid: &Txid,
     replace: bool,

src/db/search/mod.rs

@@ -106,12 +106,12 @@ fn token_volume(
     let result: Vec<TokenVolumeInfo> = tokens
         .into_par_iter()
         .map(|(token_id, name, ticker)| {
-
             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
@@ -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,7 +128,8 @@ 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))

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 tx.effective_timestamp >= ?
-          AND COALESCE(tx.first_seen_timestamp, tx.mtp_timestamp) < ?
+          AND tx.effective_timestamp < ?
       GROUP BY tx.txid
   )
   SELECT

src/rpc/mod.rs

@@ -41,7 +41,7 @@ fn all_time_volume(db: &Connection, end_timestamp: u64) -> Result<Vec<(String, i
     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) <= ?
+    WHERE tx.effective_timestamp <= ?
     GROUP BY p.token_id";
 
     let mut stmt = db.prepare(sql)?;
@@ -70,7 +70,7 @@ fn period_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) BETWEEN ? AND ?
+    WHERE tx.effective_timestamp BETWEEN ? AND ?
     GROUP BY p.token_id
     ";
 

src/rpc/pool.rs

@@ -43,7 +43,7 @@ fn pools_by_apy(connection: &Connection) -> Result<Vec<PoolYield>> {
                 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<Vec<PoolYield>> {
                 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<Vec<PoolYield>> {
                 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}")))?;

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<bool> {
+        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<Decimal> {
+        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 = "
+    // Create a visitor to collect price data
-  WITH max_timestamps AS (
+    let mut visitor = PriceVisitor::default();
-      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
-  ";
 
-    let mut statement = connection.prepare(sql)?;
+    // note that we may return old timestamp; but it's still the last known trade at that time
-    let mut rows = statement.query(params![timestamp, token_id.to_hex()])?;
 
-    let mut sum_sats: u64 = 0;
+    // Use db_visit_pool_entries with filters for the specific token and timestamp
-    let mut sum_tokens: u64 = 0;
+    db_visit_pool_entries(
-    let mut latest_timestamp = 0;
+        connection,
+        &mut visitor,
+        PoolFilters::new()
+            .lte(timestamp as u64)
+            .token_id(&token_id.to_hex()),
+    )?;
 
-    while let Some(row) = rows.next()? {
+    // Check if we found any data
-        let token_amount: i64 = row.get(0)?;
+    if !visitor.has_data() {
-        let sats: i64 = row.get(1)?;
+        return Err(anyhow::anyhow!(
-        let row_timestamp: i64 = row.get(2)?;
+            "No price data found for token {} at timestamp {}",
-
+            token_id.to_hex(),
-        if sats >= 0 && token_amount >= 0 {
+            timestamp
-            sum_sats += sats as u64;
+        ));
-            sum_tokens += token_amount as u64;
     }
 
-        if row_timestamp > latest_timestamp {
+    // Calculate the price
-            latest_timestamp = row_timestamp;
+    let price = visitor.price()?;
-        }
+    let price_f64 = price.to_f64().context("Failed to convert price to f64")?;
-    }
 
-    let sum_sats_decimal = Decimal::from_u64(sum_sats).unwrap_or_default();
+    Ok((visitor.max_timestamp(), price_f64))
-    let sum_tokens_decimal = Decimal::from_u64(sum_tokens).unwrap_or_default();
-
-    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))
 }
 
 /// 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::update_pool_history;
-    use crate::{
+    use crate::db::cauldron::utxo_spending::{self, insert_utxo_spending};
-        db::cauldron::{
+    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},
-        },
-        utiltest::{calc_sats_delta, calc_token_delta},
     };
+    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 utxo1 = OutPointHash::from_inner([0xe0; 32]); // pool 1
-        let utxo2 = OutPointHash::from_inner([0xe1; 32]);
+        let utxo2 = OutPointHash::from_inner([0xe1; 32]); // pool 2
-        let utxo3 = OutPointHash::from_inner([0xe2; 32]);
+        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 cauldron1 = dummy_cauldron(
-        let cauldron2 = dummy_cauldron(&txid2, &utxo2, &token_zero, 60000, 2000, &pkh_zero);
+            &txid1,
-        let cauldron3 = dummy_cauldron(&txid2, &utxo3, &token_zero, 90000, 3000, &pkh_zero);
+            &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]);
+        update_pool_history(conn, vec![cauldron1.clone()], Some(TIME_1), Some(TIME_1)).unwrap();
-        let pool2 = OutPointHash::from_inner([0x0b; 32]);
+        update_pool_history(conn, vec![cauldron2], Some(TIME_2), Some(TIME_2)).unwrap();
-        let pool3 = OutPointHash::from_inner([0x0c; 32]);
+        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,
+            vec![cauldron1_newer],
-            &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,
+            1,
-            500,
+            1,
             &PubkeyHash::all_zeros(),
+            &pool1,
         );
-        insert_pool_history_entry(&conn, &pool1, &cauldron, Some(TIME_1), Some(TIME_1), 0, 0)
+        update_pool_history(&conn, vec![cauldron], Some(TIME_1), Some(TIME_1)).unwrap();
-            .unwrap();
+        // insert_utxo_funding(&conn, &vec![cauldron], &txid, true).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
+        // Expected price based on 1 sats and 1 tokens
-        let expected_price = 60.0;
+        let expected_price = 1.0;
-        assert!((actual_price - expected_price).abs() < 0.01);
+        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)

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<usize>)
         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,
         },

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<F>(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;