riftenlabs-indexer/src/db/cauldron/pool.rs

// 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 std::{
    collections::{HashMap, VecDeque},
    sync::atomic::AtomicI64,
};

use crate::db::blob::{blob_to_display_hex, display_hex_to_blob, FromBlob, ToBlob};
use crate::def::PoolID;
use anyhow::{Context, Result};
use bitcoin_hashes::hex::ToHex;
use bitcoincash::TokenID;
use log::{debug, info, warn};
use malachite::Integer;
use riftenlabs_defi::{cauldron::ParsedContract, chainutil::OutPointHash};
use rust_decimal::prelude::Zero;
use serde::{Serialize, Serializer};
use sqlx::{Row, SqliteConnection, SqlitePool};

use crate::rpc::apy::PoolSnapshot;

fn serialize_integer_as_string<S>(integer: &Integer, serializer: S) -> Result<S::Ok, S::Error>
where
    S: Serializer,
{
    serializer.serialize_str(&integer.to_string())
}

pub async fn create_table(pool: &SqlitePool) {
    sqlx::query(
        "CREATE TABLE pool (
            creation_utxo BLOB PRIMARY KEY REFERENCES utxo_funding(new_utxo_hash) ON DELETE CASCADE,
            owner_pkh BLOB NOT NULL,
            token_id BLOB NOT NULL,
            withdrawn_in_utxo BLOB REFERENCES utxo_spending(spent_utxo_hash) ON DELETE SET NULL
        )",
    )
    .execute(pool)
    .await
    .expect("failed to create table pool");

    sqlx::query(
        "CREATE TABLE pool_history_entry (
            utxo BLOB PRIMARY KEY REFERENCES utxo_funding(new_utxo_hash) ON DELETE CASCADE,
            pool BLOB REFERENCES pool(creation_utxo) ON DELETE CASCADE,
            token_id BLOB NOT NULL,
            txid BLOB 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,
            sats_delta BIGINT NOT NULL,
            token_delta BIGINT NOT NULL
        )",
    )
    .execute(pool)
    .await
    .expect("failed to create table pool_history_entry");

    sqlx::query(
        "CREATE INDEX idx_pool_history_entry_pool_sequence ON pool_history_entry(pool, sequence)",
    )
    .execute(pool)
    .await
    .unwrap();
    sqlx::query("CREATE INDEX idx_pool_owner_pkh ON pool(owner_pkh)")
        .execute(pool)
        .await
        .unwrap();
    sqlx::query("CREATE INDEX idx_pool_withdrawn_in_utxo ON pool(withdrawn_in_utxo)")
        .execute(pool)
        .await
        .unwrap();
    sqlx::query("CREATE INDEX idx_pool_history_entry_pool_timestamp_sequence ON pool_history_entry (pool, effective_timestamp, sequence DESC)")
        .execute(pool).await.unwrap();
    sqlx::query("CREATE INDEX idx_pool_history_entry_token_id ON pool_history_entry(token_id)")
        .execute(pool)
        .await
        .unwrap();
}

async fn get_pool_by_utxo(
    conn: &mut SqliteConnection,
    utxo_hash: &OutPointHash,
) -> Result<Option<OutPointHash>> {
    let row: Option<(Vec<u8>,)> =
        sqlx::query_as("SELECT pool FROM pool_history_entry WHERE utxo = ?")
            .bind(utxo_hash.to_blob())
            .fetch_optional(&mut *conn)
            .await?;

    match row {
        Some((blob,)) => Ok(Some(
            OutPointHash::from_blob(&blob).expect("invalid original_utxo utxo in db"),
        )),
        None => Ok(None),
    }
}

pub async fn flag_as_withdrawn(
    conn: &mut SqliteConnection,
    pool_utxo: &OutPointHash,
    cauldron: &ParsedContract,
) -> Result<()> {
    sqlx::query("UPDATE pool SET withdrawn_in_utxo = ? WHERE creation_utxo = ?")
        .bind(cauldron.spent_utxo_hash.to_blob())
        .bind(pool_utxo.to_blob())
        .execute(&mut *conn)
        .await
        .map_err(|e| anyhow::anyhow!("failed flag pool as withdrawn. Original error: {:?}", e))?;

    Ok(())
}

pub async fn insert_new_pool(conn: &mut SqliteConnection, cauldron: &ParsedContract) -> Result<()> {
    sqlx::query(
        "INSERT OR REPLACE INTO pool (creation_utxo, owner_pkh, token_id, withdrawn_in_utxo) VALUES (?, ?, ?, ?)",
    )
    .bind(cauldron.new_utxo_hash.expect("outpoint hash for new pool missing").to_blob())
    .bind(cauldron.pkh.to_blob())
    .bind(cauldron.token_id.expect("token id for new pool missing").to_blob())
    .bind(None::<Vec<u8>>)
    .execute(&mut *conn)
    .await
    .map_err(|e| anyhow::anyhow!("failed to insert new pool. Original error: {:?}", e))?;

    Ok(())
}

