diff --git a/src/main.rs b/src/main.rs index 63b3199..0aab491 100644 --- a/src/main.rs +++ b/src/main.rs @@ -7,6 +7,7 @@ mod filter_parser; mod index_key; mod ingest; mod metric_primitive; +mod profile; mod settings; mod timestamp; @@ -17,6 +18,7 @@ use clap::Parser; use tokio::net::TcpListener; use tokio::sync::mpsc; +use profile::ProfileStore; use settings::Settings; /// Observability dashboard from terminal output. @@ -48,6 +50,38 @@ async fn main() { let (engine_tx, engine_rx) = mpsc::channel(1024); tokio::spawn(engine::run(engine_rx)); + + // Ticket 009: reload the active profile's param/metric declarations into the + // engine BEFORE starting ingest below. Ordering matters: declarations must be + // in place before any line is ingested so that backfill/live-derivation + // semantics (ticket 003) apply uniformly to every line, including ones that + // arrive immediately once ingest starts (there's no "declare after the fact + // and go back to reprocess early lines" step, so doing it after ingest starts + // could silently miss lines that raced ahead of the declarations). + { + let (active_profile_name, profiles_dir) = { + let settings_guard = settings.lock().unwrap(); + ( + settings_guard.profiles.active_profile.clone(), + profile::resolve_profiles_dir(&settings_guard, &settings_path), + ) + }; + if let Some(profile_name) = active_profile_name { + let store = ProfileStore::load(&profile::profiles_store_path(&profiles_dir)); + match store.get(&profile_name) { + Some(active_profile) => { + profile::load_into_engine(&engine_tx, active_profile).await; + } + None => { + eprintln!( + "warning: active profile '{profile_name}' not found under {}; skipping", + profiles_dir.display() + ); + } + } + } + } + tokio::spawn(ingest::run(engine_tx.clone())); let app = api::router(engine_tx, &args.static_dir, settings, settings_path); diff --git a/src/profile.rs b/src/profile.rs new file mode 100644 index 0000000..4ed28a5 --- /dev/null +++ b/src/profile.rs @@ -0,0 +1,761 @@ +//! Named, persistable bundles of parsing/indexing rules ("profiles" / "presets"), +//! ticket 009. +//! +//! A `Profile` bundles the param declarations (ticket 001) and metric +//! declarations (ticket 002) needed to derive a given CLI app's structure from +//! its raw log output, plus any saved dashboards for that app. Profiles are +//! collected in a `ProfileStore`, persisted as a single `profiles.json` file +//! under `Settings.profiles.profiles_dir` (resolved relative to the settings +//! file's directory — the same convention ticket 010 uses for CSS storage). A +//! single file (rather than one file per profile) is simplest and avoids +//! filename-sanitization concerns for profile names. +//! +//! `Settings.profiles` (see `settings.rs`) only tracks lightweight pointers — +//! known profile names and which one is active — while this module owns the +//! actual declaration content. That mirrors how `Settings.css_entries` holds +//! lightweight `CssEntry` records while the CSS content itself lives in +//! separate files (ticket 010). +//! +//! On startup, `main.rs` loads the active profile's declarations into a fresh +//! engine by replaying `EngineMessage::DeclareParam`/`DeclareMetric` for each +//! stored declaration (see `load_into_engine` below) — this reproduces the +//! same parsing/indexing rules the profile had when last saved, without +//! requiring the user to re-declare anything by hand after a restart. +//! +//! Several public methods here (e.g. `ProfileStore::delete`, +//! `record_param_declaration`) have no caller yet within this worktree — CRUD +//! over profiles isn't exposed over HTTP until ticket 004 lands the API layer +//! and calls into this module. `#![allow(dead_code)]` mirrors how +//! `engine.rs`/`data_value.rs` handle the same not-yet-wired situation for +//! tickets 001/002. +#![allow(dead_code)] + +use std::fmt; +use std::fs; +use std::io; +use std::path::{Path, PathBuf}; + +use serde::{Deserialize, Serialize}; + +use crate::data_value::{MetricPrimitiveKind, RegexToMetricPrimitiveMatcher}; +use crate::engine::{DeclareMetricRequest, DeclareParamRequest, EngineHandle, EngineMessage}; +use crate::settings::Settings; +use tokio::sync::oneshot; + +/// Mirrors `data_value::MetricPrimitiveKind` with `Serialize`/`Deserialize` so +/// profiles can round-trip through JSON. `MetricPrimitiveKind` itself only +/// derives `Serialize` (it's an engine-internal type owned by tickets +/// 001/002/003, out of scope for this ticket to modify), so this module keeps +/// its own mirror and converts at the boundary. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum ProfileMetricPrimitiveKind { + String, + Number, + Timestamp, +} + +impl From for ProfileMetricPrimitiveKind { + fn from(kind: MetricPrimitiveKind) -> Self { + match kind { + MetricPrimitiveKind::String => ProfileMetricPrimitiveKind::String, + MetricPrimitiveKind::Number => ProfileMetricPrimitiveKind::Number, + MetricPrimitiveKind::Timestamp => ProfileMetricPrimitiveKind::Timestamp, + } + } +} + +impl From for MetricPrimitiveKind { + fn from(kind: ProfileMetricPrimitiveKind) -> Self { + match kind { + ProfileMetricPrimitiveKind::String => MetricPrimitiveKind::String, + ProfileMetricPrimitiveKind::Number => MetricPrimitiveKind::Number, + ProfileMetricPrimitiveKind::Timestamp => MetricPrimitiveKind::Timestamp, + } + } +} + +/// A stored param declaration, matching `DeclareParamRequest`'s shape (ticket 001). +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct ParamDeclarationRecord { + pub name: String, + pub message_match_regex: String, + pub capture_group: String, + pub kind: ProfileMetricPrimitiveKind, +} + +impl From<&DeclareParamRequest> for ParamDeclarationRecord { + fn from(request: &DeclareParamRequest) -> Self { + ParamDeclarationRecord { + name: request.name.clone(), + message_match_regex: request.message_match_regex.clone(), + capture_group: request.capture_group.clone(), + kind: request.kind.into(), + } + } +} + +impl From<&ParamDeclarationRecord> for DeclareParamRequest { + fn from(record: &ParamDeclarationRecord) -> Self { + DeclareParamRequest { + name: record.name.clone(), + message_match_regex: record.message_match_regex.clone(), + capture_group: record.capture_group.clone(), + kind: record.kind.into(), + } + } +} + +/// A stored metric param matcher, matching `RegexToMetricPrimitiveMatcher`'s shape. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct MetricParamRecord { + pub param_name: String, + pub capture_group: String, + pub kind: ProfileMetricPrimitiveKind, +} + +/// A stored metric declaration, matching `DeclareMetricRequest`'s shape (ticket 002). +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct MetricDeclarationRecord { + pub name: String, + pub message_match_regex: String, + pub params: Vec, +} + +impl From<&DeclareMetricRequest> for MetricDeclarationRecord { + fn from(request: &DeclareMetricRequest) -> Self { + MetricDeclarationRecord { + name: request.name.clone(), + message_match_regex: request.message_match_regex.clone(), + params: request + .params + .iter() + .map(|matcher| MetricParamRecord { + param_name: matcher.param_name.clone(), + capture_group: matcher.capture_group.clone(), + kind: matcher.kind.into(), + }) + .collect(), + } + } +} + +impl From<&MetricDeclarationRecord> for DeclareMetricRequest { + fn from(record: &MetricDeclarationRecord) -> Self { + DeclareMetricRequest { + name: record.name.clone(), + message_match_regex: record.message_match_regex.clone(), + params: record + .params + .iter() + .map(|param| RegexToMetricPrimitiveMatcher { + param_name: param.param_name.clone(), + capture_group: param.capture_group.clone(), + kind: param.kind.into(), + }) + .collect(), + } + } +} + +/// A named bundle of parsing/indexing rules for a particular CLI app: which +/// params and metrics to derive from its log output, plus any saved +/// dashboards. +/// +/// `dashboards` is stubbed as opaque JSON: dashboards (ticket 014) don't exist +/// yet, so there's no concrete schema to store. Ticket 014 should replace this +/// with whatever concrete `Dashboard` type it introduces (or keep it loosely +/// typed, if that turns out to be the right call for that ticket). +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct Profile { + pub name: String, + #[serde(default)] + pub params: Vec, + #[serde(default)] + pub metrics: Vec, + /// Placeholder for ticket 014's dashboard storage. See the struct doc above. + #[serde(default)] + pub dashboards: Vec, +} + +impl Profile { + /// A new, empty profile with the given name. + pub fn new(name: String) -> Profile { + Profile { + name, + params: Vec::new(), + metrics: Vec::new(), + dashboards: Vec::new(), + } + } +} + +/// Sends every param/metric declaration stored in `profile` to the engine at +/// `engine`, as `EngineMessage::DeclareParam`/`DeclareMetric`. Used on startup +/// to reload the active profile's rules into a fresh engine (see `main.rs`), +/// and in tests to simulate that reload. +/// +/// A stored declaration should generally already be valid — it round-tripped +/// through the engine's own validation when originally declared (see +/// `record_param_declaration`/`record_metric_declaration` below) — but this +/// doesn't assume that: any declaration the engine rejects (or any send/recv +/// failure, e.g. the engine channel closing) is logged to stderr and skipped +/// rather than panicking, so one bad stored declaration can't take the whole +/// process down. +pub async fn load_into_engine(engine: &EngineHandle, profile: &Profile) { + for record in &profile.params { + let (respond_to, response) = oneshot::channel(); + let request: DeclareParamRequest = record.into(); + let name = request.name.clone(); + if engine + .send(EngineMessage::DeclareParam { request, respond_to }) + .await + .is_err() + { + eprintln!("warning: engine unavailable while loading profile '{}'", profile.name); + return; + } + match response.await { + Ok(Ok(())) => {} + Ok(Err(err)) => eprintln!( + "warning: profile '{}' param declaration '{name}' rejected by engine: {err}", + profile.name + ), + Err(_) => eprintln!( + "warning: engine dropped response while declaring param '{name}' from profile '{}'", + profile.name + ), + } + } + + for record in &profile.metrics { + let (respond_to, response) = oneshot::channel(); + let request: DeclareMetricRequest = record.into(); + let name = request.name.clone(); + if engine + .send(EngineMessage::DeclareMetric { request, respond_to }) + .await + .is_err() + { + eprintln!("warning: engine unavailable while loading profile '{}'", profile.name); + return; + } + match response.await { + Ok(Ok(())) => {} + Ok(Err(err)) => eprintln!( + "warning: profile '{}' metric declaration '{name}' rejected by engine: {err}", + profile.name + ), + Err(_) => eprintln!( + "warning: engine dropped response while declaring metric '{name}' from profile '{}'", + profile.name + ), + } + } +} + +/// Error returned by `ProfileStore` mutations and lookups. +#[derive(Debug)] +pub enum ProfileError { + /// No profile with the given name exists. + NotFound, + /// A profile with the given name already exists. + DuplicateName, + /// Underlying I/O failure reading/writing the profile store or settings file. + Io(io::Error), +} + +impl fmt::Display for ProfileError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + ProfileError::NotFound => write!(f, "no profile with that name"), + ProfileError::DuplicateName => write!(f, "a profile with that name already exists"), + ProfileError::Io(err) => write!(f, "I/O error: {err}"), + } + } +} + +impl std::error::Error for ProfileError {} + +impl From for ProfileError { + fn from(err: io::Error) -> Self { + ProfileError::Io(err) + } +} + +/// Resolves `settings.profiles.profiles_dir` relative to `settings_path`'s +/// parent directory (or treats it as absolute if it already is one) — the +/// same directory convention ticket 010 uses for CSS storage. +pub fn resolve_profiles_dir(settings: &Settings, settings_path: &Path) -> PathBuf { + let profiles_dir = Path::new(&settings.profiles.profiles_dir); + if profiles_dir.is_absolute() { + profiles_dir.to_path_buf() + } else { + settings_path + .parent() + .map(|parent| parent.join(profiles_dir)) + .unwrap_or_else(|| profiles_dir.to_path_buf()) + } +} + +/// Path to the single file holding all profiles, within `profiles_dir` (see +/// `resolve_profiles_dir`). +pub fn profiles_store_path(profiles_dir: &Path) -> PathBuf { + profiles_dir.join("profiles.json") +} + +/// On-disk collection of all known profiles, persisted as a single JSON file +/// (see module docs for why one file rather than one-per-profile). +#[derive(Debug, Clone, Default)] +pub struct ProfileStore { + profiles: Vec, +} + +impl ProfileStore { + /// Loads the store from `store_path`. A missing file is treated as an + /// empty store (not an error) — there's no reason to fail startup just + /// because no profiles have been created yet. Any I/O or parse error + /// while reading an existing file is logged to stderr and treated as an + /// empty store, mirroring `Settings::load`'s non-fatal-error handling. + pub fn load(store_path: &Path) -> ProfileStore { + match fs::read_to_string(store_path) { + Ok(contents) => match serde_json::from_str::>(&contents) { + Ok(profiles) => ProfileStore { profiles }, + Err(err) => { + eprintln!( + "warning: failed to parse profile store at {}: {err}; using empty store", + store_path.display() + ); + ProfileStore::default() + } + }, + Err(err) if err.kind() == io::ErrorKind::NotFound => ProfileStore::default(), + Err(err) => { + eprintln!( + "warning: failed to read profile store at {}: {err}; using empty store", + store_path.display() + ); + ProfileStore::default() + } + } + } + + /// Persists the store to `store_path` as pretty-printed JSON, creating + /// any missing parent directories first. + pub fn save(&self, store_path: &Path) -> io::Result<()> { + if let Some(parent) = store_path.parent() { + fs::create_dir_all(parent)?; + } + let contents = serde_json::to_string_pretty(&self.profiles) + .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err))?; + fs::write(store_path, contents) + } + + /// All known profiles. + pub fn list(&self) -> &[Profile] { + &self.profiles + } + + /// Looks up a profile by name. + pub fn get(&self, name: &str) -> Option<&Profile> { + self.profiles.iter().find(|p| p.name == name) + } + + /// Creates a new, empty profile named `name`, persists the store to + /// `store_path`, and records the name in `settings.profiles.names` + /// (persisting `settings` to `settings_path` too, keeping the two files in + /// sync). Fails with `ProfileError::DuplicateName` if a profile with that + /// name already exists. + pub fn create( + &mut self, + store_path: &Path, + settings: &mut Settings, + settings_path: &Path, + name: String, + ) -> Result<(), ProfileError> { + if self.get(&name).is_some() { + return Err(ProfileError::DuplicateName); + } + self.profiles.push(Profile::new(name.clone())); + self.save(store_path)?; + + if !settings.profiles.names.contains(&name) { + settings.profiles.names.push(name); + } + settings.save(settings_path)?; + Ok(()) + } + + /// Deletes the profile named `name`, persists the store to `store_path`, + /// and drops it from `settings.profiles.names` (persisting `settings` to + /// `settings_path` too). If the deleted profile was active, clears + /// `settings.profiles.active_profile`. Fails with `ProfileError::NotFound` + /// if no such profile exists. + pub fn delete( + &mut self, + store_path: &Path, + settings: &mut Settings, + settings_path: &Path, + name: &str, + ) -> Result<(), ProfileError> { + let index = self.profiles.iter().position(|p| p.name == name).ok_or(ProfileError::NotFound)?; + self.profiles.remove(index); + self.save(store_path)?; + + settings.profiles.names.retain(|n| n != name); + if settings.profiles.active_profile.as_deref() == Some(name) { + settings.profiles.active_profile = None; + } + settings.save(settings_path)?; + Ok(()) + } + + /// Sets (or clears, with `name: None`) the active profile: updates + /// `settings.profiles.active_profile` and persists `settings` to + /// `settings_path`. Fails with `ProfileError::NotFound` if `name` is + /// `Some` and no such profile exists in this store. + /// + /// This only updates `Settings`; it does not itself reload the newly + /// active profile's declarations into a running engine (that happens on + /// the next startup — see `load_into_engine` and `main.rs`). + pub fn set_active( + &self, + settings: &mut Settings, + settings_path: &Path, + name: Option, + ) -> Result<(), ProfileError> { + if let Some(name) = &name { + if self.get(name).is_none() { + return Err(ProfileError::NotFound); + } + } + settings.profiles.active_profile = name; + settings.save(settings_path)?; + Ok(()) + } + + /// Records a newly-declared param into `profile_name`'s stored + /// declarations (replacing any existing record with the same param name, + /// so re-declaring updates it in place), persisting the store to + /// `store_path`. + /// + /// **Integration hook for ticket 004 (HTTP API)**: when the HTTP API + /// declares a param against the engine (sending + /// `EngineMessage::DeclareParam`), it should also call this — with the + /// same `DeclareParamRequest` — against the currently-active profile, so + /// the declaration survives a restart (see `load_into_engine`). This + /// ticket (009) cannot wire that call itself: there's no declare-param + /// HTTP route yet for it to hook into. Call shape: + /// + /// ```ignore + /// profile_store.record_param_declaration(&store_path, &active_profile_name, &request)?; + /// ``` + pub fn record_param_declaration( + &mut self, + store_path: &Path, + profile_name: &str, + request: &DeclareParamRequest, + ) -> Result<(), ProfileError> { + let profile = self + .profiles + .iter_mut() + .find(|p| p.name == profile_name) + .ok_or(ProfileError::NotFound)?; + let record: ParamDeclarationRecord = request.into(); + match profile.params.iter_mut().find(|r| r.name == record.name) { + Some(existing) => *existing = record, + None => profile.params.push(record), + } + self.save(store_path)?; + Ok(()) + } + + /// Records a newly-declared metric into `profile_name`'s stored + /// declarations (replacing any existing record with the same metric name), + /// persisting the store to `store_path`. + /// + /// **Integration hook for ticket 004 (HTTP API)**: same as + /// `record_param_declaration` above, but for + /// `EngineMessage::DeclareMetric`/`DeclareMetricRequest`. + pub fn record_metric_declaration( + &mut self, + store_path: &Path, + profile_name: &str, + request: &DeclareMetricRequest, + ) -> Result<(), ProfileError> { + let profile = self + .profiles + .iter_mut() + .find(|p| p.name == profile_name) + .ok_or(ProfileError::NotFound)?; + let record: MetricDeclarationRecord = request.into(); + match profile.metrics.iter_mut().find(|r| r.name == record.name) { + Some(existing) => *existing = record, + None => profile.metrics.push(record), + } + self.save(store_path)?; + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::data_value::MetricPrimitiveKind; + use crate::dto::LogsParams; + use crate::engine::{EngineMessage, EngineQuery, EngineResponse, IngestedLine}; + use crate::timestamp::Timestamp; + use tokio::sync::mpsc; + + fn param_request(name: &str, pattern: &str, capture_group: &str) -> DeclareParamRequest { + DeclareParamRequest { + name: name.to_string(), + message_match_regex: pattern.to_string(), + capture_group: capture_group.to_string(), + kind: MetricPrimitiveKind::String, + } + } + + fn metric_request(name: &str, pattern: &str, param_name: &str, capture_group: &str) -> DeclareMetricRequest { + DeclareMetricRequest { + name: name.to_string(), + message_match_regex: pattern.to_string(), + params: vec![RegexToMetricPrimitiveMatcher { + param_name: param_name.to_string(), + capture_group: capture_group.to_string(), + kind: MetricPrimitiveKind::Number, + }], + } + } + + async fn ask(engine: &EngineHandle, query: EngineQuery) -> EngineResponse { + let (respond_to, response) = oneshot::channel(); + engine.send(EngineMessage::Query { query, respond_to }).await.unwrap(); + response.await.unwrap() + } + + #[test] + fn resolve_profiles_dir_is_relative_to_settings_parent() { + let settings_path = Path::new("/config/ottyl/settings.json"); + let mut settings = Settings::default(); + settings.profiles.profiles_dir = "profiles".to_string(); + + let dir = resolve_profiles_dir(&settings, settings_path); + assert_eq!(dir, Path::new("/config/ottyl/profiles")); + } + + #[test] + fn resolve_profiles_dir_respects_absolute_path() { + let settings_path = Path::new("/config/ottyl/settings.json"); + let mut settings = Settings::default(); + settings.profiles.profiles_dir = "/elsewhere/profiles".to_string(); + + let dir = resolve_profiles_dir(&settings, settings_path); + assert_eq!(dir, Path::new("/elsewhere/profiles")); + } + + #[test] + fn create_list_get_delete_round_trip() { + let dir = tempfile::tempdir().unwrap(); + let store_path = dir.path().join("profiles.json"); + let settings_path = dir.path().join("settings.json"); + let mut settings = Settings::default(); + + let mut store = ProfileStore::load(&store_path); + store.create(&store_path, &mut settings, &settings_path, "web".to_string()).unwrap(); + store.create(&store_path, &mut settings, &settings_path, "worker".to_string()).unwrap(); + + assert_eq!(store.list().len(), 2); + assert!(store.get("web").is_some()); + assert_eq!(settings.profiles.names, vec!["web".to_string(), "worker".to_string()]); + + // Duplicate creation is rejected. + let dup = store.create(&store_path, &mut settings, &settings_path, "web".to_string()); + assert!(matches!(dup, Err(ProfileError::DuplicateName))); + + // Reload from disk: profiles and settings both persisted. + let reloaded = ProfileStore::load(&store_path); + assert_eq!(reloaded.list().len(), 2); + let reloaded_settings = Settings::load(&settings_path); + assert_eq!(reloaded_settings.profiles.names, settings.profiles.names); + + store.delete(&store_path, &mut settings, &settings_path, "worker").unwrap(); + assert!(store.get("worker").is_none()); + assert_eq!(settings.profiles.names, vec!["web".to_string()]); + + let missing = store.delete(&store_path, &mut settings, &settings_path, "worker"); + assert!(matches!(missing, Err(ProfileError::NotFound))); + } + + #[test] + fn set_active_updates_and_persists_settings() { + let dir = tempfile::tempdir().unwrap(); + let store_path = dir.path().join("profiles.json"); + let settings_path = dir.path().join("settings.json"); + let mut settings = Settings::default(); + + let mut store = ProfileStore::load(&store_path); + store.create(&store_path, &mut settings, &settings_path, "web".to_string()).unwrap(); + store.create(&store_path, &mut settings, &settings_path, "worker".to_string()).unwrap(); + + store.set_active(&mut settings, &settings_path, Some("web".to_string())).unwrap(); + assert_eq!(settings.profiles.active_profile, Some("web".to_string())); + + // Switching to another known profile works and persists. + store.set_active(&mut settings, &settings_path, Some("worker".to_string())).unwrap(); + let reloaded = Settings::load(&settings_path); + assert_eq!(reloaded.profiles.active_profile, Some("worker".to_string())); + + // Setting an unknown profile active is rejected, leaving the previous value. + let err = store.set_active(&mut settings, &settings_path, Some("ghost".to_string())); + assert!(matches!(err, Err(ProfileError::NotFound))); + assert_eq!(settings.profiles.active_profile, Some("worker".to_string())); + + // Clearing the active profile works. + store.set_active(&mut settings, &settings_path, None).unwrap(); + assert_eq!(settings.profiles.active_profile, None); + } + + #[test] + fn record_param_and_metric_declarations_persist_into_profile() { + let dir = tempfile::tempdir().unwrap(); + let store_path = dir.path().join("profiles.json"); + let settings_path = dir.path().join("settings.json"); + let mut settings = Settings::default(); + + let mut store = ProfileStore::load(&store_path); + store.create(&store_path, &mut settings, &settings_path, "web".to_string()).unwrap(); + + let param_req = param_request("level", r"level=(?P\w+)", "level"); + store.record_param_declaration(&store_path, "web", ¶m_req).unwrap(); + + let metric_req = metric_request( + "requests", + r"duration=(?P\d+)", + "duration", + "duration", + ); + store.record_metric_declaration(&store_path, "web", &metric_req).unwrap(); + + let profile = store.get("web").unwrap(); + assert_eq!(profile.params.len(), 1); + assert_eq!(profile.params[0].name, "level"); + assert_eq!(profile.metrics.len(), 1); + assert_eq!(profile.metrics[0].name, "requests"); + + // Re-declaring the same name updates in place rather than duplicating. + let param_req_v2 = param_request("level", r"lvl=(?P\w+)", "level"); + store.record_param_declaration(&store_path, "web", ¶m_req_v2).unwrap(); + let profile = store.get("web").unwrap(); + assert_eq!(profile.params.len(), 1); + assert_eq!(profile.params[0].message_match_regex, r"lvl=(?P\w+)"); + + // Reload from disk: recorded declarations survive a restart. + let reloaded = ProfileStore::load(&store_path); + let profile = reloaded.get("web").unwrap(); + assert_eq!(profile.params.len(), 1); + assert_eq!(profile.metrics.len(), 1); + } + + /// Acceptance criterion: declaring params/metrics into a profile, "restarting" + /// (dropping and reloading the store from disk, rather than a literal process + /// restart), and replaying the saved profile against a fresh engine reproduces + /// the same declarations and still ingests lines correctly — without + /// re-declaring anything by hand. + /// + /// Note: ticket 003 (metric bucket derivation) isn't implemented yet in this + /// worktree, so there's no bucket index to compare directly. What this test + /// verifies instead is the part ticket 009 owns: that a profile's declarations + /// round-trip through disk byte-for-byte and, once replayed via + /// `load_into_engine`, produce identical engine-visible declarations (which is + /// exactly the state ticket 003's derivation logic will read from) — and that + /// the engine keeps ingesting lines normally afterwards. + #[tokio::test] + async fn profile_round_trip_reproduces_declarations_in_fresh_engine() { + let dir = tempfile::tempdir().unwrap(); + let store_path = dir.path().join("profiles.json"); + let settings_path = dir.path().join("settings.json"); + let mut settings = Settings::default(); + + // Declare directly into a profile (as ticket 004's HTTP API will do via the + // `record_*_declaration` hooks once it exists). + let mut store = ProfileStore::load(&store_path); + store.create(&store_path, &mut settings, &settings_path, "web".to_string()).unwrap(); + store + .record_param_declaration( + &store_path, + "web", + ¶m_request("level", r"level=(?P\w+)", "level"), + ) + .unwrap(); + store + .record_metric_declaration( + &store_path, + "web", + &metric_request( + "requests", + r"request duration=(?P\d+)", + "duration", + "duration", + ), + ) + .unwrap(); + store.set_active(&mut settings, &settings_path, Some("web".to_string())).unwrap(); + + // Simulate a restart: drop everything in memory, reload from disk. + drop(store); + drop(settings); + let settings = Settings::load(&settings_path); + let active_name = settings.profiles.active_profile.clone().unwrap(); + let reloaded_store = ProfileStore::load(&store_path); + let profile = reloaded_store.get(&active_name).unwrap().clone(); + + // Spawn a fresh engine and replay the profile's declarations into it. + let (engine_tx, engine_rx) = mpsc::channel(16); + tokio::spawn(crate::engine::run(engine_rx)); + load_into_engine(&engine_tx, &profile).await; + + // Ingest a line and confirm the engine is functioning with the replayed + // declarations in place (no re-declaration needed). + engine_tx + .send(EngineMessage::Ingest(IngestedLine { + sequence: 0, + timestamp: Timestamp::from_millis(0), + text: "level=info request duration=42".to_string(), + })) + .await + .unwrap(); + + let params = match ask(&engine_tx, EngineQuery::ParamDeclarations).await { + EngineResponse::ParamDeclarations(params) => params, + _ => panic!("unexpected response"), + }; + assert_eq!(params.len(), 1); + assert_eq!(params[0].name, "level"); + assert_eq!(params[0].message_match_regex, r"level=(?P\w+)"); + assert_eq!(params[0].capture_group, "level"); + assert!(matches!(params[0].kind, MetricPrimitiveKind::String)); + + let metrics = match ask(&engine_tx, EngineQuery::MetricDeclarations).await { + EngineResponse::MetricDeclarations(metrics) => metrics, + _ => panic!("unexpected response"), + }; + assert_eq!(metrics.len(), 1); + assert_eq!(metrics[0].name, "requests"); + assert_eq!(metrics[0].message_match_regex, r"request duration=(?P\d+)"); + assert_eq!(metrics[0].params.len(), 1); + assert_eq!(metrics[0].params[0].param_name, "duration"); + + let logs = match ask( + &engine_tx, + EngineQuery::Logs(LogsParams { since_sequence: None, start_ms: None, end_ms: None, limit: None }), + ) + .await + { + EngineResponse::Logs(logs) => logs, + _ => panic!("unexpected response"), + }; + assert_eq!(logs.len(), 1); + assert_eq!(logs[0].text, "level=info request duration=42"); + } +}