rskit_messaging/
event.rs

1//! CloudEvents-inspired event envelope for domain events.
2
3use chrono::{DateTime, Utc};
4use rskit_errors::{AppError, AppResult, ErrorCode};
5use serde::{Deserialize, Serialize};
6use uuid::Uuid;
7
8/// A structured event envelope following `CloudEvents` conventions.
9#[derive(Debug, Clone, Serialize, Deserialize)]
10pub struct Event {
11    /// Unique event identifier.
12    pub id: String,
13    /// The type of event (e.g. `"content.created"`).
14    #[serde(rename = "type")]
15    pub event_type: String,
16    /// The originating service or component.
17    pub source: String,
18    /// Optional subject the event relates to.
19    #[serde(default)]
20    pub subject: String,
21    /// MIME type of the `data` field.
22    #[serde(default = "default_content_type")]
23    pub content_type: String,
24    /// Schema version of the event.
25    #[serde(default)]
26    pub version: String,
27    /// When the event was created.
28    pub timestamp: DateTime<Utc>,
29    /// Arbitrary event payload.
30    pub data: serde_json::Value,
31}
32
33fn default_content_type() -> String {
34    "application/json".to_string()
35}
36
37impl Event {
38    /// Create a new event with the given type and source.
39    pub fn new(event_type: impl Into<String>, source: impl Into<String>) -> Self {
40        Self {
41            id: Uuid::new_v4().to_string(),
42            event_type: event_type.into(),
43            source: source.into(),
44            subject: String::new(),
45            content_type: "application/json".to_string(),
46            version: "1.0".to_string(),
47            timestamp: Utc::now(),
48            data: serde_json::Value::Null,
49        }
50    }
51
52    /// Set the subject.
53    #[must_use]
54    pub fn with_subject(mut self, subject: impl Into<String>) -> Self {
55        self.subject = subject.into();
56        self
57    }
58
59    /// Serialize `data` into the event payload.
60    pub fn with_data(mut self, data: impl Serialize) -> AppResult<Self> {
61        self.data = serde_json::to_value(data).map_err(|e| {
62            AppError::new(
63                ErrorCode::Internal,
64                format!("Failed to serialize event data: {e}"),
65            )
66        })?;
67        Ok(self)
68    }
69
70    /// Set the schema version.
71    #[must_use]
72    pub fn with_version(mut self, version: impl Into<String>) -> Self {
73        self.version = version.into();
74        self
75    }
76
77    /// Set the content type.
78    #[must_use]
79    pub fn with_content_type(mut self, content_type: impl Into<String>) -> Self {
80        self.content_type = content_type.into();
81        self
82    }
83
84    /// Serialize the entire event to JSON bytes.
85    pub fn to_json(&self) -> AppResult<Vec<u8>> {
86        serde_json::to_vec(self).map_err(|e| {
87            AppError::new(
88                ErrorCode::Internal,
89                format!("Failed to serialize event: {e}"),
90            )
91        })
92    }
93
94    /// Deserialize an event from JSON bytes.
95    pub fn from_json(bytes: &[u8]) -> AppResult<Self> {
96        serde_json::from_slice(bytes).map_err(|e| {
97            AppError::new(
98                ErrorCode::Internal,
99                format!("Failed to deserialize event: {e}"),
100            )
101        })
102    }
103
104    /// Parse the `data` field into a concrete type.
105    pub fn parse_data<T: serde::de::DeserializeOwned>(&self) -> AppResult<T> {
106        serde_json::from_value(self.data.clone()).map_err(|e| {
107            AppError::new(
108                ErrorCode::Internal,
109                format!("Failed to parse event data: {e}"),
110            )
111        })
112    }
113}
114
115#[cfg(test)]
116mod tests {
117    use super::*;
118
119    #[test]
120    fn new_event_has_defaults() {
121        let event = Event::new("user.created", "auth-service");
122        assert_eq!(event.event_type, "user.created");
123        assert_eq!(event.source, "auth-service");
124        assert!(!event.id.is_empty());
125        assert_eq!(event.content_type, "application/json");
126        assert_eq!(event.version, "1.0");
127        assert_eq!(event.data, serde_json::Value::Null);
128    }
129
130    #[test]
131    fn builder_methods() {
132        let event = Event::new("order.placed", "shop")
133            .with_subject("order-123")
134            .with_version("2.0")
135            .with_content_type("text/plain");
136
137        assert_eq!(event.subject, "order-123");
138        assert_eq!(event.version, "2.0");
139        assert_eq!(event.content_type, "text/plain");
140    }
141
142    #[test]
143    fn with_data_serializes_payload() {
144        #[derive(Serialize, Deserialize, Debug, PartialEq)]
145        struct Payload {
146            name: String,
147            count: u32,
148        }
149
150        let payload = Payload {
151            name: "test".to_string(),
152            count: 42,
153        };
154        let event = Event::new("test.event", "test")
155            .with_data(&payload)
156            .unwrap();
157
158        let parsed: Payload = event.parse_data().unwrap();
159        assert_eq!(parsed, payload);
160    }
161
162    #[test]
163    fn json_round_trip() {
164        let event = Event::new("round.trip", "test-source")
165            .with_subject("sub")
166            .with_data(serde_json::json!({"key": "value"}))
167            .unwrap();
168
169        let bytes = event.to_json().unwrap();
170        let restored = Event::from_json(&bytes).unwrap();
171
172        assert_eq!(restored.id, event.id);
173        assert_eq!(restored.event_type, "round.trip");
174        assert_eq!(restored.source, "test-source");
175        assert_eq!(restored.subject, "sub");
176        assert_eq!(restored.data, serde_json::json!({"key": "value"}));
177    }
178
179    #[test]
180    fn serde_renames_type_field() {
181        let event = Event::new("check.rename", "src");
182        let json = serde_json::to_string(&event).unwrap();
183        assert!(json.contains(r#""type":"check.rename""#));
184        assert!(!json.contains("event_type"));
185    }
186}
rskit_messaging/event.rs

rskit_messaging/
event.rs
