//! Handlers for query audit trail endpoints. use axum::{ extract::{Path, Query as AxumQuery, State}, Json, }; use tracing::instrument; use crate::{ dto::{AuditQueryParams, QueryAuditListResponse, QueryAuditResponse}, error::{ApiError, Result}, hex as hex_utils, state::AppState, }; use stemedb_storage::{AuditStore, GenericAuditStore}; /// List recent query audits with optional filters. /// /// Returns a paginated list of query audit records. Use the `agent_id` and /// time range filters to narrow down results for incident investigation. /// /// # Query Parameters /// /// - `agent_id` (optional): Filter by agent public key (hex-encoded, 32 bytes) /// - `subject` (optional): Filter by subject (case-insensitive substring match) /// - `predicate` (optional): Filter by predicate (case-insensitive substring match) /// - `from` (optional): Start of time range (Unix timestamp) /// - `to` (optional): End of time range (Unix timestamp) /// - `limit` (optional): Maximum number of results (default: 100) #[utoipa::path( get, path = "/v1/audit/queries", params(AuditQueryParams), responses( (status = 200, description = "List of query audits", body = QueryAuditListResponse), (status = 400, description = "Invalid request", body = crate::dto::ErrorResponse), (status = 500, description = "Internal server error", body = crate::dto::ErrorResponse) ), tag = "audit" )] #[instrument(skip(state), fields( agent_id = ?params.agent_id, subject = ?params.subject, predicate = ?params.predicate, from = ?params.from, to = ?params.to, limit = params.limit ))] pub async fn list_audits( State(state): State, AxumQuery(params): AxumQuery, ) -> Result> { let start = std::time::Instant::now(); metrics::counter!("stemedb_http_requests_total", "method" => "GET", "path" => "/v1/audit/queries").increment(1); let audit_store = GenericAuditStore::new(state.store.clone()); // Fetch a larger set to allow for subject/predicate filtering // (storage layer doesn't have an index by subject/predicate) let fetch_limit = if params.subject.is_some() || params.predicate.is_some() { params.limit * 10 // Fetch more to compensate for filtering } else { params.limit }; let mut audits = if let Some(agent_id_hex) = ¶ms.agent_id { // Filter by agent with limit pushed down to storage layer let agent_id = hex_utils::decode_agent_id(agent_id_hex)?; let from = params.from.unwrap_or(0); let to = params.to; audit_store.get_audits_for_agent(&agent_id, from, to, fetch_limit).await? } else { // List recent (limit already applied in storage layer) audit_store.list_recent_audits(fetch_limit).await? }; // Apply time range filter if not already filtered by agent if params.agent_id.is_none() { if let Some(from_ts) = params.from { audits.retain(|a| a.timestamp >= from_ts); } if let Some(to_ts) = params.to { audits.retain(|a| a.timestamp <= to_ts); } } // Apply subject filter (case-insensitive substring match) if let Some(ref subject_filter) = params.subject { let filter_lower = subject_filter.to_lowercase(); audits.retain(|a| { a.params .subject .as_ref() .map(|s| s.to_lowercase().contains(&filter_lower)) .unwrap_or(false) }); } // Apply predicate filter (case-insensitive substring match) if let Some(ref predicate_filter) = params.predicate { let filter_lower = predicate_filter.to_lowercase(); audits.retain(|a| { a.params .predicate .as_ref() .map(|p| p.to_lowercase().contains(&filter_lower)) .unwrap_or(false) }); } // Apply final limit audits.truncate(params.limit); let total_count = audits.len(); // Convert to response DTOs let audit_responses: Vec = audits.into_iter().map(QueryAuditResponse::from).collect(); // Track request duration (success case) metrics::histogram!("stemedb_http_request_duration_seconds", "method" => "GET", "path" => "/v1/audit/queries", "status" => "200" ) .record(start.elapsed().as_secs_f64()); Ok(Json(QueryAuditListResponse { audits: audit_responses, total_count })) } /// Get a specific query audit by ID. /// /// Returns the full audit record including all contributing assertions /// that influenced the query result. #[utoipa::path( get, path = "/v1/audit/query/{id}", params( ("id" = String, Path, description = "Query ID (hex-encoded, 32 bytes)") ), responses( (status = 200, description = "Query audit details", body = QueryAuditResponse), (status = 400, description = "Invalid query ID", body = crate::dto::ErrorResponse), (status = 404, description = "Audit not found", body = crate::dto::ErrorResponse), (status = 500, description = "Internal server error", body = crate::dto::ErrorResponse) ), tag = "audit" )] #[instrument(skip(state), fields(query_id = %id))] pub async fn get_audit( State(state): State, Path(id): Path, ) -> Result> { let start = std::time::Instant::now(); metrics::counter!("stemedb_http_requests_total", "method" => "GET", "path" => "/v1/audit/query/{id}").increment(1); let query_id = hex_utils::decode_hash_32(&id)?; let audit_store = GenericAuditStore::new(state.store.clone()); match audit_store.get_audit(&query_id).await? { Some(audit) => { // Track request duration (success case) metrics::histogram!("stemedb_http_request_duration_seconds", "method" => "GET", "path" => "/v1/audit/query/{id}", "status" => "200" ) .record(start.elapsed().as_secs_f64()); Ok(Json(QueryAuditResponse::from(audit))) } None => Err(ApiError::NotFound(format!("Query audit not found: {}", id))), } }