riftenlabs-indexer/src/db/search/mod.rs

// 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

use anyhow::{Context, Result};
use bitcoin_hashes::hex::FromHex;
use bitcoincash::TokenID;
use r2d2::Pool;
use r2d2_sqlite::SqliteConnectionManager;
use rayon::prelude::*;
use rusqlite::params;
use rusqlite::Connection;

type TokenBasicInfo = (String, Option<String>, Option<String>);
type TokenVolumeInfo = (String, Option<String>, Option<String>, u64);

use crate::db::bcmr::{get_token_bcmr, get_well_known_bcmr};
use crate::db::cauldron::tokenlist::db_utils::order_clause;
use crate::db::cauldron::tokenlist::db_utils::CachedSort;
use crate::db::cauldron::tokenlist::list_cached::TokenListItemCached;

fn search_bcmr(bcmr_conn: &Connection, search_query: &str) -> Result<Vec<TokenBasicInfo>> {
    let is_full_hex_token_id = TokenID::from_hex(search_query).is_ok();
    // Fetch the latest (highest) record that has BCMR data.
    let sql = if is_full_hex_token_id {
        "SELECT token_id, name, symbol
         FROM (
             SELECT a.token_id, b.name, b.symbol, a.height,
                    ROW_NUMBER() OVER (PARTITION BY a.token_id ORDER BY a.height DESC) AS rn
             FROM auth_chain_entry a
             LEFT JOIN bcmr_data b ON a.utxo = b.utxo
         ) subquery
         WHERE rn = 1
         AND (token_id = ?1);"
    } else {
        "SELECT token_id, name, symbol
         FROM (
             SELECT a.token_id, b.name, b.symbol, a.height,
                    ROW_NUMBER() OVER (PARTITION BY a.token_id ORDER BY a.height DESC) AS rn
             FROM auth_chain_entry a
             LEFT JOIN bcmr_data b ON a.utxo = b.utxo
             WHERE a.bcmr_data IS NOT NULL
         ) subquery
         WHERE rn = 1
         AND (name LIKE ?1 OR symbol LIKE ?1);"
    };

    let mut statement = bcmr_conn.prepare(sql)?;
    let search_pattern = if is_full_hex_token_id {
        search_query.to_string()
    } else {
        format!("%{search_query}%")
    };

    let mut bcmr_data: Vec<TokenBasicInfo> = Vec::new();
    let mut query = statement.query([search_pattern.as_str()])?;

    while let Some(result) = query.next()? {
        let token_id: String = result.get(0)?;
        let name: Option<String> = result.get(1)?;
        let ticker: Option<String> = result.get(2)?;
        bcmr_data.push((token_id, name, ticker));
    }

    Ok(bcmr_data)
}

fn search_crc20(crc20_conn: &Connection, search_query: &str) -> Result<Vec<TokenBasicInfo>> {
    let is_full_hex_token_id = TokenID::from_hex(search_query).is_ok();

    let sql = if is_full_hex_token_id {
        // Search by token_id if it's a full hex token ID
        "
      SELECT token_id, name, symbol
      FROM crc20
      WHERE token_id = ?1;
      "
    } else {
        // Only search by name or symbol otherwise
        "
      SELECT token_id, name, symbol
      FROM crc20
      WHERE name LIKE ?1 OR symbol LIKE ?1;
      "
    };
    let mut statement = crc20_conn.prepare(sql)?;
    let search_pattern = if is_full_hex_token_id {
        search_query.to_string()
    } else {
        format!("%{search_query}%")
    };

    let mut crc20_data: Vec<TokenBasicInfo> = Vec::new();
    let mut query = statement.query([&search_pattern])?;

    while let Some(result) = query.next()? {
        let token_id: String = result.get(0)?;
        let name: Option<String> = result.get(1)?;
        let ticker: Option<String> = result.get(2)?;
        crc20_data.push((token_id, name, ticker));
    }

    Ok(crc20_data)
}

fn token_volume(
    cauldron_pool: Pool<SqliteConnectionManager>,
    tokens: Vec<TokenBasicInfo>,
) -> Result<Vec<TokenVolumeInfo>> {
    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(
                    "
              WITH TradeData AS (
                  SELECT
                      p.token_id,
                      ABS(phe.sats_delta) as trade_volume
                  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 >= (strftime('%s', 'now') - 2592000)
                      AND p.token_id = ?
              )
              SELECT
                  token_id,
                  COALESCE(SUM(trade_volume), 0) as total_trade_volume
              FROM TradeData;
              ",
                )
                .context("Failed to prepare statement")?;

            let trade_volume: u64 = statement
                .query_row(params![&token_id], |row| row.get(1))
                .unwrap_or(0);

            // Return result as Ok tuple for successful case
            Ok((token_id, name, ticker, trade_volume))
        })
        .collect::<Result<Vec<_>>>()?; // Collect results into a Vec, propagating errors

    Ok(result)
}

