1use std::fmt;
2use std::result::Result as StdResult;
3use std::str::FromStr;
4
5use chrono::{DateTime, Utc};
6use uuid::Uuid;
7
8use crate::error::Result;
9use crate::event_key::EventKey;
10use crate::metadata::Metadata;
11use crate::namespace::Namespace;
12use crate::organization::OrganizationId;
13use crate::payload::Payload;
14use crate::topic::Topic;
15
16#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
17pub struct EventId(Uuid);
18
19impl EventId {
20 pub fn new() -> Self {
21 Self(Uuid::now_v7())
22 }
23
24 pub fn from_uuid(uuid: Uuid) -> Self {
25 Self(uuid)
26 }
27
28 pub fn as_uuid(&self) -> &Uuid {
29 &self.0
30 }
31}
32
33impl Default for EventId {
34 fn default() -> Self {
35 Self::new()
36 }
37}
38
39impl fmt::Display for EventId {
40 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
41 write!(f, "{}", self.0)
42 }
43}
44
45impl FromStr for EventId {
46 type Err = uuid::Error;
47
48 fn from_str(s: &str) -> StdResult<Self, Self::Err> {
49 Ok(Self(Uuid::parse_str(s)?))
50 }
51}
52
53#[derive(Debug, Clone)]
54pub struct Event<P = Payload> {
55 id: EventId,
56 organization: OrganizationId,
57 namespace: Namespace,
58 topic: Topic,
59 key: EventKey,
60 payload: P,
61 metadata: Metadata,
62 timestamp: DateTime<Utc>,
63 version: u64,
64 parent_id: Option<EventId>,
65 correlation_id: Option<EventKey>,
66 causation_id: Option<EventKey>,
67}
68
69pub struct EventBuilder<P = Payload> {
70 organization: OrganizationId,
71 namespace: Namespace,
72 topic: Topic,
73 key: EventKey,
74 payload: P,
75 metadata: Metadata,
76 parent_id: Option<EventId>,
77 correlation_id: Option<EventKey>,
78 causation_id: Option<EventKey>,
79}
80
81impl<P> EventBuilder<P> {
82 fn new(
83 organization: OrganizationId,
84 namespace: Namespace,
85 topic: Topic,
86 key: EventKey,
87 payload: P,
88 ) -> Self {
89 Self {
90 organization,
91 namespace,
92 topic,
93 key,
94 payload,
95 metadata: Metadata::new(),
96 parent_id: None,
97 correlation_id: None,
98 causation_id: None,
99 }
100 }
101
102 pub fn parent_id(mut self, parent_id: EventId) -> Self {
103 self.parent_id = Some(parent_id);
104 self
105 }
106
107 pub fn correlation_id(mut self, correlation_id: impl Into<String>) -> Result<Self> {
108 self.correlation_id = Some(EventKey::new(correlation_id)?);
109 Ok(self)
110 }
111
112 pub fn causation_id(mut self, causation_id: impl Into<String>) -> Result<Self> {
113 self.causation_id = Some(EventKey::new(causation_id)?);
114 Ok(self)
115 }
116
117 pub fn metadata(mut self, metadata: Metadata) -> Self {
118 self.metadata = metadata;
119 self
120 }
121
122 pub fn build(self) -> Result<Event<P>> {
123 Event::new(
124 EventId::new(),
125 self.organization,
126 self.namespace,
127 self.topic,
128 self.key,
129 self.payload,
130 self.metadata,
131 Utc::now(),
132 1,
133 self.parent_id,
134 self.correlation_id,
135 self.causation_id,
136 )
137 }
138}
139
140impl<P> Event<P> {
141 #[allow(clippy::too_many_arguments)]
142 pub fn new(
143 id: EventId,
144 organization: OrganizationId,
145 namespace: Namespace,
146 topic: Topic,
147 key: EventKey,
148 payload: P,
149 metadata: Metadata,
150 timestamp: DateTime<Utc>,
151 version: u64,
152 parent_id: Option<EventId>,
153 correlation_id: Option<EventKey>,
154 causation_id: Option<EventKey>,
155 ) -> Result<Self> {
156 Ok(Self {
157 id,
158 organization,
159 namespace,
160 topic,
161 key,
162 payload,
163 metadata,
164 timestamp,
165 version,
166 parent_id,
167 correlation_id,
168 causation_id,
169 })
170 }
171
172 pub fn builder(
173 organization: impl Into<String>,
174 namespace: impl Into<String>,
175 topic: impl Into<String>,
176 key: impl Into<String>,
177 payload: P,
178 ) -> Result<EventBuilder<P>> {
179 Ok(EventBuilder::new(
180 OrganizationId::new(organization)?,
181 Namespace::new(namespace)?,
182 Topic::new(topic)?,
183 EventKey::new(key)?,
184 payload,
185 ))
186 }
187
188 pub fn create(
189 organization: impl Into<String>,
190 namespace: impl Into<String>,
191 topic: impl Into<String>,
192 key: impl Into<String>,
193 payload: P,
194 ) -> Result<Self> {
195 Self::builder(organization, namespace, topic, key, payload)?.build()
196 }
197
198 pub fn with_metadata(mut self, metadata: Metadata) -> Self {
199 self.metadata = metadata;
200 self
201 }
202
203 pub fn id(&self) -> EventId {
204 self.id
205 }
206 pub fn organization(&self) -> &OrganizationId {
207 &self.organization
208 }
209 pub fn namespace(&self) -> &Namespace {
210 &self.namespace
211 }
212 pub fn topic(&self) -> &Topic {
213 &self.topic
214 }
215 pub fn payload(&self) -> &P {
216 &self.payload
217 }
218 pub fn into_payload(self) -> P {
219 self.payload
220 }
221 pub fn metadata(&self) -> &Metadata {
222 &self.metadata
223 }
224 pub fn timestamp(&self) -> DateTime<Utc> {
225 self.timestamp
226 }
227 pub fn version(&self) -> u64 {
228 self.version
229 }
230 pub fn key(&self) -> &EventKey {
231 &self.key
232 }
233
234 pub fn parent_id(&self) -> Option<EventId> {
235 self.parent_id
236 }
237 pub fn correlation_id(&self) -> Option<&EventKey> {
238 self.correlation_id.as_ref()
239 }
240 pub fn causation_id(&self) -> Option<&EventKey> {
241 self.causation_id.as_ref()
242 }
243
244 pub fn map_payload<Q, F>(self, f: F) -> Event<Q>
245 where
246 F: FnOnce(P) -> Q,
247 {
248 Event {
249 id: self.id,
250 organization: self.organization,
251 namespace: self.namespace,
252 topic: self.topic,
253 key: self.key,
254 payload: f(self.payload),
255 metadata: self.metadata,
256 timestamp: self.timestamp,
257 version: self.version,
258 parent_id: self.parent_id,
259 correlation_id: self.correlation_id,
260 causation_id: self.causation_id,
261 }
262 }
263
264 pub fn try_map_payload<Q, F>(self, f: F) -> Result<Event<Q>>
265 where
266 F: FnOnce(P) -> Result<Q>,
267 {
268 Ok(Event {
269 id: self.id,
270 organization: self.organization,
271 namespace: self.namespace,
272 topic: self.topic,
273 key: self.key,
274 payload: f(self.payload)?,
275 metadata: self.metadata,
276 timestamp: self.timestamp,
277 version: self.version,
278 parent_id: self.parent_id,
279 correlation_id: self.correlation_id,
280 causation_id: self.causation_id,
281 })
282 }
283
284 pub fn encode_payload<C>(&self, codec: &C) -> Result<Event<Payload>>
285 where
286 C: crate::PayloadCodec<P>,
287 {
288 Event::new(
289 self.id,
290 self.organization.clone(),
291 self.namespace.clone(),
292 self.topic.clone(),
293 self.key.clone(),
294 codec.encode(&self.payload)?,
295 self.metadata.clone(),
296 self.timestamp,
297 self.version,
298 self.parent_id,
299 self.correlation_id.clone(),
300 self.causation_id.clone(),
301 )
302 }
303}
304
305impl Event<Payload> {
306 pub fn decode_payload<P, C>(self, codec: &C) -> Result<Event<P>>
307 where
308 C: crate::PayloadCodec<P>,
309 {
310 self.try_map_payload(|payload| codec.decode(&payload))
311 }
312}
313
314#[cfg(test)]
315mod tests {
316 use super::*;
317
318 #[test]
319 fn create_event_requires_key() {
320 let payload = Payload::from_json(&serde_json::json!({"task_id": "123"})).unwrap();
321 let event = Event::create("acme", "/task", "task.created", "task-123", payload).unwrap();
322 assert_eq!(event.organization().as_str(), "acme");
323 assert_eq!(event.namespace().as_str(), "/task");
324 assert_eq!(event.topic().as_str(), "task.created");
325 assert_eq!(event.key().as_str(), "task-123");
326 assert_eq!(event.parent_id(), None);
327 assert_eq!(event.correlation_id(), None);
328 assert_eq!(event.causation_id(), None);
329 assert_eq!(event.version(), 1);
330 }
331
332 #[test]
333 fn builder_sets_optional_lineage_fields() {
334 let parent_id = EventId::new();
335 let event = Event::builder(
336 "acme",
337 "/x",
338 "thing.happened",
339 "entity-1",
340 Payload::from_string("p"),
341 )
342 .unwrap()
343 .parent_id(parent_id)
344 .correlation_id("workflow-7")
345 .unwrap()
346 .causation_id("command-9")
347 .unwrap()
348 .build()
349 .unwrap();
350
351 assert_eq!(event.key().as_str(), "entity-1");
352 assert_eq!(event.parent_id(), Some(parent_id));
353 assert_eq!(
354 event.correlation_id().map(EventKey::as_str),
355 Some("workflow-7")
356 );
357 assert_eq!(
358 event.causation_id().map(EventKey::as_str),
359 Some("command-9")
360 );
361 }
362
363 #[test]
364 fn builder_rejects_empty_required_key() {
365 assert!(
366 Event::builder(
367 "acme",
368 "/x",
369 "thing.happened",
370 "",
371 Payload::from_string("p")
372 )
373 .is_err()
374 );
375 }
376
377 #[test]
378 fn builder_rejects_empty_optional_ids() {
379 let builder = Event::builder(
380 "acme",
381 "/x",
382 "thing.happened",
383 "k",
384 Payload::from_string("p"),
385 )
386 .unwrap();
387 assert!(builder.correlation_id("").is_err());
388
389 let builder = Event::builder(
390 "acme",
391 "/x",
392 "thing.happened",
393 "k",
394 Payload::from_string("p"),
395 )
396 .unwrap();
397 assert!(builder.causation_id("").is_err());
398 }
399
400 #[test]
401 fn create_with_metadata() {
402 let payload = Payload::from_string("test");
403 let metadata = Metadata::new()
404 .with("agent_id", "abc-123")
405 .unwrap()
406 .with("project", "acme")
407 .unwrap();
408 let event = Event::builder("acme", "/agent", "agent.registered", "agent-1", payload)
409 .unwrap()
410 .metadata(metadata)
411 .build()
412 .unwrap();
413 assert_eq!(event.metadata().get("agent_id"), Some("abc-123"));
414 assert_eq!(event.metadata().get("project"), Some("acme"));
415 }
416
417 #[test]
418 fn correlation_and_causation_are_first_class_fields() {
419 let event = Event::builder(
420 "acme",
421 "/x",
422 "thing.happened",
423 "k",
424 Payload::from_string("test"),
425 )
426 .unwrap()
427 .correlation_id("corr-1")
428 .unwrap()
429 .causation_id("cause-1")
430 .unwrap()
431 .build()
432 .unwrap();
433 assert_eq!(event.correlation_id().map(EventKey::as_str), Some("corr-1"));
434 assert_eq!(event.causation_id().map(EventKey::as_str), Some("cause-1"));
435 assert!(event.metadata().is_empty());
436 }
437
438 #[derive(Debug, Clone, PartialEq, Eq)]
439 struct UserUpdated {
440 user_id: String,
441 email: String,
442 }
443
444 #[test]
445 fn create_event_with_typed_payload() {
446 let payload = UserUpdated {
447 user_id: "u-1".to_owned(),
448 email: "a@example.com".to_owned(),
449 };
450
451 let event: Event<UserUpdated> =
452 Event::create("acme", "/users", "user.updated", "user-u-1", payload).unwrap();
453
454 assert_eq!(event.payload().user_id, "u-1");
455 assert_eq!(event.organization().as_str(), "acme");
456 assert_eq!(event.namespace().as_str(), "/users");
457 assert_eq!(event.topic().as_str(), "user.updated");
458 assert_eq!(event.key().as_str(), "user-u-1");
459 }
460
461 #[test]
462 fn map_payload_changes_only_payload_type() {
463 let event: Event<UserUpdated> = Event::create(
464 "acme",
465 "/users",
466 "user.updated",
467 "user-u-1",
468 UserUpdated {
469 user_id: "u-1".to_owned(),
470 email: "a@example.com".to_owned(),
471 },
472 )
473 .unwrap()
474 .with_metadata(Metadata::new().with("source", "test").unwrap());
475
476 let id = event.id();
477 let mapped: Event<String> = event.map_payload(|payload| payload.email);
478
479 assert_eq!(mapped.id(), id);
480 assert_eq!(mapped.payload(), "a@example.com");
481 assert_eq!(mapped.metadata().get("source"), Some("test"));
482 }
483
484 #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
485 struct SerializableUserUpdated {
486 user_id: String,
487 email: String,
488 }
489
490 #[test]
491 fn event_encodes_and_decodes_payload_with_codec() {
492 use crate::JsonPayloadCodec;
493
494 let typed: Event<SerializableUserUpdated> = Event::create(
495 "acme",
496 "/users",
497 "user.updated",
498 "user-u-1",
499 SerializableUserUpdated {
500 user_id: "u-1".to_owned(),
501 email: "a@example.com".to_owned(),
502 },
503 )
504 .unwrap();
505
506 let id = typed.id();
507 let encoded = typed.encode_payload(&JsonPayloadCodec).unwrap();
508 assert_eq!(encoded.id(), id);
509 assert_eq!(encoded.payload().content_type(), crate::ContentType::Json);
510
511 let decoded: Event<SerializableUserUpdated> =
512 encoded.decode_payload(&JsonPayloadCodec).unwrap();
513 assert_eq!(decoded.id(), id);
514 assert_eq!(decoded.payload().user_id, "u-1");
515 }
516}