made_core/events/
envelope.rs

1//! Common metadata carried by every domain event.
2
3use serde::{Deserialize, Serialize};
4use time::OffsetDateTime;
5
6use crate::error::DomainError;
7use crate::value_objects::EventId;
8
9const MAX_SOURCE_LEN: usize = 256;
10
11/// Shared header attached to every domain event produced or consumed
12/// by MADE.
13#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
14pub struct EventEnvelope {
15    event_id: EventId,
16    #[serde(with = "time::serde::rfc3339")]
17    emitted_at: OffsetDateTime,
18    source: String,
19    #[serde(default)]
20    correlation_id: Option<EventId>,
21    #[serde(default)]
22    causation_id: Option<EventId>,
23}
24
25impl EventEnvelope {
26    pub fn new(
27        event_id: EventId,
28        emitted_at: OffsetDateTime,
29        source: impl Into<String>,
30        correlation_id: Option<EventId>,
31    ) -> Result<Self, DomainError> {
32        Self::new_with_causation(event_id, emitted_at, source, correlation_id, None)
33    }
34
35    pub fn new_with_causation(
36        event_id: EventId,
37        emitted_at: OffsetDateTime,
38        source: impl Into<String>,
39        correlation_id: Option<EventId>,
40        causation_id: Option<EventId>,
41    ) -> Result<Self, DomainError> {
42        let source = source.into();
43        let trimmed = source.trim();
44        if trimmed.is_empty() {
45            return Err(DomainError::EmptyField {
46                field: "event.source",
47            });
48        }
49        if trimmed.len() > MAX_SOURCE_LEN {
50            return Err(DomainError::FieldTooLong {
51                field: "event.source",
52                actual: trimmed.len(),
53                max: MAX_SOURCE_LEN,
54            });
55        }
56        Ok(Self {
57            event_id,
58            emitted_at,
59            source: trimmed.to_owned(),
60            correlation_id,
61            causation_id,
62        })
63    }
64
65    #[must_use]
66    pub fn event_id(&self) -> &EventId {
67        &self.event_id
68    }
69    #[must_use]
70    pub fn emitted_at(&self) -> OffsetDateTime {
71        self.emitted_at
72    }
73    #[must_use]
74    pub fn source(&self) -> &str {
75        &self.source
76    }
77    #[must_use]
78    pub fn correlation_id(&self) -> Option<&EventId> {
79        self.correlation_id.as_ref()
80    }
81
82    #[must_use]
83    pub fn causation_id(&self) -> Option<&EventId> {
84        self.causation_id.as_ref()
85    }
86}
87
88#[cfg(test)]
89mod tests {
90    use super::*;
91    use time::macros::datetime;
92
93    fn at() -> OffsetDateTime {
94        datetime!(2026-04-15 12:00:00 UTC)
95    }
96
97    #[test]
98    fn construction_trims_source() {
99        let env =
100            EventEnvelope::new(EventId::new("e1").unwrap(), at(), "  grafana  ", None).unwrap();
101        assert_eq!(env.source(), "grafana");
102    }
103
104    #[test]
105    fn empty_source_is_rejected() {
106        let err = EventEnvelope::new(EventId::new("e1").unwrap(), at(), "   ", None).unwrap_err();
107        assert!(matches!(
108            err,
109            DomainError::EmptyField {
110                field: "event.source"
111            }
112        ));
113    }
114
115    #[test]
116    fn overlong_source_is_rejected() {
117        let err = EventEnvelope::new(
118            EventId::new("e1").unwrap(),
119            at(),
120            "x".repeat(MAX_SOURCE_LEN + 1),
121            None,
122        )
123        .unwrap_err();
124        assert!(matches!(err, DomainError::FieldTooLong { .. }));
125    }
126
127    #[test]
128    fn correlation_id_is_optional() {
129        let env = EventEnvelope::new(EventId::new("e").unwrap(), at(), "s", None).unwrap();
130        assert!(env.correlation_id().is_none());
131        assert!(env.causation_id().is_none());
132    }
133
134    #[test]
135    fn accessors_return_fields() {
136        let corr = EventId::new("c").unwrap();
137        let cause = EventId::new("cause").unwrap();
138        let env = EventEnvelope::new_with_causation(
139            EventId::new("e").unwrap(),
140            at(),
141            "src",
142            Some(corr.clone()),
143            Some(cause.clone()),
144        )
145        .unwrap();
146        assert_eq!(env.event_id().as_str(), "e");
147        assert_eq!(env.emitted_at(), at());
148        assert_eq!(env.correlation_id(), Some(&corr));
149        assert_eq!(env.causation_id(), Some(&cause));
150    }
151
152    #[test]
153    fn json_shape_matches_asyncapi_allof() {
154        // AsyncAPI composes EventEnvelope into each event via `allOf`,
155        // which produces a flat JSON object. Events use
156        // `#[serde(flatten)]` on their `envelope` field so the wire
157        // shape matches this expectation. This test locks in the
158        // EventEnvelope keys at the JSON root; breaking it is a
159        // breaking change on the event bus contract.
160        let env = EventEnvelope::new(
161            EventId::new("e").unwrap(),
162            at(),
163            "src",
164            Some(EventId::new("c").unwrap()),
165        )
166        .unwrap();
167        let json = serde_json::to_value(&env).unwrap();
168        let obj = json.as_object().unwrap();
169        assert!(obj.contains_key("event_id"));
170        assert!(obj.contains_key("emitted_at"));
171        assert!(obj.contains_key("source"));
172        assert!(obj.contains_key("correlation_id"));
173        assert!(obj.contains_key("causation_id"));
174    }
175}
made_core/events/envelope.rs

made_core/events/
envelope.rs
