pub mod dto; pub mod mcp_server; use std::collections::BTreeMap; use std::path::PathBuf; use std::sync::Arc; use crate::api::dto::{ BucketValuesParams, CreateAlertRuleRequest, CreateDashboardRequest, CreateProfileRequest, ErrorDto, MetricDeclarationDto, ParamDeclarationDto, LogsParams, ProfileSummaryDto, PromqlParams, SaveDashboardRequest, SettingsDto, TestAlertNotificationRequest, TimelineParams, UpdateAlertRuleRequest, UpdateSettingsRequest, }; use crate::engine::data_value::MetricPrimitiveKind; use crate::engine::{ DeclareMetricRequest, DeclareParamRequest, EngineHandle, EngineMessage, EngineQuery, EngineResponse, }; use crate::engine::alerts::{deliver_ntfy, AlertError}; use crate::config::profile::{Dashboard, ParamDeclarationRecord, Profile, ProfileError}; use crate::config::SettingsHandle; use axum::extract::{Path as AxumPath, Query, State}; use axum::http::{header, StatusCode}; use axum::response::{IntoResponse, Json, Response}; use axum::routing::{delete, get, post, put}; use axum::Router; use fs_embed::Dir as EmbedDir; use tokio::sync::oneshot; static WEBAPP_DIST: EmbedDir = fs_embed::fs_embed!("webapp/dist"); // Every file here is a selectable theme: a complete stylesheet (colors, spacing, fonts, ...) // applied on top of the base app shell. Add a new theme by dropping another `.css` file in // `themes/` and rebuilding — no code changes needed (see `webapp/src/index.css`). static THEMES: EmbedDir = fs_embed::fs_embed!("themes"); const SKILL_MD: &str = include_str!("../../SKILL.md"); #[derive(Clone)] struct ApiState { engine: EngineHandle, settings: SettingsHandle, settings_path: Arc, } impl ApiState { fn persist_param_declaration(&self, request: &DeclareParamRequest) { let mut settings = self.settings.lock().unwrap(); let Some(profile_name) = settings.active_profile.clone() else { return; }; if let Err(err) = settings.record_param_declaration(&self.settings_path, &profile_name, request) { eprintln!( "warning: failed to persist param declaration '{}' into profile '{profile_name}': {err}", request.name ); } } fn persist_clear_params(&self) { let mut settings = self.settings.lock().unwrap(); let Some(profile_name) = settings.active_profile.clone() else { return; }; if let Err(err) = settings.clear_param_declarations(&self.settings_path, &profile_name) { eprintln!( "warning: failed to clear persisted param declarations for profile '{profile_name}': {err}" ); } } async fn persist_metric_declaration(&self, request: &DeclareMetricRequest) { let profile_name = { let settings = self.settings.lock().unwrap(); let Some(profile_name) = settings.active_profile.clone() else { return; }; profile_name }; let existing_params = { let settings = self.settings.lock().unwrap(); settings.profile(&profile_name).map(|p| p.params.clone()).unwrap_or_default() }; let promoted = promoted_param_records(&self.engine, &existing_params, request).await; let mut settings = self.settings.lock().unwrap(); if let Err(err) = settings.record_metric_declaration( &self.settings_path, &profile_name, request, &promoted, ) { eprintln!( "warning: failed to persist metric declaration '{}' into profile '{profile_name}': {err}", request.name ); } } async fn persist_metric_declaration_update(&self, old_name: &str, request: &DeclareMetricRequest) { let profile_name = { let settings = self.settings.lock().unwrap(); let Some(profile_name) = settings.active_profile.clone() else { return; }; profile_name }; let existing_params = { let settings = self.settings.lock().unwrap(); settings.profile(&profile_name).map(|p| p.params.clone()).unwrap_or_default() }; let promoted = promoted_param_records(&self.engine, &existing_params, request).await; let mut settings = self.settings.lock().unwrap(); if let Err(err) = settings.record_metric_declaration_update( &self.settings_path, &profile_name, old_name, request, &promoted, ) { eprintln!( "warning: failed to persist metric declaration update '{}' into profile '{profile_name}': {err}", request.name ); } } fn persist_metric_declaration_removal(&self, name: &str) { let mut settings = self.settings.lock().unwrap(); let Some(profile_name) = settings.active_profile.clone() else { return; }; if let Err(err) = settings.remove_metric_declaration(&self.settings_path, &profile_name, name) { eprintln!( "warning: failed to remove metric declaration '{name}' from profile '{profile_name}': {err}" ); } } } pub fn router(engine: EngineHandle, settings: SettingsHandle, settings_path: PathBuf) -> Router { let state = ApiState { engine, settings, settings_path: Arc::new(settings_path), }; Router::new() .route("/SKILL.md", get(get_skill_md)) .route("/api/logs", get(get_logs)) .route("/api/timeline", get(get_timeline)) .route("/api/stats", get(get_stats)) .route("/api/params", get(list_params).post(declare_param).delete(clear_params)) .route("/api/metrics", get(list_metrics).post(declare_metric)) .route( "/api/metrics/{name}", get(get_metric_bucket_values).put(update_metric).delete(delete_metric), ) .route("/api/query", get(run_promql_query)) .route("/api/settings", get(get_settings).patch(update_settings)) .route("/api/themes", get(list_themes)) .route("/api/themes/{name}", get(get_theme_content)) .route("/api/settings/alerts", get(list_alert_rules).post(create_alert_rule)) .route( "/api/settings/alerts/{id}", put(update_alert_rule).delete(delete_alert_rule), ) .route("/api/settings/alerts/test", post(test_alert_notification)) .route("/api/settings/profiles", get(list_profiles).post(create_profile)) .route("/api/settings/profiles/{name}", delete(delete_profile)) .route("/api/settings/profiles/{name}/activate", post(activate_profile)) .route( "/api/settings/profiles/{name}/dashboards", get(list_dashboards).post(create_dashboard), ) .route( "/api/settings/profiles/{name}/dashboards/{dashboard_id}", get(get_dashboard).put(save_dashboard).delete(delete_dashboard), ) .with_state(state) .fallback(serve_webapp) } async fn get_skill_md() -> impl IntoResponse { ([(header::CONTENT_TYPE, "text/markdown; charset=utf-8")], SKILL_MD) } async fn serve_webapp(uri: axum::http::Uri) -> Response { let path = uri.path().trim_start_matches('/'); let file = WEBAPP_DIST .get_file(path) .or_else(|| WEBAPP_DIST.get_file("index.html")); match file { Some(file) => { let mime = mime_for_extension(file.extension().unwrap_or("")); let bytes = file.read_bytes().unwrap_or_default(); ([(header::CONTENT_TYPE, mime)], bytes).into_response() } None => StatusCode::NOT_FOUND.into_response(), } } fn mime_for_extension(extension: &str) -> &'static str { match extension { "html" => "text/html; charset=utf-8", "js" => "text/javascript; charset=utf-8", "css" => "text/css; charset=utf-8", "svg" => "image/svg+xml", "json" => "application/json", "png" => "image/png", "ico" => "image/x-icon", "woff" => "font/woff", "woff2" => "font/woff2", _ => "application/octet-stream", } } impl EngineHandle { async fn ask(&self, query: EngineQuery) -> Result { let read_handle = self.reader.handle(); if let Some(sort_by) = query.sort_by() { let needs_build = match read_handle.enter() { Some(guard) => !guard.has_sort_helper_index(sort_by), None => return Err(StatusCode::SERVICE_UNAVAILABLE), }; if needs_build { let (respond_to, response) = oneshot::channel(); self.tx .send(EngineMessage::EnsureSortHelperIndex { param_name: sort_by.to_string(), respond_to }) .await .map_err(|_| StatusCode::SERVICE_UNAVAILABLE)?; response.await.map_err(|_| StatusCode::SERVICE_UNAVAILABLE)?; } } let guard = read_handle.enter().ok_or(StatusCode::SERVICE_UNAVAILABLE)?; Ok(guard.handle_query_ro(query)) } } async fn promoted_param_records( engine: &EngineHandle, existing_params: &[ParamDeclarationRecord], request: &DeclareMetricRequest, ) -> Vec { let missing: Vec<&str> = request .label_params .iter() .map(String::as_str) .chain(request.value_param.as_deref()) .filter(|name| !existing_params.iter().any(|p| p.name == **name)) .collect(); if missing.is_empty() { return Vec::new(); } let Ok(EngineResponse::ParamDeclarations(declarations)) = engine.ask(EngineQuery::ParamDeclarations).await else { return Vec::new(); }; declarations .into_iter() .filter(|param| missing.contains(¶m.name.as_str())) .map(|param| ParamDeclarationRecord { name: param.name, message_match_regex: param.message_match_regex, capture_group: param.capture_group, json_field_path: param.json_field_path, kind: param.kind.into(), elapsed_since_last: param.elapsed_since_last, }) .collect() } async fn get_logs( State(state): State, Query(params): Query, ) -> impl IntoResponse { match state.engine.ask(EngineQuery::Logs(params)).await { Ok(EngineResponse::Logs(entries)) => Json(entries).into_response(), Ok(EngineResponse::Error(err)) => { (StatusCode::BAD_REQUEST, Json(ErrorDto { error: err })).into_response() } Ok(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(), Err(status) => status.into_response(), } } async fn get_timeline( State(state): State, Query(params): Query, ) -> impl IntoResponse { match state.engine.ask(EngineQuery::Timeline(params)).await { Ok(EngineResponse::Timeline(buckets)) => Json(buckets).into_response(), Ok(EngineResponse::Error(err)) => { (StatusCode::BAD_REQUEST, Json(ErrorDto { error: err })).into_response() } Ok(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(), Err(status) => status.into_response(), } } async fn get_stats(State(state): State) -> impl IntoResponse { match state.engine.ask(EngineQuery::Stats).await { Ok(EngineResponse::Stats(stats)) => Json(stats).into_response(), Ok(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(), Err(status) => status.into_response(), } } async fn list_params(State(state): State) -> impl IntoResponse { match state.engine.ask(EngineQuery::ParamDeclarations).await { Ok(EngineResponse::ParamDeclarations(mut declarations)) => { declarations.extend(ParamDeclarationDto::reserved()); Json(declarations).into_response() } Ok(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(), Err(status) => status.into_response(), } } impl ParamDeclarationDto { fn reserved() -> [ParamDeclarationDto; 3] { [ ParamDeclarationDto { name: "ottyl_seq".to_string(), message_match_regex: String::new(), capture_group: String::new(), json_field_path: None, kind: MetricPrimitiveKind::Number, elapsed_since_last: false, auto_derived: false, reserved: true, }, ParamDeclarationDto { name: "ottyl_ts".to_string(), message_match_regex: String::new(), capture_group: String::new(), json_field_path: None, kind: MetricPrimitiveKind::Timestamp, elapsed_since_last: false, auto_derived: false, reserved: true, }, ParamDeclarationDto { name: "ottyl_fmt".to_string(), message_match_regex: String::new(), capture_group: String::new(), json_field_path: None, kind: MetricPrimitiveKind::String, elapsed_since_last: false, auto_derived: false, reserved: true, }, ] } } async fn declare_param( State(state): State, Json(body): Json, ) -> impl IntoResponse { let (respond_to, response) = oneshot::channel(); let request = body.clone(); if state .engine .tx .send(EngineMessage::DeclareParam { request: body, respond_to }) .await .is_err() { return StatusCode::SERVICE_UNAVAILABLE.into_response(); } match response.await { Ok(Ok(())) => { state.persist_param_declaration(&request); let dto = ParamDeclarationDto { name: request.name, message_match_regex: request.message_match_regex, capture_group: request.capture_group, json_field_path: request.json_field_path, kind: request.kind, elapsed_since_last: request.elapsed_since_last, auto_derived: false, reserved: false, }; (StatusCode::CREATED, Json(dto)).into_response() } Ok(Err(err)) => (StatusCode::BAD_REQUEST, Json(ErrorDto { error: err })).into_response(), Err(_) => StatusCode::SERVICE_UNAVAILABLE.into_response(), } } async fn clear_params(State(state): State) -> impl IntoResponse { let (respond_to, response) = oneshot::channel(); if state.engine.tx.send(EngineMessage::ClearParams { respond_to }).await.is_err() { return StatusCode::SERVICE_UNAVAILABLE.into_response(); } match response.await { Ok(result) => { state.persist_clear_params(); Json(result).into_response() } Err(_) => StatusCode::SERVICE_UNAVAILABLE.into_response(), } } async fn list_metrics(State(state): State) -> impl IntoResponse { match state.engine.ask(EngineQuery::MetricDeclarations).await { Ok(EngineResponse::MetricDeclarations(declarations)) => Json(declarations).into_response(), Ok(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(), Err(status) => status.into_response(), } } async fn declare_metric( State(state): State, Json(body): Json, ) -> impl IntoResponse { let (respond_to, response) = oneshot::channel(); let request = body.clone(); if state .engine .tx .send(EngineMessage::DeclareMetric { request: body, respond_to }) .await .is_err() { return StatusCode::SERVICE_UNAVAILABLE.into_response(); } match response.await { Ok(Ok(())) => { state.persist_metric_declaration(&request).await; let dto = MetricDeclarationDto { name: request.name, message_match_regex: request.message_match_regex, label_params: request.label_params, value_param: request.value_param, matched_labels: request.matched_labels, unmatched_labels: request.unmatched_labels, }; (StatusCode::CREATED, Json(dto)).into_response() } Ok(Err(err)) => (StatusCode::BAD_REQUEST, Json(ErrorDto { error: err })).into_response(), Err(_) => StatusCode::SERVICE_UNAVAILABLE.into_response(), } } async fn update_metric( State(state): State, AxumPath(name): AxumPath, Json(body): Json, ) -> impl IntoResponse { let (respond_to, response) = oneshot::channel(); let request = body.clone(); if state .engine .tx .send(EngineMessage::UpdateMetric { name: name.clone(), request: body, respond_to }) .await .is_err() { return StatusCode::SERVICE_UNAVAILABLE.into_response(); } match response.await { Ok(Ok(())) => { state.persist_metric_declaration_update(&name, &request).await; let dto = MetricDeclarationDto { name: request.name, message_match_regex: request.message_match_regex, label_params: request.label_params, value_param: request.value_param, matched_labels: request.matched_labels, unmatched_labels: request.unmatched_labels, }; (StatusCode::OK, Json(dto)).into_response() } Ok(Err(err)) => (StatusCode::BAD_REQUEST, Json(ErrorDto { error: err })).into_response(), Err(_) => StatusCode::SERVICE_UNAVAILABLE.into_response(), } } async fn delete_metric( State(state): State, AxumPath(name): AxumPath, ) -> impl IntoResponse { let (respond_to, response) = oneshot::channel(); if state .engine .tx .send(EngineMessage::RemoveMetricDeclaration { name: name.clone(), respond_to }) .await .is_err() { return StatusCode::SERVICE_UNAVAILABLE.into_response(); } match response.await { Ok(true) => { state.persist_metric_declaration_removal(&name); StatusCode::NO_CONTENT.into_response() } Ok(false) => StatusCode::NOT_FOUND.into_response(), Err(_) => StatusCode::SERVICE_UNAVAILABLE.into_response(), } } fn parse_labels_filter(raw: &str) -> BTreeMap { raw.split(',') .filter_map(|pair| pair.split_once('=')) .map(|(name, value)| (name.trim().to_string(), value.trim().to_string())) .collect() } async fn get_metric_bucket_values( State(state): State, AxumPath(metric_name): AxumPath, Query(params): Query, ) -> impl IntoResponse { let labels_filter = params.labels.as_deref().map(parse_labels_filter); let query = EngineQuery::BucketValues { metric_name, labels_filter, start_ms: params.start_ms, end_ms: params.end_ms, since_sequence: params.since_sequence, limit: params.limit, }; match state.engine.ask(query).await { Ok(EngineResponse::BucketValues(values)) => Json(values).into_response(), Ok(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(), Err(status) => status.into_response(), } } async fn run_promql_query( State(state): State, Query(params): Query, ) -> impl IntoResponse { let query = EngineQuery::Promql { query: params.promql, start_ms: params.start_ms, end_ms: params.end_ms, }; match state.engine.ask(query).await { Ok(EngineResponse::Promql(Ok(series))) => Json(series).into_response(), Ok(EngineResponse::Promql(Err(err))) => { (StatusCode::BAD_REQUEST, Json(ErrorDto { error: err })).into_response() } Ok(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(), Err(status) => status.into_response(), } } async fn list_themes() -> impl IntoResponse { let mut names: Vec = THEMES .walk() .filter_map(|file| file.file_name().map(|name| name.trim_end_matches(".css").to_string())) .collect(); names.sort(); Json(names).into_response() } async fn get_theme_content(AxumPath(name): AxumPath) -> impl IntoResponse { // Themes are just filenames within a flat embedded dir, not filesystem paths, so reject // anything that isn't a plain name to rule out path traversal. if name.is_empty() || !name.chars().all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-') { return StatusCode::NOT_FOUND.into_response(); } match THEMES.get_file(&format!("{name}.css")) { Some(file) => { let bytes = file.read_bytes().unwrap_or_default(); ([(header::CONTENT_TYPE, "text/css; charset=utf-8")], bytes).into_response() } None => StatusCode::NOT_FOUND.into_response(), } } impl AlertError { fn into_response(self) -> axum::response::Response { match self { AlertError::NotFound => (StatusCode::NOT_FOUND, self.to_string()).into_response(), AlertError::Io(_) => (StatusCode::INTERNAL_SERVER_ERROR, self.to_string()).into_response(), } } } async fn list_alert_rules(State(state): State) -> impl IntoResponse { let settings = state.settings.lock().unwrap(); Json(settings.alert_rules.clone()).into_response() } async fn create_alert_rule( State(state): State, Json(body): Json, ) -> impl IntoResponse { let mut settings = state.settings.lock().unwrap(); match settings.create_alert_rule(&state.settings_path, body.name, body.enabled, body.condition) { Ok(rule) => (StatusCode::CREATED, Json(rule)).into_response(), Err(err) => err.into_response(), } } async fn update_alert_rule( State(state): State, AxumPath(id): AxumPath, Json(body): Json, ) -> impl IntoResponse { let mut settings = state.settings.lock().unwrap(); match settings.update_alert_rule(&state.settings_path, &id, body.name, body.enabled, body.condition) { Ok(rule) => Json(rule).into_response(), Err(err) => err.into_response(), } } async fn delete_alert_rule(State(state): State, AxumPath(id): AxumPath) -> impl IntoResponse { let mut settings = state.settings.lock().unwrap(); match settings.remove_alert_rule(&state.settings_path, &id) { Ok(()) => StatusCode::NO_CONTENT.into_response(), Err(err) => err.into_response(), } } async fn test_alert_notification(Json(body): Json) -> impl IntoResponse { let server = body.server.unwrap_or_else(|| "https://ntfy.sh".to_string()); match deliver_ntfy(&server, &body.topic, "Test notification from ottyl").await { Ok(()) => StatusCode::NO_CONTENT.into_response(), Err(err) => (StatusCode::BAD_GATEWAY, err).into_response(), } } async fn get_settings(State(state): State) -> impl IntoResponse { let settings = state.settings.lock().unwrap(); Json(SettingsDto::from(&*settings)).into_response() } async fn update_settings( State(state): State, Json(body): Json, ) -> impl IntoResponse { let max_logs_changed = body.max_logs.is_some(); let new_max_logs = { let mut settings = state.settings.lock().unwrap(); if let Some(theme) = body.theme { settings.theme = theme; } if let Some(max_logs) = body.max_logs { settings.max_logs = max_logs; } if let Some(ntfy_topic) = body.ntfy_topic { settings.ntfy_topic = ntfy_topic; } if let Some(ntfy_server) = body.ntfy_server { settings.ntfy_server = ntfy_server; } match settings.save(&state.settings_path) { Ok(()) => (), Err(err) => return (StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response(), } (SettingsDto::from(&*settings), settings.max_logs) }; if max_logs_changed { let _ = state .engine .tx .send(EngineMessage::SetMaxLogs(new_max_logs.1)) .await; state.engine.barrier().await; } Json(new_max_logs.0).into_response() } impl ProfileError { fn into_response(self) -> axum::response::Response { match self { ProfileError::NotFound => (StatusCode::NOT_FOUND, self.to_string()).into_response(), ProfileError::DuplicateName => (StatusCode::BAD_REQUEST, self.to_string()).into_response(), ProfileError::CannotDeleteDefault => (StatusCode::BAD_REQUEST, self.to_string()).into_response(), ProfileError::Io(_) => (StatusCode::INTERNAL_SERVER_ERROR, self.to_string()).into_response(), } } } async fn list_profiles(State(state): State) -> impl IntoResponse { let settings = state.settings.lock().unwrap(); let summaries: Vec = settings.profiles.iter().map(ProfileSummaryDto::from).collect(); Json(summaries).into_response() } async fn create_profile( State(state): State, Json(body): Json, ) -> impl IntoResponse { let mut settings = state.settings.lock().unwrap(); match settings.create_profile(&state.settings_path, body.name.clone()) { Ok(()) => { let profile = settings.profile(&body.name).expect("profile was just created by create_profile"); (StatusCode::CREATED, Json(ProfileSummaryDto::from(profile))).into_response() } Err(err) => err.into_response(), } } async fn delete_profile( State(state): State, AxumPath(name): AxumPath, ) -> impl IntoResponse { let mut settings = state.settings.lock().unwrap(); match settings.delete_profile(&state.settings_path, &name) { Ok(()) => StatusCode::NO_CONTENT.into_response(), Err(err) => err.into_response(), } } async fn reload_engine_declarations(engine: &EngineHandle, profile: &Profile) { if let Ok(EngineResponse::ParamDeclarations(params)) = engine.ask(EngineQuery::ParamDeclarations).await { for param in params.into_iter().filter(|param| !param.auto_derived) { let (respond_to, response) = oneshot::channel(); if engine .tx .send(EngineMessage::RemoveParamDeclaration { name: param.name, respond_to }) .await .is_ok() { let _ = response.await; } } } if let Ok(EngineResponse::MetricDeclarations(metrics)) = engine.ask(EngineQuery::MetricDeclarations).await { for metric in metrics { let (respond_to, response) = oneshot::channel(); if engine .tx .send(EngineMessage::RemoveMetricDeclaration { name: metric.name, respond_to }) .await .is_ok() { let _ = response.await; } } } profile.load_into_engine(engine).await; } async fn activate_profile( State(state): State, AxumPath(name): AxumPath, ) -> impl IntoResponse { let activated_profile = { let mut settings = state.settings.lock().unwrap(); match settings.set_active_profile(&state.settings_path, Some(name.clone())) { Ok(()) => Ok(settings.profile(&name).cloned()), Err(err) => Err(err), } }; match activated_profile { Ok(profile) => { if let Some(profile) = profile { reload_engine_declarations(&state.engine, &profile).await; } let settings = state.settings.lock().unwrap(); Json(SettingsDto::from(&*settings)).into_response() } Err(err) => err.into_response(), } } async fn list_dashboards( State(state): State, AxumPath(name): AxumPath, ) -> impl IntoResponse { let settings = state.settings.lock().unwrap(); match settings.list_dashboards(&name) { Ok(dashboards) => Json(dashboards).into_response(), Err(err) => err.into_response(), } } async fn create_dashboard( State(state): State, AxumPath(name): AxumPath, Json(body): Json, ) -> impl IntoResponse { let mut settings = state.settings.lock().unwrap(); match settings.create_dashboard(&state.settings_path, &name, body.name) { Ok(dashboard) => (StatusCode::CREATED, Json(dashboard)).into_response(), Err(err) => err.into_response(), } } async fn get_dashboard( State(state): State, AxumPath((name, dashboard_id)): AxumPath<(String, String)>, ) -> impl IntoResponse { let settings = state.settings.lock().unwrap(); match settings.list_dashboards(&name) { Ok(dashboards) => match dashboards.iter().find(|d| d.id == dashboard_id) { Some(dashboard) => Json(dashboard).into_response(), None => ProfileError::NotFound.into_response(), }, Err(err) => err.into_response(), } } async fn save_dashboard( State(state): State, AxumPath((name, dashboard_id)): AxumPath<(String, String)>, Json(body): Json, ) -> impl IntoResponse { let mut settings = state.settings.lock().unwrap(); let dashboard = Dashboard { id: dashboard_id.clone(), name: body.name, panels: body.panels, }; match settings.save_dashboard(&state.settings_path, &name, &dashboard_id, dashboard) { Ok(dashboard) => Json(dashboard).into_response(), Err(err) => err.into_response(), } } async fn delete_dashboard( State(state): State, AxumPath((name, dashboard_id)): AxumPath<(String, String)>, ) -> impl IntoResponse { let mut settings = state.settings.lock().unwrap(); match settings.delete_dashboard(&state.settings_path, &name, &dashboard_id) { Ok(()) => StatusCode::NO_CONTENT.into_response(), Err(err) => err.into_response(), } } #[cfg(test)] mod tests { use super::*; use crate::engine::{self, IngestedLine}; use crate::config::Settings; use crate::engine::timestamp::Timestamp; use axum::body::Body; use axum::http::Request; use http_body_util::BodyExt; use serde_json::{json, Value}; use std::sync::Mutex; use tower::ServiceExt; fn test_app() -> (Router, EngineHandle, SettingsHandle, tempfile::TempDir) { let dir = tempfile::tempdir().unwrap(); let settings_path = dir.path().join("settings.json"); let settings = Arc::new(Mutex::new(Settings::load(&settings_path))); let engine_tx = engine::spawn(settings.clone(), 128); let app = router(engine_tx.clone(), settings.clone(), settings_path); (app, engine_tx, settings, dir) } async fn send_json(app: &Router, method: &str, uri: &str, body: Value) -> (StatusCode, Value) { let request = Request::builder() .method(method) .uri(uri) .header(header::CONTENT_TYPE, "application/json") .body(Body::from(body.to_string())) .unwrap(); let response = app.clone().oneshot(request).await.unwrap(); let status = response.status(); let bytes = response.into_body().collect().await.unwrap().to_bytes(); let value = if bytes.is_empty() { Value::Null } else { serde_json::from_slice(&bytes).unwrap() }; (status, value) } async fn get_json(app: &Router, uri: &str) -> (StatusCode, Value) { let request = Request::builder().method("GET").uri(uri).body(Body::empty()).unwrap(); let response = app.clone().oneshot(request).await.unwrap(); let status = response.status(); let bytes = response.into_body().collect().await.unwrap().to_bytes(); let value = if bytes.is_empty() { Value::Null } else { serde_json::from_slice(&bytes).unwrap() }; (status, value) } #[tokio::test] async fn alert_rule_crud_round_trips_through_http() { let (app, _engine, _settings, _dir) = test_app(); let (status, body) = send_json( &app, "POST", "/api/settings/alerts", json!({ "name": "error spike", "enabled": true, "condition": { "kind": "immediate", "query": "regex(\"error\")" } }), ) .await; assert_eq!(status, StatusCode::CREATED); assert_eq!(body["name"], "error spike"); let id = body["id"].as_str().unwrap().to_string(); let (status, body) = get_json(&app, "/api/settings/alerts").await; assert_eq!(status, StatusCode::OK); assert_eq!(body.as_array().unwrap().len(), 1); let (status, body) = send_json( &app, "PUT", &format!("/api/settings/alerts/{id}"), json!({ "name": "renamed", "enabled": false, "condition": { "kind": "immediate", "query": "regex(\"warn\")" } }), ) .await; assert_eq!(status, StatusCode::OK); assert_eq!(body["name"], "renamed"); assert_eq!(body["enabled"], false); let status = delete_request(&app, &format!("/api/settings/alerts/{id}")).await; assert_eq!(status, StatusCode::NO_CONTENT); let (_, body) = get_json(&app, "/api/settings/alerts").await; assert!(body.as_array().unwrap().is_empty()); } #[tokio::test] async fn alert_rule_routes_404_for_unknown_id() { let (app, _engine, _settings, _dir) = test_app(); let status = delete_request(&app, "/api/settings/alerts/does-not-exist").await; assert_eq!(status, StatusCode::NOT_FOUND); let request = Request::builder() .method("PUT") .uri("/api/settings/alerts/does-not-exist") .header(header::CONTENT_TYPE, "application/json") .body(Body::from( json!({ "name": "x", "enabled": true, "condition": { "kind": "immediate", "query": "regex(\"x\")" } }) .to_string(), )) .unwrap(); let response = app.clone().oneshot(request).await.unwrap(); assert_eq!(response.status(), StatusCode::NOT_FOUND); } #[tokio::test] async fn declares_param_and_returns_201() { let (app, _engine, _settings, _dir) = test_app(); let (status, body) = send_json( &app, "POST", "/api/params", json!({ "name": "level", "message_match_regex": r"level=(?P\w+)", "capture_group": "level", "kind": "string" }), ) .await; assert_eq!(status, StatusCode::CREATED); assert_eq!(body["name"], "level"); assert_eq!(body["capture_group"], "level"); } #[tokio::test] async fn clearing_params_persists_and_reextracts_from_retained_logs() { let (app, engine, settings_handle, dir) = test_app(); let settings_path = dir.path().join("settings.json"); { let mut settings = settings_handle.lock().unwrap(); settings.create_profile(&settings_path, "web".to_string()).unwrap(); settings.set_active_profile(&settings_path, Some("web".to_string())).unwrap(); } declare_test_param(&app, "path", r"request path=(?P\S+)", "path", "string").await; ingest_lines(&engine, 0, &["request path=/a", "level=info x=1"]).await; let (_, before) = get_json(&app, "/api/params").await; let before_declared: Vec<&str> = before .as_array() .unwrap() .iter() .filter(|p| p["reserved"] != true) .map(|p| p["name"].as_str().unwrap()) .collect(); assert!(before_declared.contains(&"path")); assert!(before_declared.contains(&"level")); assert!(before_declared.contains(&"x")); let (status, result) = send_json(&app, "DELETE", "/api/params", json!({})).await; assert_eq!(status, StatusCode::OK); assert_eq!(result["removed"], before_declared.len() as u64); assert_eq!(result["reextracted"], 3); assert_eq!(result["logs_considered"], 2); let (_, after) = get_json(&app, "/api/params").await; let after_declared: Vec<_> = after.as_array().unwrap().iter().filter(|p| p["reserved"] != true).collect(); assert_eq!(after_declared.len(), 3); assert!(after_declared.iter().all(|p| p["auto_derived"] == true)); let reloaded_settings = Settings::load(&settings_path); let profile = reloaded_settings.profile("web").unwrap(); assert!(profile.params.is_empty()); } #[tokio::test] async fn rejects_invalid_param_regex_with_400_and_error_body() { let (app, _engine, _settings, _dir) = test_app(); let (status, body) = send_json( &app, "POST", "/api/params", json!({ "name": "level", "message_match_regex": "level=(", "capture_group": "level", "kind": "string" }), ) .await; assert_eq!(status, StatusCode::BAD_REQUEST); assert!(body["error"].as_str().unwrap().contains("invalid regex")); } async fn declare_test_param(app: &Router, name: &str, pattern: &str, capture_group: &str, kind: &str) { let (status, _body) = send_json( app, "POST", "/api/params", json!({ "name": name, "message_match_regex": pattern, "capture_group": capture_group, "kind": kind, }), ) .await; assert_eq!(status, StatusCode::CREATED); } #[tokio::test] async fn declares_metric_and_shows_up_in_list() { let (app, _engine, _settings, _dir) = test_app(); declare_test_param(&app, "path", r"request path=(?P\S+)", "path", "string").await; let (status, body) = send_json( &app, "POST", "/api/metrics", json!({ "name": "requests", "message_match_regex": r"request path=(?P\S+)", "label_params": ["path"] }), ) .await; assert_eq!(status, StatusCode::CREATED); assert_eq!(body["name"], "requests"); let (status, body) = get_json(&app, "/api/metrics").await; assert_eq!(status, StatusCode::OK); let metrics = body.as_array().unwrap(); assert_eq!(metrics.len(), 1); assert_eq!(metrics[0]["name"], "requests"); } #[tokio::test] async fn rejects_metric_referencing_undeclared_param() { let (app, _engine, _settings, _dir) = test_app(); let (status, body) = send_json( &app, "POST", "/api/metrics", json!({ "name": "requests", "message_match_regex": r"request path=(?P\S+)", "label_params": ["duration"] }), ) .await; assert_eq!(status, StatusCode::BAD_REQUEST); assert!(body["error"].as_str().unwrap().contains("duration")); } #[tokio::test] async fn updates_metric_declaration_in_place_rather_than_duplicating() { let (app, _engine, _settings, _dir) = test_app(); declare_test_param(&app, "path", r"request path=(?P\S+)", "path", "string").await; declare_test_param(&app, "duration", r"duration=(?P\d+)", "duration", "number").await; send_json( &app, "POST", "/api/metrics", json!({ "name": "requests", "message_match_regex": r"request path=(?P\S+)", "label_params": ["path"] }), ) .await; let (status, body) = send_json( &app, "PUT", "/api/metrics/requests", json!({ "name": "requests", "message_match_regex": r"request path=(?P\S+) duration=(?P\d+)", "label_params": ["path", "duration"] }), ) .await; assert_eq!(status, StatusCode::OK); assert_eq!(body["name"], "requests"); assert_eq!(body["label_params"].as_array().unwrap().len(), 2); let (status, body) = get_json(&app, "/api/metrics").await; assert_eq!(status, StatusCode::OK); let metrics = body.as_array().unwrap(); assert_eq!(metrics.len(), 1); assert_eq!(metrics[0]["label_params"].as_array().unwrap().len(), 2); } #[tokio::test] async fn update_metric_rejects_unknown_metric_name() { let (app, _engine, _settings, _dir) = test_app(); let (status, body) = send_json( &app, "PUT", "/api/metrics/does-not-exist", json!({ "name": "does-not-exist", "message_match_regex": r"request path=(?P\S+)", "label_params": [] }), ) .await; assert_eq!(status, StatusCode::BAD_REQUEST); assert!(body["error"].as_str().unwrap().contains("does-not-exist")); } #[tokio::test] async fn update_metric_reuses_declare_metric_validation() { let (app, _engine, _settings, _dir) = test_app(); send_json( &app, "POST", "/api/metrics", json!({ "name": "requests", "message_match_regex": r"request path=(?P\S+)", "label_params": [] }), ) .await; let (status, body) = send_json( &app, "PUT", "/api/metrics/requests", json!({ "name": "requests", "message_match_regex": r"request path=(?P\S+)", "label_params": ["duration"] }), ) .await; assert_eq!(status, StatusCode::BAD_REQUEST); assert!(body["error"].as_str().unwrap().contains("duration")); } #[tokio::test] async fn deletes_metric_and_it_disappears_from_the_list() { let (app, _engine, _settings, _dir) = test_app(); send_json( &app, "POST", "/api/metrics", json!({ "name": "requests", "message_match_regex": r"request path=(?P\S+)", "label_params": [] }), ) .await; let status = delete_request(&app, "/api/metrics/requests").await; assert_eq!(status, StatusCode::NO_CONTENT); let (status, body) = get_json(&app, "/api/metrics").await; assert_eq!(status, StatusCode::OK); assert!(body.as_array().unwrap().is_empty()); } #[tokio::test] async fn deleting_unknown_metric_name_returns_404_not_500() { let (app, _engine, _settings, _dir) = test_app(); let status = delete_request(&app, "/api/metrics/does-not-exist").await; assert_eq!(status, StatusCode::NOT_FOUND); } #[tokio::test] async fn deleting_metric_persists_removal_and_it_does_not_reappear_after_restart() { let (app, _engine, settings_handle, dir) = test_app(); let settings_path = dir.path().join("settings.json"); { let mut settings = settings_handle.lock().unwrap(); settings.create_profile(&settings_path, "web".to_string()).unwrap(); settings.set_active_profile(&settings_path, Some("web".to_string())).unwrap(); } send_json( &app, "POST", "/api/metrics", json!({ "name": "requests", "message_match_regex": r"request path=(?P\S+)", "label_params": [] }), ) .await; assert_eq!(Settings::load(&settings_path).profile("web").unwrap().metrics.len(), 1); let status = delete_request(&app, "/api/metrics/requests").await; assert_eq!(status, StatusCode::NO_CONTENT); assert!(Settings::load(&settings_path).profile("web").unwrap().metrics.is_empty()); } #[tokio::test] async fn promql_query_against_deleted_metric_fails_with_readable_error() { let (app, engine, _settings, _dir) = test_app(); declare_test_param(&app, "path", r"request path=(?P\S+)", "path", "string").await; send_json( &app, "POST", "/api/metrics", json!({ "name": "requests", "message_match_regex": r"request path=(?P\S+)", "label_params": ["path"] }), ) .await; engine .tx .send(EngineMessage::Ingest(IngestedLine { sequence: 0, timestamp: Timestamp::from_millis(0), text: "request path=/a".to_string(), })) .await .unwrap(); engine.barrier().await; let (status, _body) = get_json(&app, "/api/query?promql=requests").await; assert_eq!(status, StatusCode::OK); let status = delete_request(&app, "/api/metrics/requests").await; assert_eq!(status, StatusCode::NO_CONTENT); let (status, body) = get_json(&app, "/api/query?promql=requests").await; assert_eq!(status, StatusCode::BAD_REQUEST); assert!( body["error"].as_str().unwrap().contains("unknown metric"), "expected a readable 'unknown metric' error, got: {body}" ); } #[tokio::test] async fn queries_bucket_values_with_label_filter_and_range() { let (app, engine, _settings, _dir) = test_app(); declare_test_param(&app, "path", r"request path=(?P\S+)", "path", "string").await; send_json( &app, "POST", "/api/metrics", json!({ "name": "requests", "message_match_regex": r"request path=(?P\S+)", "label_params": ["path"] }), ) .await; for (sequence, text) in [ (0u64, "request path=/a"), (1u64, "request path=/b"), (2u64, "request path=/a"), ] { engine .tx .send(EngineMessage::Ingest(IngestedLine { sequence, timestamp: Timestamp::from_millis(sequence as i64 * 1000), text: text.to_string(), })) .await .unwrap(); } engine.barrier().await; let (status, body) = get_json(&app, "/api/metrics/requests").await; assert_eq!(status, StatusCode::OK); assert_eq!(body.as_array().unwrap().len(), 3); let (status, body) = get_json(&app, "/api/metrics/requests?labels=path=/a").await; assert_eq!(status, StatusCode::OK); let values = body.as_array().unwrap(); assert_eq!(values.len(), 2); for value in values { assert_eq!(value["labels"]["path"]["String"], "/a"); } let (status, body) = get_json(&app, "/api/metrics/requests?since_sequence=0&limit=1").await; assert_eq!(status, StatusCode::OK); let values = body.as_array().unwrap(); assert_eq!(values.len(), 1); assert_eq!(values[0]["sequence"], 2); } #[tokio::test] async fn promql_query_returns_series_for_seeded_fixture() { let (app, engine, _settings, _dir) = test_app(); declare_test_param(&app, "path", r"request path=(?P\S+)", "path", "string").await; send_json( &app, "POST", "/api/metrics", json!({ "name": "requests", "message_match_regex": r"request path=(?P\S+)", "label_params": ["path"] }), ) .await; for (sequence, text) in [ (0u64, "request path=/a"), (1u64, "request path=/b"), (2u64, "request path=/a"), ] { engine .tx .send(EngineMessage::Ingest(IngestedLine { sequence, timestamp: Timestamp::from_millis(sequence as i64 * 1000), text: text.to_string(), })) .await .unwrap(); } engine.barrier().await; let (status, body) = get_json(&app, "/api/query?promql=requests%7Bpath%3D%22%2Fa%22%7D").await; assert_eq!(status, StatusCode::OK); let series = body.as_array().unwrap(); assert_eq!(series.len(), 1); assert_eq!(series[0]["labels"]["path"], "/a"); let points = series[0]["points"].as_array().unwrap(); assert_eq!(points.len(), 2); assert_eq!(points[0], json!([0, 1.0])); assert_eq!(points[1], json!([2000, 1.0])); } #[tokio::test] async fn promql_query_rejects_malformed_query_with_400() { let (app, _engine, _settings, _dir) = test_app(); let (status, body) = get_json(&app, "/api/query?promql=not+valid(").await; assert_eq!(status, StatusCode::BAD_REQUEST); assert!(body["error"].as_str().is_some_and(|s| !s.is_empty())); } #[tokio::test] async fn declaring_param_persists_into_active_profile() { let (app, _engine, settings_handle, dir) = test_app(); let settings_path = dir.path().join("settings.json"); { let mut settings = settings_handle.lock().unwrap(); settings.create_profile(&settings_path, "web".to_string()).unwrap(); settings.set_active_profile(&settings_path, Some("web".to_string())).unwrap(); } let (status, _body) = send_json( &app, "POST", "/api/params", json!({ "name": "level", "message_match_regex": r"level=(?P\w+)", "capture_group": "level", "kind": "string" }), ) .await; assert_eq!(status, StatusCode::CREATED); let reloaded_settings = Settings::load(&settings_path); let profile = reloaded_settings.profile("web").unwrap(); assert_eq!(profile.params.len(), 1); assert_eq!(profile.params[0].name, "level"); } #[tokio::test] async fn declaring_metric_against_auto_derived_param_promotes_it_into_the_profile() { let (app, engine, settings_handle, dir) = test_app(); let settings_path = dir.path().join("settings.json"); { let mut settings = settings_handle.lock().unwrap(); settings.create_profile(&settings_path, "web".to_string()).unwrap(); settings.set_active_profile(&settings_path, Some("web".to_string())).unwrap(); } ingest_lines(&engine, 0, &["dx=1 moved"]).await; let (status, _body) = send_json( &app, "POST", "/api/metrics", json!({ "name": "direction_x", "message_match_regex": r"dx=(?P\S+) moved", "label_params": ["dx"] }), ) .await; assert_eq!(status, StatusCode::CREATED); let reloaded_settings = Settings::load(&settings_path); let profile = reloaded_settings.profile("web").unwrap(); assert_eq!(profile.metrics.len(), 1); assert_eq!(profile.metrics[0].name, "direction_x"); assert_eq!(profile.params.len(), 1); assert_eq!(profile.params[0].name, "dx"); let fresh_settings_handle = std::sync::Arc::new(std::sync::Mutex::new(Settings::default())); let fresh_engine_tx = crate::engine::spawn(fresh_settings_handle, 16); profile.load_into_engine(&fresh_engine_tx).await; let response = fresh_engine_tx.ask(EngineQuery::MetricDeclarations).await.unwrap(); let EngineResponse::MetricDeclarations(metrics) = response else { panic!("unexpected response"); }; assert_eq!(metrics.len(), 1); assert_eq!(metrics[0].name, "direction_x"); } async fn delete_request(app: &Router, uri: &str) -> StatusCode { let request = Request::builder().method("DELETE").uri(uri).body(Body::empty()).unwrap(); app.clone().oneshot(request).await.unwrap().status() } async fn get_status(app: &Router, uri: &str) -> StatusCode { let request = Request::builder().method("GET").uri(uri).body(Body::empty()).unwrap(); app.clone().oneshot(request).await.unwrap().status() } #[tokio::test] async fn dashboard_crud_round_trips_through_http() { let (app, _engine, _settings, _dir) = test_app(); let (status, _body) = send_json(&app, "POST", "/api/settings/profiles", json!({ "name": "web" })).await; assert_eq!(status, StatusCode::CREATED); let (status, body) = send_json( &app, "POST", "/api/settings/profiles/web/dashboards", json!({ "name": "Overview" }), ) .await; assert_eq!(status, StatusCode::CREATED); assert_eq!(body["name"], "Overview"); assert!(body["panels"].as_array().unwrap().is_empty()); let dashboard_id = body["id"].as_str().unwrap().to_string(); let (status, body) = get_json(&app, "/api/settings/profiles/web/dashboards").await; assert_eq!(status, StatusCode::OK); assert_eq!(body.as_array().unwrap().len(), 1); let (status, body) = send_json( &app, "PUT", &format!("/api/settings/profiles/web/dashboards/{dashboard_id}"), json!({ "name": "Overview", "panels": [ { "id": "p1", "title": "Requests", "promql": "requests", "chart_type": "line" } ] }), ) .await; assert_eq!(status, StatusCode::OK); assert_eq!(body["id"], dashboard_id); let panels = body["panels"].as_array().unwrap(); assert_eq!(panels.len(), 1); assert_eq!(panels[0]["chart_type"], "line"); let (status, body) = get_json(&app, &format!("/api/settings/profiles/web/dashboards/{dashboard_id}")).await; assert_eq!(status, StatusCode::OK); assert_eq!(body["panels"].as_array().unwrap().len(), 1); let status = delete_request(&app, &format!("/api/settings/profiles/web/dashboards/{dashboard_id}")).await; assert_eq!(status, StatusCode::NO_CONTENT); let (status, body) = get_json(&app, "/api/settings/profiles/web/dashboards").await; assert_eq!(status, StatusCode::OK); assert!(body.as_array().unwrap().is_empty()); } #[tokio::test] async fn dashboard_routes_404_for_unknown_profile_or_dashboard() { let (app, _engine, _settings, _dir) = test_app(); let status = get_status(&app, "/api/settings/profiles/ghost/dashboards").await; assert_eq!(status, StatusCode::NOT_FOUND); let (status, _body) = send_json(&app, "POST", "/api/settings/profiles", json!({ "name": "web" })).await; assert_eq!(status, StatusCode::CREATED); let status = get_status(&app, "/api/settings/profiles/web/dashboards/does-not-exist").await; assert_eq!(status, StatusCode::NOT_FOUND); } #[tokio::test] async fn saved_dashboard_persists_across_simulated_restart() { let (app, _engine, _settings_handle, dir) = test_app(); let (status, _body) = send_json(&app, "POST", "/api/settings/profiles", json!({ "name": "web" })).await; assert_eq!(status, StatusCode::CREATED); let (status, body) = send_json( &app, "POST", "/api/settings/profiles/web/dashboards", json!({ "name": "Overview" }), ) .await; assert_eq!(status, StatusCode::CREATED); let dashboard_id = body["id"].as_str().unwrap().to_string(); send_json( &app, "PUT", &format!("/api/settings/profiles/web/dashboards/{dashboard_id}"), json!({ "name": "Overview", "panels": [ { "id": "p1", "title": "Requests", "promql": "requests", "chart_type": "bar" } ] }), ) .await; let settings_path = dir.path().join("settings.json"); let reloaded_settings = Settings::load(&settings_path); let dashboards = reloaded_settings.list_dashboards("web").unwrap(); assert_eq!(dashboards.len(), 1); assert_eq!(dashboards[0].name, "Overview"); assert_eq!(dashboards[0].panels.len(), 1); assert_eq!(dashboards[0].panels[0].promql, "requests"); } async fn ingest_lines(engine: &EngineHandle, start_sequence: u64, texts: &[&str]) { for (offset, text) in texts.iter().enumerate() { let sequence = start_sequence + offset as u64; engine .tx .send(EngineMessage::Ingest(IngestedLine { sequence, timestamp: Timestamp::from_millis(sequence as i64 * 1000), text: text.to_string(), })) .await .unwrap(); } engine.barrier().await; } #[tokio::test] async fn setting_max_logs_evicts_oldest_on_next_ingest() { let (app, engine, _settings, _dir) = test_app(); ingest_lines(&engine, 0, &["a", "b", "c"]).await; let (status, body) = get_json(&app, "/api/logs").await; assert_eq!(status, StatusCode::OK); assert_eq!(body.as_array().unwrap().len(), 3); let (status, body) = send_json(&app, "PATCH", "/api/settings", json!({ "max_logs": 2 })).await; assert_eq!(status, StatusCode::OK); assert_eq!(body["max_logs"], 2); let (_, body) = get_json(&app, "/api/logs").await; assert_eq!(body.as_array().unwrap().len(), 2); ingest_lines(&engine, 3, &["d"]).await; let (_, body) = get_json(&app, "/api/logs").await; let texts: Vec<_> = body .as_array() .unwrap() .iter() .map(|entry| entry["text"].as_str().unwrap()) .collect(); assert_eq!(texts, vec!["c", "d"]); } #[tokio::test] async fn max_logs_defaults_to_unlimited() { let (app, engine, _settings, _dir) = test_app(); let (_, body) = get_json(&app, "/api/settings").await; assert_eq!(body["max_logs"], Value::Null); ingest_lines(&engine, 0, &["a", "b", "c", "d", "e"]).await; let (_, body) = get_json(&app, "/api/logs").await; assert_eq!(body.as_array().unwrap().len(), 5); } #[tokio::test] async fn max_logs_clears_back_to_unlimited() { let (app, engine, _settings, _dir) = test_app(); send_json(&app, "PATCH", "/api/settings", json!({ "max_logs": 1 })).await; ingest_lines(&engine, 0, &["a", "b"]).await; let (_, body) = get_json(&app, "/api/logs").await; assert_eq!(body.as_array().unwrap().len(), 1); let (status, body) = send_json(&app, "PATCH", "/api/settings", json!({ "max_logs": null })).await; assert_eq!(status, StatusCode::OK); assert_eq!(body["max_logs"], Value::Null); ingest_lines(&engine, 2, &["c", "d"]).await; let (_, body) = get_json(&app, "/api/logs").await; assert_eq!(body.as_array().unwrap().len(), 3); } #[tokio::test] async fn get_logs_sort_by_orders_by_param_and_toggles_missing_param_handling() { let (app, engine, _settings, _dir) = test_app(); declare_test_param(&app, "priority", r"priority=(?P\d+)", "priority", "number").await; ingest_lines( &engine, 0, &[ "priority=3 boom", "no param here", "priority=1 boom", "also missing", ], ) .await; let (status, body) = get_json(&app, "/api/logs").await; assert_eq!(status, StatusCode::OK); let sequences: Vec = body.as_array().unwrap().iter().map(|e| e["sequence"].as_u64().unwrap()).collect(); assert_eq!(sequences, vec![0, 1, 2, 3]); let (status, body) = get_json(&app, "/api/logs?sort_by=priority").await; assert_eq!(status, StatusCode::OK); let sequences: Vec = body.as_array().unwrap().iter().map(|e| e["sequence"].as_u64().unwrap()).collect(); assert_eq!(sequences, vec![2, 0, 1, 3]); let (status, body) = get_json(&app, "/api/logs?sort_by=priority&sort_missing_fallback=false").await; assert_eq!(status, StatusCode::OK); let sequences: Vec = body.as_array().unwrap().iter().map(|e| e["sequence"].as_u64().unwrap()).collect(); assert_eq!(sequences, vec![2, 0]); } #[tokio::test] async fn get_logs_sort_by_unknown_param_returns_400() { let (app, _engine, _settings, _dir) = test_app(); let (status, body) = get_json(&app, "/api/logs?sort_by=does_not_exist").await; assert_eq!(status, StatusCode::BAD_REQUEST); assert!(body["error"].as_str().unwrap().contains("does_not_exist")); } }