1use super::event::NotifyEvent;
9use super::spec::{PagerdutyConfig, SlackConfig, WebhookConfig};
10use serde_json::{Value, json};
11
12#[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
28pub 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
69pub 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 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
101pub fn webhook(_cfg: &WebhookConfig, event: &NotifyEvent) -> Value {
103 json!({
104 "event": event.kind.as_str(),
105 "severity": event.severity.as_str(),
106 "pipeline": event.pipeline,
107 "row": event.row,
108 "title": event.title,
109 "message": event.message,
110 "details": Value::Object(event.details.clone()),
111 })
112}
113
114fn scalar(v: &Value) -> String {
116 match v {
117 Value::String(s) => s.clone(),
118 other => other.to_string(),
119 }
120}
121
122#[cfg(test)]
123mod tests {
124 use super::*;
125 use crate::notify::spec::Severity;
126
127 fn slack_cfg() -> SlackConfig {
128 SlackConfig {
129 webhook_url: "http://x".into(),
130 channel: Some("#alerts".into()),
131 username: Some("faucet".into()),
132 }
133 }
134
135 #[test]
136 fn slack_body_has_blocks_and_overrides() {
137 let e = NotifyEvent::run_failure("orders", "row1", "sink", "connection refused");
138 let body = slack(&slack_cfg(), &e);
139 assert_eq!(body["channel"], "#alerts");
140 assert_eq!(body["username"], "faucet");
141 let text = body["text"].as_str().unwrap();
142 assert!(text.contains("failed"));
143 let ctx = body["blocks"][1]["elements"][0]["text"].as_str().unwrap();
145 assert!(ctx.contains("row1"));
146 assert!(ctx.contains("error_kind"));
147 }
148
149 #[test]
150 fn slack_omits_row_when_empty() {
151 let e = NotifyEvent::scheduler_stuck("p", "no heartbeat");
152 let body = slack(
153 &SlackConfig {
154 webhook_url: "u".into(),
155 channel: None,
156 username: None,
157 },
158 &e,
159 );
160 let ctx = body["blocks"][1]["elements"][0]["text"].as_str().unwrap();
161 assert!(!ctx.contains("*Row:*"));
162 assert!(body.get("channel").is_none());
163 }
164
165 #[test]
166 fn pagerduty_trigger_carries_payload() {
167 let cfg = PagerdutyConfig {
168 routing_key: "rk".into(),
169 source: None,
170 endpoint: None,
171 };
172 let e = NotifyEvent::circuit_open("p", "", 5, 30);
173 let body = pagerduty(&cfg, &e, PdAction::Trigger, "p:");
174 assert_eq!(body["event_action"], "trigger");
175 assert_eq!(body["routing_key"], "rk");
176 assert_eq!(body["dedup_key"], "p:");
177 assert_eq!(body["payload"]["severity"], "critical");
178 assert_eq!(body["payload"]["source"], "p"); assert_eq!(body["payload"]["custom_details"]["details"]["failures"], 5);
180 }
181
182 #[test]
183 fn pagerduty_resolve_is_minimal() {
184 let cfg = PagerdutyConfig {
185 routing_key: "rk".into(),
186 source: Some("svc".into()),
187 endpoint: None,
188 };
189 let e = NotifyEvent::run_success("p", "", 1);
190 let body = pagerduty(&cfg, &e, PdAction::Resolve, "p:");
191 assert_eq!(body["event_action"], "resolve");
192 assert_eq!(body["dedup_key"], "p:");
193 assert!(body.get("payload").is_none());
194 }
195
196 #[test]
197 fn webhook_envelope_is_stable() {
198 let cfg = WebhookConfig {
199 url: "u".into(),
200 method: "POST".into(),
201 headers: Default::default(),
202 hmac_secret: None,
203 signature_header: "X-Faucet-Signature".into(),
204 };
205 let e = NotifyEvent::dlq_threshold("p", "r", 42);
206 let body = webhook(&cfg, &e);
207 assert_eq!(body["event"], "dlq_threshold");
208 assert_eq!(body["severity"], Severity::Warning.as_str());
209 assert_eq!(body["details"]["records_dlq"], 42);
210 }
211}