riftenlabs-indexer/src/rpc/tokens.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 bitcoin_hashes::hex::{FromHex, ToHex};
use bitcoincash::TokenID;
use log::{info, warn};
use rocket::response::status::Custom;
use rocket::{get, State};
use serde_json::json;
use serde_json::Value;

use crate::db::cauldron::tokenlist::db_utils::{
    cache_first_pool_ts_if_empty, db_first_pool_creation_row, CachedSort,
};
use crate::db::cauldron::tokenlist::list_cached::{
    db_list_tokens_cached, db_list_tokens_cached_by_ids, TokenListItemCached,
};
use crate::db::cauldron::tokenlist::list_tokens_volume::{db_list_tokens_by_volume, TokenListItem};
use crate::db::search::db_search_tokens_cached;
use crate::db::DB;
use crate::db::{self};
use crate::rpc::ResponseCacheInner;
use crate::timeutil::time_now;
use rayon::prelude::*;

use super::err::{
    bad_request, db_error, not_found, service_unavailable, to_internal_error, ApiErrorCode,
    CachedApiResult,
};
use super::response::{cached_ok, CACHE_AGGREGATE, CACHE_IMMUTABLE};
use super::ResponseCache;

macro_rules! function_name {
    () => {{
        fn f() {}
        fn type_name_of<T>(_: T) -> &'static str {
            std::any::type_name::<T>()
        }
        let name = type_name_of(f);
        &name[..name.len() - 3] // trim trailing "::f" from the name
    }};
}

#[get("/tokens/list_by_volume")]
pub fn list_by_volume(
    db: &State<DB>,
    response_cache: &State<ResponseCache>,
) -> CachedApiResult<Value> {
    let thirty_days = 24 * 60 * 60 * 30;
    let duration = thirty_days;
    let limit = 250;

    let cache_key = function_name!().to_owned();

    let db_copy = db.inner().clone();
    let run_query = move || -> Result<Vec<TokenListItem>, Custom<String>> {
        #[allow(clippy::type_complexity)]
        let cauldron_db = db_copy.cauldron_r.get().map_err(to_internal_error)?;
        let bcmr_db = db_copy.bcmr_r.get().map_err(to_internal_error)?;

        let list: Vec<TokenListItem> =
            db_list_tokens_by_volume(&cauldron_db, &bcmr_db, duration, limit)
                .map_err(to_internal_error)?;
        Ok(list)
    };
    let (needs_update, cached_value) = {
        let mut cache = response_cache.lock().unwrap();

        match cache.get(&cache_key) {
            Some(c) => {
                let needs_update = !c.in_progress && time_now() > c.update_timestamp - 300;
                (needs_update, c.value.clone())
            }
            None => {
                cache.insert(
                    cache_key.clone(),
                    ResponseCacheInner {
                        update_timestamp: time_now(),
                        value: None,
                        in_progress: true,
                    },
                );
                (true, None)
            }
        }
    };
    if needs_update {
        info!("Update for {cache_key} triggered.");
        if let Some(entry) = response_cache.lock().unwrap().get_mut(&cache_key) {
            entry.in_progress = true;
        };
        let cache_copy = response_cache.inner().clone();

        std::thread::spawn(move || {
            let new_entry = match run_query() {
                Ok(r) => ResponseCacheInner {
                    value: Some(json!(r)),
                    in_progress: false,
                    update_timestamp: time_now(),
                },
                Err(e) => {
                    warn!("Failed to update cache for {cache_key}: {e:?}");
                    ResponseCacheInner {
                        value: None,
                        in_progress: false,
                        update_timestamp: time_now(),
                    }
                }
            };
            cache_copy
                .lock()
                .unwrap()
                .insert(cache_key.clone(), new_entry);
            info!("Updated cached value for {cache_key}");
        });
    }

    match cached_value {
        Some(r) => Ok(cached_ok(r, CACHE_AGGREGATE)),
        None => Err(service_unavailable(
            ApiErrorCode::CacheWarming,
            "Busy, try again in 30 seconds",
        )),
    }
}

#[get("/tokens/search_by_volume?<search_query>")]
pub fn search_by_volume(search_query: &str, db: &State<DB>) -> CachedApiResult<Vec<Value>> {
    let db_copy = db.inner().clone();
    let run_query = move || {
        #[allow(clippy::type_complexity)]
        let bcmr_db = db_copy.bcmr_r.get().map_err(db_error)?;
        let crc20_db = db_copy.crc20_r.get().map_err(db_error)?;

        let list: Vec<(String, Option<String>, Option<String>, u64)> =
            db::search::search_tokens_by_volume(
                &db_copy.cauldron_r,
                &bcmr_db,
                &crc20_db,
                search_query,
            )
            .map_err(db_error)?;

        let result: Vec<Value> = list
            .into_par_iter()
            .map(|(token_id, name, ticker, trade_volume)| {
                json!({
                    "token_id": token_id,
                    "name": name,
                    "ticker": ticker,
                    "trade_volume": trade_volume
                })
            })
            .collect();
        Ok(cached_ok(result, CACHE_AGGREGATE))
    };
    run_query()
}

#[get("/tokens/search_cached?<q>&<limit>&<offset>&<by>&<order>")]
pub fn search_cached(
    db: &State<DB>,
    q: Option<String>,
    limit: Option<usize>,
    offset: Option<usize>,
    by: Option<String>,
    order: Option<String>,
) -> CachedApiResult<Value> {
    let cauldron_db = db.cauldron_r.get().map_err(db_error)?;
    let bcmr_db = db.bcmr_r.get().map_err(db_error)?;

    let sort = parse_cached_sort(by, order);
    let limit = limit.unwrap_or(250);
    let offset = offset.unwrap_or(0);
    let query = q.unwrap_or_default();

    let items: Vec<TokenListItemCached> =
        db_search_tokens_cached(&cauldron_db, &bcmr_db, &query, sort, limit, offset)
            .map_err(db_error)?;

    Ok(cached_ok(json!(items), CACHE_AGGREGATE))
}

