ironflow_engine/notify/
audit_log.rs

1//! [`AuditLogSubscriber`] -- persists every event to an [`AuditLogStore`].
2
3use std::str::FromStr;
4use std::sync::Arc;
5
6use serde_json::to_value;
7use tracing::error;
8
9use ironflow_store::audit_log_store::AuditLogStore;
10use ironflow_store::entities::{EventKind, NewAuditLogEntry};
11
12use super::{Event, EventSubscriber, SubscriberFuture};
13
14/// Subscriber that persists every received event as an audit log entry.
15///
16/// Extracts contextual IDs (run, step, user) from the event payload
17/// for efficient filtering, and stores the full event as JSON.
18///
19/// # Examples
20///
21/// ```no_run
22/// use std::sync::Arc;
23/// use ironflow_engine::notify::{AuditLogSubscriber, Event, EventPublisher};
24/// use ironflow_store::memory::InMemoryStore;
25///
26/// let store = Arc::new(InMemoryStore::new());
27/// let mut publisher = EventPublisher::new();
28/// publisher.subscribe(
29///     AuditLogSubscriber::new(store),
30///     Event::ALL,
31/// );
32/// ```
33pub struct AuditLogSubscriber {
34    store: Arc<dyn AuditLogStore>,
35}
36
37impl AuditLogSubscriber {
38    /// Create a new subscriber backed by the given store.
39    ///
40    /// # Examples
41    ///
42    /// ```
43    /// use std::sync::Arc;
44    /// use ironflow_engine::notify::AuditLogSubscriber;
45    /// use ironflow_store::memory::InMemoryStore;
46    ///
47    /// let store = Arc::new(InMemoryStore::new());
48    /// let subscriber = AuditLogSubscriber::new(store);
49    /// ```
50    pub fn new(store: Arc<dyn AuditLogStore>) -> Self {
51        Self { store }
52    }
53}
54
55impl EventSubscriber for AuditLogSubscriber {
56    fn name(&self) -> &str {
57        "audit_log"
58    }
59
60    fn handle<'a>(&'a self, event: &'a Event) -> SubscriberFuture<'a> {
61        Box::pin(async move {
62            let event_kind = match EventKind::from_str(event.event_type()) {
63                Ok(k) => k,
64                Err(e) => {
65                    error!(error = %e, event_type = event.event_type(), "unknown event kind for audit log");
66                    return;
67                }
68            };
69
70            let payload = match to_value(event) {
71                Ok(v) => v,
72                Err(e) => {
73                    error!(error = %e, event_type = event.event_type(), "failed to serialize event for audit log");
74                    return;
75                }
76            };
77
78            let entry = NewAuditLogEntry {
79                event_type: event_kind,
80                payload,
81                run_id: event.run_id(),
82                step_id: event.step_id(),
83                user_id: event.user_id(),
84            };
85
86            if let Err(e) = self.store.append_audit_log(entry).await {
87                error!(error = %e, event_type = event.event_type(), "failed to persist audit log entry");
88            }
89        })
90    }
91}
92
93#[cfg(test)]
94mod tests {
95    use std::collections::HashMap;
96    use std::sync::Arc;
97    use std::time::Duration;
98
99    use chrono::Utc;
100    use rust_decimal::Decimal;
101    use uuid::Uuid;
102
103    use ironflow_store::audit_log_store::AuditLogStore;
104    use ironflow_store::entities::{AuditLogEntry, AuditLogFilter, EventKind};
105    use ironflow_store::memory::InMemoryStore;
106    use ironflow_store::models::{RunStatus, StepKind};
107
108    use super::*;
109    use crate::notify::{
110        ApprovalGrantedEvent, ApprovalRejectedEvent, EventPublisher, EventSubscriber,
111        RunFailedEvent, RunStatusChangedEvent, StepFailedEvent, UserSignedInEvent,
112        WorkflowPausedEvent, WorkflowResumedEvent,
113    };
114
115    fn sample_run_status_changed() -> Event {
116        Event::RunStatusChanged(RunStatusChangedEvent {
117            run_id: Uuid::now_v7(),
118            workflow_name: "deploy".to_string(),
119            from: RunStatus::Running,
120            to: RunStatus::Completed,
121            error: None,
122            cost_usd: Decimal::new(42, 2),
123            duration_ms: 5000,
124            labels: HashMap::new(),
125            at: Utc::now(),
126        })
127    }
128
129    fn sample_user_signed_in() -> Event {
130        Event::UserSignedIn(UserSignedInEvent {
131            user_id: Uuid::now_v7(),
132            username: "alice".to_string(),
133            at: Utc::now(),
134        })
135    }
136
137    fn sample_step_failed() -> Event {
138        Event::StepFailed(StepFailedEvent {
139            run_id: Uuid::now_v7(),
140            step_id: Uuid::now_v7(),
141            step_name: "build".to_string(),
142            kind: StepKind::Shell,
143            error: "exit code 1".to_string(),
144            at: Utc::now(),
145        })
146    }
147
148    #[test]
149    fn name_is_audit_log() {
150        let store = Arc::new(InMemoryStore::new());
151        let subscriber = AuditLogSubscriber::new(store);
152        assert_eq!(subscriber.name(), "audit_log");
153    }
154
155    #[test]
156    fn run_id_from_run_event() {
157        let event = sample_run_status_changed();
158        assert!(event.run_id().is_some());
159    }
160
161    #[test]
162    fn run_id_from_user_event_is_none() {
163        let event = sample_user_signed_in();
164        assert!(event.run_id().is_none());
165    }
166
167    #[test]
168    fn step_id_from_step_event() {
169        let event = sample_step_failed();
170        assert!(event.step_id().is_some());
171    }
172
173    #[test]
174    fn step_id_from_run_event_is_none() {
175        let event = sample_run_status_changed();
176        assert!(event.step_id().is_none());
177    }
178
179    #[test]
180    fn user_id_from_user_event() {
181        let event = sample_user_signed_in();
182        assert!(event.user_id().is_some());
183    }
184
185    #[test]
186    fn user_id_from_run_event_is_none() {
187        let event = sample_run_status_changed();
188        assert!(event.user_id().is_none());
189    }
190
191    #[tokio::test]
192    async fn handle_persists_event() {
193        let store = Arc::new(InMemoryStore::new());
194        let subscriber = AuditLogSubscriber::new(store.clone());
195
196        let event = sample_run_status_changed();
197        subscriber.handle(&event).await;
198
199        let page = store
200            .list_audit_logs(AuditLogFilter::default(), 1, 20)
201            .await
202            .unwrap();
203
204        assert_eq!(page.items.len(), 1);
205        assert_eq!(page.items[0].event_type, EventKind::RunStatusChanged);
206        assert!(page.items[0].run_id.is_some());
207        assert!(page.items[0].step_id.is_none());
208        assert!(page.items[0].user_id.is_none());
209    }
210
211    #[tokio::test]
212    async fn handle_persists_step_event_with_ids() {
213        let store = Arc::new(InMemoryStore::new());
214        let subscriber = AuditLogSubscriber::new(store.clone());
215
216        let event = sample_step_failed();
217        subscriber.handle(&event).await;
218
219        let page = store
220            .list_audit_logs(AuditLogFilter::default(), 1, 20)
221            .await
222            .unwrap();
223
224        assert_eq!(page.items.len(), 1);
225        assert_eq!(page.items[0].event_type, EventKind::StepFailed);
226        assert!(page.items[0].run_id.is_some());
227        assert!(page.items[0].step_id.is_some());
228    }
229
230    #[tokio::test]
231    async fn handle_persists_user_event_with_user_id() {
232        let store = Arc::new(InMemoryStore::new());
233        let subscriber = AuditLogSubscriber::new(store.clone());
234
235        let event = sample_user_signed_in();
236        subscriber.handle(&event).await;
237
238        let page = store
239            .list_audit_logs(AuditLogFilter::default(), 1, 20)
240            .await
241            .unwrap();
242
243        assert_eq!(page.items.len(), 1);
244        assert_eq!(page.items[0].event_type, EventKind::UserSignedIn);
245        assert!(page.items[0].user_id.is_some());
246        assert!(page.items[0].run_id.is_none());
247    }
248
249    #[tokio::test]
250    async fn publisher_dispatches_to_audit_log_subscriber() {
251        let store = Arc::new(InMemoryStore::new());
252        let mut publisher = EventPublisher::new();
253
254        publisher.subscribe(AuditLogSubscriber::new(store.clone()), Event::ALL);
255
256        publisher.publish(sample_run_status_changed());
257        publisher.publish(sample_user_signed_in());
258        publisher.publish(sample_step_failed());
259
260        tokio::time::sleep(Duration::from_millis(100)).await;
261
262        let page = store
263            .list_audit_logs(AuditLogFilter::default(), 1, 20)
264            .await
265            .unwrap();
266
267        assert_eq!(page.items.len(), 3);
268    }
269
270    #[tokio::test]
271    async fn full_event_payload_is_preserved() {
272        let store = Arc::new(InMemoryStore::new());
273        let subscriber = AuditLogSubscriber::new(store.clone());
274
275        let run_id = Uuid::now_v7();
276        let event = Event::RunFailed(RunFailedEvent {
277            run_id,
278            workflow_name: "deploy".to_string(),
279            error: Some("step crashed".to_string()),
280            cost_usd: Decimal::new(10, 2),
281            duration_ms: 3000,
282            labels: HashMap::new(),
283            at: Utc::now(),
284        });
285        subscriber.handle(&event).await;
286
287        let page = store
288            .list_audit_logs(AuditLogFilter::default(), 1, 20)
289            .await
290            .unwrap();
291
292        let payload = &page.items[0].payload;
293        assert_eq!(payload["type"], "run_failed");
294        assert_eq!(payload["workflow_name"], "deploy");
295        assert_eq!(payload["error"], "step crashed");
296    }
297
298    /// Persist `event` and return the single entry it produced.
299    async fn persist_one(event: &Event) -> AuditLogEntry {
300        let store = Arc::new(InMemoryStore::new());
301        let subscriber = AuditLogSubscriber::new(store.clone());
302        subscriber.handle(event).await;
303
304        let mut page = store
305            .list_audit_logs(AuditLogFilter::default(), 1, 20)
306            .await
307            .unwrap();
308        assert_eq!(page.items.len(), 1);
309        page.items.remove(0)
310    }
311
312    #[tokio::test]
313    async fn audit_persists_approval_granted_with_user_id() {
314        let user_id = Uuid::now_v7();
315        let run_id = Uuid::now_v7();
316        let entry = persist_one(&Event::ApprovalGranted(ApprovalGrantedEvent {
317            run_id,
318            step_id: Some(Uuid::now_v7()),
319            approved_by: "ci-key".to_string(),
320            user_id: Some(user_id),
321            approvals_received: 1,
322            approvals_required: 1,
323            requirement: None,
324            at: Utc::now(),
325        }))
326        .await;
327
328        assert_eq!(entry.event_type, EventKind::ApprovalGranted);
329        assert_eq!(entry.user_id, Some(user_id));
330        assert_eq!(entry.run_id, Some(run_id));
331        assert_eq!(entry.payload["user_id"], user_id.to_string());
332        assert_eq!(entry.payload["approved_by"], "ci-key");
333    }
334
335    #[tokio::test]
336    async fn audit_persists_approval_rejected_with_user_id() {
337        let user_id = Uuid::now_v7();
338        let entry = persist_one(&Event::ApprovalRejected(ApprovalRejectedEvent {
339            run_id: Uuid::now_v7(),
340            step_id: Some(Uuid::now_v7()),
341            rejected_by: "bob".to_string(),
342            user_id: Some(user_id),
343            reason: Some("wrong environment".to_string()),
344            requirement: None,
345            at: Utc::now(),
346        }))
347        .await;
348
349        assert_eq!(entry.event_type, EventKind::ApprovalRejected);
350        assert_eq!(entry.user_id, Some(user_id));
351        assert_eq!(entry.payload["reason"], "wrong environment");
352    }
353
354    #[tokio::test]
355    async fn audit_persists_workflow_paused_with_user_id() {
356        let user_id = Uuid::now_v7();
357        let entry = persist_one(&Event::WorkflowPaused(WorkflowPausedEvent {
358            workflow_name: "deploy".to_string(),
359            user_id: Some(user_id),
360            at: Utc::now(),
361        }))
362        .await;
363
364        assert_eq!(entry.event_type, EventKind::WorkflowPaused);
365        assert_eq!(entry.user_id, Some(user_id));
366        assert!(entry.run_id.is_none());
367        assert_eq!(entry.payload["workflow_name"], "deploy");
368    }
369
370    #[tokio::test]
371    async fn audit_persists_workflow_resumed_with_user_id() {
372        let user_id = Uuid::now_v7();
373        let entry = persist_one(&Event::WorkflowResumed(WorkflowResumedEvent {
374            workflow_name: "deploy".to_string(),
375            user_id: Some(user_id),
376            at: Utc::now(),
377        }))
378        .await;
379
380        assert_eq!(entry.event_type, EventKind::WorkflowResumed);
381        assert_eq!(entry.user_id, Some(user_id));
382        assert_eq!(entry.payload["workflow_name"], "deploy");
383    }
384
385    #[tokio::test]
386    async fn audit_system_decision_has_no_user_id() {
387        let entry = persist_one(&Event::ApprovalRejected(ApprovalRejectedEvent {
388            run_id: Uuid::now_v7(),
389            step_id: Some(Uuid::now_v7()),
390            rejected_by: "system:timeout".to_string(),
391            user_id: None,
392            reason: None,
393            requirement: None,
394            at: Utc::now(),
395        }))
396        .await;
397
398        assert_eq!(entry.event_type, EventKind::ApprovalRejected);
399        assert!(entry.user_id.is_none());
400        assert_eq!(entry.payload["rejected_by"], "system:timeout");
401    }
402}
ironflow_engine/notify/audit_log.rs

ironflow_engine/notify/
audit_log.rs