/// Next sequence number in the `pool_history_entry` table
static NEXT_SEQUENCE: AtomicI64 = AtomicI64::new(-10);

pub async fn initialize_seq(pool: &SqlitePool) {
    let row: (i64,) = sqlx::query_as("SELECT IFNULL(MAX(sequence), 0) + 1 FROM pool_history_entry")
        .fetch_one(pool)
        .await
        .unwrap();
    NEXT_SEQUENCE.store(row.0, std::sync::atomic::Ordering::SeqCst);
}

#[allow(dead_code)]
pub fn dummy_init_seq() {
    if NEXT_SEQUENCE.load(std::sync::atomic::Ordering::SeqCst) < 0 {
        NEXT_SEQUENCE.store(42, std::sync::atomic::Ordering::SeqCst);
    }
}

pub async fn insert_pool_history_entry(
    conn: &mut SqliteConnection,
    pool: &OutPointHash,
    cauldron: &ParsedContract,
    mtp_timestamp: Option<u64>,
    first_seen_timestamp: Option<u64>,
    sats_delta: i64,
    token_delta: i64,
) -> Result<()> {
    let next_seq = NEXT_SEQUENCE.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
    assert!(next_seq >= 0);

    sqlx::query(
        "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),
            first_seen_timestamp = COALESCE(excluded.first_seen_timestamp, pool_history_entry.first_seen_timestamp),
            sequence = excluded.sequence",
    )
    .bind(cauldron.new_utxo_hash.expect("utxo hash on new pool history entry").to_blob())
    .bind(pool.to_blob())
    .bind(cauldron.token_id.expect("token id on new pool history entry").to_blob())
    .bind(cauldron.new_utxo_txid.expect("txid of new pool history entry").to_blob())
    .bind(cauldron.new_utxo_n.expect("utxo index of new pool history entry") as i64)
    .bind(mtp_timestamp.map(|t| t as i64))
    .bind(first_seen_timestamp.map(|t| t as i64))
    .bind(next_seq)
    .bind(cauldron.sats.map(|s| s as i64))
    .bind(cauldron.token_amount)
    .bind(sats_delta)
    .bind(token_delta)
    .execute(&mut *conn)
    .await
    .map_err(|e| anyhow::anyhow!("failed to insert pool history entry. Original error: {:?}", e))?;
    Ok(())
}

pub async fn update_pool_history(
    conn: &mut SqliteConnection,
    cauldrons: Vec<ParsedContract>,
    mtp_timestamp: Option<u64>,
    first_seen_timestamp: Option<u64>,
) -> Result<()> {
    if cauldrons.is_empty() {
        return Ok(());
    }

    let mut queue = VecDeque::from(cauldrons);

    while let Some(current) = queue.pop_front() {
        let (pool_utxo, is_new) =
            match get_pool_by_utxo(&mut *conn, &current.spent_utxo_hash).await? {
                Some(c) => (c, false),
                None => {
                    let has_parent = queue
                        .iter()
                        .any(|parent| Some(current.spent_utxo_hash) == parent.new_utxo_hash);

                    if has_parent {
                        queue.push_back(current);
                        continue;
                    }

                    if let Some(utxo) = current.new_utxo_hash {
                        assert!(!current.is_withdrawn);
                        (utxo, true)
                    } else {
                        assert!(current.is_withdrawn);
                        continue;
                    }
                }
            };

        if current.is_withdrawn {
            info!(
                "LP {} withdrawn in {}",
                pool_utxo.to_hex(),
                current.spent_utxo_hash.to_hex()
            );
            flag_as_withdrawn(&mut *conn, &pool_utxo, &current).await?;
        } else if is_new {
            info!(
                "Cauldron LP created in tx {}",
                current
                    .new_utxo_txid
                    .expect("expected txid in new LP")
                    .to_hex()
            );
            insert_new_pool(&mut *conn, &current).await?;
            insert_pool_history_entry(
                &mut *conn,
                &pool_utxo,
                &current,
                mtp_timestamp,
                first_seen_timestamp,
                0,
                0,
            )
            .await?;
        } else {
            debug!("New entry for pool {}", pool_utxo.to_hex());
            let prev_entry = get_pool_history_entry(&mut *conn, current.spent_utxo_hash).await?;
            let sats_delta = if let Some(current_sats) = current.sats {
                current_sats as i64 - prev_entry.sats as i64
            } else {
                0
            };
            let token_delta = if let Some(current_token_amount) = current.token_amount {
                current_token_amount - prev_entry.token_amount as i64
            } else {
                0
            };
            insert_pool_history_entry(
                &mut *conn,
                &pool_utxo,
                &current,
                mtp_timestamp,
                first_seen_timestamp,
                sats_delta,
                token_delta,
            )
            .await?;
        }
    }

    Ok(())
}