fn parse_cached_sort(by: Option<String>, order: Option<String>) -> CachedSort {
    let desc = matches!(order.as_deref(), Some("desc") | Some("DESC"));
    match by.as_deref() {
        Some("name") => {
            if desc {
                CachedSort::NameDesc
            } else {
                CachedSort::NameAsc
            }
        }
        Some("symbol") => {
            if desc {
                CachedSort::SymbolDesc
            } else {
                CachedSort::SymbolAsc
            }
        }
        Some("tvl") => {
            if desc {
                CachedSort::TvlDesc
            } else {
                CachedSort::TvlAsc
            }
        }
        Some("volume") => {
            if desc {
                CachedSort::VolumeDesc
            } else {
                CachedSort::VolumeAsc
            }
        }
        Some("change_24h_bp") => {
            if desc {
                CachedSort::Change24hDesc
            } else {
                CachedSort::Change24hAsc
            }
        }
        Some("change_7d_bp") => {
            if desc {
                CachedSort::Change7dDesc
            } else {
                CachedSort::Change7dAsc
            }
        }

        // USD sorts
        Some("price_usd") => {
            if desc {
                CachedSort::PriceUsdDesc
            } else {
                CachedSort::PriceUsdAsc
            }
        }
        Some("change_24h_usd_bp") => {
            if desc {
                CachedSort::Change24hUsdDesc
            } else {
                CachedSort::Change24hUsdAsc
            }
        }
        Some("change_7d_usd_bp") => {
            if desc {
                CachedSort::Change7dUsdDesc
            } else {
                CachedSort::Change7dUsdAsc
            }
        }

        // APY (30d) sorts — accept a few aliases
        Some("apy") | Some("apy_30d") | Some("apy_30d_bp") => {
            if desc {
                CachedSort::Apy30dDesc
            } else {
                CachedSort::Apy30dAsc
            }
        }

        // default
        _ => {
            if desc {
                CachedSort::ScoreDesc
            } else {
                CachedSort::ScoreAsc
            }
        }
    }
}

#[get("/tokens/list_cached?<limit>&<offset>&<by>&<order>")]
pub fn list_cached(
    db: &State<DB>,
    limit: Option<usize>,
    offset: Option<usize>,
    by: Option<String>,
    order: Option<String>,
) -> CachedApiResult<Value> {
    let limit = limit.unwrap_or(250);
    let offset = offset.unwrap_or(0);

    let cauldron_db = db.cauldron_r.get().map_err(db_error)?;
    let bcmr_db = db.bcmr_r.get().map_err(db_error)?;

    let sort = parse_cached_sort(by, order);
    let items: Vec<TokenListItemCached> =
        db_list_tokens_cached(&cauldron_db, &bcmr_db, limit, offset, sort).map_err(db_error)?;

    Ok(cached_ok(json!(items), CACHE_AGGREGATE))
}

#[get("/tokens/list_cached_by_ids?<ids>&<by>&<order>")]
pub fn list_cached_by_ids(
    db: &State<DB>,
    ids: &str,
    by: Option<String>,
    order: Option<String>,
) -> CachedApiResult<Value> {
    let cauldron_db = db.cauldron_r.get().map_err(db_error)?;
    let bcmr_db = db.bcmr_r.get().map_err(db_error)?;

    // split, trim, and normalize (lowercase is typical for hex IDs in DB)
    let token_ids: Vec<String> = ids
        .split(',')
        .map(|s| s.trim().to_lowercase())
        .filter(|s| !s.is_empty())
        .collect();

    if token_ids.is_empty() {
        return Ok(cached_ok(json!([]), CACHE_AGGREGATE));
    }

    let sort = parse_cached_sort(by, order);
    let items =
        db_list_tokens_cached_by_ids(&cauldron_db, &bcmr_db, &token_ids, sort).map_err(db_error)?;

    Ok(cached_ok(json!(items), CACHE_AGGREGATE))
}

#[get("/token/<token>/first_pool")]
pub fn first_pool_creation(token: &str, dbp: &State<DB>) -> CachedApiResult<Value> {
    let token = TokenID::from_hex(token).map_err(|e| {
        bad_request(
            ApiErrorCode::InvalidTokenId,
            &format!("Invalid token id: {e}"),
        )
    })?;
    let token_hex = token.to_hex();

    let conn = dbp.cauldron_r.get().map_err(db_error)?;

    match db_first_pool_creation_row(&conn, &token_hex) {
        Ok(Some((creation_utxo, txid, timestamp, block_height))) => {
            // Opportunistically cache the timestamp forever
            let cw = dbp.cauldron_w.get().map_err(db_error)?;
            if let Err(e) = cache_first_pool_ts_if_empty(&cw, &token_hex, timestamp) {
                // Non-fatal: return the data
                log::warn!("Failed to cache first_pool_ts for {token_hex}: {e}");
            }

            Ok(cached_ok(
                json!({
                    "token": token_hex,
                    "creation_utxo": creation_utxo,
                    "txid": txid,
                    "timestamp": timestamp,
                    "block_height": block_height
                }),
                CACHE_IMMUTABLE,
            ))
        }
        Ok(None) => Err(not_found(ApiErrorCode::PoolNotFound, "No pools for token")),
        Err(e) => Err(db_error(e)),
    }
}