oximedia_workflow/
workflow_audit.rs

1//! Workflow audit trail for compliance and debugging.
2//!
3//! Provides `AuditEventType`, `WorkflowAuditEntry`, and `WorkflowAudit`
4//! for recording an immutable history of workflow lifecycle events.
5
6#![allow(dead_code)]
7
8use std::collections::HashMap;
9use std::time::{SystemTime, UNIX_EPOCH};
10
11// ---------------------------------------------------------------------------
12// AuditEventType
13// ---------------------------------------------------------------------------
14
15/// Category of audit event emitted by the workflow engine.
16#[derive(Debug, Clone, PartialEq, Eq, Hash)]
17pub enum AuditEventType {
18    /// Workflow was submitted for execution.
19    WorkflowSubmitted,
20    /// Workflow moved to a different execution state.
21    StateChanged,
22    /// A task within the workflow changed state.
23    TaskStateChanged,
24    /// A user-supplied parameter was updated.
25    ParameterUpdated,
26    /// Workflow was cancelled by an operator.
27    WorkflowCancelled,
28    /// An error occurred during execution.
29    ErrorOccurred,
30    /// Workflow completed (success or failure).
31    WorkflowCompleted,
32    /// A retry was attempted.
33    RetryAttempted,
34}
35
36impl AuditEventType {
37    /// Returns `true` for event types that represent a state transition.
38    #[must_use]
39    pub fn is_state_change(&self) -> bool {
40        matches!(
41            self,
42            Self::StateChanged
43                | Self::TaskStateChanged
44                | Self::WorkflowCancelled
45                | Self::WorkflowCompleted
46        )
47    }
48
49    /// Short string identifier for the event type.
50    #[must_use]
51    pub fn as_str(&self) -> &'static str {
52        match self {
53            Self::WorkflowSubmitted => "workflow_submitted",
54            Self::StateChanged => "state_changed",
55            Self::TaskStateChanged => "task_state_changed",
56            Self::ParameterUpdated => "parameter_updated",
57            Self::WorkflowCancelled => "workflow_cancelled",
58            Self::ErrorOccurred => "error_occurred",
59            Self::WorkflowCompleted => "workflow_completed",
60            Self::RetryAttempted => "retry_attempted",
61        }
62    }
63}
64
65// ---------------------------------------------------------------------------
66// WorkflowAuditEntry
67// ---------------------------------------------------------------------------
68
69/// A single entry in the workflow audit trail.
70#[derive(Debug, Clone)]
71pub struct WorkflowAuditEntry {
72    /// Workflow run this entry belongs to.
73    pub run_id: String,
74    /// Type of event recorded.
75    pub event_type: AuditEventType,
76    /// Unix timestamp (seconds since epoch) when the event occurred.
77    pub timestamp_secs: u64,
78    /// Optional actor (user or service) that triggered the event.
79    pub actor: Option<String>,
80    /// Raw context payload (e.g. serialised diff).
81    pub context: Option<String>,
82}
83
84impl WorkflowAuditEntry {
85    /// Create a new entry with the current wall-clock time.
86    #[must_use]
87    pub fn new(run_id: impl Into<String>, event_type: AuditEventType) -> Self {
88        let timestamp_secs = SystemTime::now()
89            .duration_since(UNIX_EPOCH)
90            .map(|d| d.as_secs())
91            .unwrap_or(0);
92        Self {
93            run_id: run_id.into(),
94            event_type,
95            timestamp_secs,
96            actor: None,
97            context: None,
98        }
99    }
100
101    /// Create an entry with an explicit timestamp (useful in tests).
102    #[must_use]
103    pub fn with_timestamp(
104        run_id: impl Into<String>,
105        event_type: AuditEventType,
106        timestamp_secs: u64,
107    ) -> Self {
108        Self {
109            run_id: run_id.into(),
110            event_type,
111            timestamp_secs,
112            actor: None,
113            context: None,
114        }
115    }
116
117    /// Attach an actor label.
118    #[must_use]
119    pub fn with_actor(mut self, actor: impl Into<String>) -> Self {
120        self.actor = Some(actor.into());
121        self
122    }
123
124    /// Attach a context payload.
125    #[must_use]
126    pub fn with_context(mut self, ctx: impl Into<String>) -> Self {
127        self.context = Some(ctx.into());
128        self
129    }
130
131    /// Human-readable description of this entry.
132    #[must_use]
133    pub fn description(&self) -> String {
134        format!(
135            "[{}] run={} event={}{}",
136            self.timestamp_secs,
137            self.run_id,
138            self.event_type.as_str(),
139            self.actor
140                .as_ref()
141                .map(|a| format!(" actor={a}"))
142                .unwrap_or_default(),
143        )
144    }
145}
146
147// ---------------------------------------------------------------------------
148// WorkflowAudit
149// ---------------------------------------------------------------------------
150
151/// Append-only audit log for all workflow runs.
152#[derive(Debug, Default)]
153pub struct WorkflowAudit {
154    /// Entries indexed by run ID for fast per-run queries.
155    index: HashMap<String, Vec<usize>>,
156    /// Flat list of all entries in insertion order.
157    all: Vec<WorkflowAuditEntry>,
158}
159
160impl WorkflowAudit {
161    /// Create an empty audit log.
162    #[must_use]
163    pub fn new() -> Self {
164        Self::default()
165    }
166
167    /// Append an entry to the audit log.
168    pub fn record(&mut self, entry: WorkflowAuditEntry) {
169        let idx = self.all.len();
170        self.index
171            .entry(entry.run_id.clone())
172            .or_default()
173            .push(idx);
174        self.all.push(entry);
175    }
176
177    /// Return all entries for a specific run, in insertion order.
178    #[must_use]
179    pub fn entries_for_run(&self, run_id: &str) -> Vec<&WorkflowAuditEntry> {
180        self.index
181            .get(run_id)
182            .map(|idxs| idxs.iter().map(|&i| &self.all[i]).collect())
183            .unwrap_or_default()
184    }
185
186    /// Return the most recent `n` entries across all runs.
187    #[must_use]
188    pub fn recent(&self, n: usize) -> Vec<&WorkflowAuditEntry> {
189        self.all.iter().rev().take(n).collect()
190    }
191
192    /// Total number of entries across all runs.
193    #[must_use]
194    pub fn total_entries(&self) -> usize {
195        self.all.len()
196    }
197
198    /// All entries (insertion order).
199    #[must_use]
200    pub fn all_entries(&self) -> &[WorkflowAuditEntry] {
201        &self.all
202    }
203
204    /// Entries filtered by event type.
205    #[must_use]
206    pub fn by_event_type(&self, et: &AuditEventType) -> Vec<&WorkflowAuditEntry> {
207        self.all.iter().filter(|e| &e.event_type == et).collect()
208    }
209}
210
211// ---------------------------------------------------------------------------
212// Tests
213// ---------------------------------------------------------------------------
214
215#[cfg(test)]
216mod tests {
217    use super::*;
218
219    #[test]
220    fn test_event_type_is_state_change_true() {
221        assert!(AuditEventType::StateChanged.is_state_change());
222        assert!(AuditEventType::TaskStateChanged.is_state_change());
223        assert!(AuditEventType::WorkflowCancelled.is_state_change());
224        assert!(AuditEventType::WorkflowCompleted.is_state_change());
225    }
226
227    #[test]
228    fn test_event_type_is_state_change_false() {
229        assert!(!AuditEventType::WorkflowSubmitted.is_state_change());
230        assert!(!AuditEventType::ParameterUpdated.is_state_change());
231        assert!(!AuditEventType::ErrorOccurred.is_state_change());
232        assert!(!AuditEventType::RetryAttempted.is_state_change());
233    }
234
235    #[test]
236    fn test_event_type_as_str() {
237        assert_eq!(
238            AuditEventType::WorkflowSubmitted.as_str(),
239            "workflow_submitted"
240        );
241        assert_eq!(AuditEventType::StateChanged.as_str(), "state_changed");
242        assert_eq!(AuditEventType::ErrorOccurred.as_str(), "error_occurred");
243    }
244
245    #[test]
246    fn test_entry_description_no_actor() {
247        let e = WorkflowAuditEntry::with_timestamp("run1", AuditEventType::StateChanged, 1000);
248        let desc = e.description();
249        assert!(desc.contains("run1"));
250        assert!(desc.contains("state_changed"));
251        assert!(desc.contains("1000"));
252    }
253
254    #[test]
255    fn test_entry_description_with_actor() {
256        let e = WorkflowAuditEntry::with_timestamp("run2", AuditEventType::StateChanged, 2000)
257            .with_actor("operator");
258        let desc = e.description();
259        assert!(desc.contains("actor=operator"));
260    }
261
262    #[test]
263    fn test_entry_with_context() {
264        let e = WorkflowAuditEntry::new("r", AuditEventType::ParameterUpdated)
265            .with_context(r#"{"key":"val"}"#);
266        assert!(e.context.is_some());
267    }
268
269    #[test]
270    fn test_audit_record_and_total() {
271        let mut audit = WorkflowAudit::new();
272        audit.record(WorkflowAuditEntry::new(
273            "r1",
274            AuditEventType::WorkflowSubmitted,
275        ));
276        audit.record(WorkflowAuditEntry::new("r1", AuditEventType::StateChanged));
277        audit.record(WorkflowAuditEntry::new(
278            "r2",
279            AuditEventType::WorkflowSubmitted,
280        ));
281        assert_eq!(audit.total_entries(), 3);
282    }
283
284    #[test]
285    fn test_entries_for_run() {
286        let mut audit = WorkflowAudit::new();
287        audit.record(WorkflowAuditEntry::new(
288            "run1",
289            AuditEventType::WorkflowSubmitted,
290        ));
291        audit.record(WorkflowAuditEntry::new(
292            "run2",
293            AuditEventType::WorkflowSubmitted,
294        ));
295        audit.record(WorkflowAuditEntry::new(
296            "run1",
297            AuditEventType::WorkflowCompleted,
298        ));
299        let entries = audit.entries_for_run("run1");
300        assert_eq!(entries.len(), 2);
301    }
302
303    #[test]
304    fn test_entries_for_unknown_run() {
305        let audit = WorkflowAudit::new();
306        assert!(audit.entries_for_run("ghost").is_empty());
307    }
308
309    #[test]
310    fn test_recent_entries() {
311        let mut audit = WorkflowAudit::new();
312        for i in 0..5u64 {
313            audit.record(WorkflowAuditEntry::with_timestamp(
314                format!("run{i}"),
315                AuditEventType::WorkflowSubmitted,
316                i,
317            ));
318        }
319        let recent = audit.recent(3);
320        assert_eq!(recent.len(), 3);
321        // Most recent first (run4 has timestamp 4)
322        assert_eq!(recent[0].timestamp_secs, 4);
323    }
324
325    #[test]
326    fn test_by_event_type() {
327        let mut audit = WorkflowAudit::new();
328        audit.record(WorkflowAuditEntry::new("r", AuditEventType::ErrorOccurred));
329        audit.record(WorkflowAuditEntry::new("r", AuditEventType::StateChanged));
330        audit.record(WorkflowAuditEntry::new("r", AuditEventType::ErrorOccurred));
331        let errors = audit.by_event_type(&AuditEventType::ErrorOccurred);
332        assert_eq!(errors.len(), 2);
333    }
334
335    #[test]
336    fn test_all_entries_insertion_order() {
337        let mut audit = WorkflowAudit::new();
338        audit.record(WorkflowAuditEntry::with_timestamp(
339            "r",
340            AuditEventType::WorkflowSubmitted,
341            1,
342        ));
343        audit.record(WorkflowAuditEntry::with_timestamp(
344            "r",
345            AuditEventType::WorkflowCompleted,
346            2,
347        ));
348        let all = audit.all_entries();
349        assert_eq!(all[0].timestamp_secs, 1);
350        assert_eq!(all[1].timestamp_secs, 2);
351    }
352}
oximedia_workflow/workflow_audit.rs

oximedia_workflow/
workflow_audit.rs
