faucet_cli/notify/
spec.rs

1//! Config types for the `notifications:` block (#280).
2//!
3//! One [`NotificationSpec`] is a rule: which [`EventKind`]s it fires on, a
4//! severity floor, an optional leading-edge coalesce window, and a single
5//! delivery [`ChannelSpec`]. The channel uses the project-wide adjacently
6//! tagged `{ type, config }` shape (matching connectors / shared auth /
7//! lineage transports).
8//!
9//! Everything here is pure data + validation. Dispatch lives in
10//! [`crate::notify::dispatch`]; rendering in [`crate::notify::render`].
11
12use crate::error::{CliError, CliResult};
13use schemars::JsonSchema;
14use serde::{Deserialize, Serialize};
15
16/// A single notification rule.
17#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
18#[serde(deny_unknown_fields)]
19pub struct NotificationSpec {
20    /// Stable, human-readable name. Used in metric labels, dedupe keys, and
21    /// log lines — keep it unique across rules.
22    pub name: String,
23
24    /// Event kinds this rule fires on. **Empty = every kind.**
25    #[serde(default)]
26    pub on: Vec<EventKind>,
27
28    /// Only deliver events at or above this severity. Defaults to `info`
29    /// (deliver everything the `on` selector matches).
30    #[serde(default)]
31    pub min_severity: Severity,
32
33    /// Coalesce repeated identical events (same rule + dedupe key) within this
34    /// many seconds — leading-edge: the first event fires, subsequent ones
35    /// inside the window are dropped as `coalesced`. Absent / `0` disables
36    /// coalescing.
37    #[serde(default)]
38    pub dedupe_window_secs: Option<u64>,
39
40    /// For the `dlq_threshold` event only: fire when a run routed **at least**
41    /// this many rows to the DLQ. Defaults to `1` (any DLQ traffic).
42    #[serde(default)]
43    pub dlq_threshold: Option<u64>,
44
45    /// The delivery channel.
46    pub channel: ChannelSpec,
47}
48
49/// The lifecycle / health events a rule can subscribe to.
50#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, JsonSchema)]
51#[serde(rename_all = "snake_case")]
52pub enum EventKind {
53    /// A pipeline run (or its final flush) failed.
54    RunFailure,
55    /// A pipeline run completed successfully.
56    RunSuccess,
57    /// A post-run SLA check was violated (staleness / min_rows / volume).
58    SlaBreach,
59    /// The resilience circuit breaker tripped open.
60    CircuitOpen,
61    /// A data contract breach aborted the run (`on_breach: fail`).
62    ContractAbort,
63    /// A run routed rows to the dead-letter queue at/over the configured
64    /// threshold.
65    DlqThreshold,
66    /// The cron scheduler appears stuck (no heartbeat / consecutive-failure
67    /// exit). Emitted by `faucet schedule`.
68    SchedulerStuck,
69}
70
71impl EventKind {
72    /// Stable metric-label / log string.
73    pub fn as_str(self) -> &'static str {
74        match self {
75            EventKind::RunFailure => "run_failure",
76            EventKind::RunSuccess => "run_success",
77            EventKind::SlaBreach => "sla_breach",
78            EventKind::CircuitOpen => "circuit_open",
79            EventKind::ContractAbort => "contract_abort",
80            EventKind::DlqThreshold => "dlq_threshold",
81            EventKind::SchedulerStuck => "scheduler_stuck",
82        }
83    }
84}
85
86/// Event severity — ordered so `min_severity` can gate with `>=`. A rule with
87/// no `min_severity` defaults to the lowest floor (`info`), so it delivers
88/// everything it subscribes to.
89#[derive(
90    Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Default, Serialize, Deserialize, JsonSchema,
91)]
92#[serde(rename_all = "snake_case")]
93pub enum Severity {
94    /// Informational (e.g. `run_success`).
95    #[default]
96    Info,
97    /// A degraded but non-failing condition (e.g. `sla_breach`, `dlq_threshold`).
98    Warning,
99    /// A failure that stopped the run (e.g. `run_failure`, `contract_abort`).
100    Error,
101    /// A systemic failure needing immediate attention (e.g. `circuit_open`,
102    /// `scheduler_stuck`).
103    Critical,
104}
105
106impl Severity {
107    /// PagerDuty Events-API severity string.
108    pub fn as_pagerduty(self) -> &'static str {
109        match self {
110            Severity::Info => "info",
111            Severity::Warning => "warning",
112            Severity::Error => "error",
113            Severity::Critical => "critical",
114        }
115    }
116
117    /// Stable metric-label / log string.
118    pub fn as_str(self) -> &'static str {
119        match self {
120            Severity::Info => "info",
121            Severity::Warning => "warning",
122            Severity::Error => "error",
123            Severity::Critical => "critical",
124        }
125    }
126}
127
128/// A delivery channel. Adjacently tagged `{ type, config }` to match the
129/// project-wide connector / auth / lineage-transport shape.
130#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
131#[serde(tag = "type", content = "config", rename_all = "snake_case")]
132pub enum ChannelSpec {
133    /// Slack incoming webhook.
134    Slack(SlackConfig),
135    /// PagerDuty Events API v2 (trigger + auto-resolve).
136    Pagerduty(PagerdutyConfig),
137    /// Generic HTTP POST with an optional HMAC-SHA256 signature.
138    Webhook(WebhookConfig),
139}
140
141impl ChannelSpec {
142    /// Stable channel-kind label for metrics/logs.
143    pub fn kind(&self) -> &'static str {
144        match self {
145            ChannelSpec::Slack(_) => "slack",
146            ChannelSpec::Pagerduty(_) => "pagerduty",
147            ChannelSpec::Webhook(_) => "webhook",
148        }
149    }
150}
151
152/// Slack incoming-webhook config.
153#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
154#[serde(deny_unknown_fields)]
155pub struct SlackConfig {
156    /// Incoming-webhook URL. Supply via `${env:...}` / `${secret:...}` so it is
157    /// redacted from logs.
158    pub webhook_url: String,
159    /// Optional channel override (`#alerts`).
160    #[serde(default)]
161    pub channel: Option<String>,
162    /// Optional bot username override.
163    #[serde(default)]
164    pub username: Option<String>,
165}
166
167/// PagerDuty Events API v2 config.
168#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
169#[serde(deny_unknown_fields)]
170pub struct PagerdutyConfig {
171    /// Events API v2 integration (routing) key.
172    pub routing_key: String,
173    /// Optional `source` field for the PD payload (defaults to the pipeline
174    /// name).
175    #[serde(default)]
176    pub source: Option<String>,
177    /// Override the events endpoint (tests / EU service region). Defaults to
178    /// the global US endpoint.
179    #[serde(default)]
180    pub endpoint: Option<String>,
181}
182
183/// Generic signed-webhook config.
184#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
185#[serde(deny_unknown_fields)]
186pub struct WebhookConfig {
187    /// Destination URL.
188    pub url: String,
189    /// HTTP method — defaults to `POST`.
190    #[serde(default = "default_webhook_method")]
191    pub method: String,
192    /// Extra headers sent with every request.
193    #[serde(default)]
194    pub headers: std::collections::BTreeMap<String, String>,
195    /// If set, sign the JSON body with HMAC-SHA256 and send the lowercase-hex
196    /// digest in `signature_header`. Supply via `${env:...}` / `${secret:...}`.
197    #[serde(default)]
198    pub hmac_secret: Option<String>,
199    /// Header carrying the HMAC signature (default `X-Faucet-Signature`).
200    #[serde(default = "default_signature_header")]
201    pub signature_header: String,
202}
203
204fn default_webhook_method() -> String {
205    "POST".to_string()
206}
207
208fn default_signature_header() -> String {
209    "X-Faucet-Signature".to_string()
210}
211
212impl NotificationSpec {
213    /// Fail-fast structural validation (called at load time so misconfiguration
214    /// surfaces from `faucet validate`, never mid-run).
215    pub fn validate(&self) -> CliResult<()> {
216        if self.name.trim().is_empty() {
217            return Err(CliError::Config(
218                "notifications: each rule needs a non-empty `name`".into(),
219            ));
220        }
221        let bad = |field: &str| {
222            Err(CliError::Config(format!(
223                "notifications rule `{}`: `{field}` must be non-empty",
224                self.name
225            )))
226        };
227        match &self.channel {
228            ChannelSpec::Slack(c) if c.webhook_url.trim().is_empty() => bad("webhook_url"),
229            ChannelSpec::Pagerduty(c) if c.routing_key.trim().is_empty() => bad("routing_key"),
230            ChannelSpec::Webhook(c) if c.url.trim().is_empty() => bad("url"),
231            ChannelSpec::Webhook(c) if c.method.trim().is_empty() => bad("method"),
232            _ => Ok(()),
233        }
234    }
235}
236
237/// Validate a whole `notifications:` list (unique names + per-rule validation).
238pub fn validate_all(specs: &[NotificationSpec]) -> CliResult<()> {
239    let mut seen = std::collections::HashSet::new();
240    for s in specs {
241        s.validate()?;
242        if !seen.insert(s.name.as_str()) {
243            return Err(CliError::Config(format!(
244                "notifications: duplicate rule name `{}`",
245                s.name
246            )));
247        }
248    }
249    Ok(())
250}
251
252#[cfg(test)]
253mod tests {
254    use super::*;
255
256    fn slack(name: &str, url: &str) -> NotificationSpec {
257        NotificationSpec {
258            name: name.into(),
259            on: vec![EventKind::RunFailure],
260            min_severity: Severity::default(),
261            dedupe_window_secs: None,
262            dlq_threshold: None,
263            channel: ChannelSpec::Slack(SlackConfig {
264                webhook_url: url.into(),
265                channel: None,
266                username: None,
267            }),
268        }
269    }
270
271    #[test]
272    fn severity_orders_low_to_high() {
273        assert!(Severity::Info < Severity::Warning);
274        assert!(Severity::Warning < Severity::Error);
275        assert!(Severity::Error < Severity::Critical);
276        assert_eq!(Severity::default(), Severity::Info);
277    }
278
279    #[test]
280    fn channel_kind_labels() {
281        assert_eq!(slack("a", "u").channel.kind(), "slack");
282    }
283
284    #[test]
285    fn empty_name_rejected() {
286        let s = slack("  ", "u");
287        assert!(s.validate().is_err());
288    }
289
290    #[test]
291    fn empty_channel_field_rejected() {
292        let s = slack("a", "");
293        assert!(s.validate().is_err());
294    }
295
296    #[test]
297    fn duplicate_names_rejected() {
298        let list = vec![slack("dup", "u1"), slack("dup", "u2")];
299        assert!(validate_all(&list).is_err());
300    }
301
302    #[test]
303    fn valid_list_passes() {
304        let list = vec![slack("a", "u1"), slack("b", "u2")];
305        assert!(validate_all(&list).is_ok());
306    }
307
308    #[test]
309    fn adjacently_tagged_channel_roundtrips() {
310        let json = serde_json::json!({
311            "name": "x",
312            "on": ["run_failure", "circuit_open"],
313            "channel": { "type": "pagerduty", "config": { "routing_key": "k" } }
314        });
315        let s: NotificationSpec = serde_json::from_value(json).unwrap();
316        assert_eq!(s.on, vec![EventKind::RunFailure, EventKind::CircuitOpen]);
317        assert_eq!(s.channel.kind(), "pagerduty");
318        s.validate().unwrap();
319    }
320
321    #[test]
322    fn webhook_defaults_applied() {
323        let json = serde_json::json!({
324            "name": "w",
325            "channel": { "type": "webhook", "config": { "url": "http://x" } }
326        });
327        let s: NotificationSpec = serde_json::from_value(json).unwrap();
328        match &s.channel {
329            ChannelSpec::Webhook(c) => {
330                assert_eq!(c.method, "POST");
331                assert_eq!(c.signature_header, "X-Faucet-Signature");
332            }
333            _ => panic!("expected webhook"),
334        }
335        // Empty `on` means "all kinds".
336        assert!(s.on.is_empty());
337    }
338
339    #[test]
340    fn pagerduty_severity_strings() {
341        assert_eq!(Severity::Critical.as_pagerduty(), "critical");
342        assert_eq!(Severity::Info.as_pagerduty(), "info");
343    }
344}
faucet_cli/notify/spec.rs

faucet_cli/notify/
spec.rs
