// 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::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) -> &'static str { std::any::type_name::() } 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( duration: Option, limit: Option, db: &State, response_cache: &State, ) -> Result>, Custom> { // 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)> = 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 = list .into_par_iter() .map( |( token_id, trade_volume, trade_count, tvl_sats, tvl_token, best_contract_sats, best_contract_tokens, bcmr, )| { 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 }) }, ) .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)) }