1use core::ops::Range;
2
3use bytes::Bytes;
4use mnesis::{DomainEvent, Version};
5use thiserror::Error;
6
7use crate::value::{
8 EventType, MAX_EVENT_TYPE_LEN, MAX_METADATA_LEN, Metadata, Payload, SchemaVersion, ValueError,
9};
10
11#[allow(
16 clippy::as_conversions,
17 reason = "u32→usize is lossless on all Mnesis target platforms (32-bit+)"
18)]
19#[inline]
20const fn idx(n: u32) -> usize {
21 n as usize
22}
23
24#[derive(Debug, Error)]
30#[non_exhaustive]
31pub enum EnvelopeError {
32 #[error("range {start}..{end} exceeds buffer length {len}")]
33 RangeOutOfBounds { start: u32, end: u32, len: usize },
34
35 #[error("invalid UTF-8 in event_type at bytes {start}..{end}")]
36 InvalidUtf8 {
37 start: u32,
38 end: u32,
39 #[source]
40 source: core::str::Utf8Error,
41 },
42
43 #[error("event_type range length {actual} exceeds maximum {max}")]
49 EventTypeRangeTooLong { actual: u32, max: usize },
50
51 #[error("metadata range length {actual} exceeds maximum {max}")]
55 MetadataRangeTooLong { actual: u32, max: usize },
56
57 #[error("metadata range is empty; use None to represent absent metadata")]
63 MetadataRangeEmpty,
64
65 #[error(transparent)]
66 Value(#[from] ValueError),
67}
68
69#[derive(Debug, Error)]
75#[non_exhaustive]
76pub enum ForDecodeError {
77 #[error(transparent)]
78 Value(#[from] ValueError),
79 #[error(transparent)]
80 Wire(#[from] crate::wire::WireError),
81 #[error(transparent)]
82 Envelope(#[from] EnvelopeError),
83}
84
85#[derive(Debug, Clone)]
96pub struct PendingEnvelope {
97 version: Version,
98 event_type: EventType,
99 schema_version: SchemaVersion,
100 payload: Payload,
101 metadata: Option<Metadata>,
102}
103
104impl PendingEnvelope {
105 #[must_use]
106 pub const fn version(&self) -> Version {
107 self.version
108 }
109
110 #[must_use]
112 pub fn event_type(&self) -> &str {
113 self.event_type.as_str()
114 }
115
116 #[must_use]
118 pub fn event_type_value(&self) -> EventType {
119 self.event_type.clone()
120 }
121
122 #[must_use]
123 pub fn payload(&self) -> &[u8] {
124 self.payload.as_slice()
125 }
126
127 #[must_use]
129 pub fn payload_bytes(&self) -> Bytes {
130 self.payload.clone().into_bytes()
131 }
132
133 #[must_use]
135 pub fn payload_value(&self) -> Payload {
136 self.payload.clone()
137 }
138
139 #[must_use]
140 pub fn metadata(&self) -> Option<&[u8]> {
141 self.metadata.as_ref().map(Metadata::as_slice)
142 }
143
144 #[must_use]
146 pub fn metadata_bytes(&self) -> Option<Bytes> {
147 self.metadata.as_ref().map(|m| m.clone().into_bytes())
148 }
149
150 #[must_use]
152 pub fn metadata_value(&self) -> Option<Metadata> {
153 self.metadata.clone()
154 }
155
156 #[must_use]
158 pub const fn schema_version(&self) -> u32 {
159 self.schema_version.get()
160 }
161
162 #[must_use]
164 pub const fn schema_version_value(&self) -> SchemaVersion {
165 self.schema_version
166 }
167
168 #[cfg(feature = "import")]
177 #[must_use]
178 pub(crate) fn from_persisted(persisted: &PersistedEnvelope) -> Self {
179 Self {
180 version: persisted.version(),
181 event_type: persisted.event_type_value(),
182 schema_version: persisted.schema_version_value(),
183 payload: persisted.payload_value(),
184 metadata: persisted.metadata_value(),
185 }
186 }
187}
188
189#[derive(Debug)]
195pub struct WithVersion {
196 version: Version,
197}
198
199#[derive(Debug)]
201pub struct WithEventType {
202 version: Version,
203 event_type: EventType,
204}
205
206#[derive(Debug)]
208pub struct WithPayload {
209 version: Version,
210 event_type: EventType,
211 payload: Bytes,
212 schema_version: SchemaVersion,
213 metadata: Option<Bytes>,
214}
215
216impl WithVersion {
217 #[must_use]
219 pub fn event_type(self, event_type: &'static str) -> WithEventType {
220 WithEventType {
221 version: self.version,
222 event_type: EventType::from_static_str(event_type),
223 }
224 }
225
226 #[must_use]
228 pub fn event<E: DomainEvent + ?Sized>(self, event: &E) -> WithEventType {
229 self.event_type(event.name())
230 }
231
232 pub fn event_type_bytes(self, bytes: Bytes) -> Result<WithEventType, EnvelopeError> {
239 let event_type = EventType::from_bytes(bytes)?;
240 Ok(WithEventType {
241 version: self.version,
242 event_type,
243 })
244 }
245}
246
247impl WithEventType {
248 #[must_use]
251 pub fn payload(self, payload: impl Into<Bytes>) -> WithPayload {
252 WithPayload {
253 version: self.version,
254 event_type: self.event_type,
255 payload: payload.into(),
256 schema_version: SchemaVersion::INITIAL,
257 metadata: None,
258 }
259 }
260}
261
262impl WithPayload {
263 #[must_use]
265 pub const fn schema_version(mut self, schema_version: SchemaVersion) -> Self {
266 self.schema_version = schema_version;
267 self
268 }
269
270 #[must_use]
273 pub fn metadata(mut self, metadata: impl Into<Bytes>) -> Self {
274 self.metadata = Some(metadata.into());
275 self
276 }
277
278 pub fn build(self) -> Result<PendingEnvelope, EnvelopeError> {
286 let payload = Payload::from_bytes(self.payload)?;
287 let metadata = self.metadata.map(Metadata::from_bytes).transpose()?;
288 Ok(PendingEnvelope {
289 version: self.version,
290 event_type: self.event_type,
291 schema_version: self.schema_version,
292 payload,
293 metadata,
294 })
295 }
296}
297
298#[must_use]
313pub const fn pending_envelope(version: Version) -> WithVersion {
314 WithVersion { version }
315}
316
317#[derive(Debug, Clone)]
332pub struct PersistedEnvelope {
333 version: Version,
334 schema_version: SchemaVersion,
335 value: Bytes,
336 event_type_range: Range<u32>,
337 payload_range: Range<u32>,
338 metadata_range: Option<Range<u32>>,
339}
340
341impl PersistedEnvelope {
342 #[allow(
355 clippy::too_many_arguments,
356 reason = "all 6 fields are required to construct a validated PersistedEnvelope; \
357 a builder would add indirection with no type-safety benefit here"
358 )]
359 pub fn try_new(
360 version: Version,
361 value: Bytes,
362 schema_version: SchemaVersion,
363 event_type_range: Range<u32>,
364 payload_range: Range<u32>,
365 metadata_range: Option<Range<u32>>,
366 ) -> Result<Self, EnvelopeError> {
367 let len = value.len();
368 check_range(&event_type_range, len)?;
369 check_range(&payload_range, len)?;
370 if let Some(ref m) = metadata_range {
371 check_range(m, len)?;
372 }
373
374 let et_range_len = event_type_range.end - event_type_range.start;
378 if idx(et_range_len) > MAX_EVENT_TYPE_LEN {
379 return Err(EnvelopeError::EventTypeRangeTooLong {
380 actual: et_range_len,
381 max: MAX_EVENT_TYPE_LEN,
382 });
383 }
384 if let Some(ref m) = metadata_range {
385 let meta_range_len = m.end - m.start;
386 if meta_range_len == 0 {
387 return Err(EnvelopeError::MetadataRangeEmpty);
388 }
389 if idx(meta_range_len) > MAX_METADATA_LEN {
390 return Err(EnvelopeError::MetadataRangeTooLong {
391 actual: meta_range_len,
392 max: MAX_METADATA_LEN,
393 });
394 }
395 }
396
397 let et_start = idx(event_type_range.start);
399 let et_end = idx(event_type_range.end);
400 core::str::from_utf8(&value[et_start..et_end]).map_err(|e| EnvelopeError::InvalidUtf8 {
401 start: event_type_range.start,
402 end: event_type_range.end,
403 source: e,
404 })?;
405 Ok(Self {
406 version,
407 schema_version,
408 value,
409 event_type_range,
410 payload_range,
411 metadata_range,
412 })
413 }
414
415 #[must_use]
416 pub const fn version(&self) -> Version {
417 self.version
418 }
419
420 #[must_use]
423 pub const fn schema_version(&self) -> u32 {
424 self.schema_version.get()
425 }
426
427 #[must_use]
429 pub const fn schema_version_value(&self) -> SchemaVersion {
430 self.schema_version
431 }
432
433 #[must_use]
435 pub fn event_type(&self) -> &str {
436 let start = idx(self.event_type_range.start);
437 let end = idx(self.event_type_range.end);
438 #[allow(
442 unsafe_code,
443 reason = "UTF-8 invariant established at construction; ranges validated"
444 )]
445 unsafe {
446 core::str::from_utf8_unchecked(&self.value[start..end])
447 }
448 }
449
450 #[must_use]
452 pub fn event_type_bytes(&self) -> Bytes {
453 self.slice_range(&self.event_type_range)
454 }
455
456 #[must_use]
457 pub fn payload(&self) -> &[u8] {
458 let start = idx(self.payload_range.start);
459 let end = idx(self.payload_range.end);
460 &self.value[start..end]
461 }
462
463 #[must_use]
465 pub fn payload_bytes(&self) -> Bytes {
466 self.slice_range(&self.payload_range)
467 }
468
469 #[must_use]
470 pub fn metadata(&self) -> Option<&[u8]> {
471 self.metadata_range.as_ref().map(|r| {
472 let start = idx(r.start);
473 let end = idx(r.end);
474 &self.value[start..end]
475 })
476 }
477
478 #[must_use]
483 pub fn metadata_bytes(&self) -> Option<Bytes> {
484 self.metadata_range.as_ref().map(|r| self.slice_range(r))
485 }
486
487 #[must_use]
489 pub fn event_type_value(&self) -> EventType {
490 #[allow(
496 unsafe_code,
497 reason = "UTF-8 and length cap both established by try_new"
498 )]
499 unsafe {
500 EventType::from_validated_bytes(self.event_type_bytes())
501 }
502 }
503
504 #[must_use]
506 pub fn payload_value(&self) -> Payload {
507 #[allow(
513 unsafe_code,
514 reason = "size invariant established at PersistedEnvelope::try_new"
515 )]
516 unsafe {
517 Payload::from_validated_bytes(self.payload_bytes())
518 }
519 }
520
521 #[must_use]
526 pub fn metadata_value(&self) -> Option<Metadata> {
527 self.metadata_bytes().map(|b| {
528 #[allow(
534 unsafe_code,
535 reason = "non-empty and length cap both established by try_new"
536 )]
537 unsafe {
538 Metadata::from_validated_bytes(b)
539 }
540 })
541 }
542
543 #[must_use]
547 pub fn schema_version_as_version(&self) -> Version {
548 Version::from(self.schema_version)
549 }
550
551 pub fn for_decode(event_type: &str, payload: &[u8]) -> Result<Self, ForDecodeError> {
571 let et = EventType::from_bytes(Bytes::copy_from_slice(event_type.as_bytes()))?;
572 let pl = Payload::from_bytes(Bytes::copy_from_slice(payload))?;
573 let sv = SchemaVersion::INITIAL;
574 let frame = crate::wire::encode_frame(sv, &et, &pl, None)?;
575 Ok(Self::try_new(
576 Version::INITIAL,
577 frame.value,
578 sv,
579 frame.offsets.event_type,
580 frame.offsets.payload,
581 None,
582 )?)
583 }
584
585 fn slice_range(&self, range: &Range<u32>) -> Bytes {
586 self.value.slice(idx(range.start)..idx(range.end))
587 }
588}
589
590const fn check_range(range: &Range<u32>, len: usize) -> Result<(), EnvelopeError> {
591 if idx(range.end) > len || range.start > range.end {
592 return Err(EnvelopeError::RangeOutOfBounds {
593 start: range.start,
594 end: range.end,
595 len,
596 });
597 }
598 Ok(())
599}
600
601#[cfg(test)]
602#[allow(
603 clippy::unwrap_used,
604 clippy::expect_used,
605 clippy::panic,
606 reason = "test code asserts exact values"
607)]
608mod tests {
609 use super::*;
610 use crate::value::MAX_EVENT_TYPE_LEN;
611 use bytes::Bytes;
612 use mnesis::Version;
613
614 #[derive(Debug)]
616 struct TestEvent;
617
618 impl mnesis::Message for TestEvent {}
619
620 impl DomainEvent for TestEvent {
621 fn name(&self) -> &'static str {
622 "TestEvent"
623 }
624 }
625
626 #[test]
627 fn pending_envelope_builds_with_metadata() {
628 let env = pending_envelope(Version::INITIAL)
629 .event_type("UserCreated")
630 .payload(Bytes::from_static(b"payload-bytes"))
631 .metadata(Bytes::from_static(b"meta-bytes"))
632 .build()
633 .expect("valid envelope");
634
635 assert_eq!(env.event_type(), "UserCreated");
636 assert_eq!(env.payload(), b"payload-bytes");
637 assert_eq!(env.metadata(), Some(b"meta-bytes".as_slice()));
638 assert_eq!(env.schema_version(), 1);
639 }
640
641 #[test]
642 fn pending_envelope_builds_without_metadata() {
643 let env = pending_envelope(Version::INITIAL)
644 .event_type("X")
645 .payload(Bytes::from_static(b"p"))
646 .build()
647 .expect("valid envelope");
648
649 assert_eq!(env.metadata(), None);
650 }
651
652 #[test]
653 fn pending_envelope_typed_accessors_roundtrip() {
654 let env = pending_envelope(Version::INITIAL)
655 .event_type("UserCreated")
656 .payload(Bytes::from_static(b"payload-bytes"))
657 .metadata(Bytes::from_static(b"meta-bytes"))
658 .build()
659 .expect("valid envelope");
660
661 assert_eq!(env.event_type_value().as_str(), "UserCreated");
662 assert_eq!(env.payload_value().as_slice(), b"payload-bytes");
663 assert_eq!(
664 env.metadata_value().map(|m| m.as_slice().to_vec()),
665 Some(b"meta-bytes".to_vec())
666 );
667 assert_eq!(env.schema_version_value().get(), 1);
668 }
669
670 #[test]
671 fn pending_envelope_rejects_oversize_event_type() {
672 let oversized = "x".repeat(MAX_EVENT_TYPE_LEN + 1);
673 let err = pending_envelope(Version::INITIAL)
674 .event_type_bytes(Bytes::from(oversized))
675 .expect_err("oversized must be rejected");
676 assert!(matches!(err, EnvelopeError::Value(_)));
677 }
678
679 #[test]
680 fn event_derives_same_event_type_as_event_type_name() {
681 let via_event = pending_envelope(Version::INITIAL)
682 .event(&TestEvent)
683 .payload(Bytes::from_static(b"p"))
684 .build()
685 .expect("valid envelope");
686 let via_name = pending_envelope(Version::INITIAL)
687 .event_type(TestEvent.name())
688 .payload(Bytes::from_static(b"p"))
689 .build()
690 .expect("valid envelope");
691
692 assert_eq!(via_event.event_type(), via_name.event_type());
693 assert_eq!(via_event.event_type(), "TestEvent");
694 }
695
696 #[test]
697 fn build_rejects_oversize_payload() {
698 let oversized = vec![0u8; crate::value::MAX_PAYLOAD_LEN + 1];
699 let err = pending_envelope(Version::INITIAL)
700 .event_type("X")
701 .payload(Bytes::from(oversized))
702 .build()
703 .expect_err("oversized payload must be rejected");
704 assert!(matches!(err, EnvelopeError::Value(_)));
705 }
706
707 #[test]
708 fn build_rejects_empty_metadata() {
709 let err = pending_envelope(Version::INITIAL)
710 .event_type("X")
711 .payload(Bytes::from_static(b"p"))
712 .metadata(Bytes::new())
713 .build()
714 .expect_err("empty metadata must be rejected");
715 assert!(matches!(err, EnvelopeError::Value(_)));
716 }
717
718 #[test]
719 fn persisted_envelope_accessors_return_views_into_value() {
720 let value = Bytes::from_static(b"TYPEpayloadmeta");
721 let env = PersistedEnvelope::try_new(
723 Version::INITIAL,
724 value,
725 SchemaVersion::INITIAL,
726 0..4,
727 4..11,
728 Some(11..15),
729 )
730 .expect("valid construction");
731
732 assert_eq!(env.event_type(), "TYPE");
733 assert_eq!(env.payload(), b"payload");
734 assert_eq!(env.metadata(), Some(b"meta".as_slice()));
735 }
736
737 #[test]
738 fn persisted_envelope_payload_bytes_shares_arc() {
739 let env = PersistedEnvelope::try_new(
740 Version::INITIAL,
741 Bytes::from_static(b"TYPEpayload"),
742 SchemaVersion::INITIAL,
743 0..4,
744 4..11,
745 None,
746 )
747 .unwrap();
748
749 let payload = env.payload_bytes();
750 assert_eq!(payload.as_ref(), b"payload");
751 }
752
753 #[test]
754 fn persisted_envelope_rejects_range_past_buffer() {
755 let value = Bytes::from_static(b"short");
756 let err = PersistedEnvelope::try_new(
757 Version::INITIAL,
758 value,
759 SchemaVersion::INITIAL,
760 0..4,
761 4..100,
762 None,
763 )
764 .expect_err("must reject out-of-bounds range");
765 assert!(matches!(err, EnvelopeError::RangeOutOfBounds { .. }));
766 }
767
768 #[test]
769 fn try_new_rejects_event_type_range_too_long() {
770 let len = MAX_EVENT_TYPE_LEN + 1;
771 let mut buf = vec![0u8; len];
772 buf[0] = b'A';
774 let value = Bytes::from(buf);
775 let too_long_end = u32::try_from(len).expect("len fits in u32 by construction");
776
777 let err = PersistedEnvelope::try_new(
778 Version::INITIAL,
779 value,
780 SchemaVersion::INITIAL,
781 0..too_long_end,
782 too_long_end..too_long_end,
783 None,
784 )
785 .expect_err("event_type range > MAX_EVENT_TYPE_LEN must be rejected");
786
787 let expected_actual = u32::try_from(len).expect("len fits in u32 by construction (above)");
788 assert!(matches!(
789 err,
790 EnvelopeError::EventTypeRangeTooLong { actual, max }
791 if actual == expected_actual && max == MAX_EVENT_TYPE_LEN
792 ));
793 }
794
795 #[test]
796 fn persisted_envelope_rejects_empty_metadata_range() {
797 let env = PersistedEnvelope::try_new(
798 Version::INITIAL,
799 Bytes::from_static(b"TYPEpayload"),
800 SchemaVersion::INITIAL,
801 0..4,
802 4..11,
803 Some(4..4),
804 );
805 assert!(matches!(env, Err(EnvelopeError::MetadataRangeEmpty)));
806 }
807
808 #[test]
809 fn persisted_envelope_rejects_invalid_utf8_in_event_type() {
810 let value = Bytes::from_static(&[0xFFu8, 0xFF, b'p', b'a', b'y']);
811 let err = PersistedEnvelope::try_new(
812 Version::INITIAL,
813 value,
814 SchemaVersion::INITIAL,
815 0..2,
816 2..5,
817 None,
818 )
819 .expect_err("must reject non-UTF-8 event_type");
820 assert!(matches!(err, EnvelopeError::InvalidUtf8 { .. }));
821 }
822
823 #[test]
824 fn persisted_envelope_event_type_value_returns_validated_value_newtype() {
825 let env = PersistedEnvelope::try_new(
826 Version::INITIAL,
827 Bytes::from_static(b"TYPEpayload"),
828 SchemaVersion::INITIAL,
829 0..4,
830 4..11,
831 None,
832 )
833 .expect("valid");
834
835 let et = env.event_type_value();
836 assert_eq!(et.as_str(), "TYPE");
837 }
838
839 #[test]
840 fn persisted_envelope_payload_value_returns_validated_value_newtype() {
841 let env = PersistedEnvelope::try_new(
842 Version::INITIAL,
843 Bytes::from_static(b"TYPEpayload"),
844 SchemaVersion::INITIAL,
845 0..4,
846 4..11,
847 None,
848 )
849 .expect("valid");
850
851 let p = env.payload_value();
852 assert_eq!(p.as_slice(), b"payload");
853 }
854
855 #[test]
856 fn persisted_envelope_metadata_value_returns_some_when_present() {
857 let env = PersistedEnvelope::try_new(
858 Version::INITIAL,
859 Bytes::from_static(b"TYPEpayloadMETA"),
860 SchemaVersion::INITIAL,
861 0..4,
862 4..11,
863 Some(11..15),
864 )
865 .expect("valid");
866
867 let m = env.metadata_value().expect("present");
868 assert_eq!(m.as_slice(), b"META");
869 }
870
871 #[test]
872 fn persisted_envelope_metadata_value_returns_none_when_absent() {
873 let env = PersistedEnvelope::try_new(
874 Version::INITIAL,
875 Bytes::from_static(b"TYPEpayload"),
876 SchemaVersion::INITIAL,
877 0..4,
878 4..11,
879 None,
880 )
881 .expect("valid");
882
883 assert!(env.metadata_value().is_none());
884 }
885
886 #[cfg(feature = "import")]
887 #[test]
888 fn from_persisted_preserves_fields_drops_global_seq_zero_copy() {
889 let value = Bytes::from_static(b"TYPEpayloadmeta");
891 let persisted = PersistedEnvelope::try_new(
892 Version::new(7).expect("nonzero"),
893 value,
894 crate::value::SchemaVersion::from_u32(3).expect("nonzero"),
895 0..4,
896 4..11,
897 Some(11..15),
898 )
899 .expect("valid");
900
901 let pending = PendingEnvelope::from_persisted(&persisted);
902
903 assert_eq!(pending.version(), persisted.version());
905 assert_eq!(pending.event_type(), "TYPE");
906 assert_eq!(pending.payload(), b"payload");
907 assert_eq!(pending.metadata(), Some(b"meta".as_slice()));
908 assert_eq!(pending.schema_version(), 3);
909
910 assert!(
912 std::ptr::eq(
913 pending.payload_bytes().as_ptr(),
914 persisted.payload().as_ptr()
915 ),
916 "from_persisted must reuse the Arc-shared payload, not deep-copy",
917 );
918 }
919}
920
921#[cfg(test)]
931#[allow(clippy::expect_used, reason = "test code")]
932mod persisted_exhaustive_tests {
933 use super::{EnvelopeError, PersistedEnvelope};
934 use crate::value::{MAX_EVENT_TYPE_LEN, Metadata, SchemaVersion};
935 use bytes::Bytes;
936 use mnesis::Version;
937 use proptest::prelude::*;
938 use static_assertions::assert_impl_all;
939 use std::ops::Range;
940
941 fn v(n: u64) -> Version {
944 Version::new(n).expect("test version must be nonzero")
945 }
946
947 fn sv(n: u32) -> SchemaVersion {
948 SchemaVersion::from_u32(n).expect("test schema_version must be nonzero")
949 }
950
951 fn assemble(
954 event_type: &[u8],
955 payload: &[u8],
956 metadata: Option<&[u8]>,
957 ) -> (Bytes, Range<u32>, Range<u32>, Option<Range<u32>>) {
958 let mut buf = Vec::new();
959 buf.extend_from_slice(event_type);
960 buf.extend_from_slice(payload);
961 if let Some(m) = metadata {
962 buf.extend_from_slice(m);
963 }
964 let et_end = u32::try_from(event_type.len()).expect("event_type len fits u32");
965 let pl_end = et_end + u32::try_from(payload.len()).expect("payload len fits u32");
966 let meta_range = metadata.map(|m| {
967 let end = pl_end + u32::try_from(m.len()).expect("metadata len fits u32");
968 pl_end..end
969 });
970 (Bytes::from(buf), 0..et_end, et_end..pl_end, meta_range)
971 }
972
973 fn build(
975 version: Version,
976 schema: SchemaVersion,
977 event_type: &[u8],
978 payload: &[u8],
979 metadata: Option<&[u8]>,
980 ) -> PersistedEnvelope {
981 let (value, et, pl, meta) = assemble(event_type, payload, metadata);
982 PersistedEnvelope::try_new(version, value, schema, et, pl, meta)
983 .expect("components form a valid PersistedEnvelope")
984 }
985
986 assert_impl_all!(PersistedEnvelope: Send, Sync, Clone, std::fmt::Debug);
991
992 #[test]
993 fn persisted_envelope_clones_across_thread_boundary() {
994 let event = build(
995 v(3),
996 sv(2),
997 b"MoneyDeposited",
998 b"payload-bytes",
999 Some(b"meta"),
1000 );
1001 let moved = event.clone();
1002 let (ver, schema, etype, payload, meta) = std::thread::scope(|scope| {
1005 scope
1006 .spawn(move || {
1007 (
1008 moved.version(),
1009 moved.schema_version(),
1010 moved.event_type().to_owned(),
1011 moved.payload().to_vec(),
1012 moved.metadata().map(<[u8]>::to_vec),
1013 )
1014 })
1015 .join()
1016 .expect("worker thread must not panic")
1017 });
1018
1019 assert_eq!(ver, v(3));
1020 assert_eq!(schema, 2);
1021 assert_eq!(etype, "MoneyDeposited");
1022 assert_eq!(payload, b"payload-bytes");
1023 assert_eq!(meta, Some(b"meta".to_vec()));
1024 assert_eq!(event.event_type(), "MoneyDeposited");
1026 }
1027
1028 #[test]
1033 fn repeated_accessor_calls_are_consistent() {
1034 let event = build(v(9), sv(4), b"TYPE", b"payload", Some(b"meta"));
1035
1036 assert_eq!(event.event_type(), event.event_type());
1038 assert_eq!(event.payload(), event.payload());
1039 assert_eq!(event.metadata(), event.metadata());
1040
1041 assert_eq!(
1043 event.event_type().as_bytes(),
1044 event.event_type_bytes().as_ref()
1045 );
1046 assert_eq!(event.payload(), event.payload_bytes().as_ref());
1047 assert_eq!(event.metadata(), event.metadata_bytes().as_deref());
1048
1049 assert_eq!(event.event_type(), event.event_type_value().as_str());
1051 assert_eq!(event.payload(), event.payload_value().as_slice());
1052 assert_eq!(
1053 event.metadata(),
1054 event.metadata_value().as_ref().map(Metadata::as_slice),
1055 );
1056
1057 assert_eq!(event.version(), v(9));
1059 assert_eq!(event.version(), event.version());
1060 assert_eq!(event.schema_version(), 4);
1061 assert_eq!(event.schema_version_value(), sv(4));
1062 }
1063
1064 #[test]
1067 fn clone_is_field_for_field_equal_view() {
1068 let original = build(
1069 v(42),
1070 sv(7),
1071 b"OrderPlaced",
1072 b"the-payload",
1073 Some(b"the-meta"),
1074 );
1075 let cloned = original.clone();
1076
1077 assert_eq!(cloned.version(), original.version());
1078 assert_eq!(cloned.schema_version(), original.schema_version());
1079 assert_eq!(
1080 cloned.schema_version_value(),
1081 original.schema_version_value()
1082 );
1083 assert_eq!(cloned.event_type(), original.event_type());
1084 assert_eq!(cloned.payload(), original.payload());
1085 assert_eq!(cloned.metadata(), original.metadata());
1086 assert_eq!(
1087 cloned.metadata_value().map(|m| m.as_slice().to_vec()),
1088 original.metadata_value().map(|m| m.as_slice().to_vec()),
1089 );
1090 }
1091
1092 #[test]
1093 fn clone_shares_the_same_backing_buffer() {
1094 let original = build(v(1), sv(1), b"TYPE", b"payload", None);
1097 let cloned = original.clone();
1098
1099 let from_original = original.payload_bytes();
1100 let from_clone = cloned.payload_bytes();
1101 assert!(
1102 std::ptr::eq(from_original.as_ptr(), from_clone.as_ptr()),
1103 "clone must share the parent buffer, not deep-copy it",
1104 );
1105 assert_eq!(from_original.as_ref(), from_clone.as_ref());
1106 }
1107
1108 #[test]
1111 fn rejects_inverted_range_start_after_end() {
1112 let value = Bytes::from_static(b"TYPEpayload");
1113 let (bad_start, bad_end) = (7u32, 2u32);
1116 let err = PersistedEnvelope::try_new(
1117 Version::INITIAL,
1118 value,
1119 SchemaVersion::INITIAL,
1120 0..4,
1121 bad_start..bad_end,
1122 None,
1123 )
1124 .expect_err("start > end must be rejected");
1125 assert!(matches!(
1126 err,
1127 EnvelopeError::RangeOutOfBounds {
1128 start: 7,
1129 end: 2,
1130 ..
1131 }
1132 ));
1133 }
1134
1135 #[test]
1136 fn event_type_cap_uses_range_length_not_endpoint_sum() {
1137 let start: u32 = 40_000;
1145 let end: u32 = 70_000;
1146 let buf_len = usize::try_from(end).expect("70_000 fits usize");
1147 let value = Bytes::from(vec![b'A'; buf_len]);
1148 let event = PersistedEnvelope::try_new(
1149 Version::INITIAL,
1150 value,
1151 SchemaVersion::INITIAL,
1152 start..end,
1153 0..0,
1154 None,
1155 )
1156 .expect("event_type length 30_000 is within MAX_EVENT_TYPE_LEN (65_535)");
1157 assert_eq!(event.event_type().len(), 30_000);
1158 }
1159
1160 #[test]
1165 fn rejects_event_type_range_past_buffer_end() {
1166 let value = Bytes::from_static(b"short");
1167 let err = PersistedEnvelope::try_new(
1168 Version::INITIAL,
1169 value,
1170 SchemaVersion::INITIAL,
1171 0..100,
1173 0..0,
1174 None,
1175 )
1176 .expect_err("event_type range past buffer must be rejected");
1177 assert!(matches!(
1178 err,
1179 EnvelopeError::RangeOutOfBounds {
1180 start: 0,
1181 end: 100,
1182 len: 5
1183 }
1184 ));
1185 }
1186
1187 #[test]
1188 fn rejects_metadata_range_past_buffer_end() {
1189 let value = Bytes::from_static(b"TYPEpayload");
1190 let err = PersistedEnvelope::try_new(
1191 Version::INITIAL,
1192 value,
1193 SchemaVersion::INITIAL,
1194 0..4,
1196 4..11,
1197 Some(11..100),
1198 )
1199 .expect_err("metadata range past buffer must be rejected");
1200 assert!(matches!(
1201 err,
1202 EnvelopeError::RangeOutOfBounds {
1203 start: 11,
1204 end: 100,
1205 len: 11
1206 }
1207 ));
1208 }
1209
1210 #[test]
1211 fn rejects_inverted_metadata_range() {
1212 let value = Bytes::from_static(b"TYPEpayload");
1213 let (bad_start, bad_end) = (9u32, 5u32);
1214 let err = PersistedEnvelope::try_new(
1215 Version::INITIAL,
1216 value,
1217 SchemaVersion::INITIAL,
1218 0..4,
1219 4..11,
1220 Some(bad_start..bad_end),
1221 )
1222 .expect_err("metadata start > end must be rejected");
1223 assert!(matches!(
1224 err,
1225 EnvelopeError::RangeOutOfBounds {
1226 start: 9,
1227 end: 5,
1228 ..
1229 }
1230 ));
1231 }
1232
1233 #[test]
1236 fn ranges_are_independent_and_may_overlap() {
1237 let value = Bytes::from_static(b"TYPE");
1241 let event = PersistedEnvelope::try_new(
1242 Version::INITIAL,
1243 value,
1244 SchemaVersion::INITIAL,
1245 0..4,
1246 0..4,
1247 Some(0..4),
1248 )
1249 .expect("overlapping ranges are structurally permitted");
1250 assert_eq!(event.event_type(), "TYPE");
1251 assert_eq!(event.payload(), b"TYPE");
1252 assert_eq!(event.metadata(), Some(b"TYPE".as_slice()));
1253 }
1254
1255 #[test]
1256 fn empty_payload_range_is_accepted_unlike_empty_metadata() {
1257 let value = Bytes::from_static(b"TYPE");
1261 let event = PersistedEnvelope::try_new(
1262 Version::INITIAL,
1263 value,
1264 SchemaVersion::INITIAL,
1265 0..4,
1266 4..4,
1267 None,
1268 )
1269 .expect("empty payload range is accepted");
1270 assert_eq!(event.payload(), b"");
1271 assert_eq!(event.payload_value().as_slice(), b"");
1272 }
1273
1274 #[test]
1277 fn metadata_absent_is_none_everywhere() {
1278 let event = build(v(1), sv(1), b"TYPE", b"payload", None);
1279 assert!(event.metadata().is_none());
1280 assert!(event.metadata_bytes().is_none());
1281 assert!(event.metadata_value().is_none());
1282 }
1283
1284 #[test]
1285 fn metadata_present_threads_through_every_accessor() {
1286 let event = build(v(1), sv(1), b"TYPE", b"payload", Some(b"META"));
1287 assert_eq!(event.metadata(), Some(b"META".as_slice()));
1288 assert_eq!(event.metadata_bytes().expect("some").as_ref(), b"META");
1289 assert_eq!(event.metadata_value().expect("some").as_slice(), b"META");
1290 }
1291
1292 #[test]
1295 fn empty_event_type_is_accepted() {
1296 let event = build(v(1), sv(1), b"", b"payload", None);
1297 assert_eq!(event.event_type(), "");
1298 assert_eq!(event.event_type_value().as_str(), "");
1299 assert_eq!(event.payload(), b"payload");
1300 }
1301
1302 #[test]
1305 fn boundary_version_and_schema_version_carried_verbatim() {
1306 let event = build(v(u64::MAX), sv(u32::MAX), b"TYPE", b"payload", None);
1307 assert_eq!(event.version().as_u64(), u64::MAX);
1308 assert_eq!(event.schema_version(), u32::MAX);
1309 assert_eq!(event.schema_version_value().get(), u32::MAX);
1310 }
1311
1312 #[test]
1315 fn owned_views_alias_the_single_backing_buffer() {
1316 let (value, et, pl, meta) = assemble(b"TYPE", b"payload", Some(b"meta"));
1317 let base = value.clone(); let event = PersistedEnvelope::try_new(
1319 Version::INITIAL,
1320 value,
1321 SchemaVersion::INITIAL,
1322 et,
1323 pl,
1324 meta,
1325 )
1326 .expect("valid");
1327
1328 let et_bytes = event.event_type_bytes();
1329 let pl_bytes = event.payload_bytes();
1330 let meta_bytes = event.metadata_bytes().expect("metadata present");
1331
1332 assert!(
1335 std::ptr::eq(et_bytes.as_ptr(), base.as_ptr().wrapping_add(0)),
1336 "event_type_bytes must alias buffer offset 0",
1337 );
1338 assert!(
1339 std::ptr::eq(pl_bytes.as_ptr(), base.as_ptr().wrapping_add(4)),
1340 "payload_bytes must alias buffer offset 4",
1341 );
1342 assert!(
1343 std::ptr::eq(meta_bytes.as_ptr(), base.as_ptr().wrapping_add(11)),
1344 "metadata_bytes must alias buffer offset 11",
1345 );
1346 assert!(std::ptr::eq(
1348 event.event_type_value().as_bytes().as_ptr(),
1349 base.as_ptr().wrapping_add(0),
1350 ));
1351 assert!(std::ptr::eq(
1352 event.payload_value().as_slice().as_ptr(),
1353 base.as_ptr().wrapping_add(4),
1354 ));
1355 }
1356
1357 fn boundary_u64() -> impl Strategy<Value = u64> {
1360 prop_oneof![
1361 Just(1u64),
1362 Just(2u64),
1363 Just(u64::MAX - 1),
1364 Just(u64::MAX),
1365 1u64..=u64::MAX,
1366 ]
1367 }
1368
1369 fn boundary_u32_nonzero() -> impl Strategy<Value = u32> {
1370 prop_oneof![
1371 Just(1u32),
1372 Just(2u32),
1373 Just(u32::MAX - 1),
1374 Just(u32::MAX),
1375 1u32..=u32::MAX,
1376 ]
1377 }
1378
1379 fn event_type_bytes_strategy() -> impl Strategy<Value = Vec<u8>> {
1380 prop_oneof![
1383 Just(0usize),
1384 Just(1usize),
1385 Just(255usize),
1386 Just(MAX_EVENT_TYPE_LEN),
1387 ]
1388 .prop_flat_map(|n| proptest::collection::vec(b'A'..=b'Z', n))
1389 }
1390
1391 fn payload_bytes_strategy() -> impl Strategy<Value = Vec<u8>> {
1392 prop_oneof![Just(0usize), Just(1usize), Just(1024usize)]
1393 .prop_flat_map(|n| proptest::collection::vec(any::<u8>(), n))
1394 }
1395
1396 fn metadata_bytes_strategy() -> impl Strategy<Value = Option<Vec<u8>>> {
1397 prop_oneof![
1398 Just(None),
1399 Just(Some(vec![0u8])),
1400 (1usize..=2048)
1401 .prop_flat_map(|n| proptest::collection::vec(any::<u8>(), n))
1402 .prop_map(Some),
1403 ]
1404 }
1405
1406 proptest! {
1407 #[test]
1408 fn persisted_envelope_roundtrips_every_accessor(
1409 event_type in event_type_bytes_strategy(),
1410 payload in payload_bytes_strategy(),
1411 metadata in metadata_bytes_strategy(),
1412 version_raw in boundary_u64(),
1413 schema_raw in boundary_u32_nonzero(),
1414 ) {
1415 let version = v(version_raw);
1416 let schema = sv(schema_raw);
1417 let event = build(
1418 version,
1419 schema,
1420 &event_type,
1421 &payload,
1422 metadata.as_deref(),
1423 );
1424
1425 prop_assert_eq!(event.version(), version);
1427 prop_assert_eq!(event.version().as_u64(), version_raw);
1428 prop_assert_eq!(event.schema_version(), schema_raw);
1429 prop_assert_eq!(event.schema_version_value(), schema);
1430
1431 let et_bytes = event.event_type_bytes();
1434 let et_value = event.event_type_value();
1435 prop_assert_eq!(event.event_type().as_bytes(), event_type.as_slice());
1436 prop_assert_eq!(et_bytes.as_ref(), event_type.as_slice());
1437 prop_assert_eq!(et_value.as_bytes(), event_type.as_slice());
1438
1439 let pl_bytes = event.payload_bytes();
1441 let pl_value = event.payload_value();
1442 prop_assert_eq!(event.payload(), payload.as_slice());
1443 prop_assert_eq!(pl_bytes.as_ref(), payload.as_slice());
1444 prop_assert_eq!(pl_value.as_slice(), payload.as_slice());
1445
1446 if let Some(ref m) = metadata {
1448 prop_assert_eq!(event.metadata(), Some(m.as_slice()));
1449 let meta_bytes = event.metadata_bytes().expect("some");
1450 prop_assert_eq!(meta_bytes.as_ref(), m.as_slice());
1451 let meta_value = event.metadata_value().expect("some");
1452 prop_assert_eq!(meta_value.as_slice(), m.as_slice());
1453 } else {
1454 prop_assert!(event.metadata().is_none());
1455 prop_assert!(event.metadata_bytes().is_none());
1456 prop_assert!(event.metadata_value().is_none());
1457 }
1458 }
1459 }
1460
1461 fn arbitrary_range() -> impl Strategy<Value = Range<u32>> {
1471 (0u32..=260, 0u32..=260).prop_map(|(start, end)| start..end)
1472 }
1473
1474 fn arbitrary_optional_range() -> impl Strategy<Value = Option<Range<u32>>> {
1475 prop_oneof![Just(None), arbitrary_range().prop_map(Some)]
1476 }
1477
1478 proptest! {
1479 #[test]
1480 fn try_new_never_panics_and_accepted_events_have_sound_accessors(
1481 value_bytes in proptest::collection::vec(any::<u8>(), 0..256),
1482 event_type_range in arbitrary_range(),
1483 payload_range in arbitrary_range(),
1484 metadata_range in arbitrary_optional_range(),
1485 version_raw in boundary_u64(),
1486 schema_raw in boundary_u32_nonzero(),
1487 ) {
1488 let result = PersistedEnvelope::try_new(
1489 v(version_raw),
1490 Bytes::from(value_bytes),
1491 sv(schema_raw),
1492 event_type_range,
1493 payload_range,
1494 metadata_range,
1495 );
1496
1497 match result {
1498 Ok(event) => {
1499 let et = event.event_type();
1504 prop_assert!(std::str::from_utf8(et.as_bytes()).is_ok());
1505 let _ = event.payload();
1506 let _ = event.metadata();
1507 let _ = event.event_type_value();
1508 let _ = event.payload_value();
1509 let _ = event.metadata_value();
1510 let _ = event.event_type_bytes();
1511 let _ = event.payload_bytes();
1512 let _ = event.metadata_bytes();
1513 }
1514 Err(err) => {
1515 match err {
1518 EnvelopeError::RangeOutOfBounds { .. }
1519 | EnvelopeError::InvalidUtf8 { .. }
1520 | EnvelopeError::EventTypeRangeTooLong { .. }
1521 | EnvelopeError::MetadataRangeTooLong { .. }
1522 | EnvelopeError::MetadataRangeEmpty
1523 | EnvelopeError::Value(_) => {}
1524 }
1525 }
1526 }
1527 }
1528 }
1529}