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::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>(_: 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 || -> 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(Json(r)),
        None => Err(Custom(
            Status::ServiceUnavailable,
            "Busy, try again in 30 seconds".to_string(),
        )),
    }
}

#[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 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<String>, Option<String>, 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<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()
}