dwara_core/analytics/
query.rs

1//! Read side of the analytics store (DW-043): dashboard series, Top-N
2//! reports, and the structured query endpoint's translator.
3//!
4//! Every query reads ONLY the rollup tables (the dashboard contract is
5//! "week of traffic under 100 ms" — that is rollup-shaped work, not
6//! raw-scan work). All dynamic values bind as SQL parameters; the ONLY
7//! strings interpolated into SQL are schema identifiers chosen from
8//! closed sets ([`DIM_COLUMNS`]) — the structured query endpoint never
9//! accepts SQL text.
10
11use rusqlite::Connection;
12use serde::{Deserialize, Serialize};
13
14use super::rollup::GRANULARITIES_MS;
15use super::schema::{percentile, BUCKET_COLS};
16
17/// The fixed dimension columns queries may group or filter by. The
18/// closed set the structured endpoint accepts; also the dashboard's
19/// drill-down axes.
20pub const DIM_COLUMNS: [&str; 6] = [
21    "listener",
22    "route",
23    "upstream",
24    "consumer",
25    "method",
26    "status_class",
27];
28
29/// One point of a dashboard series.
30#[derive(Debug, Serialize)]
31pub struct SeriesPoint {
32    /// Window start (ms since epoch).
33    pub window_start: i64,
34    pub requests: i64,
35    pub errors: i64,
36    pub error_rate: f64,
37    pub rate_limited: i64,
38    pub shed: i64,
39    pub avg_ms: f64,
40    pub p50_ms: f64,
41    pub p95_ms: f64,
42    pub p99_ms: f64,
43    /// The group-by key values, when the series is grouped (parallel
44    /// to the query's single group column).
45    #[serde(skip_serializing_if = "Option::is_none")]
46    pub key: Option<String>,
47}
48
49/// Equality filters on dimension columns (dashboard drill-down).
50#[derive(Debug, Default, Clone)]
51pub struct Filters {
52    pub listener: Option<String>,
53    pub route: Option<String>,
54    pub upstream: Option<String>,
55    pub consumer: Option<String>,
56    pub method: Option<String>,
57    pub status_class: Option<String>,
58}
59
60impl Filters {
61    /// WHERE fragments (leading ` AND ...` each) plus their bind
62    /// values, in the order the fragments appear.
63    fn clauses(&self) -> (String, Vec<Box<dyn rusqlite::ToSql>>) {
64        let mut sql = String::new();
65        let mut params: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
66        for (col, val) in [
67            ("listener", &self.listener),
68            ("route", &self.route),
69            ("upstream", &self.upstream),
70            ("consumer", &self.consumer),
71            ("method", &self.method),
72            ("status_class", &self.status_class),
73        ] {
74            if let Some(v) = val {
75                sql.push_str(&format!(" AND {col} = ?"));
76                params.push(Box::new(v.clone()));
77            }
78        }
79        (sql, params)
80    }
81}
82
83/// Shared aggregate read state: 5 metric sums plus 13 bucket sums.
84struct Agg {
85    requests: i64,
86    errors: i64,
87    rate_limited: i64,
88    shed: i64,
89    duration_sum_ms: f64,
90    buckets: Vec<i64>,
91}
92
93/// Shared aggregate read: 5 metric sums then 13 bucket sums starting
94/// at column `start`. NULL-tolerant: an UNGROUPED aggregate over an
95/// empty range still yields one all-NULL row (there is no GROUP BY to
96/// eliminate it), and the documented totals shape is zeros, not a
97/// type error (found by DW-120's totals statement; grouped reads are
98/// never NULL — a group exists only where rows do).
99fn read_agg(row: &rusqlite::Row<'_>, start: usize) -> rusqlite::Result<Agg> {
100    let mut i = start;
101    let requests = row.get::<_, Option<i64>>(i)?.unwrap_or(0);
102    i += 1;
103    let errors = row.get::<_, Option<i64>>(i)?.unwrap_or(0);
104    i += 1;
105    let rate_limited = row.get::<_, Option<i64>>(i)?.unwrap_or(0);
106    i += 1;
107    let shed = row.get::<_, Option<i64>>(i)?.unwrap_or(0);
108    i += 1;
109    let duration_sum_ms = row.get::<_, Option<f64>>(i)?.unwrap_or(0.0);
110    i += 1;
111    let mut buckets = Vec::with_capacity(BUCKET_COLS);
112    for _ in 0..BUCKET_COLS {
113        buckets.push(row.get::<_, Option<i64>>(i)?.unwrap_or(0));
114        i += 1;
115    }
116    Ok(Agg {
117        requests,
118        errors,
119        rate_limited,
120        shed,
121        duration_sum_ms,
122        buckets,
123    })
124}
125
126fn metrics_of(a: &Agg) -> (f64, f64, f64, f64, f64, f64, f64) {
127    let n = a.requests.max(0);
128    (
129        if n > 0 {
130            a.errors as f64 / n as f64
131        } else {
132            0.0
133        },
134        a.rate_limited as f64,
135        a.shed as f64,
136        if n > 0 {
137            a.duration_sum_ms / n as f64
138        } else {
139            0.0
140        },
141        percentile(&a.buckets, 0.50),
142        percentile(&a.buckets, 0.95),
143        percentile(&a.buckets, 0.99),
144    )
145}
146
147/// Dashboard series: per-window metric points at a granularity
148/// (0=1m..3=1d), optionally split by ONE dimension and filtered — the
149/// drill-down axes of the dashboard data API.
150pub fn dashboard(
151    conn: &Connection,
152    from_ms: i64,
153    to_ms: i64,
154    gran: usize,
155    group_by: Option<&str>,
156    filters: &Filters,
157) -> rusqlite::Result<Vec<SeriesPoint>> {
158    let group_col = group_by.and_then(|g| DIM_COLUMNS.iter().find(|c| **c == g).copied());
159    let key_sel = group_col.map(|c| format!("{c}, ")).unwrap_or_default();
160    let group_key = group_col.unwrap_or("window_start");
161    let (filter_sql, bind) = filters.clauses();
162    let sql = format!(
163        "SELECT window_start, {key_sel}
164                SUM(requests), SUM(errors), SUM(rate_limited), SUM(shed),
165                SUM(duration_sum_ms),
166                SUM(b0), SUM(b1), SUM(b2), SUM(b3), SUM(b4), SUM(b5),
167                SUM(b6), SUM(b7), SUM(b8), SUM(b9), SUM(b10), SUM(b11),
168                SUM(b12)
169         FROM rollup_fixed
170         WHERE gran = ? AND window_start >= ? AND window_start < ?{filter_sql}
171         GROUP BY window_start, {group_key}
172         ORDER BY window_start ASC"
173    );
174    let mut all: Vec<Box<dyn rusqlite::ToSql>> =
175        vec![Box::new(gran as i64), Box::new(from_ms), Box::new(to_ms)];
176    all.extend(bind);
177    let mut stmt = conn.prepare(&sql)?;
178    let mut rows = stmt.query(rusqlite::params_from_iter(all.iter().map(|p| p.as_ref())))?;
179    let key_len = usize::from(group_col.is_some());
180    let mut out = Vec::new();
181    while let Some(row) = rows.next()? {
182        let window_start: i64 = row.get(0)?;
183        let key = if key_len == 1 {
184            Some(row.get::<_, String>(1)?)
185        } else {
186            None
187        };
188        let agg = read_agg(row, 1 + key_len)?;
189        let (error_rate, rate_limited, shed, avg_ms, p50, p95, p99) = metrics_of(&agg);
190        out.push(SeriesPoint {
191            window_start,
192            requests: agg.requests,
193            errors: agg.errors,
194            error_rate,
195            rate_limited: rate_limited as i64,
196            shed: shed as i64,
197            avg_ms,
198            p50_ms: p50,
199            p95_ms: p95,
200            p99_ms: p99,
201            key,
202        });
203    }
204    Ok(out)
205}
206
207/// Top-N report kinds (the five frozen in the feature analysis).
208#[derive(Debug, Clone, Copy, PartialEq, Eq)]
209pub enum TopKind {
210    Consumers,
211    Routes,
212    Slowest,
213    ErrorProne,
214    RateLimited,
215}
216
217impl TopKind {
218    /// Parse the endpoint's `kind` parameter. The closed set.
219    pub fn parse(s: &str) -> Option<Self> {
220        match s {
221            "consumers" => Some(TopKind::Consumers),
222            "routes" => Some(TopKind::Routes),
223            "slowest" => Some(TopKind::Slowest),
224            "error_prone" => Some(TopKind::ErrorProne),
225            "rate_limited" => Some(TopKind::RateLimited),
226            _ => None,
227        }
228    }
229    fn as_str(&self) -> &'static str {
230        match self {
231            TopKind::Consumers => "consumers",
232            TopKind::Routes => "routes",
233            TopKind::Slowest => "slowest",
234            TopKind::ErrorProne => "error_prone",
235            TopKind::RateLimited => "rate_limited",
236        }
237    }
238}
239
240/// One Top-N entry.
241#[derive(Debug, Serialize)]
242pub struct TopEntry {
243    pub kind: &'static str,
244    pub name: String,
245    pub requests: i64,
246    pub errors: i64,
247    pub error_rate: f64,
248    pub avg_ms: f64,
249    pub p95_ms: f64,
250    pub rate_limited: i64,
251}
252
253/// Top-N over a time range, summed across the FINEST granularity's
254/// windows (summing finer windows loses nothing — the whole schema is
255/// additive).
256pub fn top(
257    conn: &Connection,
258    kind: TopKind,
259    from_ms: i64,
260    to_ms: i64,
261    limit: usize,
262) -> rusqlite::Result<Vec<TopEntry>> {
263    let group = match kind {
264        TopKind::Consumers | TopKind::RateLimited => "consumer",
265        TopKind::Routes | TopKind::Slowest | TopKind::ErrorProne => "route",
266    };
267    let order = match kind {
268        TopKind::Consumers | TopKind::Routes => "SUM(requests) DESC",
269        TopKind::Slowest => "avg_ms DESC",
270        TopKind::ErrorProne => "err_rate DESC, SUM(requests) DESC",
271        TopKind::RateLimited => "SUM(rate_limited) DESC",
272    };
273    let sql = format!(
274        "SELECT {group} AS name,
275                SUM(requests), SUM(errors), SUM(rate_limited), SUM(shed),
276                SUM(duration_sum_ms),
277                SUM(b0), SUM(b1), SUM(b2), SUM(b3), SUM(b4), SUM(b5),
278                SUM(b6), SUM(b7), SUM(b8), SUM(b9), SUM(b10), SUM(b11),
279                SUM(b12),
280                (SUM(errors) * 1.0 / MAX(SUM(requests), 1)) AS err_rate,
281                (SUM(duration_sum_ms) / MAX(SUM(requests), 1)) AS avg_ms
282         FROM rollup_fixed
283         WHERE gran = 0 AND window_start >= ?1 AND window_start < ?2
284         GROUP BY {group}
285         ORDER BY {order}
286         LIMIT ?3"
287    );
288    let mut stmt = conn.prepare(&sql)?;
289    let mut rows = stmt.query(rusqlite::params![from_ms, to_ms, limit as i64])?;
290    let mut out = Vec::new();
291    while let Some(row) = rows.next()? {
292        let name: String = row.get(0)?;
293        let agg = read_agg(row, 1)?;
294        let requests = agg.requests.max(0);
295        let error_rate = if requests > 0 {
296            agg.errors as f64 / requests as f64
297        } else {
298            0.0
299        };
300        out.push(TopEntry {
301            kind: kind.as_str(),
302            name,
303            requests,
304            errors: agg.errors,
305            error_rate,
306            avg_ms: if requests > 0 {
307                agg.duration_sum_ms / requests as f64
308            } else {
309                0.0
310            },
311            p95_ms: percentile(&agg.buckets, 0.95),
312            rate_limited: agg.rate_limited,
313        });
314    }
315    Ok(out)
316}
317
318/// The structured query endpoint's request body (DW-043): a closed
319/// grammar translated to SQL — never SQL text from the caller.
320///
321/// SCALE-10 (#190): `dim_group_by` opens ad-hoc grouping over a
322/// captured custom dimension (bounded cardinality). When set, the
323/// query reads from `rollup_dim` instead of `rollup_fixed`, grouping
324/// by the dimension's values.
325#[derive(Debug, Deserialize)]
326pub struct StructuredQuery {
327    pub from_ms: i64,
328    pub to_ms: i64,
329    /// Granularity index 0..=3 (1m/5m/1h/1d).
330    pub gran: usize,
331    /// Dimension columns to group by (closed set; empty = one totals
332    /// row).
333    #[serde(default)]
334    pub group_by: Vec<String>,
335    /// SCALE-12 (#190): custom dimension name to group by. When set,
336    /// the query reads from `rollup_dim` grouped by the dimension's
337    /// values. Bounded to the configured custom dimensions'
338    /// cardinality (the rollup_dim table's primary key).
339    #[serde(default, skip_serializing_if = "Option::is_none")]
340    pub dim_group_by: Option<String>,
341    /// Equality filters on dimension columns.
342    #[serde(default)]
343    pub filters: FiltersBody,
344    /// SCALE-12 (#190): equality filters on custom dimensions. The
345    /// key is the dimension name, the value is the required value.
346    /// Applied as a semi-join against `rollup_dim`.
347    #[serde(default)]
348    pub dim_filters: std::collections::BTreeMap<String, String>,
349    /// Maximum returned rows (default 1000, hard cap 10 000).
350    #[serde(default)]
351    pub limit: Option<usize>,
352}
353
354/// The filter map of a [`StructuredQuery`]: column -> required value.
355#[derive(Debug, Default, Deserialize)]
356pub struct FiltersBody {
357    pub listener: Option<String>,
358    pub route: Option<String>,
359    pub upstream: Option<String>,
360    pub consumer: Option<String>,
361    pub method: Option<String>,
362    pub status_class: Option<String>,
363}
364
365/// Why a structured query was rejected before SQL was built.
366#[derive(Debug)]
367pub enum QueryError {
368    UnknownGroupBy(String),
369    BadGranularity(usize),
370    BadRange,
371    /// SCALE-12 (#190): the dim_group_by name is empty or exceeds the
372    /// maximum length (64 chars). Dimension names are config-bounded;
373    /// this is a safety cap against abuse.
374    BadDimGroupBy(String),
375}
376
377impl std::fmt::Display for QueryError {
378    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
379        match self {
380            QueryError::UnknownGroupBy(g) => write!(
381                f,
382                "unknown group_by '{g}' (one of: {})",
383                DIM_COLUMNS.join(", ")
384            ),
385            QueryError::BadGranularity(g) => {
386                write!(f, "gran must be 0..=3 (1m/5m/1h/1d), got {g}")
387            }
388            QueryError::BadRange => write!(f, "from_ms must be < to_ms"),
389            QueryError::BadDimGroupBy(g) => {
390                write!(f, "dim_group_by '{g}' is empty or exceeds 64 characters")
391            }
392        }
393    }
394}
395
396impl std::error::Error for QueryError {}
397
398/// Maximum length of a custom dimension name for dim_group_by.
399const MAX_DIM_NAME_LEN: usize = 64;
400
401impl StructuredQuery {
402    /// Reject anything outside the closed grammar.
403    pub fn validate(&self) -> Result<(), QueryError> {
404        if self.gran >= GRANULARITIES_MS.len() {
405            return Err(QueryError::BadGranularity(self.gran));
406        }
407        if self.from_ms >= self.to_ms {
408            return Err(QueryError::BadRange);
409        }
410        for g in &self.group_by {
411            if !DIM_COLUMNS.contains(&g.as_str()) {
412                return Err(QueryError::UnknownGroupBy(g.clone()));
413            }
414        }
415        if let Some(d) = &self.dim_group_by {
416            if d.is_empty() || d.len() > MAX_DIM_NAME_LEN {
417                return Err(QueryError::BadDimGroupBy(d.clone()));
418            }
419        }
420        for k in self.dim_filters.keys() {
421            if k.is_empty() || k.len() > MAX_DIM_NAME_LEN {
422                return Err(QueryError::BadDimGroupBy(k.clone()));
423            }
424        }
425        Ok(())
426    }
427}
428
429/// The structured query endpoint's row: group key (in the query's
430/// `group_by` order) plus the full metric set.
431#[derive(Debug, Serialize)]
432pub struct QueryRow {
433    pub key: Vec<String>,
434    pub requests: i64,
435    pub errors: i64,
436    pub error_rate: f64,
437    pub rate_limited: i64,
438    pub shed: i64,
439    pub avg_ms: f64,
440    pub p50_ms: f64,
441    pub p95_ms: f64,
442    pub p99_ms: f64,
443}
444
445/// Execute a validated structured query. Range-aggregate shape: the
446/// window column is NOT grouped (per-window series is the dashboard
447/// endpoint's job); rows order by request volume descending.
448///
449/// SCALE-12 (#190): when `dim_group_by` is set, the query reads from
450/// `rollup_dim` instead of `rollup_fixed`, grouping by the custom
451/// dimension's values. When `dim_filters` are set, they are applied
452/// as semi-join subqueries against `rollup_dim`.
453pub fn structured(conn: &Connection, q: &StructuredQuery) -> rusqlite::Result<Vec<QueryRow>> {
454    // SCALE-12 (#190): custom dimension grouping path.
455    if let Some(dim_name) = &q.dim_group_by {
456        return structured_dim(conn, q, dim_name);
457    }
458    let key_len = q.group_by.len();
459    let key_sel = if key_len == 0 {
460        String::new()
461    } else {
462        format!("{}, ", q.group_by.join(", "))
463    };
464    let filters = Filters {
465        listener: q.filters.listener.clone(),
466        route: q.filters.route.clone(),
467        upstream: q.filters.upstream.clone(),
468        consumer: q.filters.consumer.clone(),
469        method: q.filters.method.clone(),
470        status_class: q.filters.status_class.clone(),
471    };
472    let (filter_sql, bind) = filters.clauses();
473    // SCALE-12 (#190): dim_filters semi-join against rollup_dim.
474    let (dim_filter_sql, dim_bind) = dim_filter_clauses(&q.dim_filters);
475    let sql = format!(
476        "SELECT {key_sel}
477                SUM(requests), SUM(errors), SUM(rate_limited), SUM(shed),
478                SUM(duration_sum_ms),
479                SUM(b0), SUM(b1), SUM(b2), SUM(b3), SUM(b4), SUM(b5),
480                SUM(b6), SUM(b7), SUM(b8), SUM(b9), SUM(b10), SUM(b11),
481                SUM(b12)
482         FROM rollup_fixed
483         WHERE gran = ? AND window_start >= ? AND window_start < ?{filter_sql}{dim_filter_sql}
484         {group_all}
485         ORDER BY SUM(requests) DESC
486         LIMIT ?",
487        group_all = if key_len == 0 {
488            // No group keys: a plain aggregate over the range — ONE
489            // totals row (the documented shape). A literal `GROUP BY
490            // 1` would resolve positionally to the leading SUM(...) —
491            // an invalid aggregate-in-GROUP-BY (found by DW-120's
492            // totals statement).
493            String::new()
494        } else {
495            format!("GROUP BY {}", q.group_by.join(", "))
496        }
497    );
498    let limit = q.limit.unwrap_or(1000).min(10_000);
499    let mut all: Vec<Box<dyn rusqlite::ToSql>> = vec![
500        Box::new(q.gran as i64),
501        Box::new(q.from_ms),
502        Box::new(q.to_ms),
503    ];
504    all.extend(bind);
505    all.extend(dim_bind);
506    all.push(Box::new(limit as i64));
507    let mut stmt = conn.prepare(&sql)?;
508    let mut rows = stmt.query(rusqlite::params_from_iter(all.iter().map(|p| p.as_ref())))?;
509    let mut out = Vec::new();
510    while let Some(row) = rows.next()? {
511        let mut key = Vec::with_capacity(key_len);
512        for i in 0..key_len {
513            key.push(row.get::<_, String>(i)?);
514        }
515        let agg = read_agg(row, key_len)?;
516        let (error_rate, rate_limited, shed, avg_ms, p50, p95, p99) = metrics_of(&agg);
517        out.push(QueryRow {
518            key,
519            requests: agg.requests,
520            errors: agg.errors,
521            error_rate,
522            rate_limited: rate_limited as i64,
523            shed: shed as i64,
524            avg_ms,
525            p50_ms: p50,
526            p95_ms: p95,
527            p99_ms: p99,
528        });
529    }
530    Ok(out)
531}
532
533/// SCALE-12 (#190): Build WHERE clauses for custom dimension filters.
534/// Each filter becomes a semi-join: the window_start must appear in
535/// the set of windows where the dimension equals the required value.
536/// Returns (SQL fragment, bind values).
537fn dim_filter_clauses(
538    dim_filters: &std::collections::BTreeMap<String, String>,
539) -> (String, Vec<Box<dyn rusqlite::ToSql>>) {
540    if dim_filters.is_empty() {
541        return (String::new(), Vec::new());
542    }
543    let mut sql = String::new();
544    let mut params: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
545    for (dim, value) in dim_filters {
546        // Semi-join: window_start must appear in the set of windows
547        // where this dimension has this value. Uses EXISTS for clarity.
548        sql.push_str(&format!(
549            " AND EXISTS (SELECT 1 FROM rollup_dim d{idx} \
550             WHERE d{idx}.gran = rollup_fixed.gran \
551             AND d{idx}.window_start = rollup_fixed.window_start \
552             AND d{idx}.dim = ? AND d{idx}.value = ?)",
553            idx = params.len() / 2
554        ));
555        params.push(Box::new(dim.clone()));
556        params.push(Box::new(value.clone()));
557    }
558    (sql, params)
559}
560
561/// SCALE-12 (#190): Execute a structured query grouped by a custom
562/// dimension. Reads from `rollup_dim` instead of `rollup_fixed`,
563/// grouping by the dimension's values. The `rollup_dim` table does
564/// not carry `rate_limited` or `shed` columns (custom dimensions
565/// aggregate every request carrying the dimension, not split by fixed
566/// dimensions), so those fields are zero in the result.
567fn structured_dim(
568    conn: &Connection,
569    q: &StructuredQuery,
570    dim_name: &str,
571) -> rusqlite::Result<Vec<QueryRow>> {
572    let limit = q.limit.unwrap_or(1000).min(10_000);
573    let (filter_sql, bind) = dim_filter_clauses(&q.dim_filters);
574    // The rollup_dim table does not have rate_limited or shed columns.
575    let sql = format!(
576        "SELECT value,
577                SUM(requests), SUM(errors), 0, 0,
578                SUM(duration_sum_ms),
579                SUM(b0), SUM(b1), SUM(b2), SUM(b3), SUM(b4), SUM(b5),
580                SUM(b6), SUM(b7), SUM(b8), SUM(b9), SUM(b10), SUM(b11),
581                SUM(b12)
582         FROM rollup_dim
583         WHERE gran = ? AND window_start >= ? AND window_start < ?
584         AND dim = ?{filter_sql}
585         GROUP BY value
586         ORDER BY SUM(requests) DESC
587         LIMIT ?"
588    );
589    let mut all: Vec<Box<dyn rusqlite::ToSql>> = vec![
590        Box::new(q.gran as i64),
591        Box::new(q.from_ms),
592        Box::new(q.to_ms),
593        Box::new(dim_name.to_string()),
594    ];
595    all.extend(bind);
596    all.push(Box::new(limit as i64));
597    let mut stmt = conn.prepare(&sql)?;
598    let mut rows = stmt.query(rusqlite::params_from_iter(all.iter().map(|p| p.as_ref())))?;
599    let mut out = Vec::new();
600    while let Some(row) = rows.next()? {
601        let key = vec![row.get::<_, String>(0)?];
602        let agg = read_agg(row, 1)?;
603        let (error_rate, _rate_limited, _shed, avg_ms, p50, p95, p99) = metrics_of(&agg);
604        out.push(QueryRow {
605            key,
606            requests: agg.requests,
607            errors: agg.errors,
608            error_rate,
609            rate_limited: 0,
610            shed: 0,
611            avg_ms,
612            p50_ms: p50,
613            p95_ms: p95,
614            p99_ms: p99,
615        });
616    }
617    Ok(out)
618}
619
620// ---------------------------------------------------------------------------
621// DW-079: spend aggregation over the `ai_spend` table.
622// ---------------------------------------------------------------------------
623
624/// The dimension columns the spend summary may group by (DW-079). The
625/// closed set the spend endpoint accepts — a subset of the `ai_spend`
626/// table's columns (no `provider`/`version` to keep the billing view
627/// focused on who/what, not the routing detail).
628pub const SPEND_GROUP_COLUMNS: [&str; 3] = ["consumer", "team", "model"];
629
630/// One spend summary row (DW-079): aggregated token and cost totals
631/// for one group key over a billing window.
632#[derive(Debug, Serialize)]
633pub struct SpendRow {
634    /// The group-by key values, in the query's `group_by` order (empty
635    /// for an ungrouped totals row).
636    pub key: Vec<String>,
637    /// Sum of prompt (input) tokens.
638    pub prompt_tokens: i64,
639    /// Sum of completion (output) tokens.
640    pub completion_tokens: i64,
641    /// Sum of total tokens.
642    pub total_tokens: i64,
643    /// Sum of priced cost (integer micro-USD).
644    pub cost_micros: i64,
645    /// Number of AI requests in the group.
646    pub request_count: i64,
647}
648
649/// The spend query endpoint's request body (DW-079): a closed grammar
650/// over the `ai_spend` table — never SQL text from the caller.
651#[derive(Debug, Deserialize)]
652pub struct SpendQuery {
653    pub from_ms: i64,
654    pub to_ms: i64,
655    /// Dimensions to group by (closed set: consumer, team, model).
656    /// Empty = one totals row.
657    #[serde(default)]
658    pub group_by: Vec<String>,
659    /// Maximum returned rows (default 10 000, hard cap 10 000).
660    #[serde(default)]
661    pub limit: Option<usize>,
662}
663
664/// Why a spend query was rejected before SQL was built.
665#[derive(Debug)]
666pub enum SpendQueryError {
667    UnknownGroupBy(String),
668    BadRange,
669}
670
671impl std::fmt::Display for SpendQueryError {
672    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
673        match self {
674            SpendQueryError::UnknownGroupBy(g) => write!(
675                f,
676                "unknown group_by '{g}' (one of: {})",
677                SPEND_GROUP_COLUMNS.join(", ")
678            ),
679            SpendQueryError::BadRange => write!(f, "from_ms must be < to_ms"),
680        }
681    }
682}
683
684impl std::error::Error for SpendQueryError {}
685
686impl SpendQuery {
687    /// Reject anything outside the closed grammar.
688    pub fn validate(&self) -> Result<(), SpendQueryError> {
689        if self.from_ms >= self.to_ms {
690            return Err(SpendQueryError::BadRange);
691        }
692        for g in &self.group_by {
693            if !SPEND_GROUP_COLUMNS.contains(&g.as_str()) {
694                return Err(SpendQueryError::UnknownGroupBy(g.clone()));
695            }
696        }
697        Ok(())
698    }
699}
700
701/// Execute a validated spend query (DW-079): aggregate the `ai_spend`
702/// table over a billing window, grouped by the requested dimensions.
703/// Rows order by cost descending (the billing-relevant axis). Reads
704/// the RAW `ai_spend` rows directly — spend is per-request, not
705/// rolled up (the volume is orders of magnitude below the access
706/// record path, and billing windows are short).
707pub fn spend_summary(conn: &Connection, q: &SpendQuery) -> rusqlite::Result<Vec<SpendRow>> {
708    let key_len = q.group_by.len();
709    let key_sel = if key_len == 0 {
710        String::new()
711    } else {
712        format!("{}, ", q.group_by.join(", "))
713    };
714    let group_all = if key_len == 0 {
715        String::new()
716    } else {
717        format!("GROUP BY {}", q.group_by.join(", "))
718    };
719    let sql = format!(
720        "SELECT {key_sel}
721                SUM(prompt_tokens), SUM(completion_tokens), SUM(total_tokens),
722                SUM(cost_micros), COUNT(*)
723         FROM ai_spend
724         WHERE ts_ms >= ? AND ts_ms < ?
725         {group_all}
726         ORDER BY SUM(cost_micros) DESC
727         LIMIT ?"
728    );
729    let limit = q.limit.unwrap_or(10_000).min(10_000);
730    let mut stmt = conn.prepare(&sql)?;
731    let mut rows = stmt.query(rusqlite::params![q.from_ms, q.to_ms, limit as i64])?;
732    let mut out = Vec::new();
733    while let Some(row) = rows.next()? {
734        let mut key = Vec::with_capacity(key_len);
735        for i in 0..key_len {
736            key.push(row.get::<_, String>(i)?);
737        }
738        let base = key_len;
739        out.push(SpendRow {
740            key,
741            prompt_tokens: row.get::<_, i64>(base)?,
742            completion_tokens: row.get::<_, i64>(base + 1)?,
743            total_tokens: row.get::<_, i64>(base + 2)?,
744            cost_micros: row.get::<_, i64>(base + 3)?,
745            request_count: row.get::<_, i64>(base + 4)?,
746        });
747    }
748    Ok(out)
749}
750
751// ---------------------------------------------------------------------------
752// DW-084: governance audit over the `ai_governance_events` table.
753// ---------------------------------------------------------------------------
754
755/// One governance audit event row (DW-084): a raw
756/// `ai_governance_events` record — the shadow-audit view of which
757/// consumer/team called which model and whether the governance layer
758/// allowed or denied it.
759#[derive(Debug, Serialize)]
760pub struct GovernanceEventRow {
761    /// Wall-clock ms since the Unix epoch.
762    pub ts_ms: i64,
763    /// The authenticated consumer name (or "anonymous").
764    pub consumer: String,
765    /// The denying policy (team) name for a denial, or the binding
766    /// allowlist's policy name for an allow (empty when no allowlist
767    /// binds the request).
768    pub team: String,
769    /// The client-facing model alias the request named.
770    pub model: String,
771    /// `allow` or `deny`.
772    pub verdict: String,
773    /// A short reason string (the denial cause; empty for an allow).
774    pub reason: String,
775}
776
777/// The governance audit endpoint's request body (DW-084): a time
778/// window over the `ai_governance_events` table.
779#[derive(Debug, Deserialize)]
780pub struct GovernanceAuditQuery {
781    pub from_ms: i64,
782    pub to_ms: i64,
783    /// Maximum returned rows (default 10 000, hard cap 10 000).
784    #[serde(default)]
785    pub limit: Option<usize>,
786}
787
788/// Why a governance audit query was rejected before SQL was built.
789#[derive(Debug)]
790pub enum GovernanceAuditError {
791    BadRange,
792}
793
794impl std::fmt::Display for GovernanceAuditError {
795    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
796        match self {
797            GovernanceAuditError::BadRange => write!(f, "from_ms must be < to_ms"),
798        }
799    }
800}
801
802impl std::error::Error for GovernanceAuditError {}
803
804impl GovernanceAuditQuery {
805    /// Reject an invalid range.
806    pub fn validate(&self) -> Result<(), GovernanceAuditError> {
807        if self.from_ms >= self.to_ms {
808            return Err(GovernanceAuditError::BadRange);
809        }
810        Ok(())
811    }
812}
813
814/// Execute a governance audit query (DW-084): return the raw
815/// `ai_governance_events` rows in a time window, newest first. Reads
816/// the RAW rows directly — governance events are per-check, not
817/// rolled up (the volume is orders of magnitude below the access
818/// record path).
819pub fn governance_audit(
820    conn: &Connection,
821    q: &GovernanceAuditQuery,
822) -> rusqlite::Result<Vec<GovernanceEventRow>> {
823    let limit = q.limit.unwrap_or(10_000).min(10_000);
824    let mut stmt = conn.prepare(
825        "SELECT ts_ms, consumer, team, model, verdict, reason \
826         FROM ai_governance_events \
827         WHERE ts_ms >= ? AND ts_ms < ? \
828         ORDER BY ts_ms DESC \
829         LIMIT ?",
830    )?;
831    let mut rows = stmt.query(rusqlite::params![q.from_ms, q.to_ms, limit as i64])?;
832    let mut out = Vec::new();
833    while let Some(row) = rows.next()? {
834        out.push(GovernanceEventRow {
835            ts_ms: row.get(0)?,
836            consumer: row.get(1)?,
837            team: row.get(2)?,
838            model: row.get(3)?,
839            verdict: row.get(4)?,
840            reason: row.get(5)?,
841        });
842    }
843    Ok(out)
844}
845
846/// One prompt log row (DW-081): a raw `ai_prompt_logs` record — the
847/// redacted prompt and response for one captured AI request.
848#[derive(Debug, Serialize)]
849pub struct PromptLogRow {
850    /// Wall-clock ms since the Unix epoch.
851    pub ts_ms: i64,
852    /// The request id (correlation handle).
853    pub request_id: String,
854    /// The authenticated consumer name (or "anonymous").
855    pub consumer: String,
856    /// The route name that served the request.
857    pub route: String,
858    /// The serving provider name.
859    pub provider: String,
860    /// The provider's own model identifier.
861    pub model: String,
862    /// The canary version label, or empty for non-canary.
863    pub version: String,
864    /// The REDACTED prompt as JSON.
865    pub prompt_json: String,
866    /// The REDACTED response as JSON.
867    pub response_json: String,
868    /// Whether the request was streaming.
869    pub stream: bool,
870}
871
872/// The prompt-logs endpoint's request body (DW-081): a time window
873/// over the `ai_prompt_logs` table, optionally filtered by consumer.
874#[derive(Debug, Deserialize)]
875pub struct PromptLogQuery {
876    pub from_ms: i64,
877    pub to_ms: i64,
878    /// Optional consumer filter.
879    #[serde(default)]
880    pub consumer: Option<String>,
881    /// Maximum returned rows (default 1 000, hard cap 10 000).
882    #[serde(default)]
883    pub limit: Option<usize>,
884}
885
886/// Why a prompt-logs query was rejected before SQL was built.
887#[derive(Debug)]
888pub enum PromptLogError {
889    BadRange,
890}
891
892impl std::fmt::Display for PromptLogError {
893    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
894        match self {
895            PromptLogError::BadRange => write!(f, "from_ms must be < to_ms"),
896        }
897    }
898}
899
900impl std::error::Error for PromptLogError {}
901
902impl PromptLogQuery {
903    /// Reject an invalid range.
904    pub fn validate(&self) -> Result<(), PromptLogError> {
905        if self.from_ms >= self.to_ms {
906            return Err(PromptLogError::BadRange);
907        }
908        Ok(())
909    }
910}
911
912/// Execute a prompt-logs query (DW-081): return the raw
913/// `ai_prompt_logs` rows in a time window, newest first. Reads the
914/// RAW rows directly — prompt logs are per-request, not rolled up.
915/// Optionally filtered by consumer.
916pub fn prompt_logs(conn: &Connection, q: &PromptLogQuery) -> rusqlite::Result<Vec<PromptLogRow>> {
917    let limit = q.limit.unwrap_or(1_000).min(10_000);
918    let mut out = Vec::new();
919    if let Some(consumer) = &q.consumer {
920        let mut stmt = conn.prepare(
921            "SELECT ts_ms, request_id, consumer, route, provider, model, version, \
922             prompt_json, response_json, stream \
923             FROM ai_prompt_logs \
924             WHERE ts_ms >= ? AND ts_ms < ? AND consumer = ? \
925             ORDER BY ts_ms DESC \
926             LIMIT ?",
927        )?;
928        let mut rows = stmt.query(rusqlite::params![
929            q.from_ms,
930            q.to_ms,
931            consumer,
932            limit as i64
933        ])?;
934        while let Some(row) = rows.next()? {
935            out.push(PromptLogRow {
936                ts_ms: row.get(0)?,
937                request_id: row.get(1)?,
938                consumer: row.get(2)?,
939                route: row.get(3)?,
940                provider: row.get(4)?,
941                model: row.get(5)?,
942                version: row.get(6)?,
943                prompt_json: row.get(7)?,
944                response_json: row.get(8)?,
945                stream: row.get::<_, i64>(9)? != 0,
946            });
947        }
948    } else {
949        let mut stmt = conn.prepare(
950            "SELECT ts_ms, request_id, consumer, route, provider, model, version, \
951             prompt_json, response_json, stream \
952             FROM ai_prompt_logs \
953             WHERE ts_ms >= ? AND ts_ms < ? \
954             ORDER BY ts_ms DESC \
955             LIMIT ?",
956        )?;
957        let mut rows = stmt.query(rusqlite::params![q.from_ms, q.to_ms, limit as i64])?;
958        while let Some(row) = rows.next()? {
959            out.push(PromptLogRow {
960                ts_ms: row.get(0)?,
961                request_id: row.get(1)?,
962                consumer: row.get(2)?,
963                route: row.get(3)?,
964                provider: row.get(4)?,
965                model: row.get(5)?,
966                version: row.get(6)?,
967                prompt_json: row.get(7)?,
968                response_json: row.get(8)?,
969                stream: row.get::<_, i64>(9)? != 0,
970            });
971        }
972    }
973    Ok(out)
974}
975
976// --- MCP tool calls (DW-087) ---------------------------------------------
977
978/// Query parameters for MCP tool call records (DW-087).
979#[derive(Debug, Clone, Default)]
980pub struct McpToolCallQuery {
981    pub from_ms: i64,
982    pub to_ms: i64,
983    pub session_id: Option<String>,
984    pub consumer: Option<String>,
985    pub tool_name: Option<String>,
986    pub limit: Option<i64>,
987}
988
989/// One MCP tool call row (DW-087).
990#[derive(Debug, Serialize)]
991pub struct McpToolCallRow {
992    pub ts_ms: i64,
993    pub request_id: String,
994    pub session_id: String,
995    pub consumer: String,
996    pub tool_name: String,
997    pub allowed: bool,
998    pub duration_ms: f64,
999    pub error_code: Option<String>,
1000    pub status: String,
1001}
1002
1003/// Execute an MCP tool call query (DW-087): return the raw
1004/// `mcp_tool_calls` rows in a time window, newest first, optionally
1005/// filtered by session id, consumer, or tool name.
1006pub fn mcp_tool_calls(
1007    conn: &Connection,
1008    q: &McpToolCallQuery,
1009) -> rusqlite::Result<Vec<McpToolCallRow>> {
1010    let limit = q.limit.unwrap_or(10_000).min(10_000);
1011    let mut sql = String::from(
1012        "SELECT ts_ms, request_id, session_id, consumer, tool_name, allowed, \
1013         duration_ms, error_code, status \
1014         FROM mcp_tool_calls \
1015         WHERE ts_ms >= ? AND ts_ms < ?",
1016    );
1017    let mut params: Vec<Box<dyn rusqlite::ToSql>> = vec![Box::new(q.from_ms), Box::new(q.to_ms)];
1018    if let Some(v) = &q.session_id {
1019        sql.push_str(" AND session_id = ?");
1020        params.push(Box::new(v.clone()));
1021    }
1022    if let Some(v) = &q.consumer {
1023        sql.push_str(" AND consumer = ?");
1024        params.push(Box::new(v.clone()));
1025    }
1026    if let Some(v) = &q.tool_name {
1027        sql.push_str(" AND tool_name = ?");
1028        params.push(Box::new(v.clone()));
1029    }
1030    sql.push_str(" ORDER BY ts_ms DESC LIMIT ?");
1031    params.push(Box::new(limit));
1032    let mut stmt = conn.prepare(&sql)?;
1033    let param_refs: Vec<&dyn rusqlite::ToSql> = params.iter().map(|p| p.as_ref()).collect();
1034    let mut rows = stmt.query(param_refs.as_slice())?;
1035    let mut out = Vec::new();
1036    while let Some(row) = rows.next()? {
1037        out.push(McpToolCallRow {
1038            ts_ms: row.get(0)?,
1039            request_id: row.get(1)?,
1040            session_id: row.get(2)?,
1041            consumer: row.get(3)?,
1042            tool_name: row.get(4)?,
1043            allowed: row.get::<_, i64>(5)? != 0,
1044            duration_ms: row.get(6)?,
1045            error_code: row.get(7)?,
1046            status: row.get(8)?,
1047        });
1048    }
1049    Ok(out)
1050}
1051
1052// ---------------------------------------------------------------------------
1053// DW-093: custom-dimension query and journey/funnel query.
1054// ---------------------------------------------------------------------------
1055
1056/// One custom-dimension query row (DW-093): aggregated metrics for one
1057/// (window, dimension name, value) tuple, read from the `rollup_dim`
1058/// table. The dimension name is config-bounded (the rollup key); the
1059/// value is the captured tag value.
1060#[derive(Debug, Serialize)]
1061pub struct DimensionRow {
1062    /// Window start (ms since epoch).
1063    pub window_start: i64,
1064    /// The dimension name (the config-declared rollup key).
1065    pub dim: String,
1066    /// The dimension value for this group.
1067    pub value: String,
1068    /// Request count in the window.
1069    pub requests: i64,
1070    /// Error count (status >= 500) in the window.
1071    pub error_count: i64,
1072    /// Average duration in milliseconds.
1073    pub avg_duration_ms: f64,
1074}
1075
1076/// The custom-dimension query endpoint's request body (DW-093): a
1077/// closed grammar over the `rollup_dim` table — never SQL text.
1078#[derive(Debug, Deserialize)]
1079pub struct DimensionQuery {
1080    pub from_ms: i64,
1081    pub to_ms: i64,
1082    /// Granularity index 0..=3 (1m/5m/1h/1d).
1083    pub gran: usize,
1084    /// The dimension name to query (the config-declared rollup key).
1085    pub dim: String,
1086    /// Optional dimension value filter (only rows with this value).
1087    #[serde(default)]
1088    pub value: Option<String>,
1089    /// Maximum returned rows (default 10 000, hard cap 10 000).
1090    #[serde(default)]
1091    pub limit: Option<usize>,
1092}
1093
1094/// Why a dimension query was rejected before SQL was built.
1095#[derive(Debug)]
1096pub enum DimensionQueryError {
1097    BadGranularity(usize),
1098    BadRange,
1099    EmptyDim,
1100}
1101
1102impl std::fmt::Display for DimensionQueryError {
1103    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1104        match self {
1105            DimensionQueryError::BadGranularity(g) => {
1106                write!(f, "gran must be 0..=3 (1m/5m/1h/1d), got {g}")
1107            }
1108            DimensionQueryError::BadRange => write!(f, "from_ms must be < to_ms"),
1109            DimensionQueryError::EmptyDim => write!(f, "dim must be non-empty"),
1110        }
1111    }
1112}
1113
1114impl std::error::Error for DimensionQueryError {}
1115
1116impl DimensionQuery {
1117    /// Reject anything outside the closed grammar.
1118    pub fn validate(&self) -> Result<(), DimensionQueryError> {
1119        if self.gran >= GRANULARITIES_MS.len() {
1120            return Err(DimensionQueryError::BadGranularity(self.gran));
1121        }
1122        if self.from_ms >= self.to_ms {
1123            return Err(DimensionQueryError::BadRange);
1124        }
1125        if self.dim.is_empty() {
1126            return Err(DimensionQueryError::EmptyDim);
1127        }
1128        Ok(())
1129    }
1130}
1131
1132/// Execute a validated custom-dimension query (DW-093): aggregate the
1133/// `rollup_dim` table over a time range at a granularity, filtered by
1134/// dimension name and optional value. Rows order by window start
1135/// ascending, then request count descending within each window.
1136pub fn dimension_query(
1137    conn: &Connection,
1138    q: &DimensionQuery,
1139) -> rusqlite::Result<Vec<DimensionRow>> {
1140    let limit = q.limit.unwrap_or(10_000).min(10_000);
1141    let mut sql = String::from(
1142        "SELECT window_start, dim, value, requests, errors, duration_sum_ms \
1143         FROM rollup_dim \
1144         WHERE gran = ? AND window_start >= ? AND window_start < ? AND dim = ?",
1145    );
1146    let mut params: Vec<Box<dyn rusqlite::ToSql>> = vec![
1147        Box::new(q.gran as i64),
1148        Box::new(q.from_ms),
1149        Box::new(q.to_ms),
1150        Box::new(q.dim.clone()),
1151    ];
1152    if let Some(v) = &q.value {
1153        sql.push_str(" AND value = ?");
1154        params.push(Box::new(v.clone()));
1155    }
1156    sql.push_str(" ORDER BY window_start ASC, requests DESC LIMIT ?");
1157    params.push(Box::new(limit as i64));
1158    let mut stmt = conn.prepare(&sql)?;
1159    let param_refs: Vec<&dyn rusqlite::ToSql> = params.iter().map(|p| p.as_ref()).collect();
1160    let mut rows = stmt.query(param_refs.as_slice())?;
1161    let mut out = Vec::new();
1162    while let Some(row) = rows.next()? {
1163        let requests: i64 = row.get(3)?;
1164        let duration_sum_ms: f64 = row.get(5)?;
1165        out.push(DimensionRow {
1166            window_start: row.get(0)?,
1167            dim: row.get(1)?,
1168            value: row.get(2)?,
1169            requests,
1170            error_count: row.get(4)?,
1171            avg_duration_ms: if requests > 0 {
1172                duration_sum_ms / requests as f64
1173            } else {
1174                0.0
1175            },
1176        });
1177    }
1178    Ok(out)
1179}
1180
1181/// One journey/funnel query row (DW-093): a raw `raw` table record for
1182/// one correlation id — the per-request step in a business journey.
1183/// Ordered by time ascending so the caller sees the request sequence.
1184#[derive(Debug, Serialize)]
1185pub struct JourneyRow {
1186    /// Wall-clock ms since the Unix epoch.
1187    pub ts_ms: i64,
1188    /// The request id (the gateway's resolved correlation handle).
1189    pub request_id: String,
1190    /// The correlation id (the business journey key).
1191    pub correlation_id: String,
1192    /// The listener that accepted the request.
1193    pub listener: String,
1194    /// The matched route name, or "unrouted".
1195    pub route: String,
1196    /// The authenticated consumer, or "anonymous".
1197    pub consumer: String,
1198    /// The upstream that served the request (or empty).
1199    pub upstream: String,
1200    /// The HTTP method.
1201    pub method: String,
1202    /// The HTTP status code.
1203    pub status: u16,
1204    /// The request duration in milliseconds.
1205    pub duration_ms: f64,
1206    /// The custom dimensions JSON object (as stored).
1207    pub dims: String,
1208}
1209
1210/// Execute a journey/funnel query (DW-093): return all raw records
1211/// matching a correlation id, ordered by time ascending. Optional
1212/// time range filter narrows the scan. Reads the RAW `raw` table
1213/// directly (journey analysis is per-request, not rolled up). The
1214/// `correlation_id` index keeps the scan bounded.
1215pub fn journey_query(
1216    conn: &Connection,
1217    correlation_id: &str,
1218    from_ms: Option<i64>,
1219    to_ms: Option<i64>,
1220) -> rusqlite::Result<Vec<JourneyRow>> {
1221    let mut sql = String::from(
1222        "SELECT ts_ms, request_id, correlation_id, listener, route, \
1223         consumer, upstream, method, status, duration_ms, dims \
1224         FROM raw_all WHERE correlation_id = ?",
1225    );
1226    let mut params: Vec<Box<dyn rusqlite::ToSql>> = vec![Box::new(correlation_id.to_string())];
1227    if let Some(from) = from_ms {
1228        sql.push_str(" AND ts_ms >= ?");
1229        params.push(Box::new(from));
1230    }
1231    if let Some(to) = to_ms {
1232        sql.push_str(" AND ts_ms < ?");
1233        params.push(Box::new(to));
1234    }
1235    sql.push_str(" ORDER BY ts_ms ASC LIMIT 10000");
1236    let mut stmt = conn.prepare(&sql)?;
1237    let param_refs: Vec<&dyn rusqlite::ToSql> = params.iter().map(|p| p.as_ref()).collect();
1238    let mut rows = stmt.query(param_refs.as_slice())?;
1239    let mut out = Vec::new();
1240    while let Some(row) = rows.next()? {
1241        out.push(JourneyRow {
1242            ts_ms: row.get(0)?,
1243            request_id: row.get(1)?,
1244            correlation_id: row.get(2)?,
1245            listener: row.get(3)?,
1246            route: row.get(4)?,
1247            consumer: row.get(5)?,
1248            upstream: row.get(6)?,
1249            method: row.get(7)?,
1250            status: row.get::<_, i64>(8)? as u16,
1251            duration_ms: row.get(9)?,
1252            dims: row.get(10)?,
1253        });
1254    }
1255    Ok(out)
1256}
dwara_core/analytics/query.rs

dwara_core/analytics/
query.rs