/// Strategy for which pool state to select when two pool states are available for a timestamp
#[derive(PartialEq, Eq)]
enum SnapshotSelection {
    UseBefore,
    UseAfter,
}

/// Get pools and their state nearest to given timestamp.
async fn get_nearest_entries(
    pool: &SqlitePool,
    timestamp: i64,
    token_id: Option<&str>,
    owner_pkh: Option<&str>,
    resolution: SnapshotSelection,
) -> Result<HashMap<String, PoolSnapshot>> {
    let token_blob = match token_id {
        Some(id) => Some(display_hex_to_blob::<TokenID>(id)?),
        None => None,
    };
    let owner_pkh_blob = match owner_pkh {
        Some(pkh) => Some(hex::decode(pkh)?),
        None => None,
    };

    let extra_filters = match (&token_blob, &owner_pkh_blob) {
        (Some(_), Some(_)) => "token_id = ?2 AND owner_pkh = ?3",
        (Some(_), None) => "token_id = ?2",
        (None, Some(_)) => "owner_pkh = ?2",
        (None, None) => "1=1",
    };

    let ts = "effective_timestamp";

    let preferred = match resolution {
        SnapshotSelection::UseBefore => "ts ASC",
        SnapshotSelection::UseAfter => "ts DESC",
    };

    let query = format!(
        "WITH NearestEntries AS (
            SELECT phe.pool, phe.sats, phe.token_amount, {ts} AS ts, phe.sequence
            FROM pool_history_entry phe
            WHERE {ts} <= ?1
            AND phe.pool IN (
                SELECT creation_utxo
                FROM pool
                WHERE {extra_filters}
                AND withdrawn_in_utxo IS NULL
            )
            GROUP BY phe.pool
            HAVING MAX({ts})

            UNION ALL

            SELECT phe.pool, phe.sats, phe.token_amount, {ts} AS ts, phe.sequence
            FROM pool_history_entry phe
            WHERE {ts} > ?1
            AND phe.pool IN (
                SELECT creation_utxo
                FROM pool
                WHERE {extra_filters}
                AND withdrawn_in_utxo IS NULL
            )
            GROUP BY phe.pool
            HAVING MIN({ts})
        )
        SELECT * FROM NearestEntries
        ORDER BY {preferred}",
    );

    let rows = match (&token_blob, &owner_pkh_blob) {
        (Some(t), Some(p)) => {
            sqlx::query(&query)
                .bind(timestamp)
                .bind(t)
                .bind(p)
                .fetch_all(pool)
                .await?
        }
        (Some(t), None) => {
            sqlx::query(&query)
                .bind(timestamp)
                .bind(t)
                .fetch_all(pool)
                .await?
        }
        (None, Some(p)) => {
            sqlx::query(&query)
                .bind(timestamp)
                .bind(p)
                .fetch_all(pool)
                .await?
        }
        (None, None) => sqlx::query(&query).bind(timestamp).fetch_all(pool).await?,
    };
    let mut pools: HashMap<String, PoolSnapshot> = HashMap::default();

    for row in rows {
        let pool_blob: Vec<u8> = row.get(0);
        let pool_id = blob_to_display_hex::<PoolID>(&pool_blob)?;
        let sats: i64 = row.get(1);
        let token_amount: i64 = row.get(2);
        let ts_val: i64 = row.get(3);

        let pool_snapshot = PoolSnapshot {
            pool_id: pool_id.clone(),
            sats: sats as u64,
            token_amount: token_amount as u64,
            timestamp: ts_val as u64,
        };

        if !pools.contains_key(&pool_snapshot.pool_id) {
            pools.insert(pool_snapshot.pool_id.clone(), pool_snapshot);
        }
    }

    Ok(pools)
}

/// A capital injection event: both-side-positive delta on a pool_history_entry row.
/// Carries the actual pool state immediately before and after the injection so callers
/// can split a pool period into clean sub-periods without synthetic values.
pub struct InjectionRecord {
    pub sats_before: u64,
    pub tokens_before: u64,
    pub sats_after: u64,
    pub tokens_after: u64,
    pub timestamp: u64,
}

