ironflow_runtime/trigger/
event.rs

1//! Domain-event trigger for workflow chaining.
2//!
3//! [`EventTrigger`] subscribes to the [`EventPublisher`](ironflow_engine::notify::EventPublisher)
4//! and creates a new run whenever a matching event fires. This allows
5//! declarative workflow chaining without external infrastructure.
6//!
7//! # Anti-loop protection
8//!
9//! Each triggered run carries a `_chain_depth` label. When the depth
10//! reaches `max_chain_depth`, the trigger ignores the event and logs a
11//! warning. This prevents two workflows from triggering each other
12//! indefinitely.
13//!
14//! # Examples
15//!
16//! ```no_run
17//! use ironflow_runtime::trigger::event::{EventTrigger, EventTriggerRule};
18//! use ironflow_store::entities::EventKind;
19//!
20//! let trigger = EventTrigger::new(vec![
21//!     EventTriggerRule {
22//!         on_event: EventKind::RunFailed,
23//!         source_workflow: "deploy".to_string(),
24//!         target_workflow: "rollback".to_string(),
25//!         max_chain_depth: 3,
26//!         conditions: vec![],
27//!     },
28//! ]);
29//! ```
30
31use std::collections::HashMap;
32use std::fmt;
33use std::panic::{AssertUnwindSafe, catch_unwind};
34use std::sync::Arc;
35
36use rust_decimal::Decimal;
37use serde_json::json;
38use tokio::sync::mpsc;
39use tokio_util::sync::CancellationToken;
40use tracing::{info, warn};
41use uuid::Uuid;
42
43use ironflow_engine::notify::{Event, EventSubscriber, SubscriberFuture};
44use ironflow_store::entities::{EventKind, TriggerKind};
45
46use super::{Trigger, TriggerEvent, TriggerFuture, TriggerSink};
47
48/// Label key used to track chaining depth on triggered runs.
49pub const CHAIN_DEPTH_LABEL: &str = "_chain_depth";
50
51/// Context available to [`TriggerCondition`]s when evaluating whether a
52/// rule should fire.
53///
54/// Exposes metadata from the source run: labels, error message,
55/// aggregated cost and duration.
56///
57/// # Examples
58///
59/// ```
60/// use std::collections::HashMap;
61/// use ironflow_runtime::trigger::event::TriggerContext;
62/// use rust_decimal::Decimal;
63///
64/// let ctx = TriggerContext {
65///     labels: HashMap::from([("env".to_string(), "prod".to_string())]),
66///     error: Some("timeout".to_string()),
67///     cost_usd: Decimal::new(42, 2),
68///     duration_ms: 5000,
69/// };
70/// assert_eq!(ctx.labels.get("env").unwrap(), "prod");
71/// ```
72#[derive(Debug, Clone)]
73pub struct TriggerContext {
74    /// Labels of the source run.
75    pub labels: HashMap<String, String>,
76    /// Error message of the source run (if any).
77    pub error: Option<String>,
78    /// Aggregated cost in USD of the source run.
79    pub cost_usd: Decimal,
80    /// Aggregated duration in milliseconds of the source run.
81    pub duration_ms: u64,
82}
83
84/// A condition that must be satisfied for an [`EventTriggerRule`] to fire.
85///
86/// All conditions on a rule are evaluated with AND semantics: the rule
87/// fires only when every condition returns `true`. For OR semantics,
88/// create separate rules.
89///
90/// # Examples
91///
92/// ```
93/// use ironflow_runtime::trigger::event::TriggerCondition;
94///
95/// let cond = TriggerCondition::Label {
96///     key: "env".to_string(),
97///     value: "prod".to_string(),
98/// };
99/// assert!(matches!(cond, TriggerCondition::Label { .. }));
100/// ```
101pub enum TriggerCondition {
102    /// The source run must carry a label with this exact key-value pair.
103    Label {
104        /// Label key to check.
105        key: String,
106        /// Expected label value.
107        value: String,
108    },
109    /// An arbitrary predicate evaluated against the [`TriggerContext`].
110    ///
111    /// Uses [`Arc`] so the condition is `Clone + Send + Sync`.
112    Expression(Arc<dyn Fn(&TriggerContext) -> bool + Send + Sync>),
113}
114
115impl TriggerCondition {
116    /// Evaluate this condition against the given context.
117    ///
118    /// # Examples
119    ///
120    /// ```
121    /// use std::collections::HashMap;
122    /// use ironflow_runtime::trigger::event::{TriggerCondition, TriggerContext};
123    /// use rust_decimal::Decimal;
124    ///
125    /// let ctx = TriggerContext {
126    ///     labels: HashMap::from([("env".to_string(), "prod".to_string())]),
127    ///     error: None,
128    ///     cost_usd: Decimal::ZERO,
129    ///     duration_ms: 0,
130    /// };
131    /// let cond = TriggerCondition::Label {
132    ///     key: "env".to_string(),
133    ///     value: "prod".to_string(),
134    /// };
135    /// assert!(cond.evaluate(&ctx));
136    /// ```
137    ///
138    /// # Panics
139    ///
140    /// Does not panic. If an [`Expression`](TriggerCondition::Expression)
141    /// closure panics, the panic is caught and the condition evaluates to
142    /// `false`.
143    pub fn evaluate(&self, ctx: &TriggerContext) -> bool {
144        match self {
145            TriggerCondition::Label { key, value } => {
146                ctx.labels.get(key).is_some_and(|v| v == value)
147            }
148            TriggerCondition::Expression(f) => match catch_unwind(AssertUnwindSafe(|| f(ctx))) {
149                Ok(result) => result,
150                Err(_) => {
151                    warn!("expression condition panicked, treating as false");
152                    false
153                }
154            },
155        }
156    }
157}
158
159impl Clone for TriggerCondition {
160    fn clone(&self) -> Self {
161        match self {
162            TriggerCondition::Label { key, value } => TriggerCondition::Label {
163                key: key.clone(),
164                value: value.clone(),
165            },
166            TriggerCondition::Expression(f) => TriggerCondition::Expression(Arc::clone(f)),
167        }
168    }
169}
170
171impl fmt::Debug for TriggerCondition {
172    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
173        match self {
174            TriggerCondition::Label { key, value } => f
175                .debug_struct("Label")
176                .field("key", key)
177                .field("value", value)
178                .finish(),
179            TriggerCondition::Expression(_) => f.write_str("Expression(<closure>)"),
180        }
181    }
182}
183
184/// A rule that maps a domain event to a workflow to trigger.
185///
186/// # Examples
187///
188/// ```
189/// use ironflow_runtime::trigger::event::EventTriggerRule;
190/// use ironflow_store::entities::EventKind;
191///
192/// let rule = EventTriggerRule {
193///     on_event: EventKind::RunFailed,
194///     source_workflow: "deploy".to_string(),
195///     target_workflow: "rollback".to_string(),
196///     max_chain_depth: 3,
197///     conditions: vec![],
198///  };
199/// assert_eq!(rule.target_workflow, "rollback");
200/// ```
201#[derive(Debug, Clone)]
202pub struct EventTriggerRule {
203    /// The event kind to react to.
204    pub on_event: EventKind,
205    /// Only react to events from this workflow.
206    pub source_workflow: String,
207    /// The workflow to trigger.
208    pub target_workflow: String,
209    /// Maximum chaining depth (default 3). Beyond this, the event is
210    /// ignored and logged.
211    pub max_chain_depth: u8,
212    /// Optional conditions that must all match for the rule to fire.
213    ///
214    /// Evaluated with AND semantics. An empty list means the rule fires
215    /// unconditionally (backward compatible).
216    pub conditions: Vec<TriggerCondition>,
217}
218
219/// A trigger that reacts to internal domain events.
220///
221/// Register this trigger with the runtime via
222/// [`Runtime::trigger`](crate::runtime::Runtime::trigger). It must also
223/// be registered as an [`EventSubscriber`] on the
224/// [`EventPublisher`](ironflow_engine::notify::EventPublisher) so it
225/// receives events.
226///
227/// # Examples
228///
229/// ```no_run
230/// use ironflow_runtime::trigger::event::{EventTrigger, EventTriggerRule};
231/// use ironflow_store::entities::EventKind;
232///
233/// let trigger = EventTrigger::new(vec![
234///     EventTriggerRule {
235///         on_event: EventKind::RunFailed,
236///         source_workflow: "deploy".to_string(),
237///         target_workflow: "rollback".to_string(),
238///         max_chain_depth: 3,
239///         conditions: vec![],
240///     },
241/// ]);
242/// ```
243pub struct EventTrigger {
244    rules: Vec<EventTriggerRule>,
245    /// Internal channel from the EventSubscriber side to the Trigger side.
246    event_tx: mpsc::Sender<InternalEvent>,
247    event_rx: tokio::sync::Mutex<mpsc::Receiver<InternalEvent>>,
248}
249
250/// Internal representation of a domain event relevant to the trigger.
251#[derive(Debug)]
252struct InternalEvent {
253    run_id: Uuid,
254    workflow_name: String,
255    event_kind: EventKind,
256    error: Option<String>,
257    labels: HashMap<String, String>,
258    cost_usd: Decimal,
259    duration_ms: u64,
260}
261
262impl EventTrigger {
263    /// Create a new event trigger with the given rules.
264    ///
265    /// # Examples
266    ///
267    /// ```
268    /// use ironflow_runtime::trigger::event::{EventTrigger, EventTriggerRule};
269    /// use ironflow_store::entities::EventKind;
270    ///
271    /// let trigger = EventTrigger::new(vec![
272    ///     EventTriggerRule {
273    ///         on_event: EventKind::RunFailed,
274    ///         source_workflow: "deploy".to_string(),
275    ///         target_workflow: "rollback".to_string(),
276    ///         max_chain_depth: 3,
277    ///         conditions: vec![],
278    ///     },
279    /// ]);
280    /// ```
281    pub fn new(rules: Vec<EventTriggerRule>) -> Self {
282        let (event_tx, event_rx) = mpsc::channel(256);
283        Self {
284            rules,
285            event_tx,
286            event_rx: tokio::sync::Mutex::new(event_rx),
287        }
288    }
289
290    /// The event kinds this trigger listens for (for EventPublisher subscription).
291    ///
292    /// # Examples
293    ///
294    /// ```
295    /// use ironflow_runtime::trigger::event::{EventTrigger, EventTriggerRule};
296    /// use ironflow_store::entities::EventKind;
297    ///
298    /// let trigger = EventTrigger::new(vec![
299    ///     EventTriggerRule {
300    ///         on_event: EventKind::RunFailed,
301    ///         source_workflow: "deploy".to_string(),
302    ///         target_workflow: "rollback".to_string(),
303    ///         max_chain_depth: 3,
304    ///         conditions: vec![],
305    ///     },
306    /// ]);
307    /// let kinds = trigger.subscribed_event_types();
308    /// assert!(kinds.contains(&"run_failed"));
309    /// ```
310    pub fn subscribed_event_types(&self) -> Vec<&'static str> {
311        self.rules.iter().map(|r| r.on_event.as_str()).collect()
312    }
313
314    /// Find matching rules for a given event.
315    fn matching_rules(&self, event_kind: EventKind, workflow_name: &str) -> Vec<&EventTriggerRule> {
316        self.rules
317            .iter()
318            .filter(|r| r.on_event == event_kind && r.source_workflow == workflow_name)
319            .collect()
320    }
321
322    /// Build the payload for a triggered run.
323    fn build_payload(
324        source_run_id: Uuid,
325        source_workflow: &str,
326        error: &Option<String>,
327    ) -> serde_json::Value {
328        json!({
329            "source_run_id": source_run_id,
330            "source_workflow": source_workflow,
331            "error": error,
332        })
333    }
334
335    /// Extract the chain depth from an event, defaulting to 0.
336    fn chain_depth_from_event(_event_kind: &EventKind) -> u8 {
337        0
338    }
339}
340
341impl Trigger for EventTrigger {
342    fn name(&self) -> &str {
343        "event-trigger"
344    }
345
346    fn start<'a>(&'a self, sink: TriggerSink, token: &'a CancellationToken) -> TriggerFuture<'a> {
347        Box::pin(async move {
348            let mut rx = self.event_rx.lock().await;
349            loop {
350                tokio::select! {
351                    _ = token.cancelled() => {
352                        info!("event trigger shutting down");
353                        return Ok(());
354                    }
355                    event = rx.recv() => {
356                        let Some(event) = event else {
357                            return Ok(());
358                        };
359                        let rules = self.matching_rules(event.event_kind, &event.workflow_name);
360                        if rules.is_empty() {
361                            continue;
362                        }
363
364                        let trigger_ctx = TriggerContext {
365                            labels: event.labels.clone(),
366                            error: event.error.clone(),
367                            cost_usd: event.cost_usd,
368                            duration_ms: event.duration_ms,
369                        };
370
371                        for rule in rules {
372                            let depth = Self::chain_depth_from_event(&rule.on_event);
373                            if depth >= rule.max_chain_depth {
374                                warn!(
375                                    source_workflow = %event.workflow_name,
376                                    target_workflow = %rule.target_workflow,
377                                    chain_depth = depth,
378                                    max_chain_depth = rule.max_chain_depth,
379                                    "chain depth exceeded, ignoring event"
380                                );
381                                continue;
382                            }
383
384                            if !rule.conditions.iter().all(|c| c.evaluate(&trigger_ctx)) {
385                                info!(
386                                    source_workflow = %event.workflow_name,
387                                    target_workflow = %rule.target_workflow,
388                                    "conditions not met, skipping rule"
389                                );
390                                continue;
391                            }
392
393                            let payload = Self::build_payload(
394                                event.run_id,
395                                &event.workflow_name,
396                                &event.error,
397                            );
398
399                            let trigger_event = TriggerEvent {
400                                workflow_name: rule.target_workflow.clone(),
401                                payload,
402                                trigger_kind: TriggerKind::RunEvent {
403                                    source_run_id: event.run_id,
404                                    event_kind: rule.on_event.as_str().to_string(),
405                                },
406                            };
407
408                            if let Err(e) = sink.send(trigger_event).await {
409                                warn!(error = %e, "failed to emit trigger event");
410                            } else {
411                                info!(
412                                    source_workflow = %event.workflow_name,
413                                    target_workflow = %rule.target_workflow,
414                                    source_run_id = %event.run_id,
415                                    "event trigger fired"
416                                );
417                            }
418                        }
419                    }
420                }
421            }
422        })
423    }
424}
425
426impl EventSubscriber for EventTrigger {
427    fn name(&self) -> &str {
428        "event-trigger"
429    }
430
431    fn handle<'a>(&'a self, event: &'a Event) -> SubscriberFuture<'a> {
432        Box::pin(async move {
433            let internal = match event {
434                Event::RunFailed {
435                    run_id,
436                    workflow_name,
437                    error,
438                    cost_usd,
439                    duration_ms,
440                    labels,
441                    ..
442                } => InternalEvent {
443                    run_id: *run_id,
444                    workflow_name: workflow_name.clone(),
445                    event_kind: EventKind::RunFailed,
446                    error: error.clone(),
447                    labels: labels.clone(),
448                    cost_usd: *cost_usd,
449                    duration_ms: *duration_ms,
450                },
451                Event::RunStatusChanged {
452                    run_id,
453                    workflow_name,
454                    error,
455                    cost_usd,
456                    duration_ms,
457                    labels,
458                    ..
459                } => InternalEvent {
460                    run_id: *run_id,
461                    workflow_name: workflow_name.clone(),
462                    event_kind: EventKind::RunStatusChanged,
463                    error: error.clone(),
464                    labels: labels.clone(),
465                    cost_usd: *cost_usd,
466                    duration_ms: *duration_ms,
467                },
468                Event::StepFailed {
469                    run_id,
470                    step_name,
471                    error,
472                    ..
473                } => InternalEvent {
474                    run_id: *run_id,
475                    workflow_name: step_name.clone(),
476                    event_kind: EventKind::StepFailed,
477                    error: Some(error.clone()),
478                    labels: HashMap::new(),
479                    cost_usd: Decimal::ZERO,
480                    duration_ms: 0,
481                },
482                Event::ApprovalRejected {
483                    run_id,
484                    rejected_by,
485                    ..
486                } => InternalEvent {
487                    run_id: *run_id,
488                    workflow_name: String::new(),
489                    event_kind: EventKind::ApprovalRejected,
490                    error: Some(format!("rejected by {rejected_by}")),
491                    labels: HashMap::new(),
492                    cost_usd: Decimal::ZERO,
493                    duration_ms: 0,
494                },
495                _ => return,
496            };
497
498            if self.event_tx.send(internal).await.is_err() {
499                warn!("event trigger receiver dropped, event lost");
500            }
501        })
502    }
503}
504
505#[cfg(test)]
506mod tests {
507    use std::time::Duration;
508
509    use chrono::Utc;
510    use rust_decimal::Decimal;
511    use tokio::time::timeout;
512
513    use super::*;
514
515    fn make_trigger(rules: Vec<EventTriggerRule>) -> EventTrigger {
516        EventTrigger::new(rules)
517    }
518
519    fn deploy_to_rollback_rule() -> EventTriggerRule {
520        EventTriggerRule {
521            on_event: EventKind::RunFailed,
522            source_workflow: "deploy".to_string(),
523            target_workflow: "rollback".to_string(),
524            max_chain_depth: 3,
525            conditions: vec![],
526        }
527    }
528
529    fn internal_event(
530        run_id: Uuid,
531        workflow_name: &str,
532        event_kind: EventKind,
533        error: Option<String>,
534    ) -> InternalEvent {
535        InternalEvent {
536            run_id,
537            workflow_name: workflow_name.to_string(),
538            event_kind,
539            error,
540            labels: HashMap::new(),
541            cost_usd: Decimal::ZERO,
542            duration_ms: 0,
543        }
544    }
545
546    #[tokio::test]
547    async fn event_trigger_fires_on_matching_run_failed() {
548        let trigger = make_trigger(vec![deploy_to_rollback_rule()]);
549        let (sink, mut rx) = TriggerSink::channel(16);
550        let token = CancellationToken::new();
551        let token_clone = token.clone();
552
553        let run_id = Uuid::now_v7();
554        trigger
555            .event_tx
556            .send(internal_event(
557                run_id,
558                "deploy",
559                EventKind::RunFailed,
560                Some("step crashed".to_string()),
561            ))
562            .await
563            .unwrap();
564
565        let handle = tokio::spawn(async move { trigger.start(sink, &token_clone).await });
566
567        let event = timeout(Duration::from_secs(2), rx.recv())
568            .await
569            .expect("timed out")
570            .expect("channel closed");
571
572        assert_eq!(event.workflow_name, "rollback");
573        assert!(matches!(event.trigger_kind, TriggerKind::RunEvent { .. }));
574        if let TriggerKind::RunEvent {
575            source_run_id,
576            event_kind,
577        } = &event.trigger_kind
578        {
579            assert_eq!(*source_run_id, run_id);
580            assert_eq!(event_kind, "run_failed");
581        }
582
583        let payload = &event.payload;
584        assert_eq!(payload["source_workflow"], "deploy");
585        assert_eq!(payload["error"], "step crashed");
586
587        token.cancel();
588        let _ = handle.await;
589    }
590
591    #[tokio::test]
592    async fn event_trigger_ignores_non_matching_workflow() {
593        let trigger = make_trigger(vec![deploy_to_rollback_rule()]);
594        let (sink, mut rx) = TriggerSink::channel(16);
595        let token = CancellationToken::new();
596        let token_clone = token.clone();
597
598        trigger
599            .event_tx
600            .send(internal_event(
601                Uuid::now_v7(),
602                "build",
603                EventKind::RunFailed,
604                None,
605            ))
606            .await
607            .unwrap();
608
609        let handle = tokio::spawn(async move { trigger.start(sink, &token_clone).await });
610
611        // Give it time to process
612        tokio::time::sleep(Duration::from_millis(100)).await;
613        token.cancel();
614        let _ = handle.await;
615
616        // No event should have been emitted
617        assert!(rx.try_recv().is_err());
618    }
619
620    #[tokio::test]
621    async fn event_trigger_ignores_non_matching_event_kind() {
622        let trigger = make_trigger(vec![deploy_to_rollback_rule()]);
623        let (sink, mut rx) = TriggerSink::channel(16);
624        let token = CancellationToken::new();
625        let token_clone = token.clone();
626
627        trigger
628            .event_tx
629            .send(internal_event(
630                Uuid::now_v7(),
631                "deploy",
632                EventKind::RunStatusChanged,
633                None,
634            ))
635            .await
636            .unwrap();
637
638        let handle = tokio::spawn(async move { trigger.start(sink, &token_clone).await });
639
640        tokio::time::sleep(Duration::from_millis(100)).await;
641        token.cancel();
642        let _ = handle.await;
643
644        assert!(rx.try_recv().is_err());
645    }
646
647    #[tokio::test]
648    async fn event_trigger_payload_contains_source_info() {
649        let trigger = make_trigger(vec![deploy_to_rollback_rule()]);
650        let (sink, mut rx) = TriggerSink::channel(16);
651        let token = CancellationToken::new();
652        let token_clone = token.clone();
653
654        let run_id = Uuid::now_v7();
655        trigger
656            .event_tx
657            .send(internal_event(
658                run_id,
659                "deploy",
660                EventKind::RunFailed,
661                Some("timeout".to_string()),
662            ))
663            .await
664            .unwrap();
665
666        let handle = tokio::spawn(async move { trigger.start(sink, &token_clone).await });
667
668        let event = timeout(Duration::from_secs(2), rx.recv())
669            .await
670            .expect("timed out")
671            .expect("channel closed");
672
673        assert_eq!(event.payload["source_run_id"], run_id.to_string());
674        assert_eq!(event.payload["source_workflow"], "deploy");
675        assert_eq!(event.payload["error"], "timeout");
676
677        token.cancel();
678        let _ = handle.await;
679    }
680
681    #[test]
682    fn subscribed_event_types_reflects_rules() {
683        let trigger = make_trigger(vec![
684            EventTriggerRule {
685                on_event: EventKind::RunFailed,
686                source_workflow: "a".to_string(),
687                target_workflow: "b".to_string(),
688                max_chain_depth: 3,
689                conditions: vec![],
690            },
691            EventTriggerRule {
692                on_event: EventKind::StepFailed,
693                source_workflow: "c".to_string(),
694                target_workflow: "d".to_string(),
695                max_chain_depth: 3,
696                conditions: vec![],
697            },
698        ]);
699        let types = trigger.subscribed_event_types();
700        assert!(types.contains(&"run_failed"));
701        assert!(types.contains(&"step_failed"));
702    }
703
704    #[tokio::test]
705    async fn event_subscriber_forwards_run_failed() {
706        let trigger = make_trigger(vec![deploy_to_rollback_rule()]);
707
708        let event = Event::RunFailed {
709            run_id: Uuid::now_v7(),
710            workflow_name: "deploy".to_string(),
711            error: Some("crash".to_string()),
712            cost_usd: Decimal::ZERO,
713            duration_ms: 0,
714            labels: HashMap::new(),
715            at: Utc::now(),
716        };
717
718        // Call the EventSubscriber::handle method
719        EventSubscriber::handle(&trigger, &event).await;
720
721        // The internal channel should have the event
722        let mut rx = trigger.event_rx.lock().await;
723        let internal = rx.try_recv().unwrap();
724        assert_eq!(internal.workflow_name, "deploy");
725        assert_eq!(internal.event_kind, EventKind::RunFailed);
726    }
727
728    #[tokio::test]
729    async fn event_subscriber_ignores_irrelevant_events() {
730        let trigger = make_trigger(vec![deploy_to_rollback_rule()]);
731
732        let event = Event::RunCreated {
733            run_id: Uuid::now_v7(),
734            workflow_name: "deploy".to_string(),
735            at: Utc::now(),
736        };
737
738        EventSubscriber::handle(&trigger, &event).await;
739
740        let mut rx = trigger.event_rx.lock().await;
741        assert!(rx.try_recv().is_err());
742    }
743
744    #[tokio::test]
745    async fn graceful_shutdown() {
746        let trigger = make_trigger(vec![deploy_to_rollback_rule()]);
747        let (sink, _rx) = TriggerSink::channel(16);
748        let token = CancellationToken::new();
749        let token_clone = token.clone();
750
751        let handle = tokio::spawn(async move { trigger.start(sink, &token_clone).await });
752
753        // Trigger should be running
754        tokio::time::sleep(Duration::from_millis(50)).await;
755        assert!(!handle.is_finished());
756
757        // Cancel and verify clean shutdown
758        token.cancel();
759        let result = timeout(Duration::from_secs(2), handle)
760            .await
761            .expect("timed out")
762            .expect("task panicked");
763        assert!(result.is_ok());
764    }
765
766    fn internal_event_with_labels(
767        run_id: Uuid,
768        workflow_name: &str,
769        event_kind: EventKind,
770        error: Option<String>,
771        labels: HashMap<String, String>,
772    ) -> InternalEvent {
773        InternalEvent {
774            run_id,
775            workflow_name: workflow_name.to_string(),
776            event_kind,
777            error,
778            labels,
779            cost_usd: Decimal::new(42, 2),
780            duration_ms: 5000,
781        }
782    }
783
784    #[tokio::test]
785    async fn condition_label_matches() {
786        let rule = EventTriggerRule {
787            on_event: EventKind::RunFailed,
788            source_workflow: "deploy".to_string(),
789            target_workflow: "rollback".to_string(),
790            max_chain_depth: 3,
791            conditions: vec![TriggerCondition::Label {
792                key: "env".to_string(),
793                value: "prod".to_string(),
794            }],
795        };
796        let trigger = make_trigger(vec![rule]);
797        let (sink, mut rx) = TriggerSink::channel(16);
798        let token = CancellationToken::new();
799        let token_clone = token.clone();
800
801        let run_id = Uuid::now_v7();
802        trigger
803            .event_tx
804            .send(internal_event_with_labels(
805                run_id,
806                "deploy",
807                EventKind::RunFailed,
808                Some("crash".to_string()),
809                HashMap::from([("env".to_string(), "prod".to_string())]),
810            ))
811            .await
812            .unwrap();
813
814        let handle = tokio::spawn(async move { trigger.start(sink, &token_clone).await });
815
816        let event = timeout(Duration::from_secs(2), rx.recv())
817            .await
818            .expect("timed out")
819            .expect("channel closed");
820
821        assert_eq!(event.workflow_name, "rollback");
822        token.cancel();
823        let _ = handle.await;
824    }
825
826    #[tokio::test]
827    async fn condition_label_absent_no_fire() {
828        let rule = EventTriggerRule {
829            on_event: EventKind::RunFailed,
830            source_workflow: "deploy".to_string(),
831            target_workflow: "rollback".to_string(),
832            max_chain_depth: 3,
833            conditions: vec![TriggerCondition::Label {
834                key: "env".to_string(),
835                value: "prod".to_string(),
836            }],
837        };
838        let trigger = make_trigger(vec![rule]);
839        let (sink, mut rx) = TriggerSink::channel(16);
840        let token = CancellationToken::new();
841        let token_clone = token.clone();
842
843        trigger
844            .event_tx
845            .send(internal_event_with_labels(
846                Uuid::now_v7(),
847                "deploy",
848                EventKind::RunFailed,
849                None,
850                HashMap::new(),
851            ))
852            .await
853            .unwrap();
854
855        let handle = tokio::spawn(async move { trigger.start(sink, &token_clone).await });
856
857        tokio::time::sleep(Duration::from_millis(100)).await;
858        token.cancel();
859        let _ = handle.await;
860
861        assert!(rx.try_recv().is_err());
862    }
863
864    #[tokio::test]
865    async fn condition_label_wrong_value_no_fire() {
866        let rule = EventTriggerRule {
867            on_event: EventKind::RunFailed,
868            source_workflow: "deploy".to_string(),
869            target_workflow: "rollback".to_string(),
870            max_chain_depth: 3,
871            conditions: vec![TriggerCondition::Label {
872                key: "env".to_string(),
873                value: "prod".to_string(),
874            }],
875        };
876        let trigger = make_trigger(vec![rule]);
877        let (sink, mut rx) = TriggerSink::channel(16);
878        let token = CancellationToken::new();
879        let token_clone = token.clone();
880
881        trigger
882            .event_tx
883            .send(internal_event_with_labels(
884                Uuid::now_v7(),
885                "deploy",
886                EventKind::RunFailed,
887                None,
888                HashMap::from([("env".to_string(), "staging".to_string())]),
889            ))
890            .await
891            .unwrap();
892
893        let handle = tokio::spawn(async move { trigger.start(sink, &token_clone).await });
894
895        tokio::time::sleep(Duration::from_millis(100)).await;
896        token.cancel();
897        let _ = handle.await;
898
899        assert!(rx.try_recv().is_err());
900    }
901
902    #[tokio::test]
903    async fn multiple_conditions_all_match() {
904        let rule = EventTriggerRule {
905            on_event: EventKind::RunFailed,
906            source_workflow: "deploy".to_string(),
907            target_workflow: "rollback".to_string(),
908            max_chain_depth: 3,
909            conditions: vec![
910                TriggerCondition::Label {
911                    key: "env".to_string(),
912                    value: "prod".to_string(),
913                },
914                TriggerCondition::Label {
915                    key: "region".to_string(),
916                    value: "eu-west-1".to_string(),
917                },
918            ],
919        };
920        let trigger = make_trigger(vec![rule]);
921        let (sink, mut rx) = TriggerSink::channel(16);
922        let token = CancellationToken::new();
923        let token_clone = token.clone();
924
925        trigger
926            .event_tx
927            .send(internal_event_with_labels(
928                Uuid::now_v7(),
929                "deploy",
930                EventKind::RunFailed,
931                None,
932                HashMap::from([
933                    ("env".to_string(), "prod".to_string()),
934                    ("region".to_string(), "eu-west-1".to_string()),
935                ]),
936            ))
937            .await
938            .unwrap();
939
940        let handle = tokio::spawn(async move { trigger.start(sink, &token_clone).await });
941
942        let event = timeout(Duration::from_secs(2), rx.recv())
943            .await
944            .expect("timed out")
945            .expect("channel closed");
946
947        assert_eq!(event.workflow_name, "rollback");
948        token.cancel();
949        let _ = handle.await;
950    }
951
952    #[tokio::test]
953    async fn multiple_conditions_one_fails() {
954        let rule = EventTriggerRule {
955            on_event: EventKind::RunFailed,
956            source_workflow: "deploy".to_string(),
957            target_workflow: "rollback".to_string(),
958            max_chain_depth: 3,
959            conditions: vec![
960                TriggerCondition::Label {
961                    key: "env".to_string(),
962                    value: "prod".to_string(),
963                },
964                TriggerCondition::Label {
965                    key: "region".to_string(),
966                    value: "eu-west-1".to_string(),
967                },
968            ],
969        };
970        let trigger = make_trigger(vec![rule]);
971        let (sink, mut rx) = TriggerSink::channel(16);
972        let token = CancellationToken::new();
973        let token_clone = token.clone();
974
975        trigger
976            .event_tx
977            .send(internal_event_with_labels(
978                Uuid::now_v7(),
979                "deploy",
980                EventKind::RunFailed,
981                None,
982                HashMap::from([("env".to_string(), "prod".to_string())]),
983            ))
984            .await
985            .unwrap();
986
987        let handle = tokio::spawn(async move { trigger.start(sink, &token_clone).await });
988
989        tokio::time::sleep(Duration::from_millis(100)).await;
990        token.cancel();
991        let _ = handle.await;
992
993        assert!(rx.try_recv().is_err());
994    }
995
996    #[tokio::test]
997    async fn empty_conditions_backward_compat() {
998        let rule = EventTriggerRule {
999            on_event: EventKind::RunFailed,
1000            source_workflow: "deploy".to_string(),
1001            target_workflow: "rollback".to_string(),
1002            max_chain_depth: 3,
1003            conditions: vec![],
1004        };
1005        let trigger = make_trigger(vec![rule]);
1006        let (sink, mut rx) = TriggerSink::channel(16);
1007        let token = CancellationToken::new();
1008        let token_clone = token.clone();
1009
1010        trigger
1011            .event_tx
1012            .send(internal_event(
1013                Uuid::now_v7(),
1014                "deploy",
1015                EventKind::RunFailed,
1016                Some("boom".to_string()),
1017            ))
1018            .await
1019            .unwrap();
1020
1021        let handle = tokio::spawn(async move { trigger.start(sink, &token_clone).await });
1022
1023        let event = timeout(Duration::from_secs(2), rx.recv())
1024            .await
1025            .expect("timed out")
1026            .expect("channel closed");
1027
1028        assert_eq!(event.workflow_name, "rollback");
1029        token.cancel();
1030        let _ = handle.await;
1031    }
1032
1033    #[tokio::test]
1034    async fn expression_condition_with_context() {
1035        let rule = EventTriggerRule {
1036            on_event: EventKind::RunFailed,
1037            source_workflow: "deploy".to_string(),
1038            target_workflow: "rollback".to_string(),
1039            max_chain_depth: 3,
1040            conditions: vec![TriggerCondition::Expression(Arc::new(|ctx| {
1041                ctx.cost_usd > Decimal::new(10, 2) && ctx.duration_ms > 1000
1042            }))],
1043        };
1044        let trigger = make_trigger(vec![rule]);
1045        let (sink, mut rx) = TriggerSink::channel(16);
1046        let token = CancellationToken::new();
1047        let token_clone = token.clone();
1048
1049        trigger
1050            .event_tx
1051            .send(internal_event_with_labels(
1052                Uuid::now_v7(),
1053                "deploy",
1054                EventKind::RunFailed,
1055                None,
1056                HashMap::new(),
1057            ))
1058            .await
1059            .unwrap();
1060
1061        let handle = tokio::spawn(async move { trigger.start(sink, &token_clone).await });
1062
1063        let event = timeout(Duration::from_secs(2), rx.recv())
1064            .await
1065            .expect("timed out")
1066            .expect("channel closed");
1067
1068        assert_eq!(event.workflow_name, "rollback");
1069        token.cancel();
1070        let _ = handle.await;
1071    }
1072
1073    #[tokio::test]
1074    async fn expression_returns_false_no_fire() {
1075        let rule = EventTriggerRule {
1076            on_event: EventKind::RunFailed,
1077            source_workflow: "deploy".to_string(),
1078            target_workflow: "rollback".to_string(),
1079            max_chain_depth: 3,
1080            conditions: vec![TriggerCondition::Expression(Arc::new(|ctx| {
1081                ctx.cost_usd > Decimal::new(100, 0)
1082            }))],
1083        };
1084        let trigger = make_trigger(vec![rule]);
1085        let (sink, mut rx) = TriggerSink::channel(16);
1086        let token = CancellationToken::new();
1087        let token_clone = token.clone();
1088
1089        trigger
1090            .event_tx
1091            .send(internal_event_with_labels(
1092                Uuid::now_v7(),
1093                "deploy",
1094                EventKind::RunFailed,
1095                None,
1096                HashMap::new(),
1097            ))
1098            .await
1099            .unwrap();
1100
1101        let handle = tokio::spawn(async move { trigger.start(sink, &token_clone).await });
1102
1103        tokio::time::sleep(Duration::from_millis(100)).await;
1104        token.cancel();
1105        let _ = handle.await;
1106
1107        assert!(rx.try_recv().is_err());
1108    }
1109}
ironflow_runtime/trigger/event.rs

ironflow_runtime/trigger/
event.rs
