lean_ctx/gateway_server/
admin_timeseries.rs

1//! `GET /api/admin/timeseries` (enterprise#46) — per-day usage/savings series
2//! for the admin dashboard's trend charts.
3//!
4//! Same window semantics as the usage breakdown (`admin_api::resolve_window`
5//! rules): RFC-3339 `from`/`to`, defaulting to the last 30 days. Buckets are
6//! UTC days (`date_trunc('day', ts)`); empty days are filled in so charts get
7//! a gapless series — an empty day is a real "0", not missing data.
8
9use std::sync::Arc;
10
11use axum::extract::{Query, State};
12use axum::http::StatusCode;
13use axum::response::{IntoResponse, Json, Response};
14use deadpool_postgres::Pool;
15use serde::{Deserialize, Serialize};
16
17use super::admin_api::{AdminState, UsageQuery, resolve_window};
18
19/// One UTC-day bucket of the series.
20#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
21pub struct TimeseriesPoint {
22    /// UTC day, `YYYY-MM-DD`.
23    pub day: String,
24    pub requests: i64,
25    pub cost_usd: f64,
26    pub saved_usd: f64,
27    pub reference_cost_usd: f64,
28}
29
30/// Response of `GET /api/admin/timeseries`.
31#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
32pub struct TimeseriesResponse {
33    pub from: String,
34    pub to: String,
35    pub points: Vec<TimeseriesPoint>,
36}
37
38/// Deterministic per-day rollup; bounds are bound parameters (injection-free).
39const TIMESERIES_SQL: &str = "
40SELECT date_trunc('day', ts)              AS day,
41       count(*)                           AS requests,
42       coalesce(sum(cost_usd), 0)         AS cost_usd,
43       coalesce(sum(saved_usd), 0)        AS saved_usd,
44       coalesce(sum(reference_cost_usd), 0) AS reference_cost_usd
45FROM usage_events
46WHERE ts >= $1 AND ts <= $2
47GROUP BY 1
48ORDER BY 1";
49
50pub(super) async fn get_timeseries(
51    State(state): State<Arc<AdminState>>,
52    Query(q): Query<UsageQuery>,
53) -> Response {
54    let (from, to) = match resolve_window(q.from.as_deref(), q.to.as_deref()) {
55        Ok(w) => w,
56        Err(msg) => {
57            return (
58                StatusCode::BAD_REQUEST,
59                Json(serde_json::json!({"error": msg})),
60            )
61                .into_response();
62        }
63    };
64    match timeseries(&state.pool, from, to).await {
65        Ok(resp) => Json(resp).into_response(),
66        Err(e) => {
67            tracing::warn!("admin timeseries query failed: {e:#}");
68            (
69                StatusCode::SERVICE_UNAVAILABLE,
70                Json(serde_json::json!({"error": "usage store unavailable"})),
71            )
72                .into_response()
73        }
74    }
75}
76
77/// Runs the rollup and fills day gaps with zero points.
78///
79/// # Errors
80/// Propagates pool/query errors (the handler maps them to 503).
81pub async fn timeseries(
82    pool: &Pool,
83    from: chrono::DateTime<chrono::Utc>,
84    to: chrono::DateTime<chrono::Utc>,
85) -> anyhow::Result<TimeseriesResponse> {
86    let client = pool.get().await?;
87    let rows = client.query(TIMESERIES_SQL, &[&from, &to]).await?;
88    let measured: Vec<TimeseriesPoint> = rows
89        .iter()
90        .map(|r| {
91            let day: chrono::DateTime<chrono::Utc> = r.get("day");
92            TimeseriesPoint {
93                day: day.format("%Y-%m-%d").to_string(),
94                requests: r.get("requests"),
95                cost_usd: r.get("cost_usd"),
96                saved_usd: r.get("saved_usd"),
97                reference_cost_usd: r.get("reference_cost_usd"),
98            }
99        })
100        .collect();
101    Ok(TimeseriesResponse {
102        from: from.to_rfc3339(),
103        to: to.to_rfc3339(),
104        points: fill_gaps(&measured, from, to),
105    })
106}
107
108/// Produces one point per UTC day in `[from, to]`, taking measured values
109/// where present and zeros elsewhere. Pure (unit-tested). Shared with the
110/// personal view (`user_api`), which serves the same gapless-series contract.
111pub(super) fn fill_gaps(
112    measured: &[TimeseriesPoint],
113    from: chrono::DateTime<chrono::Utc>,
114    to: chrono::DateTime<chrono::Utc>,
115) -> Vec<TimeseriesPoint> {
116    let mut by_day: std::collections::BTreeMap<String, &TimeseriesPoint> =
117        measured.iter().map(|p| (p.day.clone(), p)).collect();
118    let mut out = Vec::new();
119    let mut day = from.date_naive();
120    let last = to.date_naive();
121    while day <= last {
122        let key = day.format("%Y-%m-%d").to_string();
123        out.push(by_day.remove(&key).cloned().unwrap_or(TimeseriesPoint {
124            day: key,
125            requests: 0,
126            cost_usd: 0.0,
127            saved_usd: 0.0,
128            reference_cost_usd: 0.0,
129        }));
130        day = day.succ_opt().expect("date within chrono range");
131    }
132    out
133}
134
135#[cfg(test)]
136mod tests {
137    use super::*;
138
139    fn ts(s: &str) -> chrono::DateTime<chrono::Utc> {
140        chrono::DateTime::parse_from_rfc3339(s)
141            .expect("test timestamp")
142            .with_timezone(&chrono::Utc)
143    }
144
145    #[test]
146    fn gaps_are_filled_with_zero_days() {
147        let measured = vec![
148            TimeseriesPoint {
149                day: "2026-07-01".into(),
150                requests: 5,
151                cost_usd: 1.0,
152                saved_usd: 0.5,
153                reference_cost_usd: 2.0,
154            },
155            TimeseriesPoint {
156                day: "2026-07-03".into(),
157                requests: 2,
158                cost_usd: 0.4,
159                saved_usd: 0.1,
160                reference_cost_usd: 0.9,
161            },
162        ];
163        let filled = fill_gaps(
164            &measured,
165            ts("2026-07-01T08:00:00Z"),
166            ts("2026-07-04T02:00:00Z"),
167        );
168        let days: Vec<&str> = filled.iter().map(|p| p.day.as_str()).collect();
169        assert_eq!(
170            days,
171            ["2026-07-01", "2026-07-02", "2026-07-03", "2026-07-04"]
172        );
173        assert_eq!(filled[0].requests, 5);
174        assert_eq!(filled[1].requests, 0, "gap day is an explicit zero");
175        assert_eq!(filled[2].requests, 2);
176        assert_eq!(filled[3].requests, 0);
177    }
178
179    #[test]
180    fn single_day_window_yields_one_point() {
181        let filled = fill_gaps(&[], ts("2026-07-02T00:00:00Z"), ts("2026-07-02T23:59:59Z"));
182        assert_eq!(filled.len(), 1);
183        assert_eq!(filled[0].day, "2026-07-02");
184        assert_eq!(filled[0].requests, 0);
185    }
186
187    #[test]
188    fn response_shape_round_trips() {
189        let resp = TimeseriesResponse {
190            from: "2026-07-01T00:00:00+00:00".into(),
191            to: "2026-07-02T00:00:00+00:00".into(),
192            points: vec![TimeseriesPoint {
193                day: "2026-07-01".into(),
194                requests: 10,
195                cost_usd: 3.2,
196                saved_usd: 1.1,
197                reference_cost_usd: 5.0,
198            }],
199        };
200        let json = serde_json::to_value(&resp).expect("serializes");
201        let parsed: TimeseriesResponse = serde_json::from_value(json).expect("round-trips");
202        assert_eq!(parsed, resp);
203    }
204}
lean_ctx/gateway_server/admin_timeseries.rs

lean_ctx/gateway_server/
admin_timeseries.rs
