riftenlabs-indexer/src/rpc/apy/mod.rs

// Copyright (C) 2024-2026 Whiterun LLC
//
// 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 poolperiod::PoolPeriod;
use rocket::{get, State};
use serde_json::Value;

use crate::{
    db::{
        blob::display_hex_to_blob,
        cauldron::pool::{
            get_injections_between, get_pool_period_snapshot, get_pool_period_snapshot_by_pool_ids,
        },
        DB,
    },
    def::PoolID,
    rpc::err::{bad_request, db_error, ApiErrorCode, CachedApiResult},
    rpc::response::{cached_ok, CACHE_AGGREGATE},
    rpc::validate::{parse_pkh, parse_token_id},
    timeutil::time_now,
};
use serde_json::json;

pub mod apyaggregator;
pub mod poolperiod;

#[derive(Clone, Debug)]
pub struct PoolSnapshot {
    pub pool_id: String,
    pub timestamp: u64,
    pub sats: u64,
    pub token_amount: u64,
}

impl PoolSnapshot {
    #[allow(dead_code)] // used in unit tests
    pub fn dummy(timestamp: u64, sats: u64, token_amount: u64) -> Self {
        Self {
            pool_id: "dummy".to_string(),
            timestamp,
            sats,
            token_amount,
        }
    }
}

/// Fetch apy for a token and/or an account within a given time interval. All variables are optional.
/// A query with no variables will return the AAPY based on all users and all tokens aggregated.
///
/// Status: Stable
///
/// - token: The 32 byte token ID
/// - pkh: Public key hash for a single wallet account
/// - pool: One or more pool IDs to scope the APY calculation to specific pools.
///         Use this for per-user APY: first resolve pool IDs via `/pool/active?pkh=<pkh>`,
///         then pass them here. Supports multiple wallets (collect IDs from each PKH first).
///         **Cannot be combined with token or pkh** — returns 400 if mixed.
/// - start: Unix timestamp for period start (default 30 days)
/// - end: Unix timestamp for period end (default NOW)
///
/// **Response Example:**
///
/// ```json
/// {"apy":"10.00","pools":100}
/// ```
///
#[get("/pool/aggregated_apy?<token>&<pkh>&<pool>&<start>&<end>")]
pub async fn aggregate_apy(
    token: Option<&str>,
    pkh: Option<&str>,
    pool: Vec<&str>,
    start: Option<i64>,
    end: Option<i64>,
    db: &State<DB>,
) -> CachedApiResult<Value> {
    let end = end.unwrap_or(time_now());
    let start = start.unwrap_or(end - (3600 * 24 * 30)); // 30 days

    if !pool.is_empty() && (token.is_some() || pkh.is_some()) {
        return Err(bad_request(
            ApiErrorCode::InvalidParameters,
            "pool cannot be combined with token or pkh; use pool alone to filter by specific pools",
        ));
    }

    if end < start {
        return Err(bad_request(
            ApiErrorCode::InvalidTimeRange,
            "end time cannot be less than start time",
        ));
    }
    if start < 0 {
        return Err(bad_request(
            ApiErrorCode::InvalidTimeRange,
            "start time cannot be negative",
        ));
    }

    let pool_snapshots = if !pool.is_empty() {
        let ids: Vec<String> = pool.into_iter().map(|s| s.to_string()).collect();
        get_pool_period_snapshot_by_pool_ids(&db.cauldron_r, &ids, start, end)
            .await
            .map_err(db_error)?
    } else {
        let token = token.map(parse_token_id).transpose()?;
        let pkh = pkh.map(parse_pkh).transpose()?;

        get_pool_period_snapshot(&db.cauldron_r, token.as_deref(), pkh.as_deref(), start, end)
            .await
            .map_err(db_error)?
    };

    // Fetch injection events (capital additions) for all pools in the result.
    // Each pool period is split into sub-periods at injection boundaries so the
    // APY calculation only sees real fee growth, never injected capital.
    let pool_id_blobs: Vec<Vec<u8>> = pool_snapshots
        .iter()
        .filter_map(|(s, _)| display_hex_to_blob::<PoolID>(&s.pool_id).ok())
        .collect();

    let min_start_ts = pool_snapshots
        .iter()
        .map(|(s, _)| s.timestamp as i64)
        .min()
        .unwrap_or(start);

    let injections_map = get_injections_between(&db.cauldron_r, &pool_id_blobs, min_start_ts, end)
        .await
        .map_err(db_error)?;

    let pools_count = pool_snapshots.len();
    let pools: anyhow::Result<Vec<PoolPeriod>> = pool_snapshots
        .into_iter()
        .flat_map(|(start_snap, end_snap)| {
            let injections = injections_map
                .get(&start_snap.pool_id)
                .map(|entries| {
                    entries
                        .iter()
                        .filter(|r| {
                            r.timestamp > start_snap.timestamp && r.timestamp < end_snap.timestamp
                        })
                        .collect::<Vec<_>>()
                })
                .unwrap_or_default();
            poolperiod::split_at_injections(start_snap, end_snap, injections)
        })
        .collect();

    let pools = pools.map_err(db_error)?;
    let apy = apyaggregator::APYAggregator::aggregate_apy(pools.into_iter(), Some(start as u64))
        .map_err(db_error)?;

    Ok(cached_ok(
        json!({
            "apy": apy.to_string(),
            "pools": pools_count,
        }),
        CACHE_AGGREGATE,
    ))
}