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