arc_core/event.rs
1//! # Event Module
2//!
3//! Core event type for event sourcing. Events represent immutable facts about
4//! things that have happened in the system.
5//!
6//! ## Audit metadata
7//!
8//! Every event carries an [`AuditMetadata`](crate::audit::AuditMetadata) value.
9//! Aggregates produce events with `AuditMetadata::pending()`; the
10//! `CommandBus::dispatch` implementation overwrites that placeholder with a
11//! validated audit struct sourced from the request `CommandContext` before
12//! calling `EventStore::append`. Stores reject events whose audit fails
13//! validation.
14//!
15//! ## Design Principles
16//!
17//! - **Immutable**: Once persisted, events cannot be changed.
18//! - **Serializable**: All events can be stored as JSON.
19//! - **Self-describing**: Events contain all metadata needed to understand them.
20//! - **Ordered**: Events have a sequence number within their aggregate.
21//! - **Audited**: Every persisted event carries `who/when/where/why` audit data.
22
23use crate::audit::AuditMetadata;
24use serde::{Deserialize, Serialize};
25use std::time::{SystemTime, UNIX_EPOCH};
26use uuid::Uuid;
27
28/// Core event type representing an immutable domain event.
29///
30/// # Fields
31///
32/// - `event_id`: Unique identifier for this specific event occurrence
33/// - `aggregate_type`: Type of aggregate this event belongs to (e.g., "User")
34/// - `aggregate_id`: ID of the specific aggregate instance
35/// - `sequence`: Sequential number within the aggregate stream (starts at 1)
36/// - `event_type`: Type of event (e.g., "UserRegistered")
37/// - `payload`: Event data as JSON (flexible, evolvable schema)
38/// - `audit`: HIPAA audit metadata. `pending()` until the bus stamps it.
39/// - `timestamp`: When the event occurred (milliseconds since UNIX epoch)
40#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
41pub struct Event {
42 /// Unique identifier for this event
43 pub event_id: Uuid,
44
45 /// Aggregate type (e.g., "User", "Order")
46 pub aggregate_type: String,
47
48 /// Aggregate instance identifier
49 pub aggregate_id: String,
50
51 /// Sequence number within the aggregate (starts at 1)
52 pub sequence: i64,
53
54 /// Event type (e.g., "UserCreated", "ProfileUpdated")
55 /// Convention: Past tense, PascalCase
56 pub event_type: String,
57
58 /// Event payload as JSON
59 pub payload: serde_json::Value,
60
61 /// HIPAA audit metadata. `AuditMetadata::pending()` until the
62 /// `CommandBus` overwrites it before `append`. `EventStore::append`
63 /// implementations call `audit.validate()` and reject pending values.
64 pub audit: AuditMetadata,
65
66 /// Wall-clock timestamp (milliseconds since UNIX epoch). For HIPAA
67 /// audit-quality time, use `audit.timestamp_utc_us` (microsecond precision).
68 pub timestamp: u64,
69}
70
71/// Named fields required to create a new domain event.
72///
73/// The string fields accept either borrowed values such as `"Product"` or
74/// owned `String` values. `Event::new` converts them into owned strings.
75#[derive(Debug, Clone)]
76pub struct NewEvent<AggregateType, AggregateId, EventType> {
77 /// Aggregate type (for example, `"Product"`).
78 pub aggregate_type: AggregateType,
79
80 /// Aggregate instance identifier.
81 pub aggregate_id: AggregateId,
82
83 /// Next sequence number in this aggregate's event stream.
84 pub sequence: i64,
85
86 /// Event type in past-tense PascalCase (for example, `"ProductCreated"`).
87 pub event_type: EventType,
88
89 /// Event-specific data.
90 pub payload: serde_json::Value,
91}
92
93impl Event {
94 /// Create a new event with `audit = AuditMetadata::pending()`.
95 ///
96 /// Aggregates call this from `handle()` and the `CommandBus` overwrites the
97 /// audit field with a request-scoped value before persisting. The
98 /// pending placeholder fails store-side validation, so a forgotten stamp
99 /// is impossible to commit.
100 ///
101 /// # Example
102 ///
103 /// ```rust
104 /// use arc_core::event::{Event, NewEvent};
105 /// use serde_json::json;
106 ///
107 /// let event = Event::new(NewEvent {
108 /// aggregate_type: "User",
109 /// aggregate_id: "user-456",
110 /// sequence: 1,
111 /// event_type: "UserCreated",
112 /// payload: json!({ "name": "Bob", "email": "bob@example.com" }),
113 /// });
114 /// assert!(event.audit.is_pending());
115 /// ```
116 pub fn new<AggregateType, AggregateId, EventType>(
117 event: NewEvent<AggregateType, AggregateId, EventType>,
118 ) -> Self
119 where
120 AggregateType: Into<String>,
121 AggregateId: Into<String>,
122 EventType: Into<String>,
123 {
124 Self {
125 event_id: Uuid::new_v4(),
126 aggregate_type: event.aggregate_type.into(),
127 aggregate_id: event.aggregate_id.into(),
128 sequence: event.sequence,
129 event_type: event.event_type.into(),
130 payload: event.payload,
131 audit: AuditMetadata::pending(),
132 timestamp: SystemTime::now()
133 .duration_since(UNIX_EPOCH)
134 .expect("Time went backwards")
135 .as_millis() as u64,
136 }
137 }
138
139 /// Replace `audit` with a fully-stamped value. Used by `CommandBus`
140 /// before calling `EventStore::append`.
141 pub fn with_audit(mut self, audit: AuditMetadata) -> Self {
142 self.audit = audit;
143 self
144 }
145
146 /// Serialize event to JSON string.
147 pub fn to_json(&self) -> Result<String, serde_json::Error> {
148 serde_json::to_string(self)
149 }
150
151 /// Deserialize event from JSON string.
152 pub fn from_json(json_str: &str) -> Result<Self, serde_json::Error> {
153 serde_json::from_str(json_str)
154 }
155}
156
157#[cfg(test)]
158mod tests {
159 use super::*;
160 use crate::audit::AuditMetadata;
161 use serde_json::json;
162
163 #[test]
164 fn test_event_creation() {
165 let event = Event::new(NewEvent {
166 aggregate_type: "User",
167 aggregate_id: "user-123",
168 sequence: 1,
169 event_type: "UserCreated",
170 payload: json!({
171 "name": "Test User",
172 "email": "test@example.com"
173 }),
174 });
175
176 assert_eq!(event.aggregate_type, "User");
177 assert_eq!(event.aggregate_id, "user-123");
178 assert_eq!(event.sequence, 1);
179 assert_eq!(event.event_type, "UserCreated");
180 assert_eq!(event.payload["name"], "Test User");
181 assert!(event.timestamp > 0);
182 assert!(event.audit.is_pending());
183 }
184
185 #[test]
186 fn test_with_audit_overwrites_pending() {
187 let stamp = AuditMetadata::test_default();
188 let event = Event::new(NewEvent {
189 aggregate_type: "User",
190 aggregate_id: "user-1",
191 sequence: 1,
192 event_type: "UserCreated",
193 payload: json!({}),
194 })
195 .with_audit(stamp.clone());
196 assert!(!event.audit.is_pending());
197 assert_eq!(event.audit, stamp);
198 }
199
200 #[test]
201 fn test_event_serialization_roundtrips_audit() {
202 let event = Event::new(NewEvent {
203 aggregate_type: "User",
204 aggregate_id: "user-789",
205 sequence: 3,
206 event_type: "ProfileUpdated",
207 payload: json!({"name": "X"}),
208 })
209 .with_audit(AuditMetadata::test_default());
210
211 let json_str = event.to_json().unwrap();
212 let deserialized = Event::from_json(&json_str).unwrap();
213
214 assert_eq!(event.event_id, deserialized.event_id);
215 assert_eq!(event.audit, deserialized.audit);
216 }
217
218 #[test]
219 fn test_event_ordering() {
220 let event1 = Event::new(NewEvent {
221 aggregate_type: "User",
222 aggregate_id: "user-1",
223 sequence: 1,
224 event_type: "UserCreated",
225 payload: json!({}),
226 });
227 let event2 = Event::new(NewEvent {
228 aggregate_type: "User",
229 aggregate_id: "user-1",
230 sequence: 2,
231 event_type: "ProfileUpdated",
232 payload: json!({}),
233 });
234
235 assert!(event1.sequence < event2.sequence);
236 }
237
238 #[test]
239 fn test_event_uniqueness() {
240 let event1 = Event::new(NewEvent {
241 aggregate_type: "User",
242 aggregate_id: "user-1",
243 sequence: 1,
244 event_type: "UserCreated",
245 payload: json!({}),
246 });
247 let event2 = Event::new(NewEvent {
248 aggregate_type: "User",
249 aggregate_id: "user-1",
250 sequence: 1,
251 event_type: "UserCreated",
252 payload: json!({}),
253 });
254 assert_ne!(event1.event_id, event2.event_id);
255 }
256}