appcore_core/
audit.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: audit.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/05/31 13:38:42 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/06/07 12:00:00 by dnettoRaw
8//      ###########      S: 0.6.1
9// =============================================================================
10
11//! Bounded in-memory audit log for command dispatch outcomes.
12
13use crate::ids::{AppId, CommandName, NodeId};
14use crate::operational_journal::FileOperationalJournal;
15use crate::redact_text;
16use crate::trace::TraceContext;
17use parking_lot::Mutex;
18use std::collections::VecDeque;
19use std::sync::Arc;
20
21const MAX_AUDIT_RECORDS: usize = 10_000;
22
23/// Controlled outcome recorded for an audited operation.
24#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
25#[serde(rename_all = "snake_case")]
26pub enum AuditOutcome {
27    /// Operation completed successfully.
28    Accepted,
29    /// Operation was rejected by policy or validation.
30    Rejected,
31    /// Operation failed during execution.
32    Error,
33}
34
35/// Command-specific audit record.
36#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
37pub struct AuditRecord {
38    /// Command identity.
39    pub command_id: String,
40    /// Command name.
41    pub command_name: CommandName,
42    /// Application scope.
43    pub app_id: AppId,
44    /// Runtime node scope.
45    pub node_id: NodeId,
46    /// Command start timestamp in Unix milliseconds.
47    pub timestamp_ms: u64,
48    /// Recorded command outcome.
49    pub outcome: AuditOutcome,
50    /// Optional redacted detail.
51    pub message: Option<String>,
52    /// Optional distributed trace context.
53    pub trace: Option<TraceContext>,
54}
55
56/// Generic operational category associated with an audit entry.
57#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
58#[serde(rename_all = "snake_case")]
59pub enum AuditCategory {
60    /// Command dispatch.
61    Command,
62    /// Query dispatch.
63    Query,
64    /// Event processing.
65    Event,
66    /// Scheduler execution.
67    Scheduler,
68    /// Control-plane operation.
69    ControlPlane,
70    /// Direct peer RPC operation.
71    PeerRpc,
72    /// Runtime lifecycle or infrastructure operation.
73    Runtime,
74}
75
76/// Transport-neutral append-only operational audit entry.
77#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
78pub struct AuditEntry {
79    /// Generic operation category.
80    pub category: AuditCategory,
81    /// Stable operation identity.
82    pub operation_id: String,
83    /// Stable operation name.
84    pub operation_name: String,
85    /// Optional application scope.
86    pub app_id: Option<String>,
87    /// Optional node scope.
88    pub node_id: Option<String>,
89    /// Start timestamp in Unix milliseconds.
90    pub started_at_ms: u64,
91    /// Completion timestamp in Unix milliseconds.
92    pub completed_at_ms: u64,
93    /// Saturating elapsed time in milliseconds.
94    pub latency_ms: u64,
95    /// Recorded operation outcome.
96    pub outcome: AuditOutcome,
97    /// Optional redacted detail.
98    pub message: Option<String>,
99    /// Optional distributed trace context.
100    pub trace: Option<TraceContext>,
101}
102
103impl AuditEntry {
104    /// Creates an unscoped operational audit entry.
105    pub fn new(
106        category: AuditCategory,
107        operation_id: impl Into<String>,
108        operation_name: impl Into<String>,
109        started_at_ms: u64,
110        completed_at_ms: u64,
111        outcome: AuditOutcome,
112    ) -> Self {
113        Self {
114            category,
115            operation_id: operation_id.into(),
116            operation_name: operation_name.into(),
117            app_id: None,
118            node_id: None,
119            started_at_ms,
120            completed_at_ms,
121            latency_ms: completed_at_ms.saturating_sub(started_at_ms),
122            outcome,
123            message: None,
124            trace: None,
125        }
126    }
127
128    /// Adds application and node scope.
129    pub fn with_runtime_scope(mut self, app_id: &AppId, node_id: &NodeId) -> Self {
130        self.app_id = Some(app_id.as_str().to_string());
131        self.node_id = Some(node_id.as_str().to_string());
132        self
133    }
134
135    /// Adds a redacted optional message.
136    pub fn with_message(mut self, message: Option<String>) -> Self {
137        self.message = message.map(|value| redact_text(&value));
138        self
139    }
140
141    /// Adds distributed trace context.
142    pub fn with_trace(mut self, trace: Option<TraceContext>) -> Self {
143        self.trace = trace;
144        self
145    }
146}
147
148/// Bounded process-local audit log.
149#[derive(Debug, Default)]
150pub struct AuditLog {
151    records: Mutex<VecDeque<AuditRecord>>,
152    entries: Mutex<VecDeque<AuditEntry>>,
153    journal: Mutex<Option<Arc<FileOperationalJournal>>>,
154    journal_error: Mutex<Option<String>>,
155}
156
157impl Clone for AuditLog {
158    fn clone(&self) -> Self {
159        let guard = self.records.lock();
160        Self {
161            records: Mutex::new(guard.clone()),
162            entries: Mutex::new(self.entries.lock().clone()),
163            journal: Mutex::new(self.journal.lock().clone()),
164            journal_error: Mutex::new(self.journal_error.lock().clone()),
165        }
166    }
167}
168
169impl AuditLog {
170    /// Creates an empty audit log.
171    pub fn new() -> Self {
172        Self::default()
173    }
174
175    /// Attaches a durable journal and loads its retained audit entries.
176    pub fn attach_journal(&self, journal: Arc<FileOperationalJournal>) {
177        let mut entries = journal.audit_entries();
178        if entries.len() > MAX_AUDIT_RECORDS {
179            entries.drain(..entries.len() - MAX_AUDIT_RECORDS);
180        }
181        *self.entries.lock() = entries.into();
182        *self.journal.lock() = Some(journal);
183        *self.journal_error.lock() = None;
184    }
185
186    /// Returns the last durable journal failure, when persistence degraded.
187    pub fn durability_error(&self) -> Option<String> {
188        self.journal_error.lock().clone()
189    }
190
191    /// Appends a command record and its generic audit projection.
192    pub fn push(&self, mut record: AuditRecord) {
193        record.message = record.message.map(|message| redact_text(&message));
194        let completed_at_ms = now_ms();
195        let entry = AuditEntry::new(
196            AuditCategory::Command,
197            record.command_id.clone(),
198            record.command_name.as_str(),
199            record.timestamp_ms,
200            completed_at_ms,
201            record.outcome,
202        )
203        .with_runtime_scope(&record.app_id, &record.node_id)
204        .with_message(record.message.clone())
205        .with_trace(record.trace.clone());
206        let mut guard = self.records.lock();
207        while guard.len() >= MAX_AUDIT_RECORDS {
208            guard.pop_front();
209        }
210        guard.push_back(record);
211        drop(guard);
212        self.push_entry(entry);
213    }
214
215    /// Appends one generic audit entry after redaction.
216    pub fn push_entry(&self, mut entry: AuditEntry) {
217        entry.message = entry.message.map(|message| redact_text(&message));
218        if let Some(journal) = self.journal.lock().clone() {
219            if let Err(error) = journal.append_audit(entry.clone()) {
220                *self.journal_error.lock() = Some(redact_text(&format!("{error:?}")));
221            }
222        }
223        let mut entries = self.entries.lock();
224        while entries.len() >= MAX_AUDIT_RECORDS {
225            entries.pop_front();
226        }
227        entries.push_back(entry);
228    }
229
230    /// Returns the number of command records.
231    pub fn len(&self) -> usize {
232        self.records.lock().len()
233    }
234
235    /// Reports whether no command records exist.
236    pub fn is_empty(&self) -> bool {
237        self.records.lock().is_empty()
238    }
239
240    /// Returns a point-in-time copy of command records.
241    pub fn records(&self) -> Vec<AuditRecord> {
242        self.records.lock().iter().cloned().collect()
243    }
244
245    /// Returns a point-in-time copy of generic audit entries.
246    pub fn entries(&self) -> Vec<AuditEntry> {
247        self.entries.lock().iter().cloned().collect()
248    }
249
250    /// Exports generic entries as newline-delimited JSON.
251    pub fn export_jsonl(&self) -> Result<String, serde_json::Error> {
252        let entries = self.entries.lock();
253        let mut output = String::new();
254        for entry in entries.iter() {
255            output.push_str(&serde_json::to_string(entry)?);
256            output.push('\n');
257        }
258        Ok(output)
259    }
260}
261
262fn now_ms() -> u64 {
263    std::time::SystemTime::now()
264        .duration_since(std::time::UNIX_EPOCH)
265        .map(|duration| duration.as_millis() as u64)
266        .unwrap_or(0)
267}
268
269#[cfg(test)]
270#[path = "audit_tests.rs"]
271mod tests;
appcore_core/audit.rs

appcore_core/
audit.rs
