faucet_cli/notify/
event.rs

1//! The runtime event a pipeline emits toward the notifier (#280).
2//!
3//! [`NotifyEvent`] is the pure, channel-agnostic description of something that
4//! happened. Constructors fix the canonical severity per kind so callers at the
5//! emit sites (executor, SLA pass, scheduler) don't have to. Rendering into a
6//! Slack / PagerDuty / webhook body lives in [`crate::notify::render`].
7
8use super::spec::{EventKind, Severity};
9use serde_json::{Map, Value};
10
11/// A single thing worth notifying about.
12#[derive(Debug, Clone)]
13pub struct NotifyEvent {
14    pub kind: EventKind,
15    pub severity: Severity,
16    /// Pipeline name (metric-label identity).
17    pub pipeline: String,
18    /// Matrix row id (`""` for non-matrix runs).
19    pub row: String,
20    /// Short one-line summary.
21    pub title: String,
22    /// Human-readable detail.
23    pub message: String,
24    /// Structured context rendered into channel payloads (never contains
25    /// secrets — the emit sites pass only safe scalars).
26    pub details: Map<String, Value>,
27}
28
29impl NotifyEvent {
30    fn base(
31        kind: EventKind,
32        severity: Severity,
33        pipeline: impl Into<String>,
34        row: impl Into<String>,
35        title: impl Into<String>,
36        message: impl Into<String>,
37    ) -> Self {
38        Self {
39            kind,
40            severity,
41            pipeline: pipeline.into(),
42            row: row.into(),
43            title: title.into(),
44            message: message.into(),
45            details: Map::new(),
46        }
47    }
48
49    fn with(mut self, key: &str, value: Value) -> Self {
50        self.details.insert(key.to_string(), value);
51        self
52    }
53
54    /// Correlation key for PagerDuty incident open/resolve pairing and for
55    /// coalescing per (pipeline, row). A `run_success` resolves the incident a
56    /// prior `run_failure` opened on the same key.
57    pub fn incident_key(&self) -> String {
58        format!("{}:{}", self.pipeline, self.row)
59    }
60
61    /// Per-rule coalesce key: the same kind repeating for the same (pipeline,
62    /// row) coalesces within a rule's `dedupe_window_secs`.
63    pub fn dedupe_key(&self) -> String {
64        format!("{}:{}:{}", self.kind.as_str(), self.pipeline, self.row)
65    }
66
67    /// True when this event opens an incident (a failure-class event that a
68    /// later success should resolve).
69    pub fn opens_incident(&self) -> bool {
70        matches!(
71            self.kind,
72            EventKind::RunFailure | EventKind::CircuitOpen | EventKind::ContractAbort
73        )
74    }
75
76    /// True when this event closes any open incident for its `incident_key`.
77    pub fn closes_incident(&self) -> bool {
78        matches!(self.kind, EventKind::RunSuccess)
79    }
80
81    // ── Constructors (canonical severity per kind) ───────────────────────────
82
83    pub fn run_failure(
84        pipeline: impl Into<String>,
85        row: impl Into<String>,
86        error_kind: &str,
87        message: impl Into<String>,
88    ) -> Self {
89        let p = pipeline.into();
90        Self::base(
91            EventKind::RunFailure,
92            Severity::Error,
93            p.clone(),
94            row,
95            format!("Pipeline `{p}` failed"),
96            message,
97        )
98        .with("error_kind", Value::String(error_kind.to_string()))
99    }
100
101    pub fn run_success(
102        pipeline: impl Into<String>,
103        row: impl Into<String>,
104        rows_written: u64,
105    ) -> Self {
106        let p = pipeline.into();
107        Self::base(
108            EventKind::RunSuccess,
109            Severity::Info,
110            p.clone(),
111            row,
112            format!("Pipeline `{p}` succeeded"),
113            format!("Run completed, {rows_written} records written."),
114        )
115        .with("records_written", Value::from(rows_written))
116    }
117
118    pub fn sla_breach(
119        pipeline: impl Into<String>,
120        row: impl Into<String>,
121        sla_kind: &str,
122        message: impl Into<String>,
123    ) -> Self {
124        let p = pipeline.into();
125        Self::base(
126            EventKind::SlaBreach,
127            Severity::Warning,
128            p.clone(),
129            row,
130            format!("SLA breach ({sla_kind}) on `{p}`"),
131            message,
132        )
133        .with("sla_kind", Value::String(sla_kind.to_string()))
134    }
135
136    pub fn circuit_open(
137        pipeline: impl Into<String>,
138        row: impl Into<String>,
139        failures: u32,
140        cooldown_secs: u64,
141    ) -> Self {
142        let p = pipeline.into();
143        Self::base(
144            EventKind::CircuitOpen,
145            Severity::Critical,
146            p.clone(),
147            row,
148            format!("Circuit breaker open on `{p}`"),
149            format!(
150                "Tripped after {failures} consecutive failures; cooling down {cooldown_secs}s."
151            ),
152        )
153        .with("failures", Value::from(failures))
154        .with("cooldown_secs", Value::from(cooldown_secs))
155    }
156
157    pub fn contract_abort(
158        pipeline: impl Into<String>,
159        row: impl Into<String>,
160        message: impl Into<String>,
161    ) -> Self {
162        let p = pipeline.into();
163        Self::base(
164            EventKind::ContractAbort,
165            Severity::Error,
166            p.clone(),
167            row,
168            format!("Data contract breach aborted `{p}`"),
169            message,
170        )
171    }
172
173    pub fn dlq_threshold(
174        pipeline: impl Into<String>,
175        row: impl Into<String>,
176        records_dlq: u64,
177    ) -> Self {
178        let p = pipeline.into();
179        Self::base(
180            EventKind::DlqThreshold,
181            Severity::Warning,
182            p.clone(),
183            row,
184            format!("DLQ threshold reached on `{p}`"),
185            format!("{records_dlq} records were routed to the dead-letter queue."),
186        )
187        .with("records_dlq", Value::from(records_dlq))
188    }
189
190    pub fn scheduler_stuck(pipeline: impl Into<String>, message: impl Into<String>) -> Self {
191        let p = pipeline.into();
192        Self::base(
193            EventKind::SchedulerStuck,
194            Severity::Critical,
195            p.clone(),
196            String::new(),
197            format!("Scheduler stuck for `{p}`"),
198            message,
199        )
200    }
201}
202
203#[cfg(test)]
204mod tests {
205    use super::*;
206
207    #[test]
208    fn constructors_fix_severity_and_kind() {
209        assert_eq!(
210            NotifyEvent::run_failure("p", "", "sink", "boom").severity,
211            Severity::Error
212        );
213        assert_eq!(
214            NotifyEvent::circuit_open("p", "", 5, 30).severity,
215            Severity::Critical
216        );
217        assert_eq!(
218            NotifyEvent::run_success("p", "", 10).severity,
219            Severity::Info
220        );
221        assert_eq!(
222            NotifyEvent::sla_breach("p", "", "staleness", "old").severity,
223            Severity::Warning
224        );
225        assert_eq!(
226            NotifyEvent::scheduler_stuck("p", "no beat").kind,
227            EventKind::SchedulerStuck
228        );
229    }
230
231    #[test]
232    fn incident_and_dedupe_keys() {
233        let f = NotifyEvent::run_failure("p", "r1", "sink", "boom");
234        assert_eq!(f.incident_key(), "p:r1");
235        assert_eq!(f.dedupe_key(), "run_failure:p:r1");
236        assert!(f.opens_incident());
237        assert!(!f.closes_incident());
238
239        let s = NotifyEvent::run_success("p", "r1", 3);
240        assert_eq!(s.incident_key(), "p:r1"); // matches the failure it resolves
241        assert!(s.closes_incident());
242        assert!(!s.opens_incident());
243    }
244
245    #[test]
246    fn details_carry_structured_context() {
247        let e = NotifyEvent::dlq_threshold("p", "", 42);
248        assert_eq!(e.details.get("records_dlq").unwrap(), &Value::from(42u64));
249    }
250}
faucet_cli/notify/event.rs

faucet_cli/notify/
event.rs
