ironflow_engine/notify/
event_bus.rs

1//! Per-workflow broadcast event bus for real-time monitoring.
2//!
3//! [`WorkflowEventBus`] maintains one [`tokio::sync::broadcast`] channel per
4//! workflow run. Consumers (dashboards, SSE routes) subscribe to a specific
5//! `run_id` and receive only the events for that run.
6//!
7//! # Architecture
8//!
9//! - [`WorkflowEvent`] -- granular step-level events (started, completed,
10//!   failed, approval, token usage).
11//! - [`WorkflowEventBus`] -- per-run broadcast channels with subscribe /
12//!   publish / remove lifecycle.
13//!
14//! # Examples
15//!
16//! ```
17//! use ironflow_engine::notify::{WorkflowEvent, WorkflowEventBus, WorkflowStepStartedEvent};
18//! use uuid::Uuid;
19//! use chrono::Utc;
20//!
21//! let bus = WorkflowEventBus::new();
22//! let run_id = Uuid::now_v7();
23//!
24//! let mut rx = bus.subscribe(run_id);
25//!
26//! bus.publish(run_id, WorkflowEvent::StepStarted(WorkflowStepStartedEvent {
27//!     step_name: "build".to_string(),
28//!     step_index: 0,
29//!     timestamp: Utc::now(),
30//! }));
31//! ```
32
33use std::collections::HashMap;
34use std::sync::RwLock;
35
36use chrono::{DateTime, Utc};
37use rust_decimal::Decimal;
38use serde::{Deserialize, Serialize};
39use tokio::sync::broadcast;
40use uuid::Uuid;
41
42/// Default broadcast channel buffer size per run.
43const DEFAULT_BUFFER_SIZE: usize = 64;
44
45/// Payload of the `WorkflowEvent::StepStarted` workflow event.
46///
47/// # Examples
48///
49/// ```
50/// use chrono::Utc;
51/// use ironflow_engine::notify::WorkflowStepStartedEvent;
52///
53/// let payload = WorkflowStepStartedEvent {
54///     step_name: "build".to_string(),
55///     step_index: 0,
56///     timestamp: Utc::now(),
57/// };
58/// assert_eq!(payload.step_index, 0);
59/// ```
60#[derive(Debug, Clone, Serialize, Deserialize)]
61#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
62pub struct WorkflowStepStartedEvent {
63    /// Human-readable step name.
64    pub step_name: String,
65    /// Zero-based position in the workflow.
66    pub step_index: u32,
67    /// When the step started.
68    pub timestamp: DateTime<Utc>,
69}
70
71/// Payload of the `WorkflowEvent::StepCompleted` workflow event.
72///
73/// # Examples
74///
75/// ```
76/// use ironflow_engine::notify::WorkflowStepCompletedEvent;
77///
78/// let payload = WorkflowStepCompletedEvent {
79///     step_name: "deploy".to_string(),
80///     step_index: 1,
81///     duration_ms: 5000,
82///     output_summary: Some("deployed v1.2.3".to_string()),
83/// };
84/// assert_eq!(payload.duration_ms, 5000);
85/// ```
86#[derive(Debug, Clone, Serialize, Deserialize)]
87#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
88pub struct WorkflowStepCompletedEvent {
89    /// Human-readable step name.
90    pub step_name: String,
91    /// Zero-based position in the workflow.
92    pub step_index: u32,
93    /// Step duration in milliseconds.
94    pub duration_ms: u64,
95    /// Optional summary of the step output.
96    pub output_summary: Option<String>,
97}
98
99/// Payload of the `WorkflowEvent::StepFailed` workflow event.
100///
101/// # Examples
102///
103/// ```
104/// use ironflow_engine::notify::WorkflowStepFailedEvent;
105///
106/// let payload = WorkflowStepFailedEvent {
107///     step_name: "test".to_string(),
108///     step_index: 2,
109///     error: "exit code 1".to_string(),
110///     duration_ms: 3000,
111/// };
112/// assert_eq!(payload.error, "exit code 1");
113/// ```
114#[derive(Debug, Clone, Serialize, Deserialize)]
115#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
116pub struct WorkflowStepFailedEvent {
117    /// Human-readable step name.
118    pub step_name: String,
119    /// Zero-based position in the workflow.
120    pub step_index: u32,
121    /// Error description.
122    pub error: String,
123    /// Step duration in milliseconds.
124    pub duration_ms: u64,
125}
126
127/// Payload of the `WorkflowEvent::ApprovalRequired` workflow event.
128///
129/// # Examples
130///
131/// ```
132/// use ironflow_engine::notify::WorkflowApprovalRequiredEvent;
133/// use uuid::Uuid;
134///
135/// let payload = WorkflowApprovalRequiredEvent {
136///     step_name: "prod-gate".to_string(),
137///     step_index: 3,
138///     approval_id: Uuid::now_v7(),
139/// };
140/// assert_eq!(payload.step_name, "prod-gate");
141/// ```
142#[derive(Debug, Clone, Serialize, Deserialize)]
143#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
144pub struct WorkflowApprovalRequiredEvent {
145    /// Human-readable step name.
146    pub step_name: String,
147    /// Zero-based position in the workflow.
148    pub step_index: u32,
149    /// Identifier of the approval gate.
150    pub approval_id: Uuid,
151}
152
153/// Payload of the `WorkflowEvent::AgentStepTokensUsed` workflow event.
154///
155/// # Examples
156///
157/// ```
158/// use ironflow_engine::notify::WorkflowAgentStepTokensUsedEvent;
159/// use rust_decimal::Decimal;
160///
161/// let payload = WorkflowAgentStepTokensUsedEvent {
162///     step_name: "review".to_string(),
163///     tokens: 15_000,
164///     cost_usd: Decimal::new(42, 4),
165/// };
166/// assert_eq!(payload.tokens, 15_000);
167/// ```
168#[derive(Debug, Clone, Serialize, Deserialize)]
169#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
170pub struct WorkflowAgentStepTokensUsedEvent {
171    /// Human-readable step name.
172    pub step_name: String,
173    /// Total tokens consumed.
174    pub tokens: u64,
175    /// Estimated cost in USD.
176    pub cost_usd: Decimal,
177}
178
179/// A granular step-level event for real-time workflow monitoring.
180///
181/// Unlike [`Event`](super::Event) which covers the full system lifecycle
182/// (runs, auth, audit), `WorkflowEvent` tracks individual step transitions
183/// within a single run. Serialized with a `type` discriminant for UI
184/// consumption.
185///
186/// Each variant wraps a dedicated payload struct; the serialized form stays
187/// flat, with `type` sitting next to the payload fields.
188///
189/// # Examples
190///
191/// ```
192/// use ironflow_engine::notify::{WorkflowEvent, WorkflowStepStartedEvent};
193/// use chrono::Utc;
194///
195/// let event = WorkflowEvent::StepStarted(WorkflowStepStartedEvent {
196///     step_name: "deploy".to_string(),
197///     step_index: 0,
198///     timestamp: Utc::now(),
199/// });
200/// assert_eq!(event.event_type(), "step_started");
201///
202/// let json = serde_json::to_string(&event)?;
203/// assert!(json.contains("\"type\":\"step_started\""));
204/// # Ok::<(), serde_json::Error>(())
205/// ```
206#[derive(Debug, Clone, Serialize, Deserialize)]
207#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
208#[serde(tag = "type", rename_all = "snake_case")]
209pub enum WorkflowEvent {
210    /// A step began execution.
211    StepStarted(WorkflowStepStartedEvent),
212
213    /// A step completed successfully.
214    StepCompleted(WorkflowStepCompletedEvent),
215
216    /// A step failed.
217    StepFailed(WorkflowStepFailedEvent),
218
219    /// A step requires human approval before the run can continue.
220    ApprovalRequired(WorkflowApprovalRequiredEvent),
221
222    /// Token usage report for an agent step.
223    AgentStepTokensUsed(WorkflowAgentStepTokensUsedEvent),
224}
225
226impl WorkflowEvent {
227    /// Event type constant for [`StepStarted`](WorkflowEvent::StepStarted).
228    pub const STEP_STARTED: &'static str = "step_started";
229    /// Event type constant for [`StepCompleted`](WorkflowEvent::StepCompleted).
230    pub const STEP_COMPLETED: &'static str = "step_completed";
231    /// Event type constant for [`StepFailed`](WorkflowEvent::StepFailed).
232    pub const STEP_FAILED: &'static str = "step_failed";
233    /// Event type constant for [`ApprovalRequired`](WorkflowEvent::ApprovalRequired).
234    pub const APPROVAL_REQUIRED: &'static str = "approval_required";
235    /// Event type constant for [`AgentStepTokensUsed`](WorkflowEvent::AgentStepTokensUsed).
236    pub const AGENT_STEP_TOKENS_USED: &'static str = "agent_step_tokens_used";
237
238    /// Returns the event type as a static string (e.g. `"step_started"`).
239    ///
240    /// # Examples
241    ///
242    /// ```
243    /// use ironflow_engine::notify::{WorkflowEvent, WorkflowStepStartedEvent};
244    /// use chrono::Utc;
245    ///
246    /// let event = WorkflowEvent::StepStarted(WorkflowStepStartedEvent {
247    ///     step_name: "build".to_string(),
248    ///     step_index: 0,
249    ///     timestamp: Utc::now(),
250    /// });
251    /// assert_eq!(event.event_type(), "step_started");
252    /// ```
253    #[deny(unreachable_patterns)]
254    pub fn event_type(&self) -> &'static str {
255        match self {
256            WorkflowEvent::StepStarted(_) => Self::STEP_STARTED,
257            WorkflowEvent::StepCompleted(_) => Self::STEP_COMPLETED,
258            WorkflowEvent::StepFailed(_) => Self::STEP_FAILED,
259            WorkflowEvent::ApprovalRequired(_) => Self::APPROVAL_REQUIRED,
260            WorkflowEvent::AgentStepTokensUsed(_) => Self::AGENT_STEP_TOKENS_USED,
261        }
262    }
263}
264
265/// Per-workflow broadcast event bus for real-time monitoring.
266///
267/// Maintains one [`tokio::sync::broadcast`] channel per workflow run.
268/// Consumers call [`subscribe`](Self::subscribe) to receive events for a
269/// specific run; producers call [`publish`](Self::publish) to broadcast
270/// an event to all subscribers of that run.
271///
272/// Thread-safe and cheaply cloneable (`Clone` shares the same inner state).
273///
274/// # Examples
275///
276/// ```
277/// use ironflow_engine::notify::{WorkflowEvent, WorkflowEventBus, WorkflowStepStartedEvent};
278/// use uuid::Uuid;
279/// use chrono::Utc;
280///
281/// let bus = WorkflowEventBus::new();
282/// let run_id = Uuid::now_v7();
283///
284/// let mut rx = bus.subscribe(run_id);
285/// bus.publish(run_id, WorkflowEvent::StepStarted(WorkflowStepStartedEvent {
286///     step_name: "build".to_string(),
287///     step_index: 0,
288///     timestamp: Utc::now(),
289/// }));
290/// ```
291#[derive(Clone)]
292pub struct WorkflowEventBus {
293    channels: std::sync::Arc<RwLock<HashMap<Uuid, broadcast::Sender<WorkflowEvent>>>>,
294}
295
296impl WorkflowEventBus {
297    /// Create a new empty event bus.
298    ///
299    /// # Examples
300    ///
301    /// ```
302    /// use ironflow_engine::notify::WorkflowEventBus;
303    ///
304    /// let bus = WorkflowEventBus::new();
305    /// ```
306    pub fn new() -> Self {
307        Self {
308            channels: std::sync::Arc::new(RwLock::new(HashMap::new())),
309        }
310    }
311
312    /// Subscribe to events for a specific workflow run.
313    ///
314    /// If no channel exists for this `run_id`, one is created on demand.
315    /// Returns a broadcast receiver that yields [`WorkflowEvent`]s for
316    /// that run only.
317    ///
318    /// # Examples
319    ///
320    /// ```
321    /// use ironflow_engine::notify::WorkflowEventBus;
322    /// use uuid::Uuid;
323    ///
324    /// let bus = WorkflowEventBus::new();
325    /// let run_id = Uuid::now_v7();
326    /// let _rx = bus.subscribe(run_id);
327    /// ```
328    pub fn subscribe(&self, run_id: Uuid) -> broadcast::Receiver<WorkflowEvent> {
329        let mut channels = self.channels.write().expect("event bus lock poisoned");
330        let sender = channels
331            .entry(run_id)
332            .or_insert_with(|| broadcast::channel(DEFAULT_BUFFER_SIZE).0);
333        sender.subscribe()
334    }
335
336    /// Broadcast an event to all subscribers of a specific workflow run.
337    ///
338    /// If no channel exists for `run_id` (no subscriber has called
339    /// [`subscribe`](Self::subscribe)), the event is silently dropped.
340    /// If subscribers exist but none are actively listening, the send
341    /// error is ignored.
342    ///
343    /// # Examples
344    ///
345    /// ```
346    /// use ironflow_engine::notify::{WorkflowEvent, WorkflowEventBus, WorkflowStepStartedEvent};
347    /// use uuid::Uuid;
348    /// use chrono::Utc;
349    ///
350    /// let bus = WorkflowEventBus::new();
351    /// let run_id = Uuid::now_v7();
352    ///
353    /// // No subscriber -- silently dropped.
354    /// bus.publish(run_id, WorkflowEvent::StepStarted(WorkflowStepStartedEvent {
355    ///     step_name: "build".to_string(),
356    ///     step_index: 0,
357    ///     timestamp: Utc::now(),
358    /// }));
359    /// ```
360    pub fn publish(&self, run_id: Uuid, event: WorkflowEvent) {
361        let channels = self.channels.read().expect("event bus lock poisoned");
362        if let Some(sender) = channels.get(&run_id) {
363            let _ = sender.send(event);
364        }
365    }
366
367    /// Remove the channel for a workflow run.
368    ///
369    /// Call this when a run completes or is cleaned up to free resources.
370    /// If no channel exists for `run_id`, this is a no-op.
371    ///
372    /// # Examples
373    ///
374    /// ```
375    /// use ironflow_engine::notify::WorkflowEventBus;
376    /// use uuid::Uuid;
377    ///
378    /// let bus = WorkflowEventBus::new();
379    /// let run_id = Uuid::now_v7();
380    /// let _rx = bus.subscribe(run_id);
381    /// bus.remove(run_id);
382    /// ```
383    pub fn remove(&self, run_id: Uuid) {
384        let mut channels = self.channels.write().expect("event bus lock poisoned");
385        channels.remove(&run_id);
386    }
387}
388
389impl Default for WorkflowEventBus {
390    fn default() -> Self {
391        Self::new()
392    }
393}
394
395impl std::fmt::Debug for WorkflowEventBus {
396    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
397        let count = self.channels.read().map(|c| c.len()).unwrap_or(0);
398        f.debug_struct("WorkflowEventBus")
399            .field("active_channels", &count)
400            .finish()
401    }
402}
403
404#[cfg(test)]
405mod tests {
406    use super::*;
407
408    fn step_started(step_name: &str) -> WorkflowEvent {
409        WorkflowEvent::StepStarted(WorkflowStepStartedEvent {
410            step_name: step_name.to_string(),
411            step_index: 0,
412            timestamp: Utc::now(),
413        })
414    }
415
416    #[tokio::test]
417    async fn subscribe_receives_published_events() {
418        let bus = WorkflowEventBus::new();
419        let run_id = Uuid::now_v7();
420
421        let mut rx = bus.subscribe(run_id);
422
423        bus.publish(run_id, step_started("build"));
424
425        let received = rx.recv().await.expect("should receive event");
426        assert_eq!(received.event_type(), "step_started");
427        match received {
428            WorkflowEvent::StepStarted(e) => {
429                assert_eq!(e.step_name, "build");
430                assert_eq!(e.step_index, 0);
431            }
432            _ => panic!("expected StepStarted"),
433        }
434    }
435
436    #[test]
437    fn subscribe_creates_channel_on_demand() {
438        let bus = WorkflowEventBus::new();
439        let run_id = Uuid::now_v7();
440
441        let count_before = bus.channels.read().unwrap().len();
442        assert_eq!(count_before, 0);
443
444        let _rx = bus.subscribe(run_id);
445
446        let count_after = bus.channels.read().unwrap().len();
447        assert_eq!(count_after, 1);
448    }
449
450    #[test]
451    fn publish_unknown_run_is_noop() {
452        let bus = WorkflowEventBus::new();
453        let unknown_run = Uuid::now_v7();
454
455        bus.publish(unknown_run, step_started("build"));
456    }
457
458    #[test]
459    fn remove_cleans_up_channel() {
460        let bus = WorkflowEventBus::new();
461        let run_id = Uuid::now_v7();
462
463        let _rx = bus.subscribe(run_id);
464        assert_eq!(bus.channels.read().unwrap().len(), 1);
465
466        bus.remove(run_id);
467        assert_eq!(bus.channels.read().unwrap().len(), 0);
468    }
469
470    #[test]
471    fn remove_unknown_is_noop() {
472        let bus = WorkflowEventBus::new();
473        bus.remove(Uuid::now_v7());
474    }
475
476    #[test]
477    fn workflow_event_serde_roundtrip() {
478        let cases: Vec<WorkflowEvent> = vec![
479            WorkflowEvent::StepStarted(WorkflowStepStartedEvent {
480                step_name: "build".to_string(),
481                step_index: 0,
482                timestamp: Utc::now(),
483            }),
484            WorkflowEvent::StepCompleted(WorkflowStepCompletedEvent {
485                step_name: "deploy".to_string(),
486                step_index: 1,
487                duration_ms: 5000,
488                output_summary: Some("deployed v1.2.3".to_string()),
489            }),
490            WorkflowEvent::StepFailed(WorkflowStepFailedEvent {
491                step_name: "test".to_string(),
492                step_index: 2,
493                error: "exit code 1".to_string(),
494                duration_ms: 3000,
495            }),
496            WorkflowEvent::ApprovalRequired(WorkflowApprovalRequiredEvent {
497                step_name: "prod-gate".to_string(),
498                step_index: 3,
499                approval_id: Uuid::now_v7(),
500            }),
501            WorkflowEvent::AgentStepTokensUsed(WorkflowAgentStepTokensUsedEvent {
502                step_name: "review".to_string(),
503                tokens: 15000,
504                cost_usd: Decimal::new(42, 4),
505            }),
506        ];
507
508        for event in &cases {
509            let json = serde_json::to_string(event).expect("serialize");
510            let back: WorkflowEvent = serde_json::from_str(&json).expect("deserialize");
511
512            assert_eq!(back.event_type(), event.event_type());
513            assert!(json.contains(&format!("\"type\":\"{}\"", event.event_type())));
514        }
515    }
516
517    /// The pre-refactor wire format used flat inline-struct variants. Newtype
518    /// variants produce and accept the same JSON, so SSE consumers and stored
519    /// payloads need no migration.
520    #[test]
521    fn workflow_event_legacy_flat_json_deserializes() {
522        let approval_id: Uuid = "01890000-0000-7000-8000-000000000002"
523            .parse()
524            .expect("valid uuid");
525
526        let raw = r#"{"type":"step_started","step_name":"build","step_index":0,"timestamp":"2026-01-01T00:00:00Z"}"#;
527        match serde_json::from_str::<WorkflowEvent>(raw).expect("legacy payload") {
528            WorkflowEvent::StepStarted(e) => {
529                assert_eq!(e.step_name, "build");
530                assert_eq!(e.step_index, 0);
531            }
532            other => panic!("expected StepStarted, got {other:?}"),
533        }
534
535        let raw = r#"{"type":"step_completed","step_name":"deploy","step_index":1,"duration_ms":5000,"output_summary":"deployed v1.2.3"}"#;
536        match serde_json::from_str::<WorkflowEvent>(raw).expect("legacy payload") {
537            WorkflowEvent::StepCompleted(e) => {
538                assert_eq!(e.duration_ms, 5000);
539                assert_eq!(e.output_summary.as_deref(), Some("deployed v1.2.3"));
540            }
541            other => panic!("expected StepCompleted, got {other:?}"),
542        }
543
544        let raw = r#"{"type":"step_failed","step_name":"test","step_index":2,"error":"exit code 1","duration_ms":3000}"#;
545        match serde_json::from_str::<WorkflowEvent>(raw).expect("legacy payload") {
546            WorkflowEvent::StepFailed(e) => {
547                assert_eq!(e.error, "exit code 1");
548                assert_eq!(e.duration_ms, 3000);
549            }
550            other => panic!("expected StepFailed, got {other:?}"),
551        }
552
553        let raw = r#"{"type":"approval_required","step_name":"prod-gate","step_index":3,"approval_id":"01890000-0000-7000-8000-000000000002"}"#;
554        match serde_json::from_str::<WorkflowEvent>(raw).expect("legacy payload") {
555            WorkflowEvent::ApprovalRequired(e) => {
556                assert_eq!(e.approval_id, approval_id);
557            }
558            other => panic!("expected ApprovalRequired, got {other:?}"),
559        }
560
561        let raw = r#"{"type":"agent_step_tokens_used","step_name":"review","tokens":15000,"cost_usd":0.5}"#;
562        match serde_json::from_str::<WorkflowEvent>(raw).expect("legacy payload") {
563            WorkflowEvent::AgentStepTokensUsed(e) => {
564                assert_eq!(e.tokens, 15000);
565                assert_eq!(e.cost_usd, Decimal::new(5, 1));
566            }
567            other => panic!("expected AgentStepTokensUsed, got {other:?}"),
568        }
569    }
570
571    /// Guards the internally-tagged representation: payload fields must stay
572    /// siblings of `type`, never nested under a variant key.
573    #[test]
574    fn serialized_workflow_event_is_flat_with_type_tag() {
575        let event = WorkflowEvent::StepFailed(WorkflowStepFailedEvent {
576            step_name: "test".to_string(),
577            step_index: 2,
578            error: "exit code 1".to_string(),
579            duration_ms: 3000,
580        });
581
582        let value: serde_json::Value = serde_json::to_value(&event).expect("serialize");
583        let object = value.as_object().expect("event serializes to an object");
584
585        assert_eq!(
586            object.get("type").and_then(|v| v.as_str()),
587            Some("step_failed")
588        );
589        assert_eq!(
590            object.get("step_name").and_then(|v| v.as_str()),
591            Some("test")
592        );
593        assert_eq!(object.get("step_index").and_then(|v| v.as_u64()), Some(2));
594        assert_eq!(
595            object.get("error").and_then(|v| v.as_str()),
596            Some("exit code 1")
597        );
598        assert_eq!(
599            object.get("duration_ms").and_then(|v| v.as_u64()),
600            Some(3000)
601        );
602        assert_eq!(object.len(), 5, "no nesting: {object:?}");
603    }
604
605    #[test]
606    fn event_type_all_variants() {
607        let cases: Vec<(WorkflowEvent, &str)> = vec![
608            (
609                WorkflowEvent::StepStarted(WorkflowStepStartedEvent {
610                    step_name: "s".to_string(),
611                    step_index: 0,
612                    timestamp: Utc::now(),
613                }),
614                "step_started",
615            ),
616            (
617                WorkflowEvent::StepCompleted(WorkflowStepCompletedEvent {
618                    step_name: "s".to_string(),
619                    step_index: 0,
620                    duration_ms: 0,
621                    output_summary: None,
622                }),
623                "step_completed",
624            ),
625            (
626                WorkflowEvent::StepFailed(WorkflowStepFailedEvent {
627                    step_name: "s".to_string(),
628                    step_index: 0,
629                    error: "e".to_string(),
630                    duration_ms: 0,
631                }),
632                "step_failed",
633            ),
634            (
635                WorkflowEvent::ApprovalRequired(WorkflowApprovalRequiredEvent {
636                    step_name: "s".to_string(),
637                    step_index: 0,
638                    approval_id: Uuid::now_v7(),
639                }),
640                "approval_required",
641            ),
642            (
643                WorkflowEvent::AgentStepTokensUsed(WorkflowAgentStepTokensUsedEvent {
644                    step_name: "s".to_string(),
645                    tokens: 0,
646                    cost_usd: Decimal::ZERO,
647                }),
648                "agent_step_tokens_used",
649            ),
650        ];
651
652        for (event, expected) in cases {
653            assert_eq!(event.event_type(), expected);
654        }
655    }
656
657    #[tokio::test]
658    async fn multiple_subscribers_receive_same_event() {
659        let bus = WorkflowEventBus::new();
660        let run_id = Uuid::now_v7();
661
662        let mut rx1 = bus.subscribe(run_id);
663        let mut rx2 = bus.subscribe(run_id);
664
665        bus.publish(run_id, step_started("build"));
666
667        let e1 = rx1.recv().await.expect("rx1 should receive");
668        let e2 = rx2.recv().await.expect("rx2 should receive");
669
670        assert_eq!(e1.event_type(), "step_started");
671        assert_eq!(e2.event_type(), "step_started");
672    }
673
674    #[tokio::test]
675    async fn events_isolated_between_runs() {
676        let bus = WorkflowEventBus::new();
677        let run_a = Uuid::now_v7();
678        let run_b = Uuid::now_v7();
679
680        let mut rx_a = bus.subscribe(run_a);
681        let mut rx_b = bus.subscribe(run_b);
682
683        bus.publish(run_a, step_started("only-for-a"));
684
685        let received = rx_a.recv().await.expect("rx_a should receive");
686        match received {
687            WorkflowEvent::StepStarted(e) => {
688                assert_eq!(e.step_name, "only-for-a");
689            }
690            _ => panic!("expected StepStarted"),
691        }
692
693        // rx_b should have nothing -- try_recv returns Empty.
694        assert!(rx_b.try_recv().is_err());
695    }
696}
ironflow_engine/notify/event_bus.rs

ironflow_engine/notify/
event_bus.rs