/// Returns injection events (both-side-positive deltas) for a set of pools within a time window.
/// An injection is a pool state where both sats_delta >= 0 and token_delta >= 0, meaning
/// capital was added rather than a normal trade occurring.
/// Callers use the k_before/k_after ratio to neutralize the K jump in APY calculations.
pub async fn get_injections_between(
    pool: &SqlitePool,
    pool_id_blobs: &[Vec<u8>],
    min_start_ts: i64,
    end_ts: i64,
) -> Result<HashMap<String, Vec<InjectionRecord>>> {
    if pool_id_blobs.is_empty() {
        return Ok(HashMap::new());
    }

    let placeholders = (1..=pool_id_blobs.len())
        .map(|i| format!("?{i}"))
        .collect::<Vec<_>>()
        .join(",");
    let n = pool_id_blobs.len();
    let query = format!(
        "SELECT pool, sats, token_amount, sats_delta, token_delta, effective_timestamp
         FROM pool_history_entry
         WHERE pool IN ({placeholders})
           AND effective_timestamp > ?{start_bind}
           AND effective_timestamp <= ?{end_bind}
           AND sats_delta >= 0 AND token_delta >= 0
           AND (sats_delta > 0 OR token_delta > 0)",
        start_bind = n + 1,
        end_bind = n + 2,
    );

    let mut query_builder = sqlx::query(&query);
    for blob in pool_id_blobs {
        query_builder = query_builder.bind(blob.clone());
    }
    query_builder = query_builder.bind(min_start_ts).bind(end_ts);

    let rows = query_builder.fetch_all(pool).await?;
    let mut result: HashMap<String, Vec<InjectionRecord>> = HashMap::new();

    for row in rows {
        let pool_blob: Vec<u8> = row.get(0);
        let pool_id = blob_to_display_hex::<PoolID>(&pool_blob)?;
        let sats: i64 = row.get(1);
        let token_amount: i64 = row.get(2);
        let sats_delta: i64 = row.get(3);
        let token_delta: i64 = row.get(4);
        let ts: i64 = row.get(5);

        result.entry(pool_id).or_default().push(InjectionRecord {
            sats_before: (sats - sats_delta).max(0) as u64,
            tokens_before: (token_amount - token_delta).max(0) as u64,
            sats_after: sats.max(0) as u64,
            tokens_after: token_amount.max(0) as u64,
            timestamp: ts as u64,
        });
    }

    Ok(result)
}

/// Returns pool period snapshots filtered by token and/or owner PKH.
pub async fn get_pool_period_snapshot(
    pool: &SqlitePool,
    token_id: Option<&str>,
    owner_pkh: Option<&str>,
    start: i64,
    end: i64,
) -> Result<Vec<(PoolSnapshot, PoolSnapshot)>> {
    let pools_start = get_nearest_entries(
        pool,
        start,
        token_id,
        owner_pkh,
        SnapshotSelection::UseBefore,
    )
    .await?;
    let mut pools_end =
        get_nearest_entries(pool, end, token_id, owner_pkh, SnapshotSelection::UseAfter).await?;

    let window_end = end as u64;
    let mut pools: Vec<(PoolSnapshot, PoolSnapshot)> = Vec::default();
    for (start_pool_id, start_pool) in pools_start {
        let mut end_pool = match pools_end.remove(&start_pool_id) {
            Some(end) => end,
            None => {
                warn!("Found no 'end pool' for {start_pool_id}");
                continue;
            }
        };

        // Clamp the end snapshot timestamp to the window boundary.
        // For dormant pools, the nearest entry before `end` may have a timestamp far
        // in the past (e.g. the pool's only trade was hours after creation). Without
        // clamping, yield would be annualised over that short active window rather than
        // the full dormancy period, grossly inflating the reported APY.
        //
        // Guard: only clamp when the end snapshot is actually a newer state than the start.
        // If both snapshots point to the same pre-window trade (pool had no activity
        // during the window), their timestamps are equal and duration stays zero →
        // the pool is correctly excluded. Without this guard, clamping would give those
        // pools a non-zero duration and a zero yield, diluting the AAPY of active tokens.
        if end_pool.timestamp < window_end && end_pool.timestamp > start_pool.timestamp {
            end_pool.timestamp = window_end;
        }

        let duration = end_pool.timestamp.saturating_sub(start_pool.timestamp);
        if !duration.is_zero() {
            pools.push((start_pool, end_pool))
        }
    }

    if !pools_end.is_empty() {
        warn!("Found {} end pools not in start pools", pools_end.len());
    }

    Ok(pools)
}

