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::db;
use crate::db::cauldron::tokenlist::{db_list_tokens_by_volume, TokenListItem};
use crate::db::DB;
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")]
pub fn list_by_volume(
    db: &State<DB>,
    response_cache: &State<ResponseCache>,
) -> Result<Json<Value>, Custom<String>> {
    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 || {
        #[allow(clippy::type_complexity)]
        let list: Vec<TokenListItem> = db_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)))?;

        Ok(list)
    };

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

    if let Some(t) = response_age {
        // (potentially) update query for next request
        if time_now() > t - 120 {
            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(), json!(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()
}