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 /// A manual retry was forced despite a handler version mismatch.
164 ///
165 /// Emitted when a caller passes `force=true` on a retry where the
166 /// handler version differs from the run's recorded version. This is
167 /// an audit event: it means the new code will execute on the old
168 /// payload without the handler explicitly declaring compatibility.
169 RetryForced {
170 /// The new run created by the forced retry.
171 run_id: Uuid,
172 /// Workflow name.
173 workflow_name: String,
174 /// Version stored on the original run.
175 original_version: String,
176 /// Current version of the handler.
177 current_version: String,
178 /// When the forced retry occurred.
179 at: DateTime<Utc>,
180 },
181
182 // -- Step lifecycle --
183 /// A step completed successfully.
184 StepCompleted {
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 /// Step duration in milliseconds.
195 duration_ms: u64,
196 /// Step cost in USD.
197 cost_usd: Decimal,
198 /// When the step completed.
199 at: DateTime<Utc>,
200 },
201
202 /// A step failed.
203 StepFailed {
204 /// Run identifier.
205 run_id: Uuid,
206 /// Step identifier.
207 step_id: Uuid,
208 /// Human-readable step name.
209 step_name: String,
210 /// Step operation kind.
211 #[cfg_attr(feature = "openapi", schema(value_type = String))]
212 kind: StepKind,
213 /// Error message.
214 error: String,
215 /// When the step failed.
216 at: DateTime<Utc>,
217 },
218
219 // -- Approval --
220 /// A run is waiting for human approval.
221 ApprovalRequested {
222 /// Run identifier.
223 run_id: Uuid,
224 /// Approval step identifier.
225 step_id: Uuid,
226 /// Message displayed to reviewers.
227 message: String,
228 /// When the approval was requested.
229 at: DateTime<Utc>,
230 },
231
232 /// A run was approved by a human.
233 ApprovalGranted {
234 /// Run identifier.
235 run_id: Uuid,
236 /// User who approved (ID or username).
237 approved_by: String,
238 /// When the approval was granted.
239 at: DateTime<Utc>,
240 },
241
242 /// A run was rejected by a human.
243 ApprovalRejected {
244 /// Run identifier.
245 run_id: Uuid,
246 /// User who rejected (ID or username).
247 rejected_by: String,
248 /// When the rejection occurred.
249 at: DateTime<Utc>,
250 },
251
252 // -- Log streaming --
253 /// A log line emitted during step execution.
254 ///
255 /// Pushed by the worker in real time so that SSE clients can stream
256 /// step output as it happens, without waiting for step completion.
257 LogLine {
258 /// Run identifier.
259 run_id: Uuid,
260 /// Step identifier.
261 step_id: Uuid,
262 /// Human-readable step name.
263 step_name: String,
264 /// Output stream.
265 stream: LogStream,
266 /// The log line content.
267 line: String,
268 /// When the line was emitted.
269 at: DateTime<Utc>,
270 },
271
272 // -- Authentication --
273 /// A user signed in.
274 UserSignedIn {
275 /// User identifier.
276 user_id: Uuid,
277 /// Username.
278 username: String,
279 /// When the sign-in occurred.
280 at: DateTime<Utc>,
281 },
282
283 /// A new user signed up.
284 UserSignedUp {
285 /// User identifier.
286 user_id: Uuid,
287 /// Username.
288 username: String,
289 /// When the sign-up occurred.
290 at: DateTime<Utc>,
291 },
292
293 /// A user signed out.
294 UserSignedOut {
295 /// User identifier.
296 user_id: Uuid,
297 /// When the sign-out occurred.
298 at: DateTime<Utc>,
299 },
300}
301
302impl Event {
303 /// Event type constant for [`RunCreated`](Event::RunCreated).
304 pub const RUN_CREATED: &'static str = "run_created";
305 /// Event type constant for [`RunStatusChanged`](Event::RunStatusChanged).
306 pub const RUN_STATUS_CHANGED: &'static str = "run_status_changed";
307 /// Event type constant for [`RunFailed`](Event::RunFailed).
308 pub const RUN_FAILED: &'static str = "run_failed";
309 /// Event type constant for [`RunBudgetExceeded`](Event::RunBudgetExceeded).
310 pub const RUN_BUDGET_EXCEEDED: &'static str = "run_budget_exceeded";
311 /// Event type constant for [`RetryForced`](Event::RetryForced).
312 pub const RETRY_FORCED: &'static str = "retry_forced";
313 /// Event type constant for [`StepCompleted`](Event::StepCompleted).
314 pub const STEP_COMPLETED: &'static str = "step_completed";
315 /// Event type constant for [`StepFailed`](Event::StepFailed).
316 pub const STEP_FAILED: &'static str = "step_failed";
317 /// Event type constant for [`ApprovalRequested`](Event::ApprovalRequested).
318 pub const APPROVAL_REQUESTED: &'static str = "approval_requested";
319 /// Event type constant for [`ApprovalGranted`](Event::ApprovalGranted).
320 pub const APPROVAL_GRANTED: &'static str = "approval_granted";
321 /// Event type constant for [`ApprovalRejected`](Event::ApprovalRejected).
322 pub const APPROVAL_REJECTED: &'static str = "approval_rejected";
323 /// Event type constant for [`LogLine`](Event::LogLine).
324 pub const LOG_LINE: &'static str = "log_line";
325 /// Event type constant for [`UserSignedIn`](Event::UserSignedIn).
326 pub const USER_SIGNED_IN: &'static str = "user_signed_in";
327 /// Event type constant for [`UserSignedUp`](Event::UserSignedUp).
328 pub const USER_SIGNED_UP: &'static str = "user_signed_up";
329 /// Event type constant for [`UserSignedOut`](Event::UserSignedOut).
330 pub const USER_SIGNED_OUT: &'static str = "user_signed_out";
331
332 /// All event types. Pass this to
333 /// [`EventPublisher::subscribe`](super::EventPublisher::subscribe) to
334 /// receive every event.
335 ///
336 /// # Examples
337 ///
338 /// ```no_run
339 /// use ironflow_engine::notify::{Event, EventPublisher, WebhookSubscriber};
340 ///
341 /// let mut publisher = EventPublisher::new();
342 /// publisher.subscribe(
343 /// WebhookSubscriber::new("https://example.com/all"),
344 /// Event::ALL,
345 /// );
346 /// ```
347 pub const ALL: &'static [&'static str] = &[
348 Self::RUN_CREATED,
349 Self::RUN_STATUS_CHANGED,
350 Self::RUN_FAILED,
351 Self::RUN_BUDGET_EXCEEDED,
352 Self::STEP_COMPLETED,
353 Self::STEP_FAILED,
354 Self::APPROVAL_REQUESTED,
355 Self::APPROVAL_GRANTED,
356 Self::APPROVAL_REJECTED,
357 Self::LOG_LINE,
358 Self::USER_SIGNED_IN,
359 Self::USER_SIGNED_UP,
360 Self::USER_SIGNED_OUT,
361 Self::RETRY_FORCED,
362 ];
363
364 /// Returns the event type as a static string (e.g. `"run_status_changed"`).
365 ///
366 /// Useful for filtering and logging without deserializing.
367 ///
368 /// # Examples
369 ///
370 /// ```
371 /// use ironflow_engine::notify::Event;
372 /// use uuid::Uuid;
373 /// use chrono::Utc;
374 ///
375 /// let event = Event::UserSignedIn {
376 /// user_id: Uuid::now_v7(),
377 /// username: "alice".to_string(),
378 /// at: Utc::now(),
379 /// };
380 /// assert_eq!(event.event_type(), "user_signed_in");
381 /// ```
382 pub fn event_type(&self) -> &'static str {
383 match self {
384 Event::RunCreated { .. } => Self::RUN_CREATED,
385 Event::RunStatusChanged { .. } => Self::RUN_STATUS_CHANGED,
386 Event::RunFailed { .. } => Self::RUN_FAILED,
387 Event::RunBudgetExceeded { .. } => Self::RUN_BUDGET_EXCEEDED,
388 Event::StepCompleted { .. } => Self::STEP_COMPLETED,
389 Event::StepFailed { .. } => Self::STEP_FAILED,
390 Event::ApprovalRequested { .. } => Self::APPROVAL_REQUESTED,
391 Event::ApprovalGranted { .. } => Self::APPROVAL_GRANTED,
392 Event::ApprovalRejected { .. } => Self::APPROVAL_REJECTED,
393 Event::LogLine { .. } => Self::LOG_LINE,
394 Event::UserSignedIn { .. } => Self::USER_SIGNED_IN,
395 Event::UserSignedUp { .. } => Self::USER_SIGNED_UP,
396 Event::UserSignedOut { .. } => Self::USER_SIGNED_OUT,
397 Event::RetryForced { .. } => Self::RETRY_FORCED,
398 }
399 }
400}
401
402#[cfg(test)]
403mod tests {
404 use super::*;
405
406 #[test]
407 fn run_status_changed_serde_roundtrip() {
408 let event = Event::RunStatusChanged {
409 run_id: Uuid::now_v7(),
410 workflow_name: "deploy".to_string(),
411 from: RunStatus::Running,
412 to: RunStatus::Completed,
413 error: None,
414 cost_usd: Decimal::new(42, 2),
415 duration_ms: 5000,
416 at: Utc::now(),
417 };
418
419 let json = serde_json::to_string(&event).expect("serialize");
420 let back: Event = serde_json::from_str(&json).expect("deserialize");
421
422 assert_eq!(back.event_type(), "run_status_changed");
423 assert!(json.contains("\"type\":\"run_status_changed\""));
424 }
425
426 #[test]
427 fn run_failed_serde_roundtrip() {
428 let event = Event::RunFailed {
429 run_id: Uuid::now_v7(),
430 workflow_name: "deploy".to_string(),
431 error: Some("step crashed".to_string()),
432 cost_usd: Decimal::new(10, 2),
433 duration_ms: 3000,
434 at: Utc::now(),
435 };
436
437 let json = serde_json::to_string(&event).expect("serialize");
438 let back: Event = serde_json::from_str(&json).expect("deserialize");
439
440 assert_eq!(back.event_type(), "run_failed");
441 assert!(json.contains("\"type\":\"run_failed\""));
442 assert!(json.contains("step crashed"));
443 }
444
445 #[test]
446 fn run_budget_exceeded_serde_roundtrip() {
447 let event = Event::RunBudgetExceeded {
448 run_id: Uuid::now_v7(),
449 workflow_name: "deploy".to_string(),
450 limit_usd: Decimal::new(200, 2),
451 spent_usd: Decimal::new(180, 2),
452 step_budget_usd: Decimal::new(50, 2),
453 at: Utc::now(),
454 };
455
456 let json = serde_json::to_string(&event).expect("serialize");
457 let back: Event = serde_json::from_str(&json).expect("deserialize");
458
459 assert_eq!(back.event_type(), "run_budget_exceeded");
460 assert!(json.contains("\"type\":\"run_budget_exceeded\""));
461 assert!(json.contains("limit_usd"));
462 assert!(json.contains("step_budget_usd"));
463 }
464
465 #[test]
466 fn all_contains_run_budget_exceeded() {
467 assert!(Event::ALL.contains(&Event::RUN_BUDGET_EXCEEDED));
468 }
469
470 #[test]
471 fn user_signed_in_serde_roundtrip() {
472 let event = Event::UserSignedIn {
473 user_id: Uuid::now_v7(),
474 username: "alice".to_string(),
475 at: Utc::now(),
476 };
477
478 let json = serde_json::to_string(&event).expect("serialize");
479 let back: Event = serde_json::from_str(&json).expect("deserialize");
480
481 assert_eq!(back.event_type(), "user_signed_in");
482 assert!(json.contains("alice"));
483 }
484
485 #[test]
486 fn step_failed_serde_roundtrip() {
487 let event = Event::StepFailed {
488 run_id: Uuid::now_v7(),
489 step_id: Uuid::now_v7(),
490 step_name: "build".to_string(),
491 kind: StepKind::Shell,
492 error: "exit code 1".to_string(),
493 at: Utc::now(),
494 };
495
496 let json = serde_json::to_string(&event).expect("serialize");
497 let back: Event = serde_json::from_str(&json).expect("deserialize");
498
499 assert_eq!(back.event_type(), "step_failed");
500 }
501
502 #[test]
503 fn approval_requested_serde_roundtrip() {
504 let event = Event::ApprovalRequested {
505 run_id: Uuid::now_v7(),
506 step_id: Uuid::now_v7(),
507 message: "Deploy to prod?".to_string(),
508 at: Utc::now(),
509 };
510
511 let json = serde_json::to_string(&event).expect("serialize");
512 assert!(json.contains("approval_requested"));
513 }
514
515 #[test]
516 fn log_line_serde_roundtrip() {
517 let event = Event::LogLine {
518 run_id: Uuid::now_v7(),
519 step_id: Uuid::now_v7(),
520 step_name: "build".to_string(),
521 stream: LogStream::Stdout,
522 line: "Compiling ironflow v0.1.0".to_string(),
523 at: Utc::now(),
524 };
525
526 let json = serde_json::to_string(&event).expect("serialize");
527 let back: Event = serde_json::from_str(&json).expect("deserialize");
528
529 assert_eq!(back.event_type(), "log_line");
530 assert!(json.contains("\"type\":\"log_line\""));
531 assert!(json.contains("Compiling ironflow"));
532 }
533
534 #[test]
535 fn event_type_all_variants() {
536 let id = Uuid::now_v7();
537 let now = Utc::now();
538
539 let cases: Vec<(Event, &str)> = vec![
540 (
541 Event::RunCreated {
542 run_id: id,
543 workflow_name: "w".to_string(),
544 at: now,
545 },
546 "run_created",
547 ),
548 (
549 Event::RunStatusChanged {
550 run_id: id,
551 workflow_name: "w".to_string(),
552 from: RunStatus::Pending,
553 to: RunStatus::Running,
554 error: None,
555 cost_usd: Decimal::ZERO,
556 duration_ms: 0,
557 at: now,
558 },
559 "run_status_changed",
560 ),
561 (
562 Event::RunFailed {
563 run_id: id,
564 workflow_name: "w".to_string(),
565 error: Some("boom".to_string()),
566 cost_usd: Decimal::ZERO,
567 duration_ms: 0,
568 at: now,
569 },
570 "run_failed",
571 ),
572 (
573 Event::StepCompleted {
574 run_id: id,
575 step_id: id,
576 step_name: "s".to_string(),
577 kind: StepKind::Shell,
578 duration_ms: 0,
579 cost_usd: Decimal::ZERO,
580 at: now,
581 },
582 "step_completed",
583 ),
584 (
585 Event::StepFailed {
586 run_id: id,
587 step_id: id,
588 step_name: "s".to_string(),
589 kind: StepKind::Shell,
590 error: "err".to_string(),
591 at: now,
592 },
593 "step_failed",
594 ),
595 (
596 Event::ApprovalRequested {
597 run_id: id,
598 step_id: id,
599 message: "ok?".to_string(),
600 at: now,
601 },
602 "approval_requested",
603 ),
604 (
605 Event::ApprovalGranted {
606 run_id: id,
607 approved_by: "alice".to_string(),
608 at: now,
609 },
610 "approval_granted",
611 ),
612 (
613 Event::ApprovalRejected {
614 run_id: id,
615 rejected_by: "bob".to_string(),
616 at: now,
617 },
618 "approval_rejected",
619 ),
620 (
621 Event::LogLine {
622 run_id: id,
623 step_id: id,
624 step_name: "build".to_string(),
625 stream: LogStream::Stdout,
626 line: "Compiling ironflow v0.1.0".to_string(),
627 at: now,
628 },
629 "log_line",
630 ),
631 (
632 Event::UserSignedIn {
633 user_id: id,
634 username: "u".to_string(),
635 at: now,
636 },
637 "user_signed_in",
638 ),
639 (
640 Event::UserSignedUp {
641 user_id: id,
642 username: "u".to_string(),
643 at: now,
644 },
645 "user_signed_up",
646 ),
647 (
648 Event::UserSignedOut {
649 user_id: id,
650 at: now,
651 },
652 "user_signed_out",
653 ),
654 ];
655
656 for (event, expected_type) in cases {
657 assert_eq!(event.event_type(), expected_type);
658 }
659 }
660}