faucet_cli/notify/
render.rs

1//! Pure per-channel payload rendering (#280).
2//!
3//! Each function turns a [`NotifyEvent`] into the JSON body a channel expects.
4//! No I/O, no secrets — the caller ([`crate::notify::dispatch`]) owns the HTTP
5//! client and injects credentials as headers / URL. Kept pure so the rendering
6//! is fully unit-testable.
7
8use super::event::NotifyEvent;
9use super::spec::{PagerdutyConfig, SlackConfig, WebhookConfig};
10use serde_json::{Value, json};
11
12/// PagerDuty Events API v2 action.
13#[derive(Debug, Clone, Copy, PartialEq, Eq)]
14pub enum PdAction {
15    Trigger,
16    Resolve,
17}
18
19impl PdAction {
20    fn as_str(self) -> &'static str {
21        match self {
22            PdAction::Trigger => "trigger",
23            PdAction::Resolve => "resolve",
24        }
25    }
26}
27
28/// Render a Slack incoming-webhook body (Block Kit).
29pub fn slack(cfg: &SlackConfig, event: &NotifyEvent) -> Value {
30    let emoji = match event.severity {
31        super::spec::Severity::Critical => "🚨",
32        super::spec::Severity::Error => "❌",
33        super::spec::Severity::Warning => "⚠️",
34        super::spec::Severity::Info => "✅",
35    };
36    let mut context_fields = vec![
37        format!("*Pipeline:* {}", event.pipeline),
38        format!("*Severity:* {}", event.severity.as_str()),
39    ];
40    if !event.row.is_empty() {
41        context_fields.push(format!("*Row:* {}", event.row));
42    }
43    for (k, v) in &event.details {
44        context_fields.push(format!("*{k}:* {}", scalar(v)));
45    }
46
47    let mut body = json!({
48        "text": format!("{emoji} {}", event.title),
49        "blocks": [
50            {
51                "type": "section",
52                "text": { "type": "mrkdwn", "text": format!("{emoji} *{}*\n{}", event.title, event.message) }
53            },
54            {
55                "type": "context",
56                "elements": [ { "type": "mrkdwn", "text": context_fields.join("  •  ") } ]
57            }
58        ]
59    });
60    if let Some(ch) = &cfg.channel {
61        body["channel"] = Value::String(ch.clone());
62    }
63    if let Some(user) = &cfg.username {
64        body["username"] = Value::String(user.clone());
65    }
66    body
67}
68
69/// Render a PagerDuty Events API v2 payload. `dedup_key` correlates a
70/// `resolve` with the `trigger` that opened the incident.
71pub fn pagerduty(
72    cfg: &PagerdutyConfig,
73    event: &NotifyEvent,
74    action: PdAction,
75    dedup_key: &str,
76) -> Value {
77    let source = cfg.source.clone().unwrap_or_else(|| event.pipeline.clone());
78    let mut body = json!({
79        "routing_key": cfg.routing_key,
80        "event_action": action.as_str(),
81        "dedup_key": dedup_key,
82    });
83    // A `resolve` carries only routing_key + action + dedup_key; a `trigger`
84    // carries the full payload.
85    if action == PdAction::Trigger {
86        body["payload"] = json!({
87            "summary": event.title,
88            "source": source,
89            "severity": event.severity.as_pagerduty(),
90            "custom_details": {
91                "message": event.message,
92                "pipeline": event.pipeline,
93                "row": event.row,
94                "details": Value::Object(event.details.clone()),
95            }
96        });
97    }
98    body
99}
100
101/// Render a generic webhook body — a stable, machine-readable envelope.
102pub fn webhook(cfg: &WebhookConfig, event: &NotifyEvent) -> Value {
103    let run = event.run.as_ref();
104    let mut body = json!({
105        "event": event.kind.as_str(),
106        "severity": event.severity.as_str(),
107        "pipeline": event.pipeline,
108        "row": event.row,
109        "title": event.title,
110        "message": event.message,
111        "details": Value::Object(event.details.clone()),
112        // Emitted as explicit nulls rather than omitted, so receivers can rely
113        // on a stable key set regardless of whether the event had an owning
114        // invocation (#480).
115        "run_id": run.and_then(|r| r.run_id.clone()).map_or(Value::Null, Value::String),
116        "invocation_id": run
117            .and_then(|r| r.invocation_id.clone())
118            .map_or(Value::Null, Value::String),
119        "started_at": run
120            .and_then(|r| r.started_at)
121            .map_or(Value::Null, |t| Value::String(t.to_rfc3339())),
122        "finished_at": run
123            .and_then(|r| r.finished_at)
124            .map_or(Value::Null, |t| Value::String(t.to_rfc3339())),
125        "duration_secs": run
126            .and_then(|r| r.duration)
127            .and_then(|d| serde_json::Number::from_f64(d.as_secs_f64()))
128            .map_or(Value::Null, Value::Number),
129    });
130
131    // Operator-authored static metadata. Reserved-key collisions are rejected at
132    // config-load time (`NotificationSpec::validate`), so this cannot shadow a
133    // faucet-emitted field.
134    if !cfg.extra_fields.is_empty()
135        && let Some(map) = body.as_object_mut()
136    {
137        for (k, v) in &cfg.extra_fields {
138            map.insert(k.clone(), v.clone());
139        }
140    }
141    body
142}
143
144/// Best-effort scalar stringification for Slack context lines.
145fn scalar(v: &Value) -> String {
146    match v {
147        Value::String(s) => s.clone(),
148        other => other.to_string(),
149    }
150}
151
152#[cfg(test)]
153mod tests {
154    use super::*;
155    use crate::notify::spec::Severity;
156
157    fn slack_cfg() -> SlackConfig {
158        SlackConfig {
159            webhook_url: "http://x".into(),
160            channel: Some("#alerts".into()),
161            username: Some("faucet".into()),
162        }
163    }
164
165    #[test]
166    fn slack_body_has_blocks_and_overrides() {
167        let e = NotifyEvent::run_failure("orders", "row1", "sink", "connection refused");
168        let body = slack(&slack_cfg(), &e);
169        assert_eq!(body["channel"], "#alerts");
170        assert_eq!(body["username"], "faucet");
171        let text = body["text"].as_str().unwrap();
172        assert!(text.contains("failed"));
173        // context block mentions the row + error_kind detail
174        let ctx = body["blocks"][1]["elements"][0]["text"].as_str().unwrap();
175        assert!(ctx.contains("row1"));
176        assert!(ctx.contains("error_kind"));
177    }
178
179    #[test]
180    fn slack_omits_row_when_empty() {
181        let e = NotifyEvent::scheduler_stuck("p", "no heartbeat");
182        let body = slack(
183            &SlackConfig {
184                webhook_url: "u".into(),
185                channel: None,
186                username: None,
187            },
188            &e,
189        );
190        let ctx = body["blocks"][1]["elements"][0]["text"].as_str().unwrap();
191        assert!(!ctx.contains("*Row:*"));
192        assert!(body.get("channel").is_none());
193    }
194
195    #[test]
196    fn pagerduty_trigger_carries_payload() {
197        let cfg = PagerdutyConfig {
198            routing_key: "rk".into(),
199            source: None,
200            endpoint: None,
201        };
202        let e = NotifyEvent::circuit_open("p", "", 5, 30);
203        let body = pagerduty(&cfg, &e, PdAction::Trigger, "p:");
204        assert_eq!(body["event_action"], "trigger");
205        assert_eq!(body["routing_key"], "rk");
206        assert_eq!(body["dedup_key"], "p:");
207        assert_eq!(body["payload"]["severity"], "critical");
208        assert_eq!(body["payload"]["source"], "p"); // defaults to pipeline
209        assert_eq!(body["payload"]["custom_details"]["details"]["failures"], 5);
210    }
211
212    #[test]
213    fn pagerduty_resolve_is_minimal() {
214        let cfg = PagerdutyConfig {
215            routing_key: "rk".into(),
216            source: Some("svc".into()),
217            endpoint: None,
218        };
219        let e = NotifyEvent::run_success("p", "", 1);
220        let body = pagerduty(&cfg, &e, PdAction::Resolve, "p:");
221        assert_eq!(body["event_action"], "resolve");
222        assert_eq!(body["dedup_key"], "p:");
223        assert!(body.get("payload").is_none());
224    }
225
226    #[test]
227    fn webhook_envelope_is_stable() {
228        let cfg = WebhookConfig {
229            url: "u".into(),
230            method: "POST".into(),
231            headers: Default::default(),
232            hmac_secret: None,
233            signature_header: "X-Faucet-Signature".into(),
234            extra_fields: Default::default(),
235        };
236        let e = NotifyEvent::dlq_threshold("p", "r", 42);
237        let body = webhook(&cfg, &e);
238        assert_eq!(body["event"], "dlq_threshold");
239        assert_eq!(body["severity"], Severity::Warning.as_str());
240        assert_eq!(body["details"]["records_dlq"], 42);
241    }
242
243    fn webhook_cfg() -> WebhookConfig {
244        WebhookConfig {
245            url: "u".into(),
246            method: "POST".into(),
247            headers: Default::default(),
248            hmac_secret: None,
249            signature_header: "X-Faucet-Signature".into(),
250            extra_fields: Default::default(),
251        }
252    }
253
254    #[test]
255    fn webhook_carries_run_identity_and_timing() {
256        // #480: the payload must let a receiver correlate the callback back to
257        // the run it triggered. `pipeline` + `row` alone cannot — two
258        // overlapping runs of one pipeline share both.
259        let run =
260            crate::notify::RunContext::start(Some("run-abc".into()), Some("invocation-1".into()))
261                .finish(std::time::Instant::now());
262        let e = NotifyEvent::run_success("p", "r", 7).with_run(run);
263        let body = webhook(&webhook_cfg(), &e);
264
265        assert_eq!(body["run_id"], "run-abc");
266        assert_eq!(body["invocation_id"], "invocation-1");
267        assert!(body["started_at"].is_string(), "started_at must be RFC3339");
268        assert!(body["finished_at"].is_string());
269        assert!(
270            body["duration_secs"].as_f64().unwrap() >= 0.0,
271            "monotonic duration is never negative"
272        );
273    }
274
275    #[test]
276    fn webhook_emits_explicit_nulls_when_no_run_context() {
277        // Stable key set regardless of whether the event had an owning
278        // invocation — receivers must not have to branch on key presence.
279        let e = NotifyEvent::scheduler_stuck("p", "no heartbeat");
280        let body = webhook(&webhook_cfg(), &e);
281        for k in [
282            "run_id",
283            "invocation_id",
284            "started_at",
285            "finished_at",
286            "duration_secs",
287        ] {
288            assert!(body.get(k).is_some(), "{k} key must be present");
289            assert!(body[k].is_null(), "{k} must be null, not omitted");
290        }
291    }
292
293    #[test]
294    fn webhook_merges_extra_fields() {
295        let mut cfg = webhook_cfg();
296        cfg.extra_fields
297            .insert("tenant".into(), Value::String("acme".into()));
298        cfg.extra_fields.insert("attempt".into(), Value::from(2));
299        let e = NotifyEvent::run_success("p", "r", 1);
300        let body = webhook(&cfg, &e);
301        assert_eq!(body["tenant"], "acme");
302        assert_eq!(body["attempt"], 2);
303        // Faucet-emitted fields still intact.
304        assert_eq!(body["event"], "run_success");
305    }
306
307    #[test]
308    fn two_runs_of_one_pipeline_are_distinguishable() {
309        // The regression this feature exists to prevent: identical pipeline+row,
310        // different runs.
311        let mk = |id: &str| {
312            NotifyEvent::run_success("p", "r", 1).with_run(
313                crate::notify::RunContext::start(Some(id.into()), Some(id.into()))
314                    .finish(std::time::Instant::now()),
315            )
316        };
317        let a = webhook(&webhook_cfg(), &mk("run-a"));
318        let b = webhook(&webhook_cfg(), &mk("run-b"));
319        assert_eq!(a["pipeline"], b["pipeline"]);
320        assert_eq!(a["row"], b["row"]);
321        assert_ne!(a["run_id"], b["run_id"]);
322    }
323}
faucet_cli/notify/render.rs

faucet_cli/notify/
render.rs