pub fn search_tokens_by_volume(
    cauldron_pool: &Pool<SqliteConnectionManager>,
    bcmr_conn: &Connection,
    crc20_conn: &Connection,
    search_query: &str,
) -> Result<Vec<TokenVolumeInfo>> {
    let bcmr_data = search_bcmr(bcmr_conn, search_query)?;
    let crc20_data = search_crc20(crc20_conn, search_query)?;
    let combined_tokens = [bcmr_data, crc20_data].concat();

    let mut result = token_volume(cauldron_pool.clone(), combined_tokens)?;

    result.sort_by(|a, b| b.3.cmp(&a.3));

    Ok(result)
}

/// Search the cached list by token_id (exact), or display_name/symbol (LIKE),
/// then sort by *whatever* using CachedSort.
///
/// - `query`:
///     - empty/whitespace => returns everything
///     - full 32-byte hex => exact token_id match
///     - otherwise        => display_name LIKE or display_symbol LIKE
/// - `filter_zero_tvl`: skip rows with tvl_sats=0 (usually you want `true`)
pub fn db_search_tokens_cached(
    cauldron_conn: &Connection,
    bcmr_conn: &Connection,
    q: &str,
    sort: CachedSort,
    limit: usize,
    offset: usize,
) -> Result<Vec<TokenListItemCached>> {
    use rusqlite::types::ToSql;

    let order_sql = order_clause(sort);

    enum Filter {
        None,
        Id(String),
        Pat(String),
    }

    let mut where_sql = String::new();
    let filter = {
        let q = q.trim();
        if q.is_empty() {
            Filter::None
        } else if TokenID::from_hex(q).is_ok() {
            where_sql.push_str("WHERE token_id = :id");
            Filter::Id(q.to_lowercase())
        } else {
            where_sql.push_str(
                "WHERE (display_name LIKE :pat COLLATE NOCASE OR display_symbol LIKE :pat COLLATE NOCASE)"
            );
            Filter::Pat(format!("%{}%", q))
        }
    };

    let sql = format!(
        "SELECT
            token_id,
            trade_volume,
            tvl_sats,
            tvl_tokens,
            score,
            price_now,
            price_24h,
            price_7d,
            change_24h_bp,
            change_7d_bp,
            display_name,
            display_symbol,
            price_now_usd,
            price_24h_usd,
            price_7d_usd,
            change_24h_usd_bp,
            change_7d_usd_bp,
            apy_30d_bp,
            score_rank
         FROM cached_token_metrics
         {where_sql}
         {order_sql}
         LIMIT :limit OFFSET :offset"
    );

    let mut stmt = cauldron_conn.prepare(&sql)?;

    let limit_i64 = limit as i64;
    let offset_i64 = offset as i64;
    let mut params_vec: Vec<(&str, &dyn ToSql)> = Vec::with_capacity(3);

    match &filter {
        Filter::Id(s) => params_vec.push((":id", s as &dyn ToSql)),
        Filter::Pat(s) => params_vec.push((":pat", s as &dyn ToSql)),
        Filter::None => {}
    }
    params_vec.push((":limit", &limit_i64));
    params_vec.push((":offset", &offset_i64));

    let mut rows = stmt.query(&params_vec[..])?;
    let mut out = Vec::with_capacity(limit);

    while let Some(row) = rows.next()? {
        let token_id: String = row.get("token_id")?;
        let trade_volume: u64 = row.get("trade_volume")?;
        let tvl_sats: u64 = row.get::<_, i64>("tvl_sats")? as u64;
        let tvl_tokens: u64 = row.get::<_, i64>("tvl_tokens")? as u64;
        let score: i64 = row.get("score")?;
        let score_rank: i64 = row.get("score_rank")?;

        let price_now: f64 = row.get("price_now")?;
        let price_24h: Option<f64> = row.get("price_24h")?;
        let price_7d: Option<f64> = row.get("price_7d")?;
        let change_24h_bp: Option<i64> = row.get("change_24h_bp")?;
        let change_7d_bp: Option<i64> = row.get("change_7d_bp")?;

        let display_name: Option<String> = row.get("display_name")?;
        let display_symbol: Option<String> = row.get("display_symbol")?;

        let price_now_usd: f64 = row.get("price_now_usd")?;
        let price_24h_usd: Option<f64> = row.get("price_24h_usd")?;
        let price_7d_usd: Option<f64> = row.get("price_7d_usd")?;
        let change_24h_usd_bp: Option<i64> = row.get("change_24h_usd_bp")?;
        let change_7d_usd_bp: Option<i64> = row.get("change_7d_usd_bp")?;

        let apy_30d_bp: Option<i64> = row.get("apy_30d_bp")?;
        let bcmr = get_token_bcmr(bcmr_conn, &token_id)?;
        let bcmr_well_known = get_well_known_bcmr(bcmr_conn, &token_id)?;

        out.push(TokenListItemCached {
            token_id,
            trade_volume,
            tvl_sats,
            tvl_tokens,
            bcmr,
            bcmr_well_known,
            trade_count: 0,
            score,
            score_rank,
            price_now,
            price_24h,
            price_7d,
            change_24h_bp,
            change_7d_bp,
            display_name,
            display_symbol,
            price_now_usd,
            price_24h_usd,
            price_7d_usd,
            change_24h_usd_bp,
            change_7d_usd_bp,
            apy_30d_bp,
        });
    }

    Ok(out)
}

#[cfg(test)]
mod test;