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

// Copyright (C) 2025 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

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

pub(crate) struct PoolFilters {
    pub timestamp_less_than: Option<u64>,
    pub token_id: Option<String>,
    pub owner: Option<String>,
}

#[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(crate) trait PoolVisitor {
    fn optional_fields_wanted(&self) -> u64;

    fn visit(&mut self, sats: u64, tokens: u64, optional_fields: OptionalFields) -> Result<bool>;
}

pub(crate) fn db_visit_pool_entries<T: PoolVisitor>(
    conn: &Connection,
    visitor: &mut T,
    filters: PoolFilters,
) -> Result<()> {
    let mut sql_filters = Vec::new();
    let mut params: Vec<rusqlite::types::ToSqlOutput> = Vec::new();
    let mut param_index = 1;

    if let Some(token_id) = &filters.token_id {
        sql_filters.push(format!("p.token_id = ?{}", param_index));
        params.push(rusqlite::types::ToSqlOutput::Owned(
            token_id.to_owned().into(),
        ));
        param_index += 1;
    }

    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,
            phe.sats,
            phe.token_amount,
            phe.txid,
            phe.tx_pos,
            p.token_id,
            p.creation_utxo
        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}
            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
                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);

    let mut final_params = vec![rusqlite::types::ToSqlOutput::Owned(
        (time_filter.unwrap_or(i64::MAX as u64) 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 !visitor.visit(sats, tokens, optional)? {
            break;
        }
    }

    Ok(())
}