/// Get the nearest pool history entries for a specific set of pool IDs.
async fn get_nearest_entries_by_pool_ids(
    pool: &SqlitePool,
    timestamp: i64,
    pool_id_blobs: &[Vec<u8>],
    resolution: SnapshotSelection,
) -> Result<HashMap<String, PoolSnapshot>> {
    let placeholders: String = (2..=(pool_id_blobs.len() + 1))
        .map(|i| format!("?{i}"))
        .collect::<Vec<_>>()
        .join(",");
    let pool_filter = format!("phe.pool IN ({placeholders})");

    let ts = "effective_timestamp";
    let preferred = match resolution {
        SnapshotSelection::UseBefore => "ts ASC",
        SnapshotSelection::UseAfter => "ts DESC",
    };

    let query = format!(
        "WITH NearestEntries AS (
            SELECT phe.pool, phe.sats, phe.token_amount, {ts} AS ts, phe.sequence
            FROM pool_history_entry phe
            WHERE {ts} <= ?1 AND {pool_filter}
            GROUP BY phe.pool
            HAVING MAX({ts})

            UNION ALL

            SELECT phe.pool, phe.sats, phe.token_amount, {ts} AS ts, phe.sequence
            FROM pool_history_entry phe
            WHERE {ts} > ?1 AND {pool_filter}
            GROUP BY phe.pool
            HAVING MIN({ts})
        )
        SELECT * FROM NearestEntries
        ORDER BY {preferred}",
    );

    let mut query_builder = sqlx::query(&query).bind(timestamp);
    for blob in pool_id_blobs {
        query_builder = query_builder.bind(blob.clone());
    }

    let rows = query_builder.fetch_all(pool).await?;
    let mut pools: HashMap<String, PoolSnapshot> = HashMap::default();

    for row in rows {
        let pool_blob: Vec<u8> = row.get(0);
        let pool_id = blob_to_display_hex::<PoolID>(&pool_blob)?;
        let sats: i64 = row.get(1);
        let token_amount: i64 = row.get(2);
        let ts_val: i64 = row.get(3);

        let pool_snapshot = PoolSnapshot {
            pool_id: pool_id.clone(),
            sats: sats as u64,
            token_amount: token_amount as u64,
            timestamp: ts_val as u64,
        };

        if !pools.contains_key(&pool_snapshot.pool_id) {
            pools.insert(pool_snapshot.pool_id.clone(), pool_snapshot);
        }
    }

    Ok(pools)
}

/// Returns pool period snapshots for a specific list of pool IDs.
/// Use this instead of `get_pool_period_snapshot` when filtering by pool ID —
/// it avoids a join to the `pool` table and works across future contract types.
pub async fn get_pool_period_snapshot_by_pool_ids(
    pool: &SqlitePool,
    pool_ids: &[String],
    start: i64,
    end: i64,
) -> Result<Vec<(PoolSnapshot, PoolSnapshot)>> {
    let pool_id_blobs: Result<Vec<Vec<u8>>> = pool_ids
        .iter()
        .map(|id| display_hex_to_blob::<PoolID>(id))
        .collect();
    let pool_id_blobs = pool_id_blobs?;

    let pools_start =
        get_nearest_entries_by_pool_ids(pool, start, &pool_id_blobs, SnapshotSelection::UseBefore)
            .await?;
    let mut pools_end =
        get_nearest_entries_by_pool_ids(pool, end, &pool_id_blobs, SnapshotSelection::UseAfter)
            .await?;

    let window_end = end as u64;
    let mut pools: Vec<(PoolSnapshot, PoolSnapshot)> = Vec::default();
    for (start_pool_id, start_pool) in pools_start {
        let mut end_pool = match pools_end.remove(&start_pool_id) {
            Some(end) => end,
            None => {
                warn!("Found no 'end pool' for {start_pool_id}");
                continue;
            }
        };

        // Clamp end snapshot to window boundary — see comment in get_pool_period_snapshot.
        if end_pool.timestamp < window_end && end_pool.timestamp > start_pool.timestamp {
            end_pool.timestamp = window_end;
        }

        let duration = end_pool.timestamp.saturating_sub(start_pool.timestamp);
        if !duration.is_zero() {
            pools.push((start_pool, end_pool))
        }
    }

    if !pools_end.is_empty() {
        warn!("Found {} end pools not in start pools", pools_end.len());
    }

    Ok(pools)
}

#[derive(Serialize)]
pub struct PoolHistoryEntry {
    txid: String,
    sats: u64,
    token_amount: u64,
    timestamp: u64,
    #[serde(serialize_with = "serialize_integer_as_string")]
    k: Integer,
}

async fn get_pool_history_entry(
    conn: &mut SqliteConnection,
    utxo_hash: OutPointHash,
) -> Result<PoolHistoryEntry> {
    let row = sqlx::query(
        "SELECT txid, sats, token_amount, effective_timestamp as timestamp FROM pool_history_entry WHERE utxo = ?",
    )
    .bind(utxo_hash.to_blob())
    .fetch_optional(&mut *conn)
    .await?
    .context("no pool history entry found for UTXO hash")?;

    let txid_blob: Vec<u8> = row.get(0);
    let txid = blob_to_display_hex::<bitcoincash::Txid>(&txid_blob)?;
    let sats: i64 = row.get(1);
    let token_amount: i64 = row.get(2);
    let timestamp: i64 = row.get(3);
    let sats = sats as u64;
    let token_amount = token_amount as u64;
    Ok(PoolHistoryEntry {
        txid,
        sats,
        token_amount,
        timestamp: timestamp as u64,
        k: Integer::from(sats) * Integer::from(token_amount),
    })
}

