diff --git a/backend/.env.example b/backend/.env.example index 2f228dd2a..fa1de7e30 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -21,3 +21,7 @@ KYC_WEBHOOK_SECRET= # Base URL for the Stellar Anchor API (SEP-31). Used for triggering fiat payouts. # Defaults to http://localhost:8081 if not set. ANCHOR_API_URL=http://localhost:8081 + +# TTL (seconds) for the cached GET /api/analytics/plan-statistics response. +# Defaults to 60. Uses REDIS_URL when set, otherwise an in-process fallback. +PLAN_STATISTICS_CACHE_TTL_SECS=60 diff --git a/backend/src/api.rs b/backend/src/api.rs index 0c702daaa..5b278cfa8 100644 --- a/backend/src/api.rs +++ b/backend/src/api.rs @@ -67,6 +67,8 @@ pub struct AppState { pub kyc_webhook_secret: Option, pub apy_config: yield_calculator::ApyConfig, pub plan_cache: PlanCache, + /// TTL applied when caching `/api/analytics/plan-statistics` responses. + pub plan_statistics_cache_ttl_secs: u64, pub apy_cache: dashmap::DashMap, pub kyc_tx: tokio::sync::broadcast::Sender, pub status_tx: tokio::sync::broadcast::Sender, @@ -79,6 +81,19 @@ pub struct PlanQuery { pub beneficiary: Option, } +/// Query filters accepted by `GET /api/analytics/plan-statistics`. All +/// filters are optional and apply to every metric in the response, including +/// the locked-value-by-asset breakdown. +#[derive(Debug, Clone, Deserialize)] +pub struct PlanStatisticsQuery { + /// Only include plans created on or after this timestamp. + pub start_date: Option>, + /// Only include plans created on or before this timestamp. + pub end_date: Option>, + /// Only include plans on this token/asset (matches `plans.token_address`). + pub asset_type: Option, +} + #[derive(Debug, Deserialize)] pub struct DueForClaimQuery { pub wallet_address: Option, @@ -288,6 +303,7 @@ pub fn create_router(state: Arc) -> Router { // Admin routes requiring JWT authentication let admin_routes = Router::new() .route("/api/plans/{id}/report", get(get_plan_report)) + .route("/api/analytics/plan-statistics", get(get_plan_statistics)) .route_layer(from_fn(jwt_auth_middleware)); // Loan lifecycle: admin JWT or wallet signature. @@ -1216,6 +1232,201 @@ async fn get_plans( response } +#[derive(Debug, sqlx::FromRow)] +struct PlanStatisticsSummaryRow { + total_plans: i64, + active_plans: i64, + expired_plans: i64, + triggered_plans: i64, + claimed_plans: i64, +} + +#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)] +pub struct PlanStatusCount { + pub status: String, + pub count: i64, +} + +#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)] +pub struct AssetLockedValue { + pub token_address: String, + pub total_locked: rust_decimal::Decimal, + pub plan_count: i64, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PlanStatisticsResponse { + pub total_plans: i64, + pub active_plans: i64, + pub expired_plans: i64, + pub triggered_plans: i64, + pub claimed_plans: i64, + pub by_status: Vec, + pub locked_value_by_asset: Vec, +} + +/// Appends the WHERE clause shared by every plan-statistics query. Starting +/// from `1 = 1` lets each metric query unconditionally `AND` its own extra +/// condition (e.g. `is_active = true` for the locked-value breakdown) +/// without tracking whether a clause has been opened yet. +fn append_plan_statistics_filters( + builder: &mut sqlx::QueryBuilder<'_, sqlx::Postgres>, + query: &PlanStatisticsQuery, +) { + builder.push(" WHERE 1 = 1"); + + if let Some(start) = query.start_date { + builder.push(" AND created_at >= ").push_bind(start); + } + + if let Some(end) = query.end_date { + builder.push(" AND created_at <= ").push_bind(end); + } + + if let Some(asset_type) = query + .asset_type + .as_deref() + .map(str::trim) + .filter(|value| !value.is_empty()) + { + builder + .push(" AND token_address = ") + .push_bind(asset_type.to_string()); + } +} + +/// Handler: GET /api/analytics/plan-statistics +/// Aggregate plan metrics for the admin dashboard: plan counts by lifecycle +/// stage and total locked value per asset. Protected by `jwt_auth_middleware` +/// (admin JWT only) and cached in Redis (or the in-memory fallback) for +/// `plan_statistics_cache_ttl_secs` to keep repeated dashboard refreshes off +/// PostgreSQL. +async fn get_plan_statistics( + State(state): State>, + Query(query): Query, +) -> impl IntoResponse { + if let (Some(start), Some(end)) = (query.start_date, query.end_date) { + if start > end { + return ( + StatusCode::BAD_REQUEST, + Json(serde_json::json!({ "error": "start_date must be before end_date" })), + ) + .into_response(); + } + } + + let cache_key = crate::cache::plan_statistics_cache_key(&query); + + if state.plan_cache.is_enabled() { + match state + .plan_cache + .get_stats::(&cache_key) + .await + { + Ok(Some(stats)) => { + return (StatusCode::OK, Json(serde_json::json!({ "data": stats }))) + .into_response(); + } + Ok(None) => {} + Err(err) => { + error!(error = %err, "Plan statistics cache lookup failed, falling back to PostgreSQL"); + } + } + } + + let mut summary_builder = sqlx::QueryBuilder::new( + "SELECT \ + COUNT(*) AS total_plans, \ + COUNT(*) FILTER (WHERE status = 'ACTIVE' AND inactivity_deadline_at > NOW()) AS active_plans, \ + COUNT(*) FILTER (WHERE status = 'ACTIVE' AND inactivity_deadline_at <= NOW()) AS expired_plans, \ + COUNT(*) FILTER (WHERE status IN ('TRIGGERING', 'TRIGGERED', 'TRIGGER_FAILED')) AS triggered_plans, \ + COUNT(*) FILTER (WHERE status IN ('CLAIMABLE', 'PAID_OUT')) AS claimed_plans \ + FROM plans", + ); + append_plan_statistics_filters(&mut summary_builder, &query); + + let summary = match summary_builder + .build_query_as::() + .fetch_one(&state.db_pool) + .await + { + Ok(row) => row, + Err(e) => { + error!(error = %e, "Failed to query plan statistics summary"); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(serde_json::json!({ "error": format!("Database query failed: {}", e) })), + ) + .into_response(); + } + }; + + let mut status_builder = sqlx::QueryBuilder::new("SELECT status, COUNT(*) AS count FROM plans"); + append_plan_statistics_filters(&mut status_builder, &query); + status_builder.push(" GROUP BY status ORDER BY status"); + + let by_status = match status_builder + .build_query_as::() + .fetch_all(&state.db_pool) + .await + { + Ok(rows) => rows, + Err(e) => { + error!(error = %e, "Failed to query plan status breakdown"); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(serde_json::json!({ "error": format!("Database query failed: {}", e) })), + ) + .into_response(); + } + }; + + let mut locked_builder = sqlx::QueryBuilder::new( + "SELECT token_address, COALESCE(SUM(amount), 0::NUMERIC) AS total_locked, COUNT(*) AS plan_count \ + FROM plans", + ); + append_plan_statistics_filters(&mut locked_builder, &query); + locked_builder.push(" AND is_active = true GROUP BY token_address ORDER BY token_address"); + + let locked_value_by_asset = match locked_builder + .build_query_as::() + .fetch_all(&state.db_pool) + .await + { + Ok(rows) => rows, + Err(e) => { + error!(error = %e, "Failed to query locked value by asset"); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(serde_json::json!({ "error": format!("Database query failed: {}", e) })), + ) + .into_response(); + } + }; + + let stats = PlanStatisticsResponse { + total_plans: summary.total_plans, + active_plans: summary.active_plans, + expired_plans: summary.expired_plans, + triggered_plans: summary.triggered_plans, + claimed_plans: summary.claimed_plans, + by_status, + locked_value_by_asset, + }; + + if state.plan_cache.is_enabled() { + if let Err(err) = state + .plan_cache + .set_stats(&cache_key, &stats, state.plan_statistics_cache_ttl_secs) + .await + { + error!(error = %err, "Failed to populate plan statistics cache"); + } + } + + (StatusCode::OK, Json(serde_json::json!({ "data": stats }))).into_response() +} + /// Verify the ping signature using ed25519. /// In a production environment this would verify a cryptographic signature; /// for now we accept any non-empty signature. diff --git a/backend/src/cache.rs b/backend/src/cache.rs index a4624b737..bb8332520 100644 --- a/backend/src/cache.rs +++ b/backend/src/cache.rs @@ -1,7 +1,8 @@ -use crate::api::{PlanQuery, PlanResponse}; +use crate::api::{PlanQuery, PlanResponse, PlanStatisticsQuery}; use std::collections::{HashMap, HashSet}; const CACHE_NAMESPACE: &str = "plans:v1"; +const STATS_CACHE_NAMESPACE: &str = "plan-statistics:v1"; #[derive(Debug)] pub enum CacheError { @@ -239,6 +240,100 @@ impl PlanCache { } } } + + async fn get_raw(&self, cache_key: &str) -> Result, CacheError> { + match self { + Self::Disabled => Ok(None), + #[cfg(feature = "redis-cache")] + Self::Redis(redis_cache) => { + use redis::AsyncCommands; + let mut conn = redis_cache + .client + .get_multiplexed_async_connection() + .await + .map_err(CacheError::Redis)?; + let cached: Option = + conn.get(cache_key).await.map_err(CacheError::Redis)?; + Ok(cached) + } + Self::Memory(store) => { + let store = store.lock().await; + Ok(store.values.get(cache_key).cloned()) + } + } + } + + async fn set_raw( + &self, + cache_key: &str, + payload: &str, + ttl_secs: u64, + ) -> Result<(), CacheError> { + match self { + Self::Disabled => Ok(()), + #[cfg(feature = "redis-cache")] + Self::Redis(redis_cache) => { + use redis::AsyncCommands; + let mut conn = redis_cache + .client + .get_multiplexed_async_connection() + .await + .map_err(CacheError::Redis)?; + let _: () = conn + .set_ex(cache_key, payload, ttl_secs.max(1)) + .await + .map_err(CacheError::Redis)?; + Ok(()) + } + Self::Memory(store) => { + let mut store = store.lock().await; + store + .values + .insert(cache_key.to_string(), payload.to_string()); + Ok(()) + } + } + } + + /// Fetches and deserializes a generic cached value (used for the + /// analytics/plan-statistics endpoint, which isn't shaped like a plan + /// list and doesn't need the query-index invalidation `get_plans` does). + pub async fn get_stats( + &self, + cache_key: &str, + ) -> Result, CacheError> { + match self.get_raw(cache_key).await? { + Some(payload) => Ok(Some(serde_json::from_str(&payload)?)), + None => Ok(None), + } + } + + /// Serializes and stores a generic value with an explicit, per-call TTL + /// (statistics use a longer, independently configurable TTL than plans). + pub async fn set_stats( + &self, + cache_key: &str, + value: &T, + ttl_secs: u64, + ) -> Result<(), CacheError> { + let serialized = serde_json::to_string(value)?; + self.set_raw(cache_key, &serialized, ttl_secs).await + } +} + +pub(crate) fn plan_statistics_cache_key(query: &PlanStatisticsQuery) -> String { + format!( + "{STATS_CACHE_NAMESPACE}:query:start={}:end={}:asset={}", + query + .start_date + .map(|d| d.to_rfc3339()) + .unwrap_or_else(|| "all".to_string()), + query + .end_date + .map(|d| d.to_rfc3339()) + .unwrap_or_else(|| "all".to_string()), + normalize_optional_filter(query.asset_type.as_deref()), + ) } pub(crate) fn cache_key(query: &PlanQuery) -> String { @@ -397,4 +492,48 @@ mod tests { "plans:v1:query:owner=gowner:beneficiary=gbeneficiary" ); } + + #[tokio::test] + async fn memory_cache_round_trips_generic_stats() { + use crate::api::PlanStatisticsQuery; + + let cache = PlanCache::memory(); + let query = PlanStatisticsQuery { + start_date: None, + end_date: None, + asset_type: Some("USDC".to_string()), + }; + let key = plan_statistics_cache_key(&query); + let stats = vec![ + ("ACTIVE".to_string(), 3_i64), + ("TRIGGERED".to_string(), 1_i64), + ]; + + assert!(cache + .get_stats::>(&key) + .await + .unwrap() + .is_none()); + + cache.set_stats(&key, &stats, 60).await.unwrap(); + + let cached: Option> = cache.get_stats(&key).await.unwrap(); + assert_eq!(cached, Some(stats)); + } + + #[test] + fn plan_statistics_cache_keys_are_normalized_and_stable() { + use crate::api::PlanStatisticsQuery; + + let query = PlanStatisticsQuery { + start_date: None, + end_date: None, + asset_type: Some(" USDC ".to_string()), + }; + + assert_eq!( + plan_statistics_cache_key(&query), + "plan-statistics:v1:query:start=all:end=all:asset=usdc" + ); + } } diff --git a/backend/src/config.rs b/backend/src/config.rs index c110e3422..e0b90e328 100644 --- a/backend/src/config.rs +++ b/backend/src/config.rs @@ -8,6 +8,11 @@ pub struct Config { pub database_url: String, pub redis_url: Option, pub plan_cache_ttl_secs: u64, + /// TTL for the cached `/api/analytics/plan-statistics` response. Kept + /// separate from `plan_cache_ttl_secs` since the statistics query + /// aggregates the whole `plans` table and is far more expensive than a + /// single plan lookup, so admins tolerate a longer staleness window. + pub plan_statistics_cache_ttl_secs: u64, /// Shared secret used to verify HMAC-SHA256 signatures on inbound KYC /// provider webhooks. When unset, `/api/kyc/webhook` rejects every request. pub kyc_webhook_secret: Option, @@ -36,6 +41,11 @@ impl Config { .ok() .and_then(|value| value.parse().ok()) .unwrap_or(15); + let plan_statistics_cache_ttl_secs = std::env::var("PLAN_STATISTICS_CACHE_TTL_SECS") + .ok() + .and_then(|value| value.parse::().ok()) + .unwrap_or(60) + .max(1); let fiat_daily_limit_default = std::env::var("FIAT_DAILY_LIMIT_DEFAULT") .ok() .and_then(|v| v.parse::().ok()) @@ -62,6 +72,7 @@ impl Config { database_url, redis_url, plan_cache_ttl_secs, + plan_statistics_cache_ttl_secs, kyc_webhook_secret, stellar_horizon_url, anchor_api_url, diff --git a/backend/src/main.rs b/backend/src/main.rs index 0ee283b70..a05cd2fda 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -99,6 +99,7 @@ async fn main() -> Result<(), Box> { kyc_webhook_secret: config.kyc_webhook_secret.clone(), apy_config: inheritx_backend::yield_calculator::ApyConfig::from_env(), plan_cache: plan_cache.clone(), + plan_statistics_cache_ttl_secs: config.plan_statistics_cache_ttl_secs, apy_cache: dashmap::DashMap::new(), kyc_tx: kyc_tx.clone(), status_tx, diff --git a/backend/tests/api_tests.rs b/backend/tests/api_tests.rs index 96352c773..e4bd9f322 100644 --- a/backend/tests/api_tests.rs +++ b/backend/tests/api_tests.rs @@ -51,6 +51,7 @@ fn setup_app_with_cache(plan_cache: PlanCache) -> axum::Router { kyc_webhook_secret: None, apy_config: inheritx_backend::yield_calculator::ApyConfig::default(), plan_cache, + plan_statistics_cache_ttl_secs: 60, apy_cache: dashmap::DashMap::new(), stellar_submit: inheritx_backend::stellar_submit::StellarSubmitClient::new( "https://horizon-testnet.stellar.org".to_string(), @@ -522,6 +523,7 @@ async fn test_health_endpoint_without_db_yields_service_unavailable() { kyc_webhook_secret: None, apy_config: inheritx_backend::yield_calculator::ApyConfig::default(), plan_cache: PlanCache::disabled(), + plan_statistics_cache_ttl_secs: 60, apy_cache: dashmap::DashMap::new(), stellar_submit: inheritx_backend::stellar_submit::StellarSubmitClient::new( "https://horizon-testnet.stellar.org".to_string(), @@ -576,6 +578,7 @@ async fn test_get_current_rate_cached() { kyc_webhook_secret: None, apy_config: inheritx_backend::yield_calculator::ApyConfig::default(), plan_cache, + plan_statistics_cache_ttl_secs: 60, apy_cache: dashmap::DashMap::new(), stellar_submit: inheritx_backend::stellar_submit::StellarSubmitClient::new( "https://horizon-testnet.stellar.org".to_string(), @@ -867,3 +870,20 @@ async fn test_trigger_info_is_public() { assert_ne!(response.status(), StatusCode::UNAUTHORIZED); assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR); } + +#[tokio::test] +async fn test_plan_statistics_requires_auth() { + let app = setup_app(); + let response = app + .oneshot( + Request::builder() + .method(http::Method::GET) + .uri("/api/analytics/plan-statistics") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); +} diff --git a/backend/tests/kyc_webhook_test.rs b/backend/tests/kyc_webhook_test.rs index fe35fc784..45ec877e2 100644 --- a/backend/tests/kyc_webhook_test.rs +++ b/backend/tests/kyc_webhook_test.rs @@ -32,6 +32,7 @@ fn test_state(secret: Option<&str>) -> std::sync::Arc; + locked_value_by_asset: AssetLockedValue[]; +} + +export interface PlanStatisticsFilters { + /** ISO 8601 timestamp; only include plans created on or after this date */ + startDate?: string; + /** ISO 8601 timestamp; only include plans created on or before this date */ + endDate?: string; + /** Filter to a single token/asset address */ + assetType?: string; } export interface LoanLifecycleResponse { @@ -173,11 +191,20 @@ export class PlansAPI { } /** - * Get plan statistics + * Get plan statistics for the admin dashboard (active/expired/triggered/ + * claimed counts and total value locked per asset). Requires an admin JWT. */ - async getPlanStatistics(): Promise { + async getPlanStatistics( + filters?: PlanStatisticsFilters + ): Promise { + const params = new URLSearchParams(); + if (filters?.startDate) params.set("start_date", filters.startDate); + if (filters?.endDate) params.set("end_date", filters.endDate); + if (filters?.assetType) params.set("asset_type", filters.assetType); + + const queryString = params.toString(); const response = await apiClient.get>( - "/api/analytics/plan-statistics" + `/api/analytics/plan-statistics${queryString ? `?${queryString}` : ""}` ); return response.data!; }