1mod canonical;
8mod descriptor;
9mod occurrence;
10
11pub use canonical::{
12 DOMAIN_EVENT_BODY_CODEC, DOMAIN_EVENT_BODY_CODEC_VERSION, MAX_DOMAIN_EVENT_BODY_BYTES,
13 MAX_DOMAIN_EVENT_OCCURRENCE_WIRE_BYTES,
14};
15pub use descriptor::{
16 DomainDeletion, DomainDeletionError, DomainEvent, DomainEventBodyContract,
17 DomainEventBodyDescriptor, DomainEventBodyKind, DomainEventContract, DomainEventDescriptor,
18 DomainState, DomainStateDescriptor,
19};
20pub use occurrence::{
21 DomainEventCaptureError, DomainEventCaptureOutcome, DomainEventCapturePoison,
22 DomainEventCommitGuardError, DomainEventEnvelope, DomainEventOccurrence,
23 DOMAIN_EVENT_OCCURRENCE_VERSION,
24};
25
26pub(crate) use canonical::canonical_json_bytes;
27pub(crate) use occurrence::state_descriptor_matches;
28
29#[cfg(test)]
30mod tests {
31 use std::borrow::Cow;
32 use std::collections::BTreeMap;
33 use std::time::{Duration, UNIX_EPOCH};
34
35 use serde::ser::Error as _;
36 use serde::{Deserialize, Serialize};
37 use sha2::{Digest, Sha256};
38
39 use super::*;
40 use crate::bus::{MAX_MESSAGE_NAME_LEN, MAX_STABLE_MESSAGE_ID_LEN};
41 use crate::Entity;
42
43 const STATE_FINGERPRINT: &str =
44 "sha256:1111111111111111111111111111111111111111111111111111111111111111";
45 const EVENT_FINGERPRINT: &str =
46 "sha256:2222222222222222222222222222222222222222222222222222222222222222";
47 const DELETE_FINGERPRINT: &str =
48 "sha256:3333333333333333333333333333333333333333333333333333333333333333";
49
50 #[derive(Debug, Deserialize, Serialize, PartialEq, Eq)]
51 struct TodoState {
52 todo_id: String,
53 status: String,
54 }
55
56 impl TodoState {
57 fn new(status: &str) -> Self {
58 Self {
59 todo_id: "todo-1".into(),
60 status: status.into(),
61 }
62 }
63 }
64
65 impl DomainState for TodoState {
66 const DESCRIPTOR: DomainStateDescriptor = DomainStateDescriptor::distributed_json(
67 "TodoState",
68 3,
69 "todo-state-v3",
70 STATE_FINGERPRINT,
71 );
72 }
73
74 #[derive(Serialize)]
75 struct TodoRenamed {
76 title: String,
77 }
78
79 impl DomainEvent for TodoRenamed {
80 const DESCRIPTOR: DomainEventDescriptor = DomainEventDescriptor {
81 name: Cow::Borrowed("todo.renamed"),
82 version: 2,
83 body: DomainEventBodyDescriptor::distributed_json(
84 DomainEventBodyKind::Event,
85 "TodoRenamed",
86 4,
87 "todo-renamed-v4",
88 EVENT_FINGERPRINT,
89 ),
90 };
91 }
92
93 struct FailingState;
94
95 impl Serialize for FailingState {
96 fn serialize<S>(&self, _serializer: S) -> Result<S::Ok, S::Error>
97 where
98 S: serde::Serializer,
99 {
100 Err(S::Error::custom("intentional domain-state failure"))
101 }
102 }
103
104 impl DomainState for FailingState {
105 const DESCRIPTOR: DomainStateDescriptor = DomainStateDescriptor::distributed_json(
106 "FailingState",
107 1,
108 "failing-state-v1",
109 STATE_FINGERPRINT,
110 );
111 }
112
113 fn state_event(name: &'static str) -> DomainEventDescriptor {
114 DomainEventDescriptor::state::<TodoState>(name, 1)
115 }
116
117 fn deletion_event(name: impl Into<Cow<'static, str>>) -> DomainEventDescriptor {
118 DomainEventDescriptor {
119 name: name.into(),
120 version: 1,
121 body: DomainEventBodyDescriptor::distributed_json(
122 DomainEventBodyKind::Deletion,
123 "DomainDeletion<String>",
124 1,
125 "domain-deletion-string-v1",
126 DELETE_FINGERPRINT,
127 ),
128 }
129 }
130
131 fn fixed_envelope(aggregate_id: impl Into<String>) -> DomainEventEnvelope {
132 DomainEventEnvelope {
133 aggregate_type: "todo".into(),
134 aggregate_id: aggregate_id.into(),
135 aggregate_sequence: 7,
136 publication_ordinal: 0,
137 occurred_at: UNIX_EPOCH + Duration::from_millis(1_700_000_000_123),
138 metadata: BTreeMap::from([
139 ("causation_id".into(), "cmd-7".into()),
140 ("correlation_id".into(), "request-4".into()),
141 (
142 "traceparent".into(),
143 "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01".into(),
144 ),
145 ]),
146 }
147 }
148
149 #[test]
150 fn transitions_retain_intermediate_and_final_state_occurrences_in_order() {
151 let mut entity = Entity::with_id("todo-1");
152 entity.set_correlation_id("request-1");
153 entity.digest("todo.created", &()).unwrap();
154 entity
155 .capture_domain_state("todo", state_event("todo.created"), &TodoState::new("open"))
156 .unwrap();
157 entity.set_correlation_id("request-2");
158 entity.digest("todo.completed", &()).unwrap();
159 entity
160 .capture_domain_state(
161 "todo",
162 state_event("todo.completed"),
163 &TodoState::new("completed"),
164 )
165 .unwrap();
166
167 let states = entity
168 .pending_domain_events()
169 .iter()
170 .map(|occurrence| occurrence.decode_body::<TodoState>().unwrap())
171 .collect::<Vec<_>>();
172 assert_eq!(
173 states,
174 vec![TodoState::new("open"), TodoState::new("completed")]
175 );
176 assert_eq!(
177 entity
178 .pending_domain_events()
179 .iter()
180 .map(DomainEventOccurrence::aggregate_sequence)
181 .collect::<Vec<_>>(),
182 vec![1, 2]
183 );
184 }
185
186 #[test]
187 fn occurrence_uses_the_causing_replay_records_timestamp_and_metadata() {
188 let mut entity = Entity::with_id("todo-1");
189 entity.set_correlation_id("request-1");
190 entity.digest("todo.completed", &()).unwrap();
191 let causing_timestamp = entity.events()[0].timestamp;
192 entity.set_correlation_id("changed-after-transition");
193
194 entity
195 .capture_domain_state(
196 "todo",
197 state_event("todo.completed"),
198 &TodoState::new("completed"),
199 )
200 .unwrap();
201
202 let occurrence = &entity.pending_domain_events()[0];
203 let expected_timestamp: u64 = causing_timestamp
204 .duration_since(UNIX_EPOCH)
205 .unwrap()
206 .as_millis()
207 .try_into()
208 .unwrap();
209 assert_eq!(occurrence.occurred_at_unix_ms(), expected_timestamp);
210 assert_eq!(occurrence.correlation_id(), Some("request-1"));
211 }
212
213 #[test]
214 fn authoritative_causal_stamp_updates_replay_and_outward_occurrences_together() {
215 let mut entity = Entity::with_id("todo-1");
216 entity.set_causation_id("handler-value");
217 entity.digest("todo.completed", &()).unwrap();
218 entity
219 .capture_domain_state(
220 "todo",
221 state_event("todo.completed"),
222 &TodoState::new("completed"),
223 )
224 .unwrap();
225
226 entity.overwrite_new_event_causation_id("ledger-causation");
227
228 assert_eq!(entity.events()[0].causation_id(), Some("ledger-causation"));
229 assert_eq!(
230 entity.pending_domain_events()[0].causation_id(),
231 Some("ledger-causation")
232 );
233 }
234
235 #[test]
236 fn replay_suppresses_outward_capture_without_poisoning_the_entity() {
237 let mut entity = Entity::with_id("todo-1");
238 entity.set_replaying(true);
239
240 let outcome = entity
241 .capture_domain_state(
242 "todo",
243 state_event("todo.completed"),
244 &TodoState::new("completed"),
245 )
246 .unwrap();
247
248 assert_eq!(outcome, DomainEventCaptureOutcome::SuppressedDuringReplay);
249 assert!(entity.pending_domain_events().is_empty());
250 assert!(entity.domain_event_poison().is_none());
251 }
252
253 #[test]
254 fn retry_identity_is_stable_while_repeated_publication_ordinal_is_distinct() {
255 fn captured() -> Entity {
256 let mut entity = Entity::with_id("todo-1");
257 entity.digest("todo.completed", &()).unwrap();
258 entity
259 .capture_domain_state(
260 "todo",
261 state_event("todo.completed"),
262 &TodoState::new("completed"),
263 )
264 .unwrap();
265 entity
266 .capture_domain_state(
267 "todo",
268 state_event("todo.completed"),
269 &TodoState::new("completed"),
270 )
271 .unwrap();
272 entity
273 }
274
275 let first = captured();
276 let retry = captured();
277 assert_eq!(
278 first.pending_domain_events()[0].id(),
279 retry.pending_domain_events()[0].id()
280 );
281 assert_ne!(
282 first.pending_domain_events()[0].id(),
283 first.pending_domain_events()[1].id()
284 );
285 assert_eq!(first.pending_domain_events()[1].publication_ordinal(), 1);
286 }
287
288 #[test]
289 fn serialization_failure_poison_blocks_manual_commit_guard() {
290 let mut entity = Entity::with_id("todo-1");
291 entity.digest("todo.failed", &()).unwrap();
292 let descriptor = FailingState::DESCRIPTOR.clone().event("todo.failed", 1);
293
294 let error = entity
295 .capture_domain_state("todo", descriptor, &FailingState)
296 .unwrap_err();
297 let guard = entity.domain_event_commit_guard().unwrap_err();
298
299 assert!(matches!(error, DomainEventCaptureError::BodyEncoding(_)));
300 assert_eq!(guard.poison().error, error);
301 assert!(entity.pending_domain_events().is_empty());
302 }
303
304 #[test]
305 fn failed_persistence_retains_bytes_and_successful_persistence_clears_them() {
306 fn persist_fails(_occurrences: &[DomainEventOccurrence]) -> Result<(), &'static str> {
307 Err("injected persistence failure")
308 }
309
310 let mut entity = Entity::with_id("todo-1");
311 entity.digest("todo.renamed", &()).unwrap();
312 entity
313 .capture_domain_event(
314 "todo",
315 &TodoRenamed {
316 title: "new".into(),
317 },
318 )
319 .unwrap();
320 let before_failure = entity.pending_domain_events_for_commit().unwrap()[0]
321 .canonical_bytes()
322 .unwrap();
323
324 assert_eq!(
325 persist_fails(entity.pending_domain_events_for_commit().unwrap()),
326 Err("injected persistence failure")
327 );
328 let after_failure = entity.pending_domain_events_for_commit().unwrap()[0]
329 .canonical_bytes()
330 .unwrap();
331 assert_eq!(after_failure, before_failure);
332
333 entity.mark_domain_events_committed().unwrap();
334 assert!(entity.pending_domain_events().is_empty());
335 }
336
337 #[test]
338 fn deletion_body_requires_nonzero_incarnation_and_explicit_deletion_kind() {
339 assert_eq!(
340 DomainDeletion::new("todo-1", 0).unwrap_err(),
341 DomainDeletionError
342 );
343 let deletion = DomainDeletion::new("todo-1", 3).unwrap();
344 let mut entity = Entity::with_id("todo-1");
345 entity.digest("todo.purged", &()).unwrap();
346 entity
347 .capture_domain_deletion("todo", deletion_event("todo.purged"), &deletion)
348 .unwrap();
349
350 assert_eq!(
351 entity.pending_domain_events()[0].descriptor().body.kind,
352 DomainEventBodyKind::Deletion
353 );
354 }
355
356 #[test]
357 fn message_name_and_aggregate_id_limits_accept_boundary_values() {
358 let fixture: serde_json::Value = serde_json::from_str(include_str!(
359 "../../tests/fixtures/domain-event-occurrence-v1.json"
360 ))
361 .unwrap();
362 let message_name_bytes = fixture["boundary_vectors"]["message_name"]["at_bytes"]
363 .as_u64()
364 .unwrap() as usize;
365 let stable_id_bytes = fixture["boundary_vectors"]["stable_id"]["at_bytes"]
366 .as_u64()
367 .unwrap() as usize;
368 let mut envelope = fixed_envelope("i".repeat(stable_id_bytes));
369 envelope.aggregate_type = "a".repeat(message_name_bytes);
370 let mut descriptor = state_event("todo.completed");
371 descriptor.name = Cow::Owned("e".repeat(message_name_bytes));
372
373 let occurrence =
374 DomainEventOccurrence::capture(descriptor, envelope, &TodoState::new("open"));
375
376 assert_eq!(message_name_bytes, MAX_MESSAGE_NAME_LEN);
377 assert_eq!(stable_id_bytes, MAX_STABLE_MESSAGE_ID_LEN);
378 assert_eq!(
379 fixture["boundary_vectors"]["message_name"]["at_result"],
380 "accepted"
381 );
382 assert_eq!(
383 fixture["boundary_vectors"]["stable_id"]["at_result"],
384 "accepted"
385 );
386 let occurrence = occurrence.expect("boundary values must be accepted");
387 assert!(occurrence.id().len() <= MAX_STABLE_MESSAGE_ID_LEN);
388 }
389
390 #[test]
391 fn message_name_and_aggregate_id_limits_reject_over_limit_values() {
392 let fixture: serde_json::Value = serde_json::from_str(include_str!(
393 "../../tests/fixtures/domain-event-occurrence-v1.json"
394 ))
395 .unwrap();
396 let message_name_bytes = fixture["boundary_vectors"]["message_name"]["over_bytes"]
397 .as_u64()
398 .unwrap() as usize;
399 let stable_id_bytes = fixture["boundary_vectors"]["stable_id"]["over_bytes"]
400 .as_u64()
401 .unwrap() as usize;
402 let descriptor = DomainEventDescriptor {
403 name: Cow::Owned("e".repeat(message_name_bytes)),
404 ..state_event("todo.completed")
405 };
406 let name_error = DomainEventOccurrence::capture(
407 descriptor,
408 fixed_envelope("todo-1"),
409 &TodoState::new("open"),
410 )
411 .unwrap_err();
412 let id_error = DomainEventOccurrence::capture(
413 state_event("todo.completed"),
414 fixed_envelope("i".repeat(stable_id_bytes)),
415 &TodoState::new("open"),
416 )
417 .unwrap_err();
418
419 assert_eq!(message_name_bytes, MAX_MESSAGE_NAME_LEN + 1);
420 assert_eq!(stable_id_bytes, MAX_STABLE_MESSAGE_ID_LEN + 1);
421 assert_eq!(
422 fixture["boundary_vectors"]["message_name"]["over_result"],
423 "event_name_too_long"
424 );
425 assert_eq!(
426 fixture["boundary_vectors"]["stable_id"]["over_result"],
427 "aggregate_id_too_long"
428 );
429 assert!(matches!(
430 name_error,
431 DomainEventCaptureError::EventName(crate::bus::MessageNameError::TooLong { .. })
432 ));
433 assert!(matches!(
434 id_error,
435 DomainEventCaptureError::AggregateId(crate::bus::StableMessageIdError::TooLong { .. })
436 ));
437 }
438
439 #[test]
440 fn canonical_body_limit_accepts_exactly_one_mib() {
441 let fixture: serde_json::Value = serde_json::from_str(include_str!(
442 "../../tests/fixtures/domain-event-occurrence-v1.json"
443 ))
444 .unwrap();
445 let body_bytes = fixture["boundary_vectors"]["body"]["at_bytes"]
446 .as_u64()
447 .unwrap() as usize;
448 let body = "a".repeat(body_bytes - 2);
449 let occurrence = DomainEventOccurrence::capture(
450 state_event("todo.completed"),
451 fixed_envelope("todo-1"),
452 &body,
453 )
454 .unwrap();
455
456 assert_eq!(body_bytes, MAX_DOMAIN_EVENT_BODY_BYTES);
457 assert_eq!(fixture["boundary_vectors"]["body"]["at_result"], "accepted");
458 assert_eq!(occurrence.body_bytes().len(), body_bytes);
459 }
460
461 #[test]
462 fn canonical_body_limit_rejects_one_byte_over_one_mib() {
463 let fixture: serde_json::Value = serde_json::from_str(include_str!(
464 "../../tests/fixtures/domain-event-occurrence-v1.json"
465 ))
466 .unwrap();
467 let body_bytes = fixture["boundary_vectors"]["body"]["over_bytes"]
468 .as_u64()
469 .unwrap() as usize;
470 let body = "a".repeat(body_bytes - 2);
471 let error = DomainEventOccurrence::capture(
472 state_event("todo.completed"),
473 fixed_envelope("todo-1"),
474 &body,
475 )
476 .unwrap_err();
477
478 assert_eq!(
479 fixture["boundary_vectors"]["body"]["over_result"],
480 "body_too_large"
481 );
482 assert_eq!(
483 error,
484 DomainEventCaptureError::BodyTooLarge { len: body_bytes }
485 );
486 }
487
488 #[test]
489 fn canonical_fixture_round_trips_and_fingerprints_exact_wire_bytes() {
490 let fixture: serde_json::Value = serde_json::from_str(include_str!(
491 "../../tests/fixtures/domain-event-occurrence-v1.json"
492 ))
493 .unwrap();
494 let occurrence = DomainEventOccurrence::capture(
495 state_event("todo.completed"),
496 fixed_envelope("todo-1"),
497 &TodoState::new("completed"),
498 )
499 .unwrap();
500 let canonical = occurrence.canonical_bytes().unwrap();
501 let actual_value: serde_json::Value = serde_json::from_slice(&canonical).unwrap();
502 let digest = Sha256::digest(&canonical);
503 let actual_digest = format!("sha256:{digest:x}");
504
505 assert_eq!(actual_value, fixture["occurrence"]);
506 assert_eq!(actual_digest, fixture["canonical_occurrence_fingerprint"]);
507 assert_eq!(
508 std::str::from_utf8(occurrence.body_bytes()).unwrap(),
509 fixture["canonical_body"].as_str().unwrap()
510 );
511 assert_eq!(
512 DomainEventOccurrence::from_canonical_bytes(&canonical).unwrap(),
513 occurrence
514 );
515 assert_eq!(
516 fixture["limits"]["body_bytes"].as_u64().unwrap() as usize,
517 MAX_DOMAIN_EVENT_BODY_BYTES
518 );
519 assert_eq!(
520 fixture["limits"]["occurrence_wire_bytes"].as_u64().unwrap() as usize,
521 MAX_DOMAIN_EVENT_OCCURRENCE_WIRE_BYTES
522 );
523 }
524}