From 823bc81d07d7aa2eefb9afbeaa6e3379bc3124e2 Mon Sep 17 00:00:00 2001 From: lda Date: Fri, 11 Sep 2026 04:23:39 +0700 Subject: [PATCH] alert history work - dedup error handling - alert now clears with observation_retenstion_seconds - evaluate recent alerts once per server per 5s - some iter stuff, some type stuff, and some store api stuff. --- ...9b5976c79bb63759bc9ecff1a7873e2d70137.json | 12 + ...5303a2784b74f49068ce8e946cedcd8d12c4e.json | 20 ++ ...2b9b7771f9bf63c45613c4b6f8b86be479d70.json | 20 ++ ...7e0a7552f1f4e38390959e14ea5cc765bf2c7.json | 12 + ...d53209c48be145562192b3a8f84903d39eac1.json | 12 + ...3ebcce1156ce0f7da2d39f054025ab9b66b6b.json | 20 ++ ...d05447558646fb2b47963751a7ab53b3a99b6.json | 12 + ...9b5976c79bb63759bc9ecff1a7873e2d70137.json | 12 + ...5303a2784b74f49068ce8e946cedcd8d12c4e.json | 20 ++ ...2b9b7771f9bf63c45613c4b6f8b86be479d70.json | 20 ++ ...7e0a7552f1f4e38390959e14ea5cc765bf2c7.json | 12 + ...d53209c48be145562192b3a8f84903d39eac1.json | 12 + ...3ebcce1156ce0f7da2d39f054025ab9b66b6b.json | 20 ++ ...d05447558646fb2b47963751a7ab53b3a99b6.json | 12 + wakey-control-plane/src/api/alerts.rs | 210 ++++++++++++------ wakey-control-plane/src/api/commands.rs | 67 +++--- wakey-control-plane/src/api/control/enroll.rs | 63 ++---- .../src/api/control/fleet/mod.rs | 187 +++++++++------- wakey-control-plane/src/api/mod.rs | 2 +- wakey-control-plane/src/api/terminals.rs | 9 +- wakey-control-plane/src/config/types.rs | 1 - wakey-control-plane/src/runtime/mod.rs | 16 ++ wakey-control-plane/src/state/mod.rs | 5 +- wakey-control-plane/src/state/store.rs | 132 ++++++++++- wakey-control-plane/src/state/store/audit.rs | 19 ++ wakey-control-plane/src/state/store/db.rs | 31 +++ .../src/state/store/enrollment.rs | 53 ++--- wakey-control-plane/src/state/types.rs | 6 + wakey-control-plane/src/ws.rs | 27 +-- 29 files changed, 750 insertions(+), 294 deletions(-) create mode 100644 .sqlx/query-6ce45cc94cbfdd948aad5fb986f9b5976c79bb63759bc9ecff1a7873e2d70137.json create mode 100644 .sqlx/query-b324587b94ee2e98dc8eff5f6735303a2784b74f49068ce8e946cedcd8d12c4e.json create mode 100644 .sqlx/query-bd1666cea8840df04ddcdc8a0c42b9b7771f9bf63c45613c4b6f8b86be479d70.json create mode 100644 .sqlx/query-ca612fee8c80f815b33dbc088cc7e0a7552f1f4e38390959e14ea5cc765bf2c7.json create mode 100644 .sqlx/query-ce850a32c8702595bbc3207bf86d53209c48be145562192b3a8f84903d39eac1.json create mode 100644 .sqlx/query-dc7056b7421dcc83648cbd360273ebcce1156ce0f7da2d39f054025ab9b66b6b.json create mode 100644 .sqlx/query-e83a43217a3bfaacd2f5e76df1fd05447558646fb2b47963751a7ab53b3a99b6.json create mode 100644 wakey-control-plane/.sqlx/query-6ce45cc94cbfdd948aad5fb986f9b5976c79bb63759bc9ecff1a7873e2d70137.json create mode 100644 wakey-control-plane/.sqlx/query-b324587b94ee2e98dc8eff5f6735303a2784b74f49068ce8e946cedcd8d12c4e.json create mode 100644 wakey-control-plane/.sqlx/query-bd1666cea8840df04ddcdc8a0c42b9b7771f9bf63c45613c4b6f8b86be479d70.json create mode 100644 wakey-control-plane/.sqlx/query-ca612fee8c80f815b33dbc088cc7e0a7552f1f4e38390959e14ea5cc765bf2c7.json create mode 100644 wakey-control-plane/.sqlx/query-ce850a32c8702595bbc3207bf86d53209c48be145562192b3a8f84903d39eac1.json create mode 100644 wakey-control-plane/.sqlx/query-dc7056b7421dcc83648cbd360273ebcce1156ce0f7da2d39f054025ab9b66b6b.json create mode 100644 wakey-control-plane/.sqlx/query-e83a43217a3bfaacd2f5e76df1fd05447558646fb2b47963751a7ab53b3a99b6.json diff --git a/.sqlx/query-6ce45cc94cbfdd948aad5fb986f9b5976c79bb63759bc9ecff1a7873e2d70137.json b/.sqlx/query-6ce45cc94cbfdd948aad5fb986f9b5976c79bb63759bc9ecff1a7873e2d70137.json new file mode 100644 index 0000000..e424e43 --- /dev/null +++ b/.sqlx/query-6ce45cc94cbfdd948aad5fb986f9b5976c79bb63759bc9ecff1a7873e2d70137.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "INSERT INTO audit_events\n (event_key, event_id, ts_unix, actor_type, actor_id, agent_id, request_id,\n event_type, outcome, latency_ms, message, metadata_json)\n VALUES (?1, ?2, ?3, 'test', NULL, NULL, NULL, 'test', 'ok', NULL, 'old', '{}')", + "describe": { + "columns": [], + "parameters": { + "Right": 3 + }, + "nullable": [] + }, + "hash": "6ce45cc94cbfdd948aad5fb986f9b5976c79bb63759bc9ecff1a7873e2d70137" +} diff --git a/.sqlx/query-b324587b94ee2e98dc8eff5f6735303a2784b74f49068ce8e946cedcd8d12c4e.json b/.sqlx/query-b324587b94ee2e98dc8eff5f6735303a2784b74f49068ce8e946cedcd8d12c4e.json new file mode 100644 index 0000000..6b66cc2 --- /dev/null +++ b/.sqlx/query-b324587b94ee2e98dc8eff5f6735303a2784b74f49068ce8e946cedcd8d12c4e.json @@ -0,0 +1,20 @@ +{ + "db_name": "SQLite", + "query": "SELECT COUNT(*) FROM alert_transitions WHERE transition_id = ?1", + "describe": { + "columns": [ + { + "name": "COUNT(*)", + "ordinal": 0, + "type_info": "Integer" + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + false + ] + }, + "hash": "b324587b94ee2e98dc8eff5f6735303a2784b74f49068ce8e946cedcd8d12c4e" +} diff --git a/.sqlx/query-bd1666cea8840df04ddcdc8a0c42b9b7771f9bf63c45613c4b6f8b86be479d70.json b/.sqlx/query-bd1666cea8840df04ddcdc8a0c42b9b7771f9bf63c45613c4b6f8b86be479d70.json new file mode 100644 index 0000000..d3879e7 --- /dev/null +++ b/.sqlx/query-bd1666cea8840df04ddcdc8a0c42b9b7771f9bf63c45613c4b6f8b86be479d70.json @@ -0,0 +1,20 @@ +{ + "db_name": "SQLite", + "query": "SELECT COUNT(*) FROM audit_events WHERE message = ?1", + "describe": { + "columns": [ + { + "name": "COUNT(*)", + "ordinal": 0, + "type_info": "Integer" + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + false + ] + }, + "hash": "bd1666cea8840df04ddcdc8a0c42b9b7771f9bf63c45613c4b6f8b86be479d70" +} diff --git a/.sqlx/query-ca612fee8c80f815b33dbc088cc7e0a7552f1f4e38390959e14ea5cc765bf2c7.json b/.sqlx/query-ca612fee8c80f815b33dbc088cc7e0a7552f1f4e38390959e14ea5cc765bf2c7.json new file mode 100644 index 0000000..f67dd7f --- /dev/null +++ b/.sqlx/query-ca612fee8c80f815b33dbc088cc7e0a7552f1f4e38390959e14ea5cc765bf2c7.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "DELETE FROM audit_events WHERE ts_unix < ?1", + "describe": { + "columns": [], + "parameters": { + "Right": 1 + }, + "nullable": [] + }, + "hash": "ca612fee8c80f815b33dbc088cc7e0a7552f1f4e38390959e14ea5cc765bf2c7" +} diff --git a/.sqlx/query-ce850a32c8702595bbc3207bf86d53209c48be145562192b3a8f84903d39eac1.json b/.sqlx/query-ce850a32c8702595bbc3207bf86d53209c48be145562192b3a8f84903d39eac1.json new file mode 100644 index 0000000..dda53b9 --- /dev/null +++ b/.sqlx/query-ce850a32c8702595bbc3207bf86d53209c48be145562192b3a8f84903d39eac1.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "DELETE FROM alert_transitions WHERE ts_unix < ?1", + "describe": { + "columns": [], + "parameters": { + "Right": 1 + }, + "nullable": [] + }, + "hash": "ce850a32c8702595bbc3207bf86d53209c48be145562192b3a8f84903d39eac1" +} diff --git a/.sqlx/query-dc7056b7421dcc83648cbd360273ebcce1156ce0f7da2d39f054025ab9b66b6b.json b/.sqlx/query-dc7056b7421dcc83648cbd360273ebcce1156ce0f7da2d39f054025ab9b66b6b.json new file mode 100644 index 0000000..9a8e6e2 --- /dev/null +++ b/.sqlx/query-dc7056b7421dcc83648cbd360273ebcce1156ce0f7da2d39f054025ab9b66b6b.json @@ -0,0 +1,20 @@ +{ + "db_name": "SQLite", + "query": "SELECT COUNT(*) FROM audit_events WHERE event_id = ?1", + "describe": { + "columns": [ + { + "name": "COUNT(*)", + "ordinal": 0, + "type_info": "Integer" + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + false + ] + }, + "hash": "dc7056b7421dcc83648cbd360273ebcce1156ce0f7da2d39f054025ab9b66b6b" +} diff --git a/.sqlx/query-e83a43217a3bfaacd2f5e76df1fd05447558646fb2b47963751a7ab53b3a99b6.json b/.sqlx/query-e83a43217a3bfaacd2f5e76df1fd05447558646fb2b47963751a7ab53b3a99b6.json new file mode 100644 index 0000000..e625dbb --- /dev/null +++ b/.sqlx/query-e83a43217a3bfaacd2f5e76df1fd05447558646fb2b47963751a7ab53b3a99b6.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "INSERT INTO alert_transitions\n (transition_key, transition_id, ts_unix, alert_id, kind, agent_id,\n from_status, to_status, message, metadata_json)\n VALUES (?1, ?2, ?3, 'alert-old', 'test', NULL, NULL, 'active', 'old', '{}')", + "describe": { + "columns": [], + "parameters": { + "Right": 3 + }, + "nullable": [] + }, + "hash": "e83a43217a3bfaacd2f5e76df1fd05447558646fb2b47963751a7ab53b3a99b6" +} diff --git a/wakey-control-plane/.sqlx/query-6ce45cc94cbfdd948aad5fb986f9b5976c79bb63759bc9ecff1a7873e2d70137.json b/wakey-control-plane/.sqlx/query-6ce45cc94cbfdd948aad5fb986f9b5976c79bb63759bc9ecff1a7873e2d70137.json new file mode 100644 index 0000000..e424e43 --- /dev/null +++ b/wakey-control-plane/.sqlx/query-6ce45cc94cbfdd948aad5fb986f9b5976c79bb63759bc9ecff1a7873e2d70137.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "INSERT INTO audit_events\n (event_key, event_id, ts_unix, actor_type, actor_id, agent_id, request_id,\n event_type, outcome, latency_ms, message, metadata_json)\n VALUES (?1, ?2, ?3, 'test', NULL, NULL, NULL, 'test', 'ok', NULL, 'old', '{}')", + "describe": { + "columns": [], + "parameters": { + "Right": 3 + }, + "nullable": [] + }, + "hash": "6ce45cc94cbfdd948aad5fb986f9b5976c79bb63759bc9ecff1a7873e2d70137" +} diff --git a/wakey-control-plane/.sqlx/query-b324587b94ee2e98dc8eff5f6735303a2784b74f49068ce8e946cedcd8d12c4e.json b/wakey-control-plane/.sqlx/query-b324587b94ee2e98dc8eff5f6735303a2784b74f49068ce8e946cedcd8d12c4e.json new file mode 100644 index 0000000..6b66cc2 --- /dev/null +++ b/wakey-control-plane/.sqlx/query-b324587b94ee2e98dc8eff5f6735303a2784b74f49068ce8e946cedcd8d12c4e.json @@ -0,0 +1,20 @@ +{ + "db_name": "SQLite", + "query": "SELECT COUNT(*) FROM alert_transitions WHERE transition_id = ?1", + "describe": { + "columns": [ + { + "name": "COUNT(*)", + "ordinal": 0, + "type_info": "Integer" + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + false + ] + }, + "hash": "b324587b94ee2e98dc8eff5f6735303a2784b74f49068ce8e946cedcd8d12c4e" +} diff --git a/wakey-control-plane/.sqlx/query-bd1666cea8840df04ddcdc8a0c42b9b7771f9bf63c45613c4b6f8b86be479d70.json b/wakey-control-plane/.sqlx/query-bd1666cea8840df04ddcdc8a0c42b9b7771f9bf63c45613c4b6f8b86be479d70.json new file mode 100644 index 0000000..d3879e7 --- /dev/null +++ b/wakey-control-plane/.sqlx/query-bd1666cea8840df04ddcdc8a0c42b9b7771f9bf63c45613c4b6f8b86be479d70.json @@ -0,0 +1,20 @@ +{ + "db_name": "SQLite", + "query": "SELECT COUNT(*) FROM audit_events WHERE message = ?1", + "describe": { + "columns": [ + { + "name": "COUNT(*)", + "ordinal": 0, + "type_info": "Integer" + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + false + ] + }, + "hash": "bd1666cea8840df04ddcdc8a0c42b9b7771f9bf63c45613c4b6f8b86be479d70" +} diff --git a/wakey-control-plane/.sqlx/query-ca612fee8c80f815b33dbc088cc7e0a7552f1f4e38390959e14ea5cc765bf2c7.json b/wakey-control-plane/.sqlx/query-ca612fee8c80f815b33dbc088cc7e0a7552f1f4e38390959e14ea5cc765bf2c7.json new file mode 100644 index 0000000..f67dd7f --- /dev/null +++ b/wakey-control-plane/.sqlx/query-ca612fee8c80f815b33dbc088cc7e0a7552f1f4e38390959e14ea5cc765bf2c7.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "DELETE FROM audit_events WHERE ts_unix < ?1", + "describe": { + "columns": [], + "parameters": { + "Right": 1 + }, + "nullable": [] + }, + "hash": "ca612fee8c80f815b33dbc088cc7e0a7552f1f4e38390959e14ea5cc765bf2c7" +} diff --git a/wakey-control-plane/.sqlx/query-ce850a32c8702595bbc3207bf86d53209c48be145562192b3a8f84903d39eac1.json b/wakey-control-plane/.sqlx/query-ce850a32c8702595bbc3207bf86d53209c48be145562192b3a8f84903d39eac1.json new file mode 100644 index 0000000..dda53b9 --- /dev/null +++ b/wakey-control-plane/.sqlx/query-ce850a32c8702595bbc3207bf86d53209c48be145562192b3a8f84903d39eac1.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "DELETE FROM alert_transitions WHERE ts_unix < ?1", + "describe": { + "columns": [], + "parameters": { + "Right": 1 + }, + "nullable": [] + }, + "hash": "ce850a32c8702595bbc3207bf86d53209c48be145562192b3a8f84903d39eac1" +} diff --git a/wakey-control-plane/.sqlx/query-dc7056b7421dcc83648cbd360273ebcce1156ce0f7da2d39f054025ab9b66b6b.json b/wakey-control-plane/.sqlx/query-dc7056b7421dcc83648cbd360273ebcce1156ce0f7da2d39f054025ab9b66b6b.json new file mode 100644 index 0000000..9a8e6e2 --- /dev/null +++ b/wakey-control-plane/.sqlx/query-dc7056b7421dcc83648cbd360273ebcce1156ce0f7da2d39f054025ab9b66b6b.json @@ -0,0 +1,20 @@ +{ + "db_name": "SQLite", + "query": "SELECT COUNT(*) FROM audit_events WHERE event_id = ?1", + "describe": { + "columns": [ + { + "name": "COUNT(*)", + "ordinal": 0, + "type_info": "Integer" + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + false + ] + }, + "hash": "dc7056b7421dcc83648cbd360273ebcce1156ce0f7da2d39f054025ab9b66b6b" +} diff --git a/wakey-control-plane/.sqlx/query-e83a43217a3bfaacd2f5e76df1fd05447558646fb2b47963751a7ab53b3a99b6.json b/wakey-control-plane/.sqlx/query-e83a43217a3bfaacd2f5e76df1fd05447558646fb2b47963751a7ab53b3a99b6.json new file mode 100644 index 0000000..e625dbb --- /dev/null +++ b/wakey-control-plane/.sqlx/query-e83a43217a3bfaacd2f5e76df1fd05447558646fb2b47963751a7ab53b3a99b6.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "INSERT INTO alert_transitions\n (transition_key, transition_id, ts_unix, alert_id, kind, agent_id,\n from_status, to_status, message, metadata_json)\n VALUES (?1, ?2, ?3, 'alert-old', 'test', NULL, NULL, 'active', 'old', '{}')", + "describe": { + "columns": [], + "parameters": { + "Right": 3 + }, + "nullable": [] + }, + "hash": "e83a43217a3bfaacd2f5e76df1fd05447558646fb2b47963751a7ab53b3a99b6" +} diff --git a/wakey-control-plane/src/api/alerts.rs b/wakey-control-plane/src/api/alerts.rs index 45defaa..323cb40 100644 --- a/wakey-control-plane/src/api/alerts.rs +++ b/wakey-control-plane/src/api/alerts.rs @@ -1,4 +1,5 @@ use std::collections::{BTreeMap, HashSet}; +use std::sync::Arc; use std::time::{SystemTime, UNIX_EPOCH}; use axum::Json; @@ -7,12 +8,13 @@ use axum::extract::{Query, State}; use axum::http::StatusCode; use axum::response::IntoResponse; use serde::Deserialize; +use tokio::sync::{RwLock, broadcast}; use tokio::time::Duration; use tracing::warn; use crate::api::ApiError; use crate::runtime::AppState; -use crate::state::{AlertState, AuditEvent, AuditEventFilter}; +use crate::state::{AlertState, AlertTransition, AuditEvent, AuditEventFilter}; struct AlertBuildContext<'a> { now_unix: u64, @@ -26,21 +28,13 @@ struct AlertBuildContext<'a> { enroll_rejected_threshold: u64, } -#[derive(Debug, Deserialize)] -pub struct ActiveAlertsQuery { - pub lookback_seconds: Option, - pub timeout_threshold: Option, - pub auth_rejected_threshold: Option, - pub enroll_rejected_threshold: Option, -} - #[derive(Debug, Deserialize)] pub struct AlertHistoryQuery { pub since_unix: Option, pub limit: Option, } -#[derive(Debug, Deserialize, Clone)] +#[derive(Debug, Clone, Deserialize)] pub struct AlertRuleConfig { pub lookback_seconds: Option, pub timeout_threshold: Option, @@ -48,25 +42,120 @@ pub struct AlertRuleConfig { pub enroll_rejected_threshold: Option, } -pub async fn active_alerts( - State(state): State, - Query(query): Query, -) -> Result { - let alerts = evaluate_alerts( - &state, - AlertRuleConfig { - lookback_seconds: query.lookback_seconds, - timeout_threshold: query.timeout_threshold, - auth_rejected_threshold: query.auth_rejected_threshold, - enroll_rejected_threshold: query.enroll_rejected_threshold, - }, - ) - .await?; +impl Default for AlertRuleConfig { + fn default() -> Self { + Self { + lookback_seconds: Some(900), + timeout_threshold: Some(3), + auth_rejected_threshold: Some(3), + enroll_rejected_threshold: Some(5), + } + } +} - if let Err(err) = state.store.sync_alert_transitions(&alerts).await { - warn!(error = %err, "failed to sync alert transitions"); +#[derive(Clone, Default)] +pub struct AlertsSnapshot { + pub ts_unix: u64, + pub alerts: Vec, + pub recent_transitions: Vec, +} + +/// Holds the latest alert snapshot computed by the background evaluator and +/// broadcasts each new snapshot to connected `alerts/ws` subscribers. +#[derive(Clone)] +pub struct AlertsHub { + tx: broadcast::Sender, + latest_payload: Arc>>, + latest_alerts: Arc>>, +} + +impl Default for AlertsHub { + fn default() -> Self { + Self::new() + } +} + +impl AlertsHub { + pub fn new() -> Self { + let (tx, _) = broadcast::channel(16); + Self { + tx, + latest_payload: Arc::new(RwLock::new(None)), + latest_alerts: Arc::new(RwLock::new(Vec::new())), + } } + pub fn subscribe(&self) -> broadcast::Receiver { + self.tx.subscribe() + } + + pub async fn latest_alerts(&self) -> Vec { + self.latest_alerts.read().await.clone() + } + + pub async fn latest_payload(&self) -> Option { + self.latest_payload.read().await.clone() + } + + pub async fn publish(&self, snapshot: AlertsSnapshot) { + let payload = serde_json::json!({ + "type": "alerts_snapshot", + "ts_unix": snapshot.ts_unix, + "alerts": snapshot.alerts, + "recent_transitions": snapshot.recent_transitions, + }); + let Ok(encoded) = serde_json::to_string(&payload) else { + warn!("failed to encode alerts snapshot payload"); + return; + }; + *self.latest_alerts.write().await = snapshot.alerts; + *self.latest_payload.write().await = Some(encoded.clone()); + let _ = self.tx.send(encoded); + } +} + +/// Background task that owns alert evaluation: on a fixed ticker it recomputes +/// the snapshot, persists transitions, and publishes to subscribers so neither +/// the GET endpoint nor individual WS clients evaluate alerts themselves. +pub async fn run_alert_evaluator(state: AppState) { + let config = AlertRuleConfig::default(); + let mut tick = tokio::time::interval(Duration::from_secs(5)); + tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + tick.tick().await; + let alerts = match evaluate_alerts(&state, &config).await { + Ok(alerts) => alerts, + Err(err) => { + warn!(code = %err.code, "failed to evaluate alerts in background task"); + continue; + } + }; + + if let Err(err) = state.store.sync_alert_transitions(&alerts).await { + warn!(error = %err, "failed to sync alert transitions in background task"); + } + + let recent_transitions = match state.store.list_alert_transitions(None, 20).await { + Ok(history) => history, + Err(err) => { + warn!(error = %err, "failed to load alert transition history for snapshot"); + Vec::new() + } + }; + + state + .alerts + .publish(AlertsSnapshot { + ts_unix: now_unix(), + alerts, + recent_transitions, + }) + .await; + } +} + +pub async fn active_alerts(State(state): State) -> Result { + let alerts = state.alerts.latest_alerts().await; Ok((StatusCode::OK, Json(alerts))) } @@ -94,64 +183,34 @@ pub async fn alert_history( pub async fn alerts_stream( ws: WebSocketUpgrade, State(state): State, - Query(query): Query, ) -> impl IntoResponse { - let config = AlertRuleConfig { - lookback_seconds: query.lookback_seconds, - timeout_threshold: query.timeout_threshold, - auth_rejected_threshold: query.auth_rejected_threshold, - enroll_rejected_threshold: query.enroll_rejected_threshold, - }; - ws.on_upgrade(move |socket| alerts_stream_socket(state, socket, config)) + ws.on_upgrade(move |socket| alerts_stream_socket(state, socket)) } -async fn alerts_stream_socket(state: AppState, mut socket: WebSocket, config: AlertRuleConfig) { - let mut tick = tokio::time::interval(Duration::from_secs(5)); - loop { - tick.tick().await; - let alerts = match evaluate_alerts(&state, config.clone()).await { - Ok(alerts) => alerts, - Err(err) => { - warn!(code = %err.code, "failed to evaluate alerts for stream"); - continue; - } - }; +async fn alerts_stream_socket(state: AppState, mut socket: WebSocket) { + let mut rx = state.alerts.subscribe(); - if let Err(err) = state.store.sync_alert_transitions(&alerts).await { - warn!(error = %err, "failed to sync alert transitions in stream"); + if let Some(encoded) = state.alerts.latest_payload().await + && socket.send(Message::Text(encoded.into())).await.is_err() { + return; } - let history = match state.store.list_alert_transitions(None, 20).await { - Ok(h) => h, - Err(err) => { - warn!(error = %err, "failed to load alert transition history for stream"); - Vec::new() + loop { + match rx.recv().await { + Ok(encoded) => { + if socket.send(Message::Text(encoded.into())).await.is_err() { + break; + } } - }; - - let payload = serde_json::json!({ - "type": "alerts_snapshot", - "ts_unix": now_unix(), - "alerts": alerts, - "recent_transitions": history, - }); - let encoded = match serde_json::to_string(&payload) { - Ok(s) => s, - Err(err) => { - warn!(error = %err, "failed to encode alerts stream payload"); - continue; - } - }; - - if socket.send(Message::Text(encoded.into())).await.is_err() { - break; + Err(broadcast::error::RecvError::Lagged(_)) => continue, + Err(broadcast::error::RecvError::Closed) => break, } } } async fn evaluate_alerts( state: &AppState, - config: AlertRuleConfig, + config: &AlertRuleConfig, ) -> Result, ApiError> { let lookback_seconds = config.lookback_seconds.unwrap_or(900).clamp(60, 86_400); let timeout_threshold = config.timeout_threshold.unwrap_or(3).max(1); @@ -161,7 +220,14 @@ async fn evaluate_alerts( let now = now_unix(); let since_unix = now.saturating_sub(lookback_seconds); - let enrolled_agents = state.store.list_agents().await; + let enrolled_agents = state.store.list_agents().await.map_err(|err| { + warn!(error = %err, "failed listing agents for alert evaluation"); + ApiError::new( + StatusCode::INTERNAL_SERVER_ERROR, + "alerts_query_failed", + err.to_string(), + ) + })?; let connected_agents = state .sessions .read() diff --git a/wakey-control-plane/src/api/commands.rs b/wakey-control-plane/src/api/commands.rs index c86c8af..2f5b317 100644 --- a/wakey-control-plane/src/api/commands.rs +++ b/wakey-control-plane/src/api/commands.rs @@ -40,7 +40,18 @@ pub struct RelayCommandResponse { } pub async fn list_agents(State(state): State) -> Result { - let enrolled = state.store.list_agents_with_nicknames().await; + let enrolled = state + .store + .list_agents_with_nicknames() + .await + .map_err(|err| { + warn!(error = %err, "failed listing agents"); + ApiError::new( + StatusCode::INTERNAL_SERVER_ERROR, + "list_agents_failed", + err.to_string(), + ) + })?; let sessions = state.sessions.read().await; let agents = enrolled @@ -120,9 +131,9 @@ pub async fn relay_agent_command( .insert(request_id_string.clone(), pending_tx); info!("dispatching command to agent"); - if let Err(err) = state + state .store - .append_audit_event(AuditEventInput { + .audit(AuditEventInput { actor_type: "admin_api".into(), actor_id: None, agent_id: Some(agent_id.to_string()), @@ -133,10 +144,7 @@ pub async fn relay_agent_command( message: "dispatched command to connected agent".into(), metadata: serde_json::json!({ "command": command_kind }), }) - .await - { - warn!(error = %err, "failed to append audit event for command dispatch"); - } + .await; if let Err(err) = tx.send(SessionEvent::Message(ServerMessage::Command { request_id, @@ -144,9 +152,9 @@ pub async fn relay_agent_command( })) { state.pending.lock().await.remove(&request_id_string); warn!(error = %err, "failed sending command to agent session"); - if let Err(audit_err) = state + state .store - .append_audit_event(AuditEventInput { + .audit(AuditEventInput { actor_type: "admin_api".into(), actor_id: None, agent_id: Some(agent_id.to_string()), @@ -157,10 +165,7 @@ pub async fn relay_agent_command( message: err.to_string(), metadata: serde_json::json!({ "command": command_kind }), }) - .await - { - warn!(error = %audit_err, "failed to append audit event for command send failure"); - } + .await; return Err(ApiError::new( StatusCode::BAD_GATEWAY, "agent_send_failed", @@ -177,9 +182,9 @@ pub async fn relay_agent_command( let response = match outcome { Ok(Ok(AgentReply::Result(result))) => { info!("agent command completed"); - if let Err(err) = state + state .store - .append_audit_event(AuditEventInput { + .audit(AuditEventInput { actor_type: "admin_api".into(), actor_id: None, agent_id: Some(agent_id.to_string()), @@ -190,10 +195,7 @@ pub async fn relay_agent_command( message: "agent command completed".into(), metadata: serde_json::json!({ "command": command_kind }), }) - .await - { - warn!(error = %err, "failed to append audit event for command success"); - } + .await; RelayCommandResponse { request_id: request_id_string, status: "ok".into(), @@ -203,9 +205,9 @@ pub async fn relay_agent_command( } Ok(Ok(AgentReply::Error(error))) => { warn!(code = %error.code, "agent command returned error"); - if let Err(err) = state + state .store - .append_audit_event(AuditEventInput { + .audit(AuditEventInput { actor_type: "admin_api".into(), actor_id: None, agent_id: Some(agent_id.to_string()), @@ -216,10 +218,7 @@ pub async fn relay_agent_command( message: error.message.clone(), metadata: serde_json::json!({ "command": command_kind, "code": error.code }), }) - .await - { - warn!(error = %err, "failed to append audit event for command error result"); - } + .await; RelayCommandResponse { request_id: request_id_string, status: "error".into(), @@ -229,9 +228,9 @@ pub async fn relay_agent_command( } Ok(Err(_)) => { warn!("agent response channel dropped"); - if let Err(err) = state + state .store - .append_audit_event(AuditEventInput { + .audit(AuditEventInput { actor_type: "admin_api".into(), actor_id: None, agent_id: Some(agent_id.to_string()), @@ -242,10 +241,7 @@ pub async fn relay_agent_command( message: "agent response channel dropped".into(), metadata: serde_json::json!({ "command": command_kind }), }) - .await - { - warn!(error = %err, "failed to append audit event for dropped response"); - } + .await; return Err(ApiError::new( StatusCode::BAD_GATEWAY, "agent_response_dropped", @@ -258,9 +254,9 @@ pub async fn relay_agent_command( timeout_ms = timeout.as_millis() as u64, "agent command timed out" ); - if let Err(err) = state + state .store - .append_audit_event(AuditEventInput { + .audit(AuditEventInput { actor_type: "admin_api".into(), actor_id: None, agent_id: Some(agent_id.to_string()), @@ -271,10 +267,7 @@ pub async fn relay_agent_command( message: "agent command timed out".into(), metadata: serde_json::json!({ "command": command_kind, "timeout_ms": timeout.as_millis() as u64 }), }) - .await - { - warn!(error = %err, "failed to append audit event for timeout"); - } + .await; return Err(ApiError::new( StatusCode::GATEWAY_TIMEOUT, "agent_timeout", diff --git a/wakey-control-plane/src/api/control/enroll.rs b/wakey-control-plane/src/api/control/enroll.rs index 77cd395..80d86cd 100644 --- a/wakey-control-plane/src/api/control/enroll.rs +++ b/wakey-control-plane/src/api/control/enroll.rs @@ -74,9 +74,9 @@ pub async fn enroll( match state.store.enroll(&req.enroll_token).await { Ok(issued) => { info!(agent_id = %issued.agent_id, "agent enrollment accepted"); - if let Err(err) = state + state .store - .append_audit_event(AuditEventInput { + .audit(AuditEventInput { actor_type: "agent".into(), actor_id: Some(issued.agent_id.clone()), agent_id: Some(issued.agent_id.clone()), @@ -87,10 +87,7 @@ pub async fn enroll( message: "agent enrollment accepted".into(), metadata: serde_json::json!({}), }) - .await - { - warn!(error = %err, "failed to append audit event for enroll success"); - } + .await; Ok(( StatusCode::OK, Json(EnrollResponse { @@ -102,9 +99,9 @@ pub async fn enroll( } Err(err) => { warn!(error = %err, "agent enrollment rejected"); - if let Err(audit_err) = state + state .store - .append_audit_event(AuditEventInput { + .audit(AuditEventInput { actor_type: "agent".into(), actor_id: None, agent_id: None, @@ -115,10 +112,7 @@ pub async fn enroll( message: err.to_string(), metadata: serde_json::json!({}), }) - .await - { - warn!(error = %audit_err, "failed to append audit event for enroll rejection"); - } + .await; Err(ApiError::new( StatusCode::UNAUTHORIZED, "enrollment_rejected", @@ -145,9 +139,9 @@ pub async fn issue_enroll_token( expires_at_unix = issued.expires_at_unix, "issued enroll token" ); - if let Err(err) = state + state .store - .append_audit_event(AuditEventInput { + .audit(AuditEventInput { actor_type: "admin_api".into(), actor_id: None, agent_id: None, @@ -161,10 +155,7 @@ pub async fn issue_enroll_token( "expires_at_unix": issued.expires_at_unix, }), }) - .await - { - warn!(error = %err, "failed to append audit event for token issuance"); - } + .await; Ok(( StatusCode::OK, Json(IssueEnrollTokenResponse { @@ -189,9 +180,9 @@ pub async fn list_enroll_tokens( ) -> Result { match state.store.list_enroll_tokens().await { Ok(tokens) => { - if let Err(err) = state + state .store - .append_audit_event(AuditEventInput { + .audit(AuditEventInput { actor_type: "admin_api".into(), actor_id: None, agent_id: None, @@ -204,10 +195,7 @@ pub async fn list_enroll_tokens( "count": tokens.len(), }), }) - .await - { - warn!(error = %err, "failed to append audit event for token listing"); - } + .await; let body = tokens .into_iter() .map(|t| EnrollTokenStatus { @@ -235,9 +223,9 @@ pub async fn revoke_enroll_token( ) -> Result { match state.store.revoke_enroll_token(&token).await { Ok(revoked) => { - if let Err(err) = state + state .store - .append_audit_event(AuditEventInput { + .audit(AuditEventInput { actor_type: "admin_api".into(), actor_id: None, agent_id: None, @@ -256,10 +244,7 @@ pub async fn revoke_enroll_token( }, metadata: serde_json::json!({ "token": token }), }) - .await - { - warn!(error = %err, "failed to append audit event for token revoke"); - } + .await; Ok(( StatusCode::OK, Json(RevokeEnrollTokenResponse { token, revoked }), @@ -289,9 +274,9 @@ pub async fn revoke_agent( } } - if let Err(err) = state + state .store - .append_audit_event(AuditEventInput { + .audit(AuditEventInput { actor_type: "admin_api".into(), actor_id: None, agent_id: Some(agent_id.clone()), @@ -310,10 +295,7 @@ pub async fn revoke_agent( }, metadata: serde_json::json!({ "agent_id": agent_id }), }) - .await - { - warn!(error = %err, "failed to append audit event for agent revoke"); - } + .await; Ok(( StatusCode::OK, @@ -349,9 +331,9 @@ pub async fn set_agent_nickname( .await { Ok(updated) => { - if let Err(err) = state + state .store - .append_audit_event(AuditEventInput { + .audit(AuditEventInput { actor_type: "admin_api".into(), actor_id: None, agent_id: Some(agent_id.clone()), @@ -373,10 +355,7 @@ pub async fn set_agent_nickname( "nickname": normalized, }), }) - .await - { - warn!(error = %err, "failed to append audit event for nickname set"); - } + .await; Ok(( StatusCode::OK, diff --git a/wakey-control-plane/src/api/control/fleet/mod.rs b/wakey-control-plane/src/api/control/fleet/mod.rs index dc57958..e0bace1 100644 --- a/wakey-control-plane/src/api/control/fleet/mod.rs +++ b/wakey-control-plane/src/api/control/fleet/mod.rs @@ -4,11 +4,12 @@ use axum::Json; use axum::extract::{Query, State}; use axum::http::StatusCode; use axum::response::IntoResponse; +use futures_util::StreamExt; use tracing::warn; use wakey_agent::protocol::{AgentCommand, InventoryRequest, WakeRequest}; use crate::api::ApiError; -use crate::api::commands::relay_agent_command; +use crate::api::commands::{RelayCommandResponse, relay_agent_command}; use crate::runtime::AppState; mod build; @@ -26,6 +27,9 @@ use types::{ RefreshFleetDevicesResponse, WakeFleetDeviceRequest, WakeFleetDeviceResponse, }; +const MAX_REFRESH_AGENTS: usize = 128; +const REFRESH_CONCURRENCY: usize = 8; + pub async fn list_fleet_devices( State(state): State, Query(query): Query, @@ -55,88 +59,43 @@ pub async fn refresh_fleet_devices( }; agent_ids.sort(); agent_ids.dedup(); - - let mut results = Vec::with_capacity(agent_ids.len()); - let mut total_accepted = 0usize; - for agent_id in agent_ids { - let response = relay_agent_command( - &state, - &agent_id, - AgentCommand::Inventory(InventoryRequest { - query: None, - name: None, - ips: Vec::new(), - devs: Vec::new(), - nuds: Vec::new(), - macs: Vec::new(), - }), - req.timeout_ms, - ) - .await; - - match response { - Ok(response) if response.status == "ok" => { - let Some(result) = response.result else { - results.push(RefreshFleetAgentResult { - agent_id, - status: "error".into(), - accepted: 0, - error: Some("inventory command returned no result".into()), - }); - continue; - }; - match serde_json::from_value::(result) - .map(|i| i.devices) - { - Ok(devices) => { - match state - .store - .replace_agent_device_snapshot(&agent_id, &devices) - .await - { - Ok(accepted) => { - total_accepted = total_accepted.saturating_add(accepted); - results.push(RefreshFleetAgentResult { - agent_id, - status: "ok".into(), - accepted, - error: None, - }); - } - Err(err) => results.push(RefreshFleetAgentResult { - agent_id, - status: "error".into(), - accepted: 0, - error: Some(err.to_string()), - }), - } - } - Err(err) => results.push(RefreshFleetAgentResult { - agent_id, - status: "error".into(), - accepted: 0, - error: Some(err.to_string()), - }), - } - } - Ok(response) => results.push(RefreshFleetAgentResult { - agent_id, - status: "error".into(), - accepted: 0, - error: response - .error - .map(|error| error.message) - .or_else(|| Some("inventory command failed".into())), - }), - Err(err) => results.push(RefreshFleetAgentResult { - agent_id, - status: "error".into(), - accepted: 0, - error: Some(format!("{}: {}", err.status, err.message)), - }), - } + if agent_ids.len() > MAX_REFRESH_AGENTS { + warn!( + requested = agent_ids.len(), + capped = MAX_REFRESH_AGENTS, + "capping fleet refresh agent count" + ); + agent_ids.truncate(MAX_REFRESH_AGENTS); } + let results = futures_util::stream::iter(agent_ids.into_iter().map(|agent_id| { + let state = state.clone(); + async move { + let response = relay_agent_command( + &state, + &agent_id, + AgentCommand::Inventory(InventoryRequest { + query: None, + name: None, + ips: Vec::new(), + devs: Vec::new(), + nuds: Vec::new(), + macs: Vec::new(), + }), + req.timeout_ms, + ) + .await; + refresh_fleet_agent_result(&state, &agent_id, response).await + } + })) + .buffer_unordered(REFRESH_CONCURRENCY) + .collect::>() + .await; + + let mut results = results; + results.sort_by(|a, b| a.agent_id.cmp(&b.agent_id)); + let total_accepted = results.iter().map(|result| result.accepted).sum::(); + Ok(( StatusCode::OK, Json(RefreshFleetDevicesResponse { @@ -146,6 +105,70 @@ pub async fn refresh_fleet_devices( )) } +async fn refresh_fleet_agent_result( + state: &AppState, + agent_id: &str, + response: Result, +) -> RefreshFleetAgentResult { + match response { + Ok(response) if response.status == "ok" => { + let Some(result) = response.result else { + return RefreshFleetAgentResult { + agent_id: agent_id.to_string(), + status: "error".into(), + accepted: 0, + error: Some("inventory command returned no result".into()), + }; + }; + match serde_json::from_value::(result) + .map(|inventory| inventory.devices) + { + Ok(devices) => { + match state + .store + .replace_agent_device_snapshot(agent_id, &devices) + .await + { + Ok(accepted) => RefreshFleetAgentResult { + agent_id: agent_id.to_string(), + status: "ok".into(), + accepted, + error: None, + }, + Err(err) => RefreshFleetAgentResult { + agent_id: agent_id.to_string(), + status: "error".into(), + accepted: 0, + error: Some(err.to_string()), + }, + } + } + Err(err) => RefreshFleetAgentResult { + agent_id: agent_id.to_string(), + status: "error".into(), + accepted: 0, + error: Some(err.to_string()), + }, + } + } + Ok(response) => RefreshFleetAgentResult { + agent_id: agent_id.to_string(), + status: "error".into(), + accepted: 0, + error: response + .error + .map(|error| error.message) + .or_else(|| Some("inventory command failed".into())), + }, + Err(err) => RefreshFleetAgentResult { + agent_id: agent_id.to_string(), + status: "error".into(), + accepted: 0, + error: Some(format!("{}: {}", err.status, err.message)), + }, + } +} + pub async fn wake_fleet_device( State(state): State, Json(req): Json, @@ -243,7 +266,7 @@ async fn load_fleet_devices( let sessions = state.sessions.read().await; sessions.keys().cloned().collect::>() }; - let agents = state.store.list_agents_with_nicknames().await; + let agents = state.store.list_agents_with_nicknames().await?; let agent_status = agents .into_iter() .map(|(agent_id, nickname)| { diff --git a/wakey-control-plane/src/api/mod.rs b/wakey-control-plane/src/api/mod.rs index 2f9224f..ebfa7e9 100644 --- a/wakey-control-plane/src/api/mod.rs +++ b/wakey-control-plane/src/api/mod.rs @@ -7,7 +7,7 @@ mod commands; mod control; mod terminals; -pub use alerts::{active_alerts, alert_history, alerts_stream}; +pub use alerts::{AlertsHub, active_alerts, alert_history, alerts_stream, run_alert_evaluator}; pub use audit::list_audit_events; pub use commands::{list_agents, run_command}; pub use control::{ diff --git a/wakey-control-plane/src/api/terminals.rs b/wakey-control-plane/src/api/terminals.rs index 2843841..26b6944 100644 --- a/wakey-control-plane/src/api/terminals.rs +++ b/wakey-control-plane/src/api/terminals.rs @@ -649,9 +649,9 @@ async fn append_terminal_audit( terminal_id: &str, audit: TerminalAudit<'_>, ) { - if let Err(err) = state + state .store - .append_audit_event(AuditEventInput { + .audit(AuditEventInput { actor_type: audit.actor_type.into(), actor_id: None, agent_id: Some(agent_id.to_string()), @@ -662,10 +662,7 @@ async fn append_terminal_audit( message: audit.message.into(), metadata: audit.metadata, }) - .await - { - warn!(terminal_id, event_type = audit.event_type, error = %err, "failed to append terminal audit event"); - } + .await; } fn validate_size(rows: u16, cols: u16) -> Result<(), ApiError> { diff --git a/wakey-control-plane/src/config/types.rs b/wakey-control-plane/src/config/types.rs index 4c15819..c2a9287 100644 --- a/wakey-control-plane/src/config/types.rs +++ b/wakey-control-plane/src/config/types.rs @@ -13,7 +13,6 @@ pub struct DaemonConfig { pub state_file: PathBuf, pub command_timeout: Duration, pub enroll_token_ttl: Duration, - #[allow(dead_code)] pub observation_retention: Duration, pub pid_file: PathBuf, pub ui_dist_dir: PathBuf, diff --git a/wakey-control-plane/src/runtime/mod.rs b/wakey-control-plane/src/runtime/mod.rs index 7053894..41b854b 100644 --- a/wakey-control-plane/src/runtime/mod.rs +++ b/wakey-control-plane/src/runtime/mod.rs @@ -43,6 +43,7 @@ pub struct AppState { pub command_timeout: Duration, pub enroll_token_ttl: Duration, pub terminals: terminals::TerminalRegistry, + pub alerts: api::AlertsHub, } #[derive(Clone)] @@ -182,8 +183,11 @@ pub async fn serve(daemon: config::DaemonConfig) -> Result<()> { command_timeout: daemon.command_timeout, enroll_token_ttl: daemon.enroll_token_ttl, terminals: terminals::TerminalRegistry::new(), + alerts: api::AlertsHub::new(), }; + tokio::spawn(api::run_alert_evaluator(app_state.clone())); + // Keep route classes explicit so edge policy can map directly: // - public_api_routes: intended internet-facing agent endpoints // - control_api_routes: intended admin-only endpoints behind Access @@ -231,6 +235,18 @@ pub async fn serve(daemon: config::DaemonConfig) -> Result<()> { } Err(err) => warn!(error = %err, "periodic gc failed"), } + match app_state.store.gc_retention(daemon.observation_retention).await { + Ok(stats) => { + if stats.audit_events_removed > 0 || stats.alert_transitions_removed > 0 { + info!( + audit_events_removed = stats.audit_events_removed, + alert_transitions_removed = stats.alert_transitions_removed, + "periodic gc pruned rows older than retention window" + ); + } + } + Err(err) => warn!(error = %err, "periodic retention gc failed"), + } } join = &mut server => { let _ = remove_pid_file(&daemon.pid_file); diff --git a/wakey-control-plane/src/state/mod.rs b/wakey-control-plane/src/state/mod.rs index 5d81019..d14a502 100644 --- a/wakey-control-plane/src/state/mod.rs +++ b/wakey-control-plane/src/state/mod.rs @@ -5,6 +5,7 @@ pub use store::Store; #[cfg(test)] pub use types::AgentDeviceRow; pub use types::{ - AgentDeviceWithChildren, AlertState, AuditEvent, AuditEventFilter, AuditEventInput, - DeviceIdentifier, DeviceIdentifierInput, KnownDevice, KnownDeviceInput, KnownDeviceSummary, + AgentDeviceWithChildren, AlertState, AlertTransition, AuditEvent, AuditEventFilter, + AuditEventInput, DeviceIdentifier, DeviceIdentifierInput, KnownDevice, KnownDeviceInput, + KnownDeviceSummary, }; diff --git a/wakey-control-plane/src/state/store.rs b/wakey-control-plane/src/state/store.rs index 8e01b82..8d4a2ce 100644 --- a/wakey-control-plane/src/state/store.rs +++ b/wakey-control-plane/src/state/store.rs @@ -10,7 +10,7 @@ use uuid::Uuid; use crate::state::types::{ AlertState, AlertTransition, AuditEvent, AuditEventFilter, AuditEventInput, DeviceIdentifier, DeviceIdentifierInput, EnrollTokenInfo, IssuedAgent, IssuedEnrollToken, KnownDevice, - KnownDeviceInput, StateStats, + KnownDeviceInput, RetentionStats, StateStats, }; pub struct Store { @@ -103,6 +103,124 @@ mod tests { assert_eq!(exists, 0); } + #[tokio::test] + async fn enroll_rejects_negative_expiry() { + let ts = TestStore::new().await; + sqlx::query!( + "INSERT INTO enroll_tokens (token, expires_at_unix) VALUES (?1, ?2)", + "enr-negative-expiry-test", + -1_i64 + ) + .execute(&ts.store().pool) + .await + .expect("insert should succeed"); + + let err = ts + .store() + .enroll("enr-negative-expiry-test") + .await + .expect_err("negative expiry should be treated as expired"); + + assert!(err.to_string().contains("expired")); + let exists = sqlx::query_scalar!( + "SELECT COUNT(*) FROM enroll_tokens WHERE token = ?1", + "enr-negative-expiry-test" + ) + .fetch_one(&ts.store().pool) + .await + .expect("read should succeed"); + assert_eq!(exists, 0); + } + + #[tokio::test] + async fn gc_retention_prunes_old_audit_and_alert_history() { + let ts = TestStore::new().await; + + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("clock should be after epoch") + .as_secs() as i64; + let old_ts = now - 86_400; + let old_evt = format!("{:020}:old-evt", old_ts); + sqlx::query!( + "INSERT INTO audit_events + (event_key, event_id, ts_unix, actor_type, actor_id, agent_id, request_id, + event_type, outcome, latency_ms, message, metadata_json) + VALUES (?1, ?2, ?3, 'test', NULL, NULL, NULL, 'test', 'ok', NULL, 'old', '{}')", + old_evt, + "old-evt", + old_ts + ) + .execute(&ts.store().pool) + .await + .expect("old audit event should insert"); + + let old_atr = format!("{:020}:old-atr", old_ts); + sqlx::query!( + "INSERT INTO alert_transitions + (transition_key, transition_id, ts_unix, alert_id, kind, agent_id, + from_status, to_status, message, metadata_json) + VALUES (?1, ?2, ?3, 'alert-old', 'test', NULL, NULL, 'active', 'old', '{}')", + old_atr, + "old-atr", + old_ts + ) + .execute(&ts.store().pool) + .await + .expect("old alert transition should insert"); + + ts.store() + .append_audit_event(crate::state::AuditEventInput { + actor_type: "test".into(), + actor_id: None, + agent_id: None, + request_id: None, + event_type: "test".into(), + outcome: "ok".into(), + latency_ms: None, + message: "fresh".into(), + metadata: serde_json::json!({}), + }) + .await + .expect("fresh audit event should insert"); + + let stats = ts + .store() + .gc_retention(Duration::from_secs(3600)) + .await + .expect("retention gc should succeed"); + + assert_eq!(stats.audit_events_removed, 1); + assert_eq!(stats.alert_transitions_removed, 1); + + let old_audit_gone = sqlx::query_scalar!( + "SELECT COUNT(*) FROM audit_events WHERE event_id = ?1", + "old-evt" + ) + .fetch_one(&ts.store().pool) + .await + .expect("read should succeed"); + assert_eq!(old_audit_gone, 0); + + let fresh_audit_remains = sqlx::query_scalar!( + "SELECT COUNT(*) FROM audit_events WHERE message = ?1", + "fresh" + ) + .fetch_one(&ts.store().pool) + .await + .expect("read should succeed"); + assert_eq!(fresh_audit_remains, 1); + + let old_transition_gone = sqlx::query_scalar!( + "SELECT COUNT(*) FROM alert_transitions WHERE transition_id = ?1", + "old-atr" + ) + .fetch_one(&ts.store().pool) + .await + .expect("read should succeed"); + assert_eq!(old_transition_gone, 0); + } + #[tokio::test] async fn enroll_rejects_expired_token() { let ts = TestStore::new().await; @@ -209,7 +327,11 @@ mod tests { .expect("nickname set should succeed"); assert!(updated); - let listed = ts.store().list_agents_with_nicknames().await; + let listed = ts + .store() + .list_agents_with_nicknames() + .await + .expect("list agents with nicknames should succeed"); assert!(listed.iter().any(|(id, name)| { id == &issued.agent_id && name.as_deref() == Some("kitchen-router") })); @@ -221,7 +343,11 @@ mod tests { .expect("nickname clear should succeed"); assert!(cleared); - let listed = ts.store().list_agents_with_nicknames().await; + let listed = ts + .store() + .list_agents_with_nicknames() + .await + .expect("list agents with nicknames should succeed"); assert!( listed .iter() diff --git a/wakey-control-plane/src/state/store/audit.rs b/wakey-control-plane/src/state/store/audit.rs index e21c6f9..f5c8791 100644 --- a/wakey-control-plane/src/state/store/audit.rs +++ b/wakey-control-plane/src/state/store/audit.rs @@ -1,6 +1,25 @@ use super::*; impl Store { + /// Best-effort audit append: logs the failure with event context instead of + /// bubbling it up. Auditing must never fail the operation that triggered it. + pub async fn audit(&self, input: AuditEventInput) { + let event_type = input.event_type.clone(); + let outcome = input.outcome.clone(); + let agent_id = input.agent_id.clone(); + let request_id = input.request_id.clone(); + if let Err(err) = self.append_audit_event(input).await { + warn!( + error = %err, + event_type = %event_type, + outcome = %outcome, + agent_id = ?agent_id, + request_id = ?request_id, + "failed to append audit event" + ); + } + } + pub async fn append_audit_event(&self, input: AuditEventInput) -> Result { let event = AuditEvent { event_id: format!("evt-{}", Uuid::new_v4()), diff --git a/wakey-control-plane/src/state/store/db.rs b/wakey-control-plane/src/state/store/db.rs index bd88690..983f741 100644 --- a/wakey-control-plane/src/state/store/db.rs +++ b/wakey-control-plane/src/state/store/db.rs @@ -154,6 +154,37 @@ impl Store { Ok(removed) } + /// Prunes audit and alert history older than the retention window. + /// + /// Runs from the daemon's periodic maintenance loop; both tables have a + /// `ts_unix` index so the deletes stay cheap as history grows. + pub async fn gc_retention(&self, older_than: Duration) -> Result { + let cutoff = i64::try_from(now_unix().saturating_sub(older_than.as_secs().max(1))) + .context("retention cutoff does not fit SQLite integer")?; + let audit_events_removed = + sqlx::query!("DELETE FROM audit_events WHERE ts_unix < ?1", cutoff) + .execute(&self.pool) + .await + .context("failed pruning old audit events")? + .rows_affected(); + let alert_transitions_removed = + sqlx::query!("DELETE FROM alert_transitions WHERE ts_unix < ?1", cutoff) + .execute(&self.pool) + .await + .context("failed pruning old alert transitions")? + .rows_affected(); + if audit_events_removed > 0 || alert_transitions_removed > 0 { + info!( + audit_events_removed, + alert_transitions_removed, "pruned state rows older than retention window" + ); + } + Ok(RetentionStats { + audit_events_removed, + alert_transitions_removed, + }) + } + async fn ensure_schema_version(&self) -> Result<()> { match sqlx::query_scalar!("SELECT value FROM meta WHERE key = ?1", SCHEMA_VERSION_KEY) .fetch_optional(&self.pool) diff --git a/wakey-control-plane/src/state/store/enrollment.rs b/wakey-control-plane/src/state/store/enrollment.rs index bbce61a..9bd3e79 100644 --- a/wakey-control-plane/src/state/store/enrollment.rs +++ b/wakey-control-plane/src/state/store/enrollment.rs @@ -17,7 +17,8 @@ impl Store { }; let now = now_unix(); - if expires_at_unix as u64 <= now { + let now_i64 = i64::try_from(now).context("current time does not fit SQLite integer")?; + if expires_at_unix <= now_i64 { sqlx::query!("DELETE FROM enroll_tokens WHERE token = ?1", enroll_token) .execute(&mut *tx) .await @@ -165,20 +166,16 @@ impl Store { } } - pub async fn list_agents(&self) -> Vec { - match sqlx::query_scalar!(r#"SELECT agent_id as "agent_id!" FROM agents ORDER BY agent_id"#) - .fetch_all(&self.pool) - .await - { - Ok(out) => out, - Err(err) => { - warn!(error = %err, "failed to list agents from state db"); - Vec::new() - } - } + pub async fn list_agents(&self) -> Result> { + let ids = + sqlx::query_scalar!(r#"SELECT agent_id as "agent_id!" FROM agents ORDER BY agent_id"#) + .fetch_all(&self.pool) + .await + .context("failed listing agents from state db")?; + Ok(ids) } - pub async fn list_agents_with_nicknames(&self) -> Vec<(String, Option)> { + pub async fn list_agents_with_nicknames(&self) -> Result)>> { let rows = sqlx::query!( r#"SELECT agents.agent_id as "agent_id!", agent_meta.nickname FROM agents @@ -186,23 +183,19 @@ impl Store { ORDER BY agents.agent_id"#, ) .fetch_all(&self.pool) - .await; + .await + .context("failed listing agents from state db")?; - match rows { - Ok(rows) => rows - .into_iter() - .map(|row| { - let nickname = row - .nickname - .map(|v| v.trim().to_string()) - .filter(|v| !v.is_empty()); - (row.agent_id, nickname) - }) - .collect(), - Err(err) => { - warn!(error = %err, "failed to list agents from state db"); - Vec::new() - } - } + let agents = rows + .into_iter() + .map(|row| { + let nickname = row + .nickname + .map(|v| v.trim().to_string()) + .filter(|v| !v.is_empty()); + (row.agent_id, nickname) + }) + .collect(); + Ok(agents) } } diff --git a/wakey-control-plane/src/state/types.rs b/wakey-control-plane/src/state/types.rs index b88387b..7cb277a 100644 --- a/wakey-control-plane/src/state/types.rs +++ b/wakey-control-plane/src/state/types.rs @@ -32,6 +32,12 @@ pub struct StateStats { pub expired_enroll_token_count: usize, } +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RetentionStats { + pub audit_events_removed: u64, + pub alert_transitions_removed: u64, +} + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct KnownDevice { pub device_id: String, diff --git a/wakey-control-plane/src/ws.rs b/wakey-control-plane/src/ws.rs index 396dcfb..3adb01f 100644 --- a/wakey-control-plane/src/ws.rs +++ b/wakey-control-plane/src/ws.rs @@ -159,9 +159,9 @@ async fn handle_agent_socket(state: AppState, socket: WebSocket) { sessions.remove(&agent_id); } drop(sessions); - if let Err(err) = state + state .store - .append_audit_event(AuditEventInput { + .audit(AuditEventInput { actor_type: "agent".into(), actor_id: Some(agent_id.clone()), agent_id: Some(agent_id), @@ -172,10 +172,7 @@ async fn handle_agent_socket(state: AppState, socket: WebSocket) { message: "agent websocket disconnected".into(), metadata: serde_json::json!({}), }) - .await - { - warn!(error = %err, "failed to append audit event for ws disconnect"); - } + .await; } writer.abort(); @@ -251,9 +248,9 @@ async fn process_agent_text( .await { warn!(agent_id = %agent_id, "agent auth rejected"); - if let Err(err) = state + state .store - .append_audit_event(AuditEventInput { + .audit(AuditEventInput { actor_type: "agent".into(), actor_id: Some(agent_id.clone()), agent_id: Some(agent_id), @@ -267,10 +264,7 @@ async fn process_agent_text( "hello_to_auth_ms": hello_to_auth_ms, }), }) - .await - { - warn!(error = %err, "failed to append audit event for auth rejection"); - } + .await; anyhow::bail!("agent auth rejected"); } state.sessions.write().await.insert( @@ -284,9 +278,9 @@ async fn process_agent_text( ); connection.authed_agent_id = Some(agent_id.clone()); info!(agent_id = %agent_id, connect_to_auth_ms, hello_to_auth_ms = hello_to_auth_ms.unwrap_or(0), "agent authenticated"); - if let Err(err) = state + state .store - .append_audit_event(AuditEventInput { + .audit(AuditEventInput { actor_type: "agent".into(), actor_id: Some(agent_id.clone()), agent_id: Some(agent_id), @@ -300,10 +294,7 @@ async fn process_agent_text( "hello_to_auth_ms": hello_to_auth_ms, }), }) - .await - { - warn!(error = %err, "failed to append audit event for auth success"); - } + .await; let _ = tx.send(SessionEvent::Message(ServerMessage::SyncDeviceSnapshot)); } IncomingClientMessage::Heartbeat { agent_id } => {