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

// Copyright (C) 2025-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 rusqlite::Connection;

use anyhow::Result;

use crate::timeutil::time_now;

#[repr(u64)]
#[derive(Copy, Clone, Debug, PartialEq)]
pub(crate) enum OptionalPoolFields {
    Owner = 1,
    Txid = 1 << 2,
    TxPos = 1 << 3,
    TokenId = 1 << 4,
    PoolId = 1 << 5,
    Timestamp = 1 << 6,
}

#[derive(Default)]
pub(crate) struct PoolFilters {
    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>,
    pub txid: Option<String>,
    pub tx_pos: Option<u32>,
    pub token_id: Option<String>,
    pub pool_id: Option<String>,
    pub timestamp: Option<i64>,
}

pub(crate) trait PoolVisitor {
    fn optional_fields_wanted(&self) -> u64;

    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 (phe_timestamp_filter, time_params) = time_filters(&filters, 2); // Start at ?2, ?1 for withdrawn

    let mut sql_filters = Vec::new();
    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 "))
    };

    let query: String = "
        SELECT
            p.owner_pkh,
            phe.sats,
            phe.token_amount,
            phe.txid,
            phe.tx_pos,
            p.token_id,
            p.creation_utxo,
            phe.effective_timestamp
        FROM
            pool p
        JOIN pool_history_entry phe ON p.creation_utxo = phe.pool
        JOIN (
            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 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}
        "
    .replace("{filters}", &sql_filters_str)
    .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(
        (withdraw_time_filter as i64).into(),
    )];
    final_params.extend(params);

    let mut stmt = conn.prepare(&query)?;
    let mut rows = stmt.query(rusqlite::params_from_iter(final_params))?;

    let wanted = visitor.optional_fields_wanted();

    while let Some(row) = rows.next()? {
        let sats: u64 = row.get(1)?;
        let tokens: u64 = row.get(2)?;

        let mut optional: OptionalFields = OptionalFields::default();

        if (wanted & OptionalPoolFields::Owner as u64) != 0 {
            optional.owner = Some(row.get(0)?);
        }
        if (wanted & OptionalPoolFields::Txid as u64) != 0 {
            optional.txid = Some(row.get(3)?);
        }
        if (wanted & OptionalPoolFields::TxPos as u64) != 0 {
            optional.tx_pos = Some(row.get(4)?);
        }
        if (wanted & OptionalPoolFields::TokenId as u64) != 0 {
            optional.token_id = Some(row.get(5)?);
        }
        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;
        }
    }

    Ok(())
}