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