dwara_core/analytics/schema.rs
1//! Analytics store schema (DW-043).
2//!
3//! The analytics database is a SEPARATE SQLite file from the state
4//! store (DW-018): different lifecycle (retention deletes vs identity
5//! upserts), different write pattern (high-churn batched appends vs
6//! request-path point lookups), and an independent bounded-disk story
7//! (incremental vacuum reclaims retention deletes here without ever
8//! compacting the identity store). It owns its own `PRAGMA user_version`
9//! namespace — a state-store database and an analytics database can
10//! never be confused for one another, and the state store's
11//! forward-only migration contract stays untouched by analytics churn.
12//!
13//! Schema v1 (this file is migration 001; v2 adds the DW-120
14//! `export_runs` ledger — see [`SCHEMA_V2`]):
15//!
16//! - `meta` — rollup cursors (see [`super::rollup`]): the exclusive
17//! upper bound each granularity has durably rolled through, written
18//! in the SAME transaction as the rollup rows it covers, so a crash
19//! between rollups never double-counts a window.
20//! - `raw` — one row per completed request, exactly the redacted
21//! [`AccessRecord`](crate::observability::AccessRecord) field set
22//! plus a JSON object of custom dimensions. Raw retention is SHORT
23//! (default 24 h): raw exists to seed rollups and answer
24//! custom-dimension ad hoc queries; the durable history is the
25//! rollup tables.
26//! - `rollup_fixed` — pre-aggregated counters per (granularity,
27//! window, fixed dimension tuple). Latency lives as 13 fixed
28//! per-bucket counts ([`MS_BUCKETS`] bounds) so any set of windows
29//! merges by summation and percentiles are estimable without
30//! storing per-request samples.
31//! - `rollup_dim` — the custom-dimension twin: one row per
32//! (granularity, window, dimension name, value), aggregated in Rust
33//! (the source is a JSON column; keeping the aggregation out of SQL
34//! avoids a JSON1 feature dependency in the bundled SQLite).
35
36/// Latency bucket UPPER bounds in milliseconds (inclusive). A request
37/// lands in exactly one bucket: the first bound it does not exceed,
38/// else the overflow bucket (index 12). The bounds are the
39/// observability histogram's shape at millisecond granularity —
40/// sub-millisecond precision is noise at gateway scale, 10 s covers
41/// every classified timeout. Counts are stored NON-cumulative
42/// (per-bucket); cumulative form is computed at read time.
43pub const MS_BUCKETS: [f64; 12] = [
44 1.0, 2.0, 5.0, 10.0, 25.0, 50.0, 100.0, 250.0, 500.0, 1000.0, 2500.0, 5000.0,
45];
46
47/// Total stored bucket counts per rollup row: 12 bounded + overflow.
48pub const BUCKET_COLS: usize = MS_BUCKETS.len() + 1;
49
50/// Bucket index for one duration (the overflow bucket for anything
51/// past the last bound).
52pub fn bucket_of(duration_ms: f64) -> usize {
53 MS_BUCKETS
54 .iter()
55 .position(|b| duration_ms <= *b)
56 .unwrap_or(MS_BUCKETS.len())
57}
58
59/// Percentile estimate from non-cumulative bucket counts: the first
60/// bound whose cumulative count reaches `p * total` (else the last
61/// bound — everything past it is reported as 10 s, the honest ceiling).
62pub fn percentile(buckets: &[i64], p: f64) -> f64 {
63 let total: i64 = buckets.iter().sum();
64 if total == 0 {
65 return 0.0;
66 }
67 let target = (p * total as f64).ceil() as i64;
68 let mut cum = 0i64;
69 for (i, b) in buckets.iter().enumerate() {
70 cum += b;
71 if cum >= target {
72 return if i < MS_BUCKETS.len() {
73 MS_BUCKETS[i]
74 } else {
75 // Overflow bucket: report the last bound as the floor.
76 MS_BUCKETS[MS_BUCKETS.len() - 1]
77 };
78 }
79 }
80 MS_BUCKETS[MS_BUCKETS.len() - 1]
81}
82
83/// The full v1 DDL. Idempotent (`IF NOT EXISTS`) so a fresh database
84/// and a re-run both land at version 1. Connection-level pragmas
85/// (WAL, synchronous, auto_vacuum) are applied at open time — several
86/// of them return rows and cannot live in an `execute_batch`.
87pub const SCHEMA_V1: &str = "
88 CREATE TABLE IF NOT EXISTS meta (
89 key TEXT PRIMARY KEY,
90 value TEXT NOT NULL
91 );
92
93 CREATE TABLE IF NOT EXISTS raw (
94 id INTEGER PRIMARY KEY AUTOINCREMENT,
95 ts_ms INTEGER NOT NULL,
96 listener TEXT NOT NULL,
97 route TEXT NOT NULL,
98 consumer TEXT NOT NULL,
99 upstream TEXT NOT NULL,
100 method TEXT NOT NULL,
101 status INTEGER NOT NULL,
102 status_class TEXT NOT NULL,
103 duration_ms REAL NOT NULL,
104 attempts INTEGER NOT NULL,
105 rate_limited INTEGER NOT NULL,
106 broken INTEGER NOT NULL,
107 shed INTEGER NOT NULL,
108 dims TEXT NOT NULL DEFAULT '{}'
109 );
110 CREATE INDEX IF NOT EXISTS idx_raw_ts ON raw(ts_ms);
111
112 CREATE TABLE IF NOT EXISTS rollup_fixed (
113 gran INTEGER NOT NULL,
114 window_start INTEGER NOT NULL,
115 listener TEXT NOT NULL,
116 route TEXT NOT NULL,
117 upstream TEXT NOT NULL,
118 consumer TEXT NOT NULL,
119 method TEXT NOT NULL,
120 status_class TEXT NOT NULL,
121 requests INTEGER NOT NULL,
122 errors INTEGER NOT NULL,
123 rate_limited INTEGER NOT NULL,
124 shed INTEGER NOT NULL,
125 duration_sum_ms REAL NOT NULL,
126 duration_max_ms REAL NOT NULL,
127 b0 INTEGER NOT NULL, b1 INTEGER NOT NULL, b2 INTEGER NOT NULL,
128 b3 INTEGER NOT NULL, b4 INTEGER NOT NULL, b5 INTEGER NOT NULL,
129 b6 INTEGER NOT NULL, b7 INTEGER NOT NULL, b8 INTEGER NOT NULL,
130 b9 INTEGER NOT NULL, b10 INTEGER NOT NULL, b11 INTEGER NOT NULL,
131 b12 INTEGER NOT NULL,
132 PRIMARY KEY (gran, window_start, listener, route, upstream,
133 consumer, method, status_class)
134 ) WITHOUT ROWID;
135
136 CREATE TABLE IF NOT EXISTS rollup_dim (
137 gran INTEGER NOT NULL,
138 window_start INTEGER NOT NULL,
139 dim TEXT NOT NULL,
140 value TEXT NOT NULL,
141 requests INTEGER NOT NULL,
142 errors INTEGER NOT NULL,
143 duration_sum_ms REAL NOT NULL,
144 b0 INTEGER NOT NULL, b1 INTEGER NOT NULL, b2 INTEGER NOT NULL,
145 b3 INTEGER NOT NULL, b4 INTEGER NOT NULL, b5 INTEGER NOT NULL,
146 b6 INTEGER NOT NULL, b7 INTEGER NOT NULL, b8 INTEGER NOT NULL,
147 b9 INTEGER NOT NULL, b10 INTEGER NOT NULL, b11 INTEGER NOT NULL,
148 b12 INTEGER NOT NULL,
149 PRIMARY KEY (gran, window_start, dim, value)
150 ) WITHOUT ROWID;
151";
152
153/// Schema v2 (DW-120): the `export_runs` ledger. One row per
154/// scheduled-or-manual usage-statement export attempt per window
155/// (primary key; a re-export REPLACES the row — the file output is
156/// equally idempotent). `status` is `ok` or `failed`; `partial` flags
157/// a window older than the query granularity's retention (undercounts
158/// are possible) so billing consumers can reject flagged statements
159/// instead of trusting them; `windows_nonempty` counts the range's
160/// rollup windows that carry any data (informational — quiet windows
161/// have no rows and are not loss).
162pub const SCHEMA_V2: &str = "
163 CREATE TABLE IF NOT EXISTS export_runs (
164 kind TEXT NOT NULL,
165 window_start_ms INTEGER NOT NULL,
166 window_end_ms INTEGER NOT NULL,
167 status TEXT NOT NULL,
168 partial INTEGER NOT NULL,
169 formats TEXT NOT NULL,
170 directory TEXT NOT NULL,
171 consumers INTEGER NOT NULL,
172 requests INTEGER NOT NULL,
173 windows_nonempty INTEGER NOT NULL,
174 error TEXT NOT NULL DEFAULT '',
175 generated_at_ms INTEGER NOT NULL,
176 PRIMARY KEY (kind, window_start_ms)
177 ) WITHOUT ROWID;
178";
179
180/// Schema v3 (DW-079): the `ai_spend` table. One row per completed AI
181/// request carrying the provider-reported token usage and the priced
182/// cost (micro-USD), attributed to the consumer/team/model/provider
183/// that served it. The spend query surface
184/// ([`super::query::spend_summary`]) aggregates these rows per
185/// consumer/team/model for billing reconciliation; the export
186/// statement (DW-120) folds the per-consumer totals into its rows.
187pub const SCHEMA_V3: &str = "
188 CREATE TABLE IF NOT EXISTS ai_spend (
189 id INTEGER PRIMARY KEY AUTOINCREMENT,
190 ts_ms INTEGER NOT NULL,
191 consumer TEXT NOT NULL,
192 team TEXT NOT NULL DEFAULT '',
193 provider TEXT NOT NULL,
194 model TEXT NOT NULL,
195 version TEXT NOT NULL DEFAULT '',
196 prompt_tokens INTEGER NOT NULL,
197 completion_tokens INTEGER NOT NULL,
198 total_tokens INTEGER NOT NULL,
199 cost_micros INTEGER NOT NULL
200 );
201 CREATE INDEX IF NOT EXISTS idx_ai_spend_ts ON ai_spend(ts_ms);
202 CREATE INDEX IF NOT EXISTS idx_ai_spend_consumer ON ai_spend(consumer);
203 CREATE INDEX IF NOT EXISTS idx_ai_spend_model ON ai_spend(model);
204";
205
206/// Schema v4 (DW-084): the `ai_governance_events` table. One row per
207/// governance check outcome (allow or deny) when the audit switch is
208/// on, plus every DENIAL even when the audit switch is off (the
209/// done-when requirement: a blocked attempt appears in the audit
210/// log). The admin `/analytics/governance-audit` endpoint reads these
211/// rows for shadow review.
212pub const SCHEMA_V4: &str = "
213 CREATE TABLE IF NOT EXISTS ai_governance_events (
214 id INTEGER PRIMARY KEY AUTOINCREMENT,
215 ts_ms INTEGER NOT NULL,
216 consumer TEXT NOT NULL,
217 team TEXT NOT NULL DEFAULT '',
218 model TEXT NOT NULL,
219 verdict TEXT NOT NULL,
220 reason TEXT NOT NULL DEFAULT ''
221 );
222 CREATE INDEX IF NOT EXISTS idx_ai_governance_ts ON ai_governance_events(ts_ms);
223 CREATE INDEX IF NOT EXISTS idx_ai_governance_consumer ON ai_governance_events(consumer);
224";
225
226/// Schema v5 (DW-081): the `ai_prompt_logs` table. One row per
227/// captured AI request carrying the REDACTED prompt and response as
228/// JSON, attributed to the consumer/route/provider/model that served
229/// it. Capture is opt-in (privacy-first); the redaction pass scrubs
230/// PII/secrets before storage. The admin `/analytics/prompt-logs`
231/// endpoint reads these rows.
232pub const SCHEMA_V5: &str = "
233 CREATE TABLE IF NOT EXISTS ai_prompt_logs (
234 id INTEGER PRIMARY KEY AUTOINCREMENT,
235 ts_ms INTEGER NOT NULL,
236 request_id TEXT NOT NULL,
237 consumer TEXT NOT NULL,
238 route TEXT NOT NULL,
239 provider TEXT NOT NULL,
240 model TEXT NOT NULL,
241 version TEXT NOT NULL DEFAULT '',
242 prompt_json TEXT NOT NULL,
243 response_json TEXT NOT NULL,
244 stream INTEGER NOT NULL DEFAULT 0
245 );
246 CREATE INDEX IF NOT EXISTS idx_ai_prompt_logs_ts ON ai_prompt_logs(ts_ms);
247 CREATE INDEX IF NOT EXISTS idx_ai_prompt_logs_consumer ON ai_prompt_logs(consumer);
248";
249
250/// Schema v8 (DW-113): add the `consumer_type` column to `ai_spend`
251/// and `mcp_tool_calls` for typed agent/user attribution. The column
252/// defaults to `'user'` so pre-DW-113 rows and records that omit the
253/// field land as user traffic (the pre-existing default). Uses ALTER
254/// TABLE (additive) so existing databases migrate in place.
255pub const SCHEMA_V8: &str = "
256 ALTER TABLE ai_spend ADD COLUMN consumer_type TEXT NOT NULL DEFAULT 'user';
257 ALTER TABLE mcp_tool_calls ADD COLUMN consumer_type TEXT NOT NULL DEFAULT 'user';
258";
259
260/// Schema v9 (DW-093): add `request_id` and `correlation_id` columns
261/// to the `raw` table for business-metrics journey/funnel analytics.
262/// Both default to `''` so pre-DW-093 rows and records that omit the
263/// fields land as empty (the journey query simply returns no rows for
264/// an empty correlation id). The `correlation_id` index supports the
265/// journey query (all raw rows for one correlation id, ordered by
266/// time). Uses ALTER TABLE (additive) so existing databases migrate
267/// in place.
268pub const SCHEMA_V9: &str = "
269 ALTER TABLE raw ADD COLUMN request_id TEXT NOT NULL DEFAULT '';
270 ALTER TABLE raw ADD COLUMN correlation_id TEXT NOT NULL DEFAULT '';
271 CREATE INDEX IF NOT EXISTS idx_raw_correlation ON raw(correlation_id, ts_ms);
272";
273
274/// Schema v10 (DW-102): add optional replay-capture columns to the
275/// `raw` table for time-travel debugging. `request_headers_redacted`
276/// holds the redacted request headers (opt-in, scrubbed via the
277/// existing PII redaction patterns) and `auth_identity` holds the
278/// authenticated consumer name (opt-in). Both default to `NULL` so
279/// pre-DW-102 rows and records that omit the fields (capture is
280/// opt-in) land as NULL — the replay reader treats NULL as "not
281/// captured". Uses ALTER TABLE (additive) so existing databases
282/// migrate in place.
283pub const SCHEMA_V10: &str = "
284 ALTER TABLE raw ADD COLUMN request_headers_redacted TEXT;
285 ALTER TABLE raw ADD COLUMN auth_identity TEXT;
286";
287
288/// Schema v11 (PERF-12, #209): add `bytes_in` and `bytes_out` columns
289/// to the `raw` table for per-route traffic accounting. `bytes_in` is
290/// the declared `Content-Length` (0 for chunked request bodies);
291/// `bytes_out` is the frame-counted response body size (counted as
292/// frames stream to the client, no buffering). Both default to 0 so
293/// pre-v11 rows land as zero. Uses ALTER TABLE (additive) so existing
294/// databases migrate in place.
295pub const SCHEMA_V11: &str = "
296 ALTER TABLE raw ADD COLUMN bytes_in INTEGER NOT NULL DEFAULT 0;
297 ALTER TABLE raw ADD COLUMN bytes_out INTEGER NOT NULL DEFAULT 0;
298";
299
300/// Latest analytics schema version this build knows.
301pub const LATEST_SCHEMA_VERSION: u32 = 12;
302
303/// Schema v6 (DW-086): three tables for prompt experimentation.
304///
305/// - `ai_experiment_assignments` — one row per request that was
306/// served by an A/B test alias, recording which variant was
307/// selected. The `request_id` ties the assignment to the access
308/// log and spend rows; the `experiment` and `variant` labels are
309/// config-bounded (the experiment and variant names from config).
310///
311/// - `ai_eval_results` — one row per eval case run, recording the
312/// input, expected, actual output, pass/fail, scorer, and
313/// attribution (eval name, model alias, variant, prompt version).
314/// The admin `/analytics/eval-results` endpoint reads these rows.
315///
316/// - `ai_feedback` — one row per feedback record ingested via the
317/// admin API, carrying the request id, label, comment, and
318/// attribution. The admin `/analytics/feedback` endpoint reads
319/// these rows.
320pub const SCHEMA_V6: &str = "
321 CREATE TABLE IF NOT EXISTS ai_experiment_assignments (
322 id INTEGER PRIMARY KEY AUTOINCREMENT,
323 ts_ms INTEGER NOT NULL,
324 request_id TEXT NOT NULL,
325 experiment TEXT NOT NULL,
326 variant TEXT NOT NULL,
327 model TEXT NOT NULL,
328 consumer TEXT NOT NULL DEFAULT ''
329 );
330 CREATE INDEX IF NOT EXISTS idx_ai_exp_assign_ts ON ai_experiment_assignments(ts_ms);
331 CREATE INDEX IF NOT EXISTS idx_ai_exp_assign_experiment ON ai_experiment_assignments(experiment);
332
333 CREATE TABLE IF NOT EXISTS ai_eval_results (
334 id INTEGER PRIMARY KEY AUTOINCREMENT,
335 ts_ms INTEGER NOT NULL,
336 eval_name TEXT NOT NULL,
337 model TEXT NOT NULL,
338 variant TEXT NOT NULL DEFAULT '',
339 prompt_version TEXT NOT NULL DEFAULT '',
340 case_index INTEGER NOT NULL,
341 input TEXT NOT NULL,
342 expected TEXT NOT NULL,
343 actual TEXT NOT NULL,
344 passed INTEGER NOT NULL,
345 scorer TEXT NOT NULL,
346 latency_ms REAL NOT NULL DEFAULT 0.0
347 );
348 CREATE INDEX IF NOT EXISTS idx_ai_eval_results_ts ON ai_eval_results(ts_ms);
349 CREATE INDEX IF NOT EXISTS idx_ai_eval_results_eval ON ai_eval_results(eval_name);
350
351 CREATE TABLE IF NOT EXISTS ai_feedback (
352 id INTEGER PRIMARY KEY AUTOINCREMENT,
353 ts_ms INTEGER NOT NULL,
354 request_id TEXT NOT NULL,
355 label TEXT NOT NULL,
356 comment TEXT NOT NULL DEFAULT '',
357 consumer TEXT NOT NULL DEFAULT '',
358 model TEXT NOT NULL DEFAULT ''
359 );
360 CREATE INDEX IF NOT EXISTS idx_ai_feedback_ts ON ai_feedback(ts_ms);
361 CREATE INDEX IF NOT EXISTS idx_ai_feedback_request_id ON ai_feedback(request_id);
362";
363
364/// Schema v7 (DW-087): the `mcp_tool_calls` table. One row per MCP
365/// tool call proxied through the gateway, carrying the session
366/// correlation id, the authenticated consumer, the tool name, the
367/// authorization outcome (`allowed`), the call duration, an optional
368/// error code, and a status string (`success`, `error`, or `denied`).
369/// The admin `/mcp/calls` endpoint reads these rows; the session and
370/// consumer indexes support per-session and per-consumer drill-down.
371pub const SCHEMA_V7: &str = "
372 CREATE TABLE IF NOT EXISTS mcp_tool_calls (
373 id INTEGER PRIMARY KEY AUTOINCREMENT,
374 ts_ms INTEGER NOT NULL,
375 request_id TEXT NOT NULL,
376 session_id TEXT NOT NULL,
377 consumer TEXT NOT NULL,
378 tool_name TEXT NOT NULL,
379 allowed INTEGER NOT NULL,
380 duration_ms REAL NOT NULL,
381 error_code TEXT,
382 status TEXT NOT NULL
383 );
384 CREATE INDEX IF NOT EXISTS idx_mcp_tool_calls_session ON mcp_tool_calls(session_id);
385 CREATE INDEX IF NOT EXISTS idx_mcp_tool_calls_consumer ON mcp_tool_calls(consumer);
386";
387
388/// Apply migrations to a fresh-or-existing analytics connection. A
389/// database at a NEWER version than this build is a hard error (the
390/// state store's forward-only rule, same rationale).
391pub fn migrate(conn: &rusqlite::Connection) -> Result<(), rusqlite::Error> {
392 let version: u32 = conn.query_row("PRAGMA user_version", [], |r| r.get(0))?;
393 if version > LATEST_SCHEMA_VERSION {
394 return Err(rusqlite::Error::ToSqlConversionFailure(Box::new(
395 std::io::Error::other(format!(
396 "analytics database schema version {version} is newer than this \
397 build's {LATEST_SCHEMA_VERSION}; forward-only: upgrade the binary"
398 )),
399 )));
400 }
401 // Forward-only sequential migrations: each version's batch is
402 // idempotent (IF NOT EXISTS) so a crash between batches re-runs
403 // safely on the next open.
404 if version < 1 {
405 conn.execute_batch(SCHEMA_V1)?;
406 conn.pragma_update(None, "user_version", 1)?;
407 }
408 if version < 2 {
409 conn.execute_batch(SCHEMA_V2)?;
410 conn.pragma_update(None, "user_version", 2)?;
411 }
412 if version < 3 {
413 conn.execute_batch(SCHEMA_V3)?;
414 conn.pragma_update(None, "user_version", 3)?;
415 }
416 if version < 4 {
417 conn.execute_batch(SCHEMA_V4)?;
418 conn.pragma_update(None, "user_version", 4)?;
419 }
420 if version < 5 {
421 conn.execute_batch(SCHEMA_V5)?;
422 conn.pragma_update(None, "user_version", 5)?;
423 }
424 if version < 6 {
425 conn.execute_batch(SCHEMA_V6)?;
426 conn.pragma_update(None, "user_version", 6)?;
427 }
428 if version < 7 {
429 conn.execute_batch(SCHEMA_V7)?;
430 conn.pragma_update(None, "user_version", 7)?;
431 }
432 if version < 8 {
433 conn.execute_batch(SCHEMA_V8)?;
434 conn.pragma_update(None, "user_version", 8)?;
435 }
436 if version < 9 {
437 conn.execute_batch(SCHEMA_V9)?;
438 conn.pragma_update(None, "user_version", 9)?;
439 }
440 if version < 10 {
441 conn.execute_batch(SCHEMA_V10)?;
442 conn.pragma_update(None, "user_version", 10)?;
443 }
444 if version < 11 {
445 conn.execute_batch(super::partition::SCHEMA_V11)?;
446 // SCALE-09 (#188): create the initial raw_all view (just
447 // `raw` — partition tables are added by `refresh_raw_all_view`
448 // as they are created). This ensures databases that migrate
449 // to v11 have the view even if they never call
450 // `refresh_raw_all_view` directly (e.g., test databases).
451 conn.execute_batch(
452 "DROP VIEW IF EXISTS raw_all;
453 CREATE VIEW raw_all AS SELECT * FROM raw;",
454 )?;
455 conn.pragma_update(None, "user_version", 11)?;
456 }
457 if version < 12 {
458 // PERF-12 (#209): add byte counters to the raw access table.
459 conn.execute_batch(SCHEMA_V11)?;
460 conn.pragma_update(None, "user_version", 12)?;
461 }
462 Ok(())
463}