ironflow_engine/notify/
event.rs

1//! Domain events emitted throughout the ironflow lifecycle.
2
3use chrono::{DateTime, Utc};
4use rust_decimal::Decimal;
5use serde::{Deserialize, Serialize};
6use uuid::Uuid;
7
8use ironflow_store::models::{RunStatus, StepKind};
9
10/// Output stream for a [`LogLine`](Event::LogLine) event.
11///
12/// # Examples
13///
14/// ```
15/// use ironflow_engine::notify::LogStream;
16///
17/// let stream: LogStream = "stdout".parse().unwrap();
18/// assert_eq!(stream.as_str(), "stdout");
19/// ```
20#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
21#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
22#[serde(rename_all = "snake_case")]
23pub enum LogStream {
24    /// Standard output.
25    Stdout,
26    /// Standard error.
27    Stderr,
28    /// System-level messages (e.g. step start/stop notifications).
29    System,
30}
31
32impl LogStream {
33    /// Returns the wire-format string for this stream.
34    pub fn as_str(&self) -> &'static str {
35        match self {
36            Self::Stdout => "stdout",
37            Self::Stderr => "stderr",
38            Self::System => "system",
39        }
40    }
41}
42
43impl std::fmt::Display for LogStream {
44    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
45        f.write_str(self.as_str())
46    }
47}
48
49impl std::str::FromStr for LogStream {
50    type Err = String;
51
52    fn from_str(s: &str) -> Result<Self, Self::Err> {
53        match s {
54            "stdout" => Ok(Self::Stdout),
55            "stderr" => Ok(Self::Stderr),
56            "system" => Ok(Self::System),
57            _ => Err(format!("unknown log stream: {s}")),
58        }
59    }
60}
61
62/// A domain event emitted by the ironflow system.
63///
64/// Covers the full lifecycle: runs, steps, approvals, and authentication.
65/// Subscribers receive these via [`EventPublisher`](super::EventPublisher)
66/// and pattern-match on the variants they care about.
67///
68/// # Examples
69///
70/// ```
71/// use ironflow_engine::notify::Event;
72/// use ironflow_store::models::RunStatus;
73/// use uuid::Uuid;
74///
75/// let event = Event::RunStatusChanged {
76///     run_id: Uuid::now_v7(),
77///     workflow_name: "deploy".to_string(),
78///     from: RunStatus::Running,
79///     to: RunStatus::Completed,
80///     error: None,
81///     cost_usd: rust_decimal::Decimal::ZERO,
82///     duration_ms: 5000,
83///     at: chrono::Utc::now(),
84/// };
85/// ```
86#[derive(Debug, Clone, Serialize, Deserialize)]
87#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
88#[serde(tag = "type", rename_all = "snake_case")]
89pub enum Event {
90    // -- Run lifecycle --
91    /// A new run was created (status: Pending).
92    RunCreated {
93        /// Run identifier.
94        run_id: Uuid,
95        /// Workflow name.
96        workflow_name: String,
97        /// When the run was created.
98        at: DateTime<Utc>,
99    },
100
101    /// A run changed status.
102    RunStatusChanged {
103        /// Run identifier.
104        run_id: Uuid,
105        /// Workflow name.
106        workflow_name: String,
107        /// Previous status.
108        from: RunStatus,
109        /// New status.
110        to: RunStatus,
111        /// Error message (when transitioning to Failed).
112        error: Option<String>,
113        /// Aggregated cost in USD at the time of transition.
114        cost_usd: Decimal,
115        /// Aggregated duration in milliseconds at the time of transition.
116        duration_ms: u64,
117        /// When the transition occurred.
118        at: DateTime<Utc>,
119    },
120
121    /// A run transitioned to [`Failed`](ironflow_store::models::RunStatus::Failed).
122    ///
123    /// This is a convenience event emitted alongside [`RunStatusChanged`](Event::RunStatusChanged)
124    /// when the target status is `Failed`. Subscribe to this instead of
125    /// `RUN_STATUS_CHANGED` when you only care about failures.
126    RunFailed {
127        /// Run identifier.
128        run_id: Uuid,
129        /// Workflow name.
130        workflow_name: String,
131        /// Error message.
132        error: Option<String>,
133        /// Aggregated cost in USD at the time of failure.
134        cost_usd: Decimal,
135        /// Aggregated duration in milliseconds at the time of failure.
136        duration_ms: u64,
137        /// When the failure occurred.
138        at: DateTime<Utc>,
139    },
140
141    /// A run was stopped because it reached its cumulative cost cap.
142    ///
143    /// Emitted when the engine refuses an agent step that would cross the run's
144    /// `max_cost_usd`. The run transitions to
145    /// [`Cancelled`](ironflow_store::models::RunStatus::Cancelled) and the step
146    /// is never launched, so the reported spend is what the run had already
147    /// consumed.
148    RunBudgetExceeded {
149        /// Run identifier.
150        run_id: Uuid,
151        /// Workflow name.
152        workflow_name: String,
153        /// The configured cost cap in USD.
154        limit_usd: Decimal,
155        /// Cost already consumed when the cap was reached, in USD.
156        spent_usd: Decimal,
157        /// Declared budget of the refused step, in USD.
158        step_budget_usd: Decimal,
159        /// When the refusal occurred.
160        at: DateTime<Utc>,
161    },
162
163    // -- Step lifecycle --
164    /// A step completed successfully.
165    StepCompleted {
166        /// Run identifier.
167        run_id: Uuid,
168        /// Step identifier.
169        step_id: Uuid,
170        /// Human-readable step name.
171        step_name: String,
172        /// Step operation kind.
173        #[cfg_attr(feature = "openapi", schema(value_type = String))]
174        kind: StepKind,
175        /// Step duration in milliseconds.
176        duration_ms: u64,
177        /// Step cost in USD.
178        cost_usd: Decimal,
179        /// When the step completed.
180        at: DateTime<Utc>,
181    },
182
183    /// A step failed.
184    StepFailed {
185        /// Run identifier.
186        run_id: Uuid,
187        /// Step identifier.
188        step_id: Uuid,
189        /// Human-readable step name.
190        step_name: String,
191        /// Step operation kind.
192        #[cfg_attr(feature = "openapi", schema(value_type = String))]
193        kind: StepKind,
194        /// Error message.
195        error: String,
196        /// When the step failed.
197        at: DateTime<Utc>,
198    },
199
200    // -- Approval --
201    /// A run is waiting for human approval.
202    ApprovalRequested {
203        /// Run identifier.
204        run_id: Uuid,
205        /// Approval step identifier.
206        step_id: Uuid,
207        /// Message displayed to reviewers.
208        message: String,
209        /// When the approval was requested.
210        at: DateTime<Utc>,
211    },
212
213    /// A run was approved by a human.
214    ApprovalGranted {
215        /// Run identifier.
216        run_id: Uuid,
217        /// User who approved (ID or username).
218        approved_by: String,
219        /// When the approval was granted.
220        at: DateTime<Utc>,
221    },
222
223    /// A run was rejected by a human.
224    ApprovalRejected {
225        /// Run identifier.
226        run_id: Uuid,
227        /// User who rejected (ID or username).
228        rejected_by: String,
229        /// When the rejection occurred.
230        at: DateTime<Utc>,
231    },
232
233    // -- Log streaming --
234    /// A log line emitted during step execution.
235    ///
236    /// Pushed by the worker in real time so that SSE clients can stream
237    /// step output as it happens, without waiting for step completion.
238    LogLine {
239        /// Run identifier.
240        run_id: Uuid,
241        /// Step identifier.
242        step_id: Uuid,
243        /// Human-readable step name.
244        step_name: String,
245        /// Output stream.
246        stream: LogStream,
247        /// The log line content.
248        line: String,
249        /// When the line was emitted.
250        at: DateTime<Utc>,
251    },
252
253    // -- Authentication --
254    /// A user signed in.
255    UserSignedIn {
256        /// User identifier.
257        user_id: Uuid,
258        /// Username.
259        username: String,
260        /// When the sign-in occurred.
261        at: DateTime<Utc>,
262    },
263
264    /// A new user signed up.
265    UserSignedUp {
266        /// User identifier.
267        user_id: Uuid,
268        /// Username.
269        username: String,
270        /// When the sign-up occurred.
271        at: DateTime<Utc>,
272    },
273
274    /// A user signed out.
275    UserSignedOut {
276        /// User identifier.
277        user_id: Uuid,
278        /// When the sign-out occurred.
279        at: DateTime<Utc>,
280    },
281}
282
283impl Event {
284    /// Event type constant for [`RunCreated`](Event::RunCreated).
285    pub const RUN_CREATED: &'static str = "run_created";
286    /// Event type constant for [`RunStatusChanged`](Event::RunStatusChanged).
287    pub const RUN_STATUS_CHANGED: &'static str = "run_status_changed";
288    /// Event type constant for [`RunFailed`](Event::RunFailed).
289    pub const RUN_FAILED: &'static str = "run_failed";
290    /// Event type constant for [`RunBudgetExceeded`](Event::RunBudgetExceeded).
291    pub const RUN_BUDGET_EXCEEDED: &'static str = "run_budget_exceeded";
292    /// Event type constant for [`StepCompleted`](Event::StepCompleted).
293    pub const STEP_COMPLETED: &'static str = "step_completed";
294    /// Event type constant for [`StepFailed`](Event::StepFailed).
295    pub const STEP_FAILED: &'static str = "step_failed";
296    /// Event type constant for [`ApprovalRequested`](Event::ApprovalRequested).
297    pub const APPROVAL_REQUESTED: &'static str = "approval_requested";
298    /// Event type constant for [`ApprovalGranted`](Event::ApprovalGranted).
299    pub const APPROVAL_GRANTED: &'static str = "approval_granted";
300    /// Event type constant for [`ApprovalRejected`](Event::ApprovalRejected).
301    pub const APPROVAL_REJECTED: &'static str = "approval_rejected";
302    /// Event type constant for [`LogLine`](Event::LogLine).
303    pub const LOG_LINE: &'static str = "log_line";
304    /// Event type constant for [`UserSignedIn`](Event::UserSignedIn).
305    pub const USER_SIGNED_IN: &'static str = "user_signed_in";
306    /// Event type constant for [`UserSignedUp`](Event::UserSignedUp).
307    pub const USER_SIGNED_UP: &'static str = "user_signed_up";
308    /// Event type constant for [`UserSignedOut`](Event::UserSignedOut).
309    pub const USER_SIGNED_OUT: &'static str = "user_signed_out";
310
311    /// All event types. Pass this to
312    /// [`EventPublisher::subscribe`](super::EventPublisher::subscribe) to
313    /// receive every event.
314    ///
315    /// # Examples
316    ///
317    /// ```no_run
318    /// use ironflow_engine::notify::{Event, EventPublisher, WebhookSubscriber};
319    ///
320    /// let mut publisher = EventPublisher::new();
321    /// publisher.subscribe(
322    ///     WebhookSubscriber::new("https://example.com/all"),
323    ///     Event::ALL,
324    /// );
325    /// ```
326    pub const ALL: &'static [&'static str] = &[
327        Self::RUN_CREATED,
328        Self::RUN_STATUS_CHANGED,
329        Self::RUN_FAILED,
330        Self::RUN_BUDGET_EXCEEDED,
331        Self::STEP_COMPLETED,
332        Self::STEP_FAILED,
333        Self::APPROVAL_REQUESTED,
334        Self::APPROVAL_GRANTED,
335        Self::APPROVAL_REJECTED,
336        Self::LOG_LINE,
337        Self::USER_SIGNED_IN,
338        Self::USER_SIGNED_UP,
339        Self::USER_SIGNED_OUT,
340    ];
341
342    /// Returns the event type as a static string (e.g. `"run_status_changed"`).
343    ///
344    /// Useful for filtering and logging without deserializing.
345    ///
346    /// # Examples
347    ///
348    /// ```
349    /// use ironflow_engine::notify::Event;
350    /// use uuid::Uuid;
351    /// use chrono::Utc;
352    ///
353    /// let event = Event::UserSignedIn {
354    ///     user_id: Uuid::now_v7(),
355    ///     username: "alice".to_string(),
356    ///     at: Utc::now(),
357    /// };
358    /// assert_eq!(event.event_type(), "user_signed_in");
359    /// ```
360    pub fn event_type(&self) -> &'static str {
361        match self {
362            Event::RunCreated { .. } => Self::RUN_CREATED,
363            Event::RunStatusChanged { .. } => Self::RUN_STATUS_CHANGED,
364            Event::RunFailed { .. } => Self::RUN_FAILED,
365            Event::RunBudgetExceeded { .. } => Self::RUN_BUDGET_EXCEEDED,
366            Event::StepCompleted { .. } => Self::STEP_COMPLETED,
367            Event::StepFailed { .. } => Self::STEP_FAILED,
368            Event::ApprovalRequested { .. } => Self::APPROVAL_REQUESTED,
369            Event::ApprovalGranted { .. } => Self::APPROVAL_GRANTED,
370            Event::ApprovalRejected { .. } => Self::APPROVAL_REJECTED,
371            Event::LogLine { .. } => Self::LOG_LINE,
372            Event::UserSignedIn { .. } => Self::USER_SIGNED_IN,
373            Event::UserSignedUp { .. } => Self::USER_SIGNED_UP,
374            Event::UserSignedOut { .. } => Self::USER_SIGNED_OUT,
375        }
376    }
377}
378
379#[cfg(test)]
380mod tests {
381    use super::*;
382
383    #[test]
384    fn run_status_changed_serde_roundtrip() {
385        let event = Event::RunStatusChanged {
386            run_id: Uuid::now_v7(),
387            workflow_name: "deploy".to_string(),
388            from: RunStatus::Running,
389            to: RunStatus::Completed,
390            error: None,
391            cost_usd: Decimal::new(42, 2),
392            duration_ms: 5000,
393            at: Utc::now(),
394        };
395
396        let json = serde_json::to_string(&event).expect("serialize");
397        let back: Event = serde_json::from_str(&json).expect("deserialize");
398
399        assert_eq!(back.event_type(), "run_status_changed");
400        assert!(json.contains("\"type\":\"run_status_changed\""));
401    }
402
403    #[test]
404    fn run_failed_serde_roundtrip() {
405        let event = Event::RunFailed {
406            run_id: Uuid::now_v7(),
407            workflow_name: "deploy".to_string(),
408            error: Some("step crashed".to_string()),
409            cost_usd: Decimal::new(10, 2),
410            duration_ms: 3000,
411            at: Utc::now(),
412        };
413
414        let json = serde_json::to_string(&event).expect("serialize");
415        let back: Event = serde_json::from_str(&json).expect("deserialize");
416
417        assert_eq!(back.event_type(), "run_failed");
418        assert!(json.contains("\"type\":\"run_failed\""));
419        assert!(json.contains("step crashed"));
420    }
421
422    #[test]
423    fn run_budget_exceeded_serde_roundtrip() {
424        let event = Event::RunBudgetExceeded {
425            run_id: Uuid::now_v7(),
426            workflow_name: "deploy".to_string(),
427            limit_usd: Decimal::new(200, 2),
428            spent_usd: Decimal::new(180, 2),
429            step_budget_usd: Decimal::new(50, 2),
430            at: Utc::now(),
431        };
432
433        let json = serde_json::to_string(&event).expect("serialize");
434        let back: Event = serde_json::from_str(&json).expect("deserialize");
435
436        assert_eq!(back.event_type(), "run_budget_exceeded");
437        assert!(json.contains("\"type\":\"run_budget_exceeded\""));
438        assert!(json.contains("limit_usd"));
439        assert!(json.contains("step_budget_usd"));
440    }
441
442    #[test]
443    fn all_contains_run_budget_exceeded() {
444        assert!(Event::ALL.contains(&Event::RUN_BUDGET_EXCEEDED));
445    }
446
447    #[test]
448    fn user_signed_in_serde_roundtrip() {
449        let event = Event::UserSignedIn {
450            user_id: Uuid::now_v7(),
451            username: "alice".to_string(),
452            at: Utc::now(),
453        };
454
455        let json = serde_json::to_string(&event).expect("serialize");
456        let back: Event = serde_json::from_str(&json).expect("deserialize");
457
458        assert_eq!(back.event_type(), "user_signed_in");
459        assert!(json.contains("alice"));
460    }
461
462    #[test]
463    fn step_failed_serde_roundtrip() {
464        let event = Event::StepFailed {
465            run_id: Uuid::now_v7(),
466            step_id: Uuid::now_v7(),
467            step_name: "build".to_string(),
468            kind: StepKind::Shell,
469            error: "exit code 1".to_string(),
470            at: Utc::now(),
471        };
472
473        let json = serde_json::to_string(&event).expect("serialize");
474        let back: Event = serde_json::from_str(&json).expect("deserialize");
475
476        assert_eq!(back.event_type(), "step_failed");
477    }
478
479    #[test]
480    fn approval_requested_serde_roundtrip() {
481        let event = Event::ApprovalRequested {
482            run_id: Uuid::now_v7(),
483            step_id: Uuid::now_v7(),
484            message: "Deploy to prod?".to_string(),
485            at: Utc::now(),
486        };
487
488        let json = serde_json::to_string(&event).expect("serialize");
489        assert!(json.contains("approval_requested"));
490    }
491
492    #[test]
493    fn log_line_serde_roundtrip() {
494        let event = Event::LogLine {
495            run_id: Uuid::now_v7(),
496            step_id: Uuid::now_v7(),
497            step_name: "build".to_string(),
498            stream: LogStream::Stdout,
499            line: "Compiling ironflow v0.1.0".to_string(),
500            at: Utc::now(),
501        };
502
503        let json = serde_json::to_string(&event).expect("serialize");
504        let back: Event = serde_json::from_str(&json).expect("deserialize");
505
506        assert_eq!(back.event_type(), "log_line");
507        assert!(json.contains("\"type\":\"log_line\""));
508        assert!(json.contains("Compiling ironflow"));
509    }
510
511    #[test]
512    fn event_type_all_variants() {
513        let id = Uuid::now_v7();
514        let now = Utc::now();
515
516        let cases: Vec<(Event, &str)> = vec![
517            (
518                Event::RunCreated {
519                    run_id: id,
520                    workflow_name: "w".to_string(),
521                    at: now,
522                },
523                "run_created",
524            ),
525            (
526                Event::RunStatusChanged {
527                    run_id: id,
528                    workflow_name: "w".to_string(),
529                    from: RunStatus::Pending,
530                    to: RunStatus::Running,
531                    error: None,
532                    cost_usd: Decimal::ZERO,
533                    duration_ms: 0,
534                    at: now,
535                },
536                "run_status_changed",
537            ),
538            (
539                Event::RunFailed {
540                    run_id: id,
541                    workflow_name: "w".to_string(),
542                    error: Some("boom".to_string()),
543                    cost_usd: Decimal::ZERO,
544                    duration_ms: 0,
545                    at: now,
546                },
547                "run_failed",
548            ),
549            (
550                Event::StepCompleted {
551                    run_id: id,
552                    step_id: id,
553                    step_name: "s".to_string(),
554                    kind: StepKind::Shell,
555                    duration_ms: 0,
556                    cost_usd: Decimal::ZERO,
557                    at: now,
558                },
559                "step_completed",
560            ),
561            (
562                Event::StepFailed {
563                    run_id: id,
564                    step_id: id,
565                    step_name: "s".to_string(),
566                    kind: StepKind::Shell,
567                    error: "err".to_string(),
568                    at: now,
569                },
570                "step_failed",
571            ),
572            (
573                Event::ApprovalRequested {
574                    run_id: id,
575                    step_id: id,
576                    message: "ok?".to_string(),
577                    at: now,
578                },
579                "approval_requested",
580            ),
581            (
582                Event::ApprovalGranted {
583                    run_id: id,
584                    approved_by: "alice".to_string(),
585                    at: now,
586                },
587                "approval_granted",
588            ),
589            (
590                Event::ApprovalRejected {
591                    run_id: id,
592                    rejected_by: "bob".to_string(),
593                    at: now,
594                },
595                "approval_rejected",
596            ),
597            (
598                Event::LogLine {
599                    run_id: id,
600                    step_id: id,
601                    step_name: "build".to_string(),
602                    stream: LogStream::Stdout,
603                    line: "Compiling ironflow v0.1.0".to_string(),
604                    at: now,
605                },
606                "log_line",
607            ),
608            (
609                Event::UserSignedIn {
610                    user_id: id,
611                    username: "u".to_string(),
612                    at: now,
613                },
614                "user_signed_in",
615            ),
616            (
617                Event::UserSignedUp {
618                    user_id: id,
619                    username: "u".to_string(),
620                    at: now,
621                },
622                "user_signed_up",
623            ),
624            (
625                Event::UserSignedOut {
626                    user_id: id,
627                    at: now,
628                },
629                "user_signed_out",
630            ),
631        ];
632
633        for (event, expected_type) in cases {
634            assert_eq!(event.event_type(), expected_type);
635        }
636    }
637}
ironflow_engine/notify/event.rs

ironflow_engine/notify/
event.rs
