// 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::db::cauldron::tokenlist::{db_list_tokens_by_volume, TokenListItem}; use crate::db::DB; use crate::db::{self}; use crate::rpc::ResponseCacheInner; use crate::timeutil::time_now; use rayon::prelude::*; use super::err::to_internal_error; 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( db: &State, response_cache: &State, ) -> Result, Custom> { 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, Custom> { #[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 = 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 {} triggered.", cache_key); 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(Json(r)), None => Err(Custom( Status::ServiceUnavailable, "Busy, try again in 30 seconds".to_string(), )), } } #[get("/tokens/search_by_volume?")] pub fn search_by_volume( search_query: &str, db: &State, ) -> Result>, Custom> { let db_copy = db.inner().clone(); let run_query = move || { #[allow(clippy::type_complexity)] let bcmr_db = db_copy.bcmr_r.get().map_err(to_internal_error)?; let crc20_db = db_copy.crc20_r.get().map_err(to_internal_error)?; let list: Vec<(String, Option, Option, u64)> = db::search::search_tokens_by_volume( &db_copy.cauldron_r, &bcmr_db, &crc20_db, search_query, ) .map_err(|e| Custom(Status::InternalServerError, format!("Error: {}", e)))?; let result: Vec = 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() }