riftenlabs-indexer/src/rpc/tokens.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 log::{info, warn};
use rocket::http::Status;
use rocket::{get, response::status::Custom, serde::json::Json, State};
use serde_json::json;
use serde_json::Value;

use crate::bcmr::parsedbcmr::ParsedBCMR;
use crate::db;
use crate::db::DB;
use crate::rpc;
use crate::timeutil::time_now;
use rayon::prelude::*;

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?<duration>&<limit>")]
pub fn list_by_volume(
    duration: Option<usize>,
    limit: Option<usize>,
    db: &State<DB>,
    response_cache: &State<ResponseCache>,
) -> Result<Json<Vec<Value>>, Custom<String>> {
    // For default query (as used on Cauldron DEX), use cache.
    let use_cache = duration.is_none() && limit.is_none();

    let thirty_days = 24 * 60 * 60 * 30;
    let duration = duration.unwrap_or(thirty_days);
    let limit = limit.unwrap_or(1000);
    let limit = std::cmp::min(limit, 1000);

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

    let db_copy = db.inner().clone();
    let run_query = move || {
        #[allow(clippy::type_complexity)]
        let list: Vec<(
            String,
            u64,
            u64,
            u64,
            u64,
            u64,
            u64,
            Option<ParsedBCMR>,
            Vec<ParsedBCMR>,
        )> = rpc::list_tokens_by_volume(
            &db_copy.cauldron_r.get().unwrap(),
            &db_copy.bcmr_r.get().unwrap(),
            duration,
            limit,
        )
        .map_err(|e| Custom(Status::InternalServerError, format!("Error: {}", e)))?;

        let result: Vec<Value> = list
            .into_par_iter()
            .map(
                |(
                    token_id,
                    trade_volume,
                    trade_count,
                    tvl_sats,
                    tvl_token,
                    best_contract_sats,
                    best_contract_tokens,
                    bcmr,
                    bcmr_well_known,
                )| {
                    json!({
                        "token_id": token_id,
                        "trade_volume": trade_volume,
                        "trade_count": trade_count,
                        "tvl_sats": tvl_sats,
                        "tvl_tokens": tvl_token,
                        "best_contract_sats": best_contract_sats,
                        "best_contract_tokens": best_contract_tokens,
                        "bcmr": bcmr,
                        "bcmr_well_known": bcmr_well_known
                    })
                },
            )
            .collect();
        Ok(result)
    };

    let (response_age, result) = if use_cache {
        let lock = response_cache.lock().unwrap();
        if let Some((time, value)) = lock.get(&cache_key) {
            (Some(*time), value.clone())
        } else {
            (None, run_query()?)
        }
    } else {
        (None, run_query()?)
    };

    if use_cache {
        if let Some(t) = response_age {
            // (potentially) update query for next request
            if time_now() > t - 60 {
                info!("update for cached value of {} triggered", cache_key);
                let cache_copy = response_cache.inner().clone();
                std::thread::spawn(move || {
                    match run_query() {
                        Ok(r) => {
                            cache_copy
                                .lock()
                                .unwrap()
                                .insert(cache_key.clone(), (time_now(), r));
                            info!("Updated cached value for {}", cache_key)
                        }
                        Err(e) => {
                            warn!("Failed to update cache for {}: {:?}", cache_key, e)
                        }
                    };
                });
            }
        } else {
            // first time querying (no old timestamp)
            response_cache
                .lock()
                .unwrap()
                .insert(cache_key, (time_now(), result.clone()));
        }
    }

    Ok(Json(result))
}

#[get("/tokens/search_by_volume?<search_query>")]
pub fn search_by_volume(
    search_query: &str,
    db: &State<DB>,
) -> Result<Json<Vec<Value>>, Custom<String>> {
    let db_copy = db.inner().clone();
    let run_query = move || {
        #[allow(clippy::type_complexity)]
        let list: Vec<(String, Option<String>, Option<String>, u64)> =
            db::search::search_tokens_by_volume(
                &db_copy.cauldron_r,
                &db_copy.bcmr_r.get().unwrap(),
                &db_copy.crc20_r.get().unwrap(),
                search_query,
            )
            .map_err(|e| Custom(Status::InternalServerError, format!("Error: {}", e)))?;

        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(Json(result))
    };
    run_query()
}