opssignal_core/
signal.rs

1use chrono::{DateTime, Utc};
2use serde::{Deserialize, Serialize};
3use sha2::{Digest, Sha256};
4use uuid::Uuid;
5
6/// Severity is a closed enum, not a free string, so the routing engine
7/// can match on it exhaustively instead of doing string comparison.
8#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
9#[serde(rename_all = "lowercase")]
10pub enum Severity {
11    #[default]
12    Info,
13    Success,
14    Warning,
15    Error,
16    Critical,
17}
18
19impl std::fmt::Display for Severity {
20    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
21        f.write_str(match self {
22            Severity::Info => "info",
23            Severity::Success => "success",
24            Severity::Warning => "warning",
25            Severity::Error => "error",
26            Severity::Critical => "critical",
27        })
28    }
29}
30
31impl std::str::FromStr for Severity {
32    type Err = String;
33
34    fn from_str(s: &str) -> Result<Self, Self::Err> {
35        match s.to_lowercase().as_str() {
36            "info" => Ok(Severity::Info),
37            "success" => Ok(Severity::Success),
38            "warning" => Ok(Severity::Warning),
39            "error" => Ok(Severity::Error),
40            "critical" => Ok(Severity::Critical),
41            other => Err(format!(
42                "unknown severity {other:?}; expected one of: info, success, warning, error, critical"
43            )),
44        }
45    }
46}
47
48/// The generic signal model. `id` and `timestamp` are always stamped by
49/// the core (never trusted from the caller) so every downstream
50/// component — sink, dedup, retry tracking — has a stable identity and
51/// ordering guarantee regardless of which language binding produced it.
52#[derive(Debug, Clone, Serialize, Deserialize)]
53pub struct Signal {
54    pub id: Uuid,
55    pub source: String,
56    pub service: Option<String>,
57    pub component: Option<String>,
58    pub event_type: String,
59    pub severity: Severity,
60    pub title: String,
61    pub message: Option<String>,
62    pub environment: Option<String>,
63    pub team: Option<String>,
64    pub run_id: Option<String>,
65    pub correlation_id: Option<String>,
66    pub trace_id: Option<String>,
67    pub log_url: Option<String>,
68    #[serde(default = "default_metadata")]
69    pub metadata: serde_json::Value,
70    pub timestamp: DateTime<Utc>,
71    pub dedup_key: Option<String>,
72}
73
74fn default_metadata() -> serde_json::Value {
75    serde_json::json!({})
76}
77
78/// Builder input — what callers actually construct. Notably missing
79/// `id` and `timestamp`: those are not caller-settable.
80#[derive(Debug, Clone, Default, Serialize, Deserialize)]
81pub struct SignalInput {
82    pub source: String,
83    pub service: Option<String>,
84    pub component: Option<String>,
85    pub event_type: String,
86    pub severity: Severity,
87    pub title: String,
88    pub message: Option<String>,
89    pub environment: Option<String>,
90    pub team: Option<String>,
91    pub run_id: Option<String>,
92    pub correlation_id: Option<String>,
93    pub trace_id: Option<String>,
94    pub log_url: Option<String>,
95    pub metadata: Option<serde_json::Value>,
96    /// Caller may supply an explicit dedup key; if absent, core computes
97    /// one from (source, event_type, title, environment).
98    pub dedup_key: Option<String>,
99}
100
101impl Signal {
102    pub fn from_input(input: SignalInput) -> Self {
103        let timestamp = Utc::now();
104        let dedup_key = input
105            .dedup_key
106            .clone()
107            .unwrap_or_else(|| compute_dedup_key(&input));
108
109        Signal {
110            id: Uuid::new_v4(),
111            source: input.source,
112            service: input.service,
113            component: input.component,
114            event_type: input.event_type,
115            severity: input.severity,
116            title: input.title,
117            message: input.message,
118            environment: input.environment,
119            team: input.team,
120            run_id: input.run_id,
121            correlation_id: input.correlation_id,
122            trace_id: input.trace_id,
123            log_url: input.log_url,
124            metadata: input.metadata.unwrap_or_else(default_metadata),
125            timestamp,
126            dedup_key: Some(dedup_key),
127        }
128    }
129}
130
131fn compute_dedup_key(input: &SignalInput) -> String {
132    let mut hasher = Sha256::new();
133    hasher.update(input.source.as_bytes());
134    hasher.update(b"|");
135    hasher.update(input.event_type.as_bytes());
136    hasher.update(b"|");
137    hasher.update(input.title.as_bytes());
138    hasher.update(b"|");
139    hasher.update(input.environment.as_deref().unwrap_or("").as_bytes());
140    format!("{:x}", hasher.finalize())
141}
142
143#[derive(Debug, thiserror::Error)]
144#[non_exhaustive]
145pub enum SignalError {
146    #[error("validation failed: {0}")]
147    Validation(String),
148    #[error("config error: {0}")]
149    Config(String),
150    #[error("delivery failed: {0}")]
151    Delivery(String),
152    #[error("sink write failed: {0}")]
153    Sink(String),
154    #[error("timeout after {0:?}")]
155    Timeout(std::time::Duration),
156    #[error("internal error (recovered from panic): {0}")]
157    Internal(String),
158}
159
160pub fn validate(input: &SignalInput) -> Result<(), SignalError> {
161    if input.source.trim().is_empty() {
162        return Err(SignalError::Validation("source must not be empty".into()));
163    }
164    if input.event_type.trim().is_empty() {
165        return Err(SignalError::Validation(
166            "event_type must not be empty".into(),
167        ));
168    }
169    if input.title.trim().is_empty() {
170        return Err(SignalError::Validation("title must not be empty".into()));
171    }
172    Ok(())
173}
174
175#[cfg(test)]
176mod tests {
177    use super::*;
178
179    #[test]
180    fn dedup_key_is_stable_for_identical_input() {
181        let input = SignalInput {
182            source: "airflow".into(),
183            event_type: "task_failed".into(),
184            title: "dbt model failed".into(),
185            environment: Some("prod".into()),
186            severity: Severity::Warning,
187            ..Default::default()
188        };
189        let k1 = compute_dedup_key(&input);
190        let k2 = compute_dedup_key(&input);
191        assert_eq!(k1, k2);
192    }
193
194    #[test]
195    fn validate_rejects_empty_title() {
196        let input = SignalInput {
197            source: "x".into(),
198            event_type: "y".into(),
199            title: "".into(),
200            severity: Severity::Info,
201            ..Default::default()
202        };
203        assert!(validate(&input).is_err());
204    }
205
206    #[test]
207    fn severity_display_is_lowercase() {
208        assert_eq!(Severity::Info.to_string(), "info");
209        assert_eq!(Severity::Critical.to_string(), "critical");
210        assert_eq!(Severity::Warning.to_string(), "warning");
211    }
212
213    #[test]
214    fn severity_from_str_roundtrips() {
215        for s in &["info", "success", "warning", "error", "critical"] {
216            let parsed: Severity = s.parse().expect("valid severity");
217            assert_eq!(parsed.to_string(), *s);
218        }
219    }
220
221    #[test]
222    fn severity_from_str_is_case_insensitive() {
223        assert_eq!("WARNING".parse::<Severity>().unwrap(), Severity::Warning);
224        assert_eq!("Error".parse::<Severity>().unwrap(), Severity::Error);
225    }
226
227    #[test]
228    fn severity_from_str_rejects_unknown() {
229        assert!("bogus".parse::<Severity>().is_err());
230    }
231
232    fn valid_input() -> SignalInput {
233        SignalInput {
234            source: "airflow".into(),
235            event_type: "task_failed".into(),
236            title: "dbt model failed".into(),
237            environment: Some("prod".into()),
238            ..Default::default()
239        }
240    }
241
242    #[test]
243    fn validate_accepts_minimal_input() {
244        assert!(validate(&valid_input()).is_ok());
245    }
246
247    #[test]
248    fn validate_rejects_blank_required_fields() {
249        for field in ["source", "event_type", "title"] {
250            let mut input = valid_input();
251            match field {
252                "source" => input.source = " \t".into(),
253                "event_type" => input.event_type = String::new(),
254                _ => input.title = "\n".into(),
255            }
256            let err = validate(&input).expect_err("blank field must be rejected");
257            assert!(matches!(err, SignalError::Validation(_)));
258            assert!(err.to_string().contains(field), "{err} should name {field}");
259        }
260    }
261
262    #[test]
263    fn id_and_timestamp_are_core_generated_even_if_the_payload_carries_them() {
264        // CLAUDE.md hard rule 4: callers cannot supply id or timestamp.
265        let caller_id = "00000000-0000-0000-0000-000000000001";
266        let json = format!(
267            r#"{{"id":"{caller_id}","timestamp":"2000-01-01T00:00:00Z","source":"ci","event_type":"build","title":"t","severity":"info"}}"#
268        );
269        let input: SignalInput = serde_json::from_str(&json).expect("input parses");
270
271        let before = Utc::now();
272        let signal = Signal::from_input(input);
273        let after = Utc::now();
274
275        assert_ne!(signal.id.to_string(), caller_id);
276        assert!(signal.timestamp >= before && signal.timestamp <= after);
277    }
278
279    #[test]
280    fn each_signal_gets_a_unique_id() {
281        let a = Signal::from_input(valid_input());
282        let b = Signal::from_input(valid_input());
283        assert_ne!(a.id, b.id);
284        assert_eq!(a.dedup_key, b.dedup_key);
285    }
286
287    #[test]
288    fn caller_supplied_dedup_key_is_kept() {
289        let mut input = valid_input();
290        input.dedup_key = Some("my-key".into());
291        assert_eq!(
292            Signal::from_input(input).dedup_key.as_deref(),
293            Some("my-key")
294        );
295    }
296
297    #[test]
298    fn dedup_key_changes_with_each_identifying_field() {
299        let base = compute_dedup_key(&valid_input());
300        let variants: [fn(&mut SignalInput); 4] = [
301            |i| i.source = "jenkins".into(),
302            |i| i.event_type = "task_retried".into(),
303            |i| i.title = "other".into(),
304            |i| i.environment = None,
305        ];
306        for change in variants {
307            let mut input = valid_input();
308            change(&mut input);
309            assert_ne!(compute_dedup_key(&input), base);
310        }
311        // Severity and message are deliberately not part of the key.
312        let mut input = valid_input();
313        input.severity = Severity::Critical;
314        input.message = Some("details".into());
315        assert_eq!(compute_dedup_key(&input), base);
316    }
317
318    #[test]
319    fn missing_metadata_defaults_to_empty_object() {
320        assert_eq!(
321            Signal::from_input(valid_input()).metadata,
322            serde_json::json!({})
323        );
324    }
325}
opssignal_core/signal.rs

opssignal_core/
signal.rs