pub async fn db_pool_history(
    pool: &SqlitePool,
    pool_id: &PoolID,
    start_time: u64,
) -> Result<Vec<PoolHistoryEntry>> {
    let query = "SELECT
        phe.txid,
        phe.sats,
        phe.token_amount,
        phe.effective_timestamp as timestamp
    FROM
        pool_history_entry phe
    WHERE
        phe.pool = ?1
        AND timestamp >= ?2
    ORDER BY
        phe.sequence ASC;
    ";

    let rows = sqlx::query(query)
        .bind(pool_id.to_blob())
        .bind(start_time as i64)
        .fetch_all(pool)
        .await?;

    let mut history: Vec<PoolHistoryEntry> = Vec::default();

    for row in rows {
        let txid_blob: Vec<u8> = row.get(0);
        let txid = blob_to_display_hex::<bitcoincash::Txid>(&txid_blob)?;
        let sats: i64 = row.get(1);
        let token_amount: i64 = row.get(2);
        let timestamp: i64 = row.get(3);
        let sats = sats as u64;
        let token_amount = token_amount as u64;

        let k = Integer::from(sats) * Integer::from(token_amount);

        history.push(PoolHistoryEntry {
            txid,
            sats,
            token_amount,
            timestamp: timestamp as u64,
            k,
        });
    }

    Ok(history)
}

pub async fn db_pool_get_details(pool: &SqlitePool, pool_id: &PoolID) -> Result<(String, String)> {
    let row: (Vec<u8>, Vec<u8>) =
        sqlx::query_as("SELECT token_id, owner_pkh FROM pool WHERE creation_utxo = ?1")
            .bind(pool_id.to_blob())
            .fetch_one(pool)
            .await?;
    let token_hex = blob_to_display_hex::<TokenID>(&row.0)?;
    let owner_pkh_hex = hex::encode(&row.1);
    Ok((token_hex, owner_pkh_hex))
}

pub async fn db_pool_id_from_utxo(
    pool: &SqlitePool,
    utxo_hash: &OutPointHash,
) -> Result<Option<String>> {
    let row: Option<(Vec<u8>,)> =
        sqlx::query_as("SELECT pool FROM pool_history_entry WHERE utxo = ?")
            .bind(utxo_hash.to_blob())
            .fetch_optional(pool)
            .await?;

    match row {
        Some((pool_blob,)) => {
            let pool_id = blob_to_display_hex::<PoolID>(&pool_blob)?;
            Ok(Some(pool_id))
        }
        None => Ok(None),
    }
}

/// Get total volume in satoshis across all tokens for a given time period
pub async fn get_total_volume_sats(
    pool: &SqlitePool,
    start_timestamp: u64,
    end_timestamp: u64,
) -> anyhow::Result<i64> {
    let sql = "
    SELECT
        COALESCE(SUM(ABS(phe.sats_delta)), 0) AS 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 tx.effective_timestamp BETWEEN ? AND ?";

    let row: (i64,) = sqlx::query_as(sql)
        .bind(start_timestamp as i64)
        .bind(end_timestamp as i64)
        .fetch_one(pool)
        .await?;

    Ok(row.0)
}

/// Get volume in both satoshis and tokens for a specific token for a given time period
pub async fn get_token_volume_sats(
    pool: &SqlitePool,
    start_timestamp: u64,
    end_timestamp: u64,
    token_id: &str,
) -> anyhow::Result<(i64, i64)> {
    // CROSS JOIN forces tx-first join order: filter by timestamp (few rows),
    // then look up phe by txid, then filter pool by token_id.
    // Without CROSS JOIN, the planner starts from pool→phe (924K rows) → tx.
    let sql = "
    SELECT
        COALESCE(SUM(ABS(phe.sats_delta)), 0) AS token_volume_sats,
        COALESCE(SUM(ABS(phe.token_delta)), 0) AS token_volume_tokens
    FROM tx
    CROSS JOIN pool_history_entry phe ON tx.txid = phe.txid
    JOIN pool p ON phe.pool = p.creation_utxo
    WHERE tx.effective_timestamp BETWEEN ? AND ?
        AND p.token_id = ?";
    let token_blob = display_hex_to_blob::<TokenID>(token_id)?;
    let row: (i64, i64) = sqlx::query_as(sql)
        .bind(start_timestamp as i64)
        .bind(end_timestamp as i64)
        .bind(token_blob)
        .fetch_one(pool)
        .await?;

    Ok(row)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::db::cauldron::prepare_tables;
    use crate::db::cauldron::tx::{insert_block_tx, insert_mempool_tx};
    use crate::db::cauldron::utxo_funding::insert_utxo_funding;
    use bitcoin_hashes::hex::ToHex;
    use bitcoin_hashes::Hash;
    use bitcoincash::{BlockHash, PubkeyHash, Txid};
    use riftenlabs_defi::cauldron::ParsedContract;
    use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
    use std::sync::atomic::{AtomicU64, Ordering};

    static POOL_TEST_COUNTER: AtomicU64 = AtomicU64::new(0);

    async fn test_pool() -> SqlitePool {
        let id = POOL_TEST_COUNTER.fetch_add(1, Ordering::SeqCst);
        let uri = format!("file:pool_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_test_db(pool: &SqlitePool) {
        prepare_tables(pool).await;
        dummy_init_seq();
    }

    /// Helper to insert a simple pool with two history entries for testing get_pool_period_snapshot
    /// Inserts entries at t0 and t1 so the period has non-zero duration
    async fn seed_test_pool(
        pool: &SqlitePool,
        token: TokenID,
        owner_pkh: PubkeyHash,
        t0: i64,
        t1: i64,
        sats: u64,
        token_amount: i64,
    ) -> OutPointHash {
        // Generate unique hashes based on owner_pkh to avoid PK conflicts in DB
        let mut seed = [0u8; 32];

        // Use the hex representation to seed the buffer if into_inner() is missing
        // This ensures a unique, deterministic seed for each owner in the test
        let pkh_hex = owner_pkh.to_hex();
        let pkh_bytes = hex::decode(pkh_hex).unwrap();
        seed[0..20].copy_from_slice(&pkh_bytes);

        let pool_utxo_0 = OutPointHash::from_inner(seed);
        let mut seed_tx = seed;
        seed_tx[31] = 0xFF; // slightly different for txid
        let txid0 = Txid::from_inner(seed_tx);

        let txid1 = Txid::from_inner(seed); // different enough for test

        let contract0 = ParsedContract {
            pkh: owner_pkh,
            is_withdrawn: false,
            spent_utxo_hash: OutPointHash::all_zeros(),
            new_utxo_hash: Some(pool_utxo_0),
            new_utxo_txid: Some(txid0),
            new_utxo_n: Some(0),
            token_id: Some(token),
            sats: Some(sats),
            token_amount: Some(token_amount),
        };

        let mut conn = pool.acquire().await.unwrap();
        insert_new_pool(&mut conn, &contract0).await.unwrap();
        insert_utxo_funding(&mut conn, &vec![contract0.clone()], &txid0, true)
            .await
            .unwrap();
        insert_block_tx(&mut conn, &txid0, &BlockHash::all_zeros(), t0)
            .await
            .unwrap();
        insert_mempool_tx(&mut conn, &txid0, t0 as u64)
            .await
            .unwrap();
        insert_pool_history_entry(
            &mut conn,
            &pool_utxo_0,
            &contract0,
            Some(t0 as u64),
            Some(t0 as u64),
            0,
            0,
        )
        .await
        .unwrap();

        // Use a different UTXO for the second entry in the history to satisfy DB constraints
        let pool_utxo_1 = OutPointHash::from_inner(seed_tx);
        let contract1 = ParsedContract {
            pkh: owner_pkh,
            is_withdrawn: false,
            spent_utxo_hash: pool_utxo_0,
            new_utxo_hash: Some(pool_utxo_1),
            new_utxo_txid: Some(txid1),
            new_utxo_n: Some(0),
            token_id: Some(token),
            sats: Some(sats + 100),
            token_amount: Some(token_amount - 10),
        };

        insert_utxo_funding(&mut conn, &vec![contract1.clone()], &txid1, true)
            .await
            .unwrap();
        insert_block_tx(&mut conn, &txid1, &BlockHash::all_zeros(), t1)
            .await
            .unwrap();
        insert_mempool_tx(&mut conn, &txid1, t1 as u64)
            .await
            .unwrap();
        insert_pool_history_entry(
            &mut conn,
            &pool_utxo_0, // Recording history for the original pool
            &contract1,
            Some(t1 as u64),
            Some(t1 as u64),
            100,
            -10,
        )
        .await
        .unwrap();

        pool_utxo_0
    }

    #[tokio::test]
    async fn test_volume_no_trades_in_period() {
        let pool = test_pool().await;
        setup_test_db(&pool).await;

        let start_timestamp = 1755508856u64;
        let end_timestamp = 1755595256u64;
        let token_id = "f6677f3d3805d70949b375d36e094ff0ec9ece2a2cb1fde6d8b0e90b368f1f63";

        let result = get_token_volume_sats(&pool, start_timestamp, end_timestamp, token_id).await;
        let (sats_volume, token_volume) = result.unwrap();
        assert_eq!(sats_volume, 0);
        assert_eq!(token_volume, 0);

        let result = get_total_volume_sats(&pool, start_timestamp, end_timestamp).await;
        let total_volume = result.unwrap();
        assert_eq!(total_volume, 0);
    }

    #[tokio::test]
    async fn test_get_pool_period_snapshot_with_token_filter() {
        let db_pool = test_pool().await;
        setup_test_db(&db_pool).await;

        let token = TokenID::from_inner([0x01; 32]);
        let owner = PubkeyHash::all_zeros();
        let t0 = 1700000000i64;
        let t1 = 1700001000i64;

        seed_test_pool(&db_pool, token, owner, t0, t1, 1000, 500).await;

        let token_hex = token.to_hex();
        let result =
            get_pool_period_snapshot(&db_pool, Some(&token_hex), None, t0 - 100, t1 + 100).await;

        let pools = result.expect("query should not fail");
        assert!(
            !pools.is_empty(),
            "token filter should return matching pools (got 0)"
        );
    }

    #[tokio::test]
    async fn test_get_pool_period_snapshot_with_pkh_filter() {
        let db_pool = test_pool().await;
        setup_test_db(&db_pool).await;

        let token = TokenID::from_inner([0x02; 32]);
        let owner = PubkeyHash::from_inner([0x11; 20]);
        let t0 = 1700000000i64;
        let t1 = 1700001000i64;

        seed_test_pool(&db_pool, token, owner, t0, t1, 2000, 1000).await;

        let pkh_hex = owner.to_hex();
        let result =
            get_pool_period_snapshot(&db_pool, None, Some(&pkh_hex), t0 - 100, t1 + 100).await;

        let pools = result.expect("query should not fail");
        assert!(
            !pools.is_empty(),
            "pkh filter should return matching pools (got 0)"
        );
    }

    #[tokio::test]
    async fn test_get_pool_period_snapshot_by_pool_ids() {
        let db_pool = test_pool().await;
        setup_test_db(&db_pool).await;

        let token = TokenID::from_inner([0x02; 32]);
        let t0 = 1700000000i64;
        let t1 = 1700001000i64;

        let owner_1 = PubkeyHash::from_inner([0x11; 20]);
        let owner_2 = PubkeyHash::from_inner([0x22; 20]);

        let pool_id_1 = seed_test_pool(&db_pool, token, owner_1, t0, t1, 2000, 1000).await;
        let pool_id_2 = seed_test_pool(&db_pool, token, owner_2, t0, t1, 3000, 1500).await;

        // Filter by a single pool ID
        let result = get_pool_period_snapshot_by_pool_ids(
            &db_pool,
            &[pool_id_1.to_hex()],
            t0 - 100,
            t1 + 100,
        )
        .await;
        let pools = result.expect("single pool_id query should not fail");
        assert_eq!(pools.len(), 1, "Should return exactly 1 pool");

        // Filter by both pool IDs
        let result = get_pool_period_snapshot_by_pool_ids(
            &db_pool,
            &[pool_id_1.to_hex(), pool_id_2.to_hex()],
            t0 - 100,
            t1 + 100,
        )
        .await;
        let pools = result.expect("multi pool_id query should not fail");
        assert_eq!(pools.len(), 2, "Should return 2 pools for both IDs");
    }

    #[tokio::test]
    async fn test_get_pool_period_snapshot_pool_id_format() {
        let db_pool = test_pool().await;
        setup_test_db(&db_pool).await;

        let token = TokenID::from_inner([0x03; 32]);
        let owner = PubkeyHash::all_zeros();
        let t0 = 1700000000i64;
        let t1 = 1700001000i64;

        let expected_pool_id = seed_test_pool(&db_pool, token, owner, t0, t1, 3000, 1500).await;

        let result = get_pool_period_snapshot(&db_pool, None, None, t0 - 100, t1 + 100).await;

        let pools = result.expect("query should not fail");
        assert!(!pools.is_empty(), "should return at least one pool");

        let (start_pool, _end_pool) = &pools[0];

        assert_eq!(
            start_pool.pool_id.len(),
            64,
            "pool_id should be 64 hex chars, got: {}",
            start_pool.pool_id
        );
        assert!(
            start_pool.pool_id.chars().all(|c| c.is_ascii_hexdigit()),
            "pool_id should be valid hex, got: {}",
            start_pool.pool_id
        );

        assert_eq!(
            start_pool.pool_id,
            expected_pool_id.to_hex(),
            "pool_id should match expected"
        );
    }
}