Skip to main content

mnesis_store/
envelope.rs

1use core::iter::{Chain, Once, once};
2use core::num::NonZeroUsize;
3use core::ops::Range;
4use core::slice::Iter;
5
6use bytes::Bytes;
7use mnesis::{DomainEvent, Version};
8use thiserror::Error;
9
10use crate::value::{
11    EventType, MAX_EVENT_TYPE_LEN, MAX_METADATA_LEN, Metadata, Payload, SchemaVersion, ValueError,
12};
13
14/// Cast `u32` index to `usize` for slice indexing.
15///
16/// `u32 as usize` is lossless on all platforms Mnesis supports (32-bit+).
17/// 16-bit platforms are not supported; this is a deliberate architecture constraint.
18#[allow(
19    clippy::as_conversions,
20    reason = "u32→usize is lossless on all Mnesis target platforms (32-bit+)"
21)]
22#[inline]
23const fn idx(n: u32) -> usize {
24    n as usize
25}
26
27// =============================================================================
28// Errors
29// =============================================================================
30
31/// Errors from envelope construction.
32#[derive(Debug, Error)]
33#[non_exhaustive]
34pub enum EnvelopeError {
35    #[error("range {start}..{end} exceeds buffer length {len}")]
36    RangeOutOfBounds { start: u32, end: u32, len: usize },
37
38    #[error("invalid UTF-8 in event_type at bytes {start}..{end}")]
39    InvalidUtf8 {
40        start: u32,
41        end: u32,
42        #[source]
43        source: core::str::Utf8Error,
44    },
45
46    /// The `event_type` range length exceeds the wire-format cap.
47    ///
48    /// Structurally rules out the only path that could otherwise hand
49    /// `EventType::from_validated_bytes` an oversize slice on the read
50    /// side, so the fast-path accessor is sound by construction.
51    #[error("event_type range length {actual} exceeds maximum {max}")]
52    EventTypeRangeTooLong { actual: u32, max: usize },
53
54    /// The metadata range length exceeds the wire-format cap.
55    ///
56    /// Mirrors `EventTypeRangeTooLong` for metadata.
57    #[error("metadata range length {actual} exceeds maximum {max}")]
58    MetadataRangeTooLong { actual: u32, max: usize },
59
60    /// `Some(range)` was passed where `range` is empty. The wire format
61    /// reserves the absent sentinel (`meta_len == u32::MAX`) for "no
62    /// metadata"; an empty range collides with the `Bytes::slice(empty)`
63    /// `STATIC_VTABLE` orphan footgun and the value newtype invariant
64    /// `!Metadata::is_empty()`.
65    #[error("metadata range is empty; use None to represent absent metadata")]
66    MetadataRangeEmpty,
67
68    #[error(transparent)]
69    Value(#[from] ValueError),
70}
71
72/// Errors from [`PersistedEnvelope::for_decode`].
73///
74/// Combines the failure modes of the underlying value-newtype
75/// construction, the wire encode, and the envelope `try_new` — `?`
76/// promotes any of the three.
77#[derive(Debug, Error)]
78#[non_exhaustive]
79pub enum ForDecodeError {
80    #[error(transparent)]
81    Value(#[from] ValueError),
82    #[error(transparent)]
83    Wire(#[from] crate::wire::WireError),
84    #[error(transparent)]
85    Envelope(#[from] EnvelopeError),
86}
87
88// =============================================================================
89// PendingEnvelope — write path, fully owned (Bytes), no lifetime, no generic
90// =============================================================================
91
92/// Event envelope for the write path.
93///
94/// Fields hold validated value newtypes — `EventType`, `Payload`, `Metadata`,
95/// `SchemaVersion` — so downstream wire encoding can skip re-validation.
96///
97/// Construction is via the typestate builder rooted at [`pending_envelope`].
98#[derive(Debug, Clone)]
99pub struct PendingEnvelope {
100    version: Version,
101    event_type: EventType,
102    schema_version: SchemaVersion,
103    payload: Payload,
104    metadata: Option<Metadata>,
105}
106
107impl PendingEnvelope {
108    #[must_use]
109    pub const fn version(&self) -> Version {
110        self.version
111    }
112
113    /// Borrowed event type as `&str`.
114    #[must_use]
115    pub fn event_type(&self) -> &str {
116        self.event_type.as_str()
117    }
118
119    /// Owned event type — one Arc share.
120    #[must_use]
121    pub fn event_type_value(&self) -> EventType {
122        self.event_type.clone()
123    }
124
125    #[must_use]
126    pub fn payload(&self) -> &[u8] {
127        self.payload.as_slice()
128    }
129
130    /// Owned payload — one Arc share.
131    #[must_use]
132    pub fn payload_bytes(&self) -> Bytes {
133        self.payload.clone().into_bytes()
134    }
135
136    /// Owned payload value newtype — one Arc share.
137    #[must_use]
138    pub fn payload_value(&self) -> Payload {
139        self.payload.clone()
140    }
141
142    #[must_use]
143    pub fn metadata(&self) -> Option<&[u8]> {
144        self.metadata.as_ref().map(Metadata::as_slice)
145    }
146
147    /// Owned metadata — one Arc share per `Some`.
148    #[must_use]
149    pub fn metadata_bytes(&self) -> Option<Bytes> {
150        self.metadata.as_ref().map(|m| m.clone().into_bytes())
151    }
152
153    /// Owned metadata value newtype — one Arc share per `Some`.
154    #[must_use]
155    pub fn metadata_value(&self) -> Option<Metadata> {
156        self.metadata.clone()
157    }
158
159    /// The raw u32 view (always > 0, by the `SchemaVersion` invariant).
160    #[must_use]
161    pub const fn schema_version(&self) -> u32 {
162        self.schema_version.get()
163    }
164
165    /// The typed schema version.
166    #[must_use]
167    pub const fn schema_version_value(&self) -> SchemaVersion {
168        self.schema_version
169    }
170
171    /// Rebuild a write-path envelope from a read-path one.
172    ///
173    /// Reuses the [`PersistedEnvelope`]'s already-validated value newtypes
174    /// (one Arc share each — zero copy, no re-validation). Infallible, because
175    /// every field of a `PersistedEnvelope` was validated at its own
176    /// construction. Used by the importer to re-append exported events; the
177    /// target store assigns the `$all` position fresh on append (it is not an
178    /// envelope field).
179    #[cfg(feature = "import")]
180    #[must_use]
181    pub(crate) fn from_persisted(persisted: &PersistedEnvelope) -> Self {
182        Self {
183            version: persisted.version(),
184            event_type: persisted.event_type_value(),
185            schema_version: persisted.schema_version_value(),
186            payload: persisted.payload_value(),
187            metadata: persisted.metadata_value(),
188        }
189    }
190}
191
192// =============================================================================
193// PendingBatch — the non-empty run handed to `RawEventStore::append`
194// =============================================================================
195
196/// The concrete iterator over a [`PendingBatch`] — the head chained to the tail.
197pub type PendingBatchIter<'a> = Chain<Once<&'a PendingEnvelope>, Iter<'a, PendingEnvelope>>;
198
199/// A **non-empty** borrowed run of [`PendingEnvelope`]s — what
200/// [`RawEventStore::append`](crate::RawEventStore::append) takes.
201///
202/// # Why the append input is non-empty by construction
203///
204/// `append` returns the [`AllPosition`](crate::AllPosition) its events landed
205/// at, and an append of nothing has no position to return. Keeping the empty
206/// case legal would force the return type to be `Option<AllPosition>`, so every
207/// caller would handle a `None` it can already prove impossible. Making the
208/// input non-empty removes the case instead of propagating it — the same
209/// discipline [`Repository::save`](crate::Repository::save) already applies by
210/// taking `&Events<E, N>` rather than a slice.
211///
212/// This deliberately removes one prior behaviour: `append(id, expected, &[])`
213/// used to validate `expected_version` and write nothing, i.e. a
214/// version-assertion probe. It had no production caller. If it is ever wanted
215/// it returns as its own method, which is an additive change.
216///
217/// # Shape
218///
219/// Head-plus-tail (`first`, `rest`) rather than a non-empty *slice*, matching
220/// [`Events`](mnesis::Events) and `ProjectedIntents`. Because the two parts
221/// need not be contiguous, a caller holding a statically non-empty collection
222/// builds a batch with no runtime check and no unprovable `unwrap` — see
223/// [`from_parts`](Self::from_parts). Callers whose length is only known at
224/// runtime use [`new`](Self::new) and handle the `None` honestly.
225#[derive(Debug, Clone, Copy)]
226pub struct PendingBatch<'a> {
227    first: &'a PendingEnvelope,
228    rest: &'a [PendingEnvelope],
229}
230
231impl<'a> PendingBatch<'a> {
232    /// A batch of exactly one envelope. Infallible — one is not zero.
233    #[must_use]
234    pub const fn of(only: &'a PendingEnvelope) -> Self {
235        Self {
236            first: only,
237            rest: &[],
238        }
239    }
240
241    /// A batch from an already-split head and tail. Infallible: the head's
242    /// existence is the non-emptiness.
243    ///
244    /// This is the constructor for callers that already know their input is
245    /// non-empty — the repository path, whose events come from a non-empty
246    /// [`Events`](mnesis::Events).
247    #[must_use]
248    pub const fn from_parts(first: &'a PendingEnvelope, rest: &'a [PendingEnvelope]) -> Self {
249        Self { first, rest }
250    }
251
252    /// A batch over a slice whose length is only known at runtime.
253    ///
254    /// `None` iff `envelopes` is empty — the one honest place the empty case is
255    /// answered, instead of every downstream caller answering it again.
256    #[must_use]
257    pub fn new(envelopes: &'a [PendingEnvelope]) -> Option<Self> {
258        envelopes
259            .split_first()
260            .map(|(first, rest)| Self { first, rest })
261    }
262
263    /// The first envelope — the lowest [`Version`] in the run.
264    #[must_use]
265    pub const fn first(&self) -> &'a PendingEnvelope {
266        self.first
267    }
268
269    /// The last envelope — the highest [`Version`] in the run, and the one whose
270    /// assigned position `append` returns.
271    #[must_use]
272    pub fn last(&self) -> &'a PendingEnvelope {
273        self.rest.last().unwrap_or(self.first)
274    }
275
276    /// How many envelopes the batch carries — always at least one, carried in
277    /// the type so adapters size buffers without re-deriving non-emptiness.
278    #[must_use]
279    pub const fn len(&self) -> NonZeroUsize {
280        // `1 + rest.len()`, which cannot overflow: `rest` is a slice of a
281        // non-zero-sized type, so its length is at most `isize::MAX`. The
282        // saturating form is the total operation on `NonZeroUsize` and the
283        // saturating branch is unreachable by that proof — not a silently
284        // capped size computation (rule 2).
285        NonZeroUsize::MIN.saturating_add(self.rest.len())
286    }
287
288    /// Iterate the run in version order, head first.
289    pub fn iter(&self) -> PendingBatchIter<'a> {
290        once(self.first).chain(self.rest)
291    }
292}
293
294impl<'a> IntoIterator for PendingBatch<'a> {
295    type Item = &'a PendingEnvelope;
296    type IntoIter = PendingBatchIter<'a>;
297
298    fn into_iter(self) -> Self::IntoIter {
299        self.iter()
300    }
301}
302
303impl<'a> IntoIterator for &PendingBatch<'a> {
304    type Item = &'a PendingEnvelope;
305    type IntoIter = PendingBatchIter<'a>;
306
307    fn into_iter(self) -> Self::IntoIter {
308        self.iter()
309    }
310}
311
312// =============================================================================
313// Typestate builder — compile-time enforced construction
314// =============================================================================
315
316/// Step 1: has `version`, needs `event_type`.
317#[derive(Debug)]
318pub struct WithVersion {
319    version: Version,
320}
321
322/// Step 2: has `version` + `event_type`, needs payload.
323#[derive(Debug)]
324pub struct WithEventType {
325    version: Version,
326    event_type: EventType,
327}
328
329/// Step 3: has all core fields; optional `schema_version`/`metadata`; finalize via `build`.
330#[derive(Debug)]
331pub struct WithPayload {
332    version: Version,
333    event_type: EventType,
334    payload: Bytes,
335    schema_version: SchemaVersion,
336    metadata: Option<Bytes>,
337}
338
339impl WithVersion {
340    /// Set the event type from a `&'static str` literal. Infallible.
341    #[must_use]
342    pub fn event_type(self, event_type: &'static str) -> WithEventType {
343        WithEventType {
344            version: self.version,
345            event_type: EventType::from_static_str(event_type),
346        }
347    }
348
349    /// Derive the event type from a [`DomainEvent`] — no restating `name()`.
350    #[must_use]
351    pub fn event<E: DomainEvent + ?Sized>(self, event: &E) -> WithEventType {
352        self.event_type(event.name())
353    }
354
355    /// Set the event type from arbitrary bytes; validates UTF-8 and size cap.
356    ///
357    /// # Errors
358    ///
359    /// Returns [`EnvelopeError::Value`] if the bytes are invalid UTF-8 or
360    /// exceed [`MAX_EVENT_TYPE_LEN`](crate::value::MAX_EVENT_TYPE_LEN).
361    pub fn event_type_bytes(self, bytes: Bytes) -> Result<WithEventType, EnvelopeError> {
362        let event_type = EventType::from_bytes(bytes)?;
363        Ok(WithEventType {
364            version: self.version,
365            event_type,
366        })
367    }
368}
369
370impl WithEventType {
371    /// Stash the payload bytes from any `Into<Bytes>` source. Infallible —
372    /// validated in [`WithPayload::build`].
373    #[must_use]
374    pub fn payload(self, payload: impl Into<Bytes>) -> WithPayload {
375        WithPayload {
376            version: self.version,
377            event_type: self.event_type,
378            payload: payload.into(),
379            schema_version: SchemaVersion::INITIAL,
380            metadata: None,
381        }
382    }
383}
384
385impl WithPayload {
386    /// Override the schema version (default: [`SchemaVersion::INITIAL`]).
387    #[must_use]
388    pub const fn schema_version(mut self, schema_version: SchemaVersion) -> Self {
389        self.schema_version = schema_version;
390        self
391    }
392
393    /// Stash metadata bytes from any `Into<Bytes>` source. Infallible —
394    /// validated in [`build`](Self::build).
395    #[must_use]
396    pub fn metadata(mut self, metadata: impl Into<Bytes>) -> Self {
397        self.metadata = Some(metadata.into());
398        self
399    }
400
401    /// Validate and finalize — the one fallible step.
402    ///
403    /// # Errors
404    ///
405    /// Returns [`EnvelopeError::Value`] if the payload exceeds
406    /// [`MAX_PAYLOAD_LEN`](crate::value::MAX_PAYLOAD_LEN), or the metadata is
407    /// empty or exceeds [`MAX_METADATA_LEN`](crate::value::MAX_METADATA_LEN).
408    pub fn build(self) -> Result<PendingEnvelope, EnvelopeError> {
409        let payload = Payload::from_bytes(self.payload)?;
410        let metadata = self.metadata.map(Metadata::from_bytes).transpose()?;
411        Ok(PendingEnvelope {
412            version: self.version,
413            event_type: self.event_type,
414            schema_version: self.schema_version,
415            payload,
416            metadata,
417        })
418    }
419}
420
421/// Start building a `PendingEnvelope`.
422///
423/// ```ignore
424/// pending_envelope(version)
425///     .event_type("UserCreated")
426///     .payload(bytes)
427///     .build()?
428/// // or, deriving the event type from a `DomainEvent`:
429/// pending_envelope(version)
430///     .event(&my_event)
431///     .payload(bytes)
432///     .metadata(meta_bytes)
433///     .build()?
434/// ```
435#[must_use]
436pub const fn pending_envelope(version: Version) -> WithVersion {
437    WithVersion { version }
438}
439
440// =============================================================================
441// PersistedEnvelope — read path, owned via Bytes + Range<u32> offsets
442// =============================================================================
443
444/// Event envelope for the read path.
445///
446/// Holds the whole row value as a single `Bytes` plus `Range<u32>` offsets
447/// for `event_type`, `payload`, and (optional) `metadata`. All views share
448/// the one Arc; accessors return `&[u8]`/`&str` cheaply or `Bytes` via
449/// `value.slice(range)` for owned views.
450///
451/// Construction validates ranges against the buffer, UTF-8 of `event_type`,
452/// and the structural per-field caps that the value newtypes own (so the
453/// fast-path `*_value()` accessors are sound without re-validation).
454#[derive(Debug, Clone)]
455pub struct PersistedEnvelope {
456    version: Version,
457    schema_version: SchemaVersion,
458    value: Bytes,
459    event_type_range: Range<u32>,
460    payload_range: Range<u32>,
461    metadata_range: Option<Range<u32>>,
462}
463
464impl PersistedEnvelope {
465    /// Construct from decoded row data, validating ranges and UTF-8.
466    ///
467    /// # Errors
468    ///
469    /// - [`EnvelopeError::RangeOutOfBounds`] if any range's `end` exceeds `value.len()`.
470    /// - [`EnvelopeError::InvalidUtf8`] if `event_type` bytes are not valid UTF-8.
471    /// - [`EnvelopeError::EventTypeRangeTooLong`] if the event-type range
472    ///   length exceeds [`MAX_EVENT_TYPE_LEN`].
473    /// - [`EnvelopeError::MetadataRangeTooLong`] if the metadata range
474    ///   length exceeds [`MAX_METADATA_LEN`].
475    /// - [`EnvelopeError::MetadataRangeEmpty`] if `Some(range)` is passed
476    ///   with an empty range.
477    #[allow(
478        clippy::too_many_arguments,
479        reason = "all 6 fields are required to construct a validated PersistedEnvelope; \
480                  a builder would add indirection with no type-safety benefit here"
481    )]
482    pub fn try_new(
483        version: Version,
484        value: Bytes,
485        schema_version: SchemaVersion,
486        event_type_range: Range<u32>,
487        payload_range: Range<u32>,
488        metadata_range: Option<Range<u32>>,
489    ) -> Result<Self, EnvelopeError> {
490        let len = value.len();
491        check_range(&event_type_range, len)?;
492        check_range(&payload_range, len)?;
493        if let Some(ref m) = metadata_range {
494            check_range(m, len)?;
495        }
496
497        // Structural per-field caps — owned upstream by the value
498        // newtypes; mirroring them here makes `event_type_value` /
499        // `metadata_value` sound without re-validation.
500        let et_range_len = event_type_range.end - event_type_range.start;
501        if idx(et_range_len) > MAX_EVENT_TYPE_LEN {
502            return Err(EnvelopeError::EventTypeRangeTooLong {
503                actual: et_range_len,
504                max: MAX_EVENT_TYPE_LEN,
505            });
506        }
507        if let Some(ref m) = metadata_range {
508            let meta_range_len = m.end - m.start;
509            if meta_range_len == 0 {
510                return Err(EnvelopeError::MetadataRangeEmpty);
511            }
512            if idx(meta_range_len) > MAX_METADATA_LEN {
513                return Err(EnvelopeError::MetadataRangeTooLong {
514                    actual: meta_range_len,
515                    max: MAX_METADATA_LEN,
516                });
517            }
518        }
519
520        // UTF-8 validation of event_type once at construction.
521        let et_start = idx(event_type_range.start);
522        let et_end = idx(event_type_range.end);
523        core::str::from_utf8(&value[et_start..et_end]).map_err(|e| EnvelopeError::InvalidUtf8 {
524            start: event_type_range.start,
525            end: event_type_range.end,
526            source: e,
527        })?;
528        Ok(Self {
529            version,
530            schema_version,
531            value,
532            event_type_range,
533            payload_range,
534            metadata_range,
535        })
536    }
537
538    #[must_use]
539    pub const fn version(&self) -> Version {
540        self.version
541    }
542
543    /// The raw `u32` view of `schema_version` (always > 0 by the
544    /// [`SchemaVersion`] invariant).
545    #[must_use]
546    pub const fn schema_version(&self) -> u32 {
547        self.schema_version.get()
548    }
549
550    /// The typed [`SchemaVersion`].
551    #[must_use]
552    pub const fn schema_version_value(&self) -> SchemaVersion {
553        self.schema_version
554    }
555
556    /// UTF-8 validated at construction; cheap accessor.
557    #[must_use]
558    pub fn event_type(&self) -> &str {
559        let start = idx(self.event_type_range.start);
560        let end = idx(self.event_type_range.end);
561        // SAFETY: UTF-8 validated once in `try_new`; range bounds checked
562        // there too. The backing `Bytes` is immutable post-construction
563        // (no setter, fields private).
564        #[allow(
565            unsafe_code,
566            reason = "UTF-8 invariant established at construction; ranges validated"
567        )]
568        unsafe {
569            core::str::from_utf8_unchecked(&self.value[start..end])
570        }
571    }
572
573    /// Owned `Bytes` view of `event_type` — one atomic refcount inc.
574    #[must_use]
575    pub fn event_type_bytes(&self) -> Bytes {
576        self.slice_range(&self.event_type_range)
577    }
578
579    #[must_use]
580    pub fn payload(&self) -> &[u8] {
581        let start = idx(self.payload_range.start);
582        let end = idx(self.payload_range.end);
583        &self.value[start..end]
584    }
585
586    /// Owned `Bytes` view of `payload` — one atomic refcount inc.
587    #[must_use]
588    pub fn payload_bytes(&self) -> Bytes {
589        self.slice_range(&self.payload_range)
590    }
591
592    #[must_use]
593    pub fn metadata(&self) -> Option<&[u8]> {
594        self.metadata_range.as_ref().map(|r| {
595            let start = idx(r.start);
596            let end = idx(r.end);
597            &self.value[start..end]
598        })
599    }
600
601    /// Owned `Bytes` view of `metadata` — one atomic refcount inc per `Some`.
602    ///
603    /// `try_new` rejects `Some(empty)`, so the `Bytes::slice(empty)`
604    /// `STATIC_VTABLE` orphan footgun is structurally unreachable here.
605    #[must_use]
606    pub fn metadata_bytes(&self) -> Option<Bytes> {
607        self.metadata_range.as_ref().map(|r| self.slice_range(r))
608    }
609
610    /// Validated event type — one Arc share over the underlying buffer.
611    #[must_use]
612    pub fn event_type_value(&self) -> EventType {
613        // SAFETY: `from_validated_bytes` requires (1) valid UTF-8 and
614        // (2) `bytes.len() <= MAX_EVENT_TYPE_LEN`. Both invariants are
615        // established by `try_new`: UTF-8 via `core::str::from_utf8`, and
616        // the cap via the `EventTypeRangeTooLong` check on the range
617        // length.
618        #[allow(
619            unsafe_code,
620            reason = "UTF-8 and length cap both established by try_new"
621        )]
622        unsafe {
623            EventType::from_validated_bytes(self.event_type_bytes())
624        }
625    }
626
627    /// Validated payload — one Arc share over the underlying buffer.
628    #[must_use]
629    pub fn payload_value(&self) -> Payload {
630        // `from_validated_bytes` is a safe fast-path (no UB contract): the
631        // length invariant is already established — `try_new` validated
632        // `payload_range` against `value.len()`, and the `Range<u32>` bounds
633        // the slice length by `u32::MAX = MAX_PAYLOAD_LEN`.
634        Payload::from_validated_bytes(self.payload_bytes())
635    }
636
637    /// Validated metadata — one Arc share per `Some` over the underlying
638    /// buffer.
639    ///
640    /// Returns `None` when the wire-level absent sentinel was present.
641    #[must_use]
642    pub fn metadata_value(&self) -> Option<Metadata> {
643        self.metadata_bytes().map(|b| {
644            // Safe fast-path (no UB contract): both invariants are already
645            // established by `try_new` — the `MetadataRangeEmpty` check rejects
646            // an empty `Some(range)`, and `MetadataRangeTooLong` enforces the
647            // length cap.
648            Metadata::from_validated_bytes(b)
649        })
650    }
651
652    /// The schema version widened to the kernel's [`Version`] for upcaster APIs.
653    ///
654    /// Total conversion — [`SchemaVersion`] is structurally nonzero.
655    #[must_use]
656    pub fn schema_version_as_version(&self) -> Version {
657        Version::from(self.schema_version)
658    }
659
660    /// Wrap raw bytes in a synthetic envelope suitable for [`Decode`].
661    ///
662    /// Builds a fresh wire-format frame via [`crate::wire::encode_frame`] so the
663    /// payload pointer lands on a 16-byte boundary. Use this when calling a
664    /// [`Decode`](crate::codec::Decode) impl outside the cursor's normal frame
665    /// buffer — snapshot decoding, upcaster post-transform decoding, codec
666    /// round-trip tests.
667    ///
668    /// Reports `Version::INITIAL` and `schema_version = SchemaVersion::INITIAL`.
669    /// Most codecs ignore those fields; when they don't (or you're bridging an
670    /// upcast back to a decode and need to preserve the original envelope's
671    /// version), construct the envelope manually via [`try_new`](Self::try_new).
672    ///
673    /// # Errors
674    ///
675    /// Returns [`ForDecodeError`] if the value newtypes reject the inputs
676    /// (oversize `event_type`/`payload`), the wire encode fails
677    /// (`FrameLengthOverflow`), or the envelope `try_new` fails (range
678    /// invariants).
679    pub fn for_decode(event_type: &str, payload: &[u8]) -> Result<Self, ForDecodeError> {
680        let et = EventType::from_bytes(Bytes::copy_from_slice(event_type.as_bytes()))?;
681        let pl = Payload::from_bytes(Bytes::copy_from_slice(payload))?;
682        let sv = SchemaVersion::INITIAL;
683        let frame = crate::wire::encode_frame(sv, &et, &pl, None)?;
684        Ok(Self::try_new(
685            Version::INITIAL,
686            frame.value,
687            sv,
688            frame.offsets.event_type,
689            frame.offsets.payload,
690            None,
691        )?)
692    }
693
694    fn slice_range(&self, range: &Range<u32>) -> Bytes {
695        self.value.slice(idx(range.start)..idx(range.end))
696    }
697}
698
699const fn check_range(range: &Range<u32>, len: usize) -> Result<(), EnvelopeError> {
700    if idx(range.end) > len || range.start > range.end {
701        return Err(EnvelopeError::RangeOutOfBounds {
702            start: range.start,
703            end: range.end,
704            len,
705        });
706    }
707    Ok(())
708}
709
710#[cfg(test)]
711#[allow(
712    clippy::unwrap_used,
713    clippy::expect_used,
714    clippy::panic,
715    reason = "test code asserts exact values"
716)]
717mod tests {
718    use super::*;
719    use crate::value::MAX_EVENT_TYPE_LEN;
720    use bytes::Bytes;
721    use mnesis::Version;
722
723    /// A minimal `DomainEvent` for exercising `.event(&e)`.
724    #[derive(Debug)]
725    struct TestEvent;
726
727    impl mnesis::Message for TestEvent {}
728
729    impl DomainEvent for TestEvent {
730        fn name(&self) -> &'static str {
731            "TestEvent"
732        }
733    }
734
735    #[test]
736    fn pending_envelope_builds_with_metadata() {
737        let env = pending_envelope(Version::INITIAL)
738            .event_type("UserCreated")
739            .payload(Bytes::from_static(b"payload-bytes"))
740            .metadata(Bytes::from_static(b"meta-bytes"))
741            .build()
742            .expect("valid envelope");
743
744        assert_eq!(env.event_type(), "UserCreated");
745        assert_eq!(env.payload(), b"payload-bytes");
746        assert_eq!(env.metadata(), Some(b"meta-bytes".as_slice()));
747        assert_eq!(env.schema_version(), 1);
748    }
749
750    #[test]
751    fn pending_envelope_builds_without_metadata() {
752        let env = pending_envelope(Version::INITIAL)
753            .event_type("X")
754            .payload(Bytes::from_static(b"p"))
755            .build()
756            .expect("valid envelope");
757
758        assert_eq!(env.metadata(), None);
759    }
760
761    #[test]
762    fn pending_envelope_typed_accessors_roundtrip() {
763        let env = pending_envelope(Version::INITIAL)
764            .event_type("UserCreated")
765            .payload(Bytes::from_static(b"payload-bytes"))
766            .metadata(Bytes::from_static(b"meta-bytes"))
767            .build()
768            .expect("valid envelope");
769
770        assert_eq!(env.event_type_value().as_str(), "UserCreated");
771        assert_eq!(env.payload_value().as_slice(), b"payload-bytes");
772        assert_eq!(
773            env.metadata_value().map(|m| m.as_slice().to_vec()),
774            Some(b"meta-bytes".to_vec())
775        );
776        assert_eq!(env.schema_version_value().get(), 1);
777    }
778
779    #[test]
780    fn pending_envelope_rejects_oversize_event_type() {
781        let oversized = "x".repeat(MAX_EVENT_TYPE_LEN + 1);
782        let err = pending_envelope(Version::INITIAL)
783            .event_type_bytes(Bytes::from(oversized))
784            .expect_err("oversized must be rejected");
785        assert!(matches!(err, EnvelopeError::Value(_)));
786    }
787
788    #[test]
789    fn event_derives_same_event_type_as_event_type_name() {
790        let via_event = pending_envelope(Version::INITIAL)
791            .event(&TestEvent)
792            .payload(Bytes::from_static(b"p"))
793            .build()
794            .expect("valid envelope");
795        let via_name = pending_envelope(Version::INITIAL)
796            .event_type(TestEvent.name())
797            .payload(Bytes::from_static(b"p"))
798            .build()
799            .expect("valid envelope");
800
801        assert_eq!(via_event.event_type(), via_name.event_type());
802        assert_eq!(via_event.event_type(), "TestEvent");
803    }
804
805    #[test]
806    fn build_rejects_oversize_payload() {
807        let oversized = vec![0u8; crate::value::MAX_PAYLOAD_LEN + 1];
808        let err = pending_envelope(Version::INITIAL)
809            .event_type("X")
810            .payload(Bytes::from(oversized))
811            .build()
812            .expect_err("oversized payload must be rejected");
813        assert!(matches!(err, EnvelopeError::Value(_)));
814    }
815
816    #[test]
817    fn build_rejects_empty_metadata() {
818        let err = pending_envelope(Version::INITIAL)
819            .event_type("X")
820            .payload(Bytes::from_static(b"p"))
821            .metadata(Bytes::new())
822            .build()
823            .expect_err("empty metadata must be rejected");
824        assert!(matches!(err, EnvelopeError::Value(_)));
825    }
826
827    #[test]
828    fn persisted_envelope_accessors_return_views_into_value() {
829        let value = Bytes::from_static(b"TYPEpayloadmeta");
830        // ranges: event_type 0..4 ("TYPE"), payload 4..11 ("payload"), metadata 11..15 ("meta")
831        let env = PersistedEnvelope::try_new(
832            Version::INITIAL,
833            value,
834            SchemaVersion::INITIAL,
835            0..4,
836            4..11,
837            Some(11..15),
838        )
839        .expect("valid construction");
840
841        assert_eq!(env.event_type(), "TYPE");
842        assert_eq!(env.payload(), b"payload");
843        assert_eq!(env.metadata(), Some(b"meta".as_slice()));
844    }
845
846    #[test]
847    fn persisted_envelope_payload_bytes_shares_arc() {
848        let env = PersistedEnvelope::try_new(
849            Version::INITIAL,
850            Bytes::from_static(b"TYPEpayload"),
851            SchemaVersion::INITIAL,
852            0..4,
853            4..11,
854            None,
855        )
856        .unwrap();
857
858        let payload = env.payload_bytes();
859        assert_eq!(payload.as_ref(), b"payload");
860    }
861
862    #[test]
863    fn persisted_envelope_rejects_range_past_buffer() {
864        let value = Bytes::from_static(b"short");
865        let err = PersistedEnvelope::try_new(
866            Version::INITIAL,
867            value,
868            SchemaVersion::INITIAL,
869            0..4,
870            4..100,
871            None,
872        )
873        .expect_err("must reject out-of-bounds range");
874        assert!(matches!(err, EnvelopeError::RangeOutOfBounds { .. }));
875    }
876
877    #[test]
878    fn try_new_rejects_event_type_range_too_long() {
879        let len = MAX_EVENT_TYPE_LEN + 1;
880        let mut buf = vec![0u8; len];
881        // Fill with valid UTF-8 (all-zeros is valid UTF-8 ASCII).
882        buf[0] = b'A';
883        let value = Bytes::from(buf);
884        let too_long_end = u32::try_from(len).expect("len fits in u32 by construction");
885
886        let err = PersistedEnvelope::try_new(
887            Version::INITIAL,
888            value,
889            SchemaVersion::INITIAL,
890            0..too_long_end,
891            too_long_end..too_long_end,
892            None,
893        )
894        .expect_err("event_type range > MAX_EVENT_TYPE_LEN must be rejected");
895
896        let expected_actual = u32::try_from(len).expect("len fits in u32 by construction (above)");
897        assert!(matches!(
898            err,
899            EnvelopeError::EventTypeRangeTooLong { actual, max }
900                if actual == expected_actual && max == MAX_EVENT_TYPE_LEN
901        ));
902    }
903
904    #[test]
905    fn persisted_envelope_rejects_empty_metadata_range() {
906        let env = PersistedEnvelope::try_new(
907            Version::INITIAL,
908            Bytes::from_static(b"TYPEpayload"),
909            SchemaVersion::INITIAL,
910            0..4,
911            4..11,
912            Some(4..4),
913        );
914        assert!(matches!(env, Err(EnvelopeError::MetadataRangeEmpty)));
915    }
916
917    #[test]
918    fn persisted_envelope_rejects_invalid_utf8_in_event_type() {
919        let value = Bytes::from_static(&[0xFFu8, 0xFF, b'p', b'a', b'y']);
920        let err = PersistedEnvelope::try_new(
921            Version::INITIAL,
922            value,
923            SchemaVersion::INITIAL,
924            0..2,
925            2..5,
926            None,
927        )
928        .expect_err("must reject non-UTF-8 event_type");
929        assert!(matches!(err, EnvelopeError::InvalidUtf8 { .. }));
930    }
931
932    #[test]
933    fn persisted_envelope_event_type_value_returns_validated_value_newtype() {
934        let env = PersistedEnvelope::try_new(
935            Version::INITIAL,
936            Bytes::from_static(b"TYPEpayload"),
937            SchemaVersion::INITIAL,
938            0..4,
939            4..11,
940            None,
941        )
942        .expect("valid");
943
944        let et = env.event_type_value();
945        assert_eq!(et.as_str(), "TYPE");
946    }
947
948    #[test]
949    fn persisted_envelope_payload_value_returns_validated_value_newtype() {
950        let env = PersistedEnvelope::try_new(
951            Version::INITIAL,
952            Bytes::from_static(b"TYPEpayload"),
953            SchemaVersion::INITIAL,
954            0..4,
955            4..11,
956            None,
957        )
958        .expect("valid");
959
960        let p = env.payload_value();
961        assert_eq!(p.as_slice(), b"payload");
962    }
963
964    #[test]
965    fn persisted_envelope_metadata_value_returns_some_when_present() {
966        let env = PersistedEnvelope::try_new(
967            Version::INITIAL,
968            Bytes::from_static(b"TYPEpayloadMETA"),
969            SchemaVersion::INITIAL,
970            0..4,
971            4..11,
972            Some(11..15),
973        )
974        .expect("valid");
975
976        let m = env.metadata_value().expect("present");
977        assert_eq!(m.as_slice(), b"META");
978    }
979
980    #[test]
981    fn persisted_envelope_metadata_value_returns_none_when_absent() {
982        let env = PersistedEnvelope::try_new(
983            Version::INITIAL,
984            Bytes::from_static(b"TYPEpayload"),
985            SchemaVersion::INITIAL,
986            0..4,
987            4..11,
988            None,
989        )
990        .expect("valid");
991
992        assert!(env.metadata_value().is_none());
993    }
994
995    #[cfg(feature = "import")]
996    #[test]
997    fn from_persisted_preserves_fields_drops_global_seq_zero_copy() {
998        // A read-path envelope at version 7, global_seq 99, schema 3, with meta.
999        let value = Bytes::from_static(b"TYPEpayloadmeta");
1000        let persisted = PersistedEnvelope::try_new(
1001            Version::new(7).expect("nonzero"),
1002            value,
1003            crate::value::SchemaVersion::from_u32(3).expect("nonzero"),
1004            0..4,
1005            4..11,
1006            Some(11..15),
1007        )
1008        .expect("valid");
1009
1010        let pending = PendingEnvelope::from_persisted(&persisted);
1011
1012        // Every field carried verbatim (global_seq has no home on the write path).
1013        assert_eq!(pending.version(), persisted.version());
1014        assert_eq!(pending.event_type(), "TYPE");
1015        assert_eq!(pending.payload(), b"payload");
1016        assert_eq!(pending.metadata(), Some(b"meta".as_slice()));
1017        assert_eq!(pending.schema_version(), 3);
1018
1019        // Zero-copy: the rebuilt payload aliases the same backing allocation.
1020        assert!(
1021            std::ptr::eq(
1022                pending.payload_bytes().as_ptr(),
1023                persisted.payload().as_ptr()
1024            ),
1025            "from_persisted must reuse the Arc-shared payload, not deep-copy",
1026        );
1027    }
1028}
1029
1030// ============================================================================
1031// Exhaustive PersistedEnvelope suite — migrated from the former PortableEvent
1032// tests (issue #145). PortableEvent's validator was a literal copy of
1033// PersistedEnvelope's, so its range/UTF-8/cap/zero-copy/fuzz tests apply here
1034// unchanged. The stream_id-specific tests were dropped (PersistedEnvelope has
1035// global_seq instead); the differential oracle was dropped (it compared the
1036// two now-unified validators). These cover the 4 cross-cutting categories:
1037// sequence/protocol, lifecycle, defensive boundary, and Send/Sync.
1038// ============================================================================
1039#[cfg(test)]
1040#[allow(clippy::expect_used, reason = "test code")]
1041mod persisted_exhaustive_tests {
1042    use super::{EnvelopeError, PersistedEnvelope};
1043    use crate::value::{MAX_EVENT_TYPE_LEN, Metadata, SchemaVersion};
1044    use bytes::Bytes;
1045    use mnesis::Version;
1046    use proptest::prelude::*;
1047    use static_assertions::assert_impl_all;
1048    use std::ops::Range;
1049
1050    // ── helpers ─────────────────────────────────────────────────────────────
1051
1052    fn v(n: u64) -> Version {
1053        Version::new(n).expect("test version must be nonzero")
1054    }
1055
1056    fn sv(n: u32) -> SchemaVersion {
1057        SchemaVersion::from_u32(n).expect("test schema_version must be nonzero")
1058    }
1059
1060    /// Assemble a contiguous `[event_type][payload][metadata?]` buffer and the
1061    /// matching ranges — the exact shape a decoder hands `try_new`.
1062    fn assemble(
1063        event_type: &[u8],
1064        payload: &[u8],
1065        metadata: Option<&[u8]>,
1066    ) -> (Bytes, Range<u32>, Range<u32>, Option<Range<u32>>) {
1067        let mut buf = Vec::new();
1068        buf.extend_from_slice(event_type);
1069        buf.extend_from_slice(payload);
1070        if let Some(m) = metadata {
1071            buf.extend_from_slice(m);
1072        }
1073        let et_end = u32::try_from(event_type.len()).expect("event_type len fits u32");
1074        let pl_end = et_end + u32::try_from(payload.len()).expect("payload len fits u32");
1075        let meta_range = metadata.map(|m| {
1076            let end = pl_end + u32::try_from(m.len()).expect("metadata len fits u32");
1077            pl_end..end
1078        });
1079        (Bytes::from(buf), 0..et_end, et_end..pl_end, meta_range)
1080    }
1081
1082    /// Build a valid `PersistedEnvelope` from byte components.
1083    fn build(
1084        version: Version,
1085        schema: SchemaVersion,
1086        event_type: &[u8],
1087        payload: &[u8],
1088        metadata: Option<&[u8]>,
1089    ) -> PersistedEnvelope {
1090        let (value, et, pl, meta) = assemble(event_type, payload, metadata);
1091        PersistedEnvelope::try_new(version, value, schema, et, pl, meta)
1092            .expect("components form a valid PersistedEnvelope")
1093    }
1094
1095    // ── Category 4: linearizability / Send + Sync ───────────────────────────
1096    // PersistedEnvelope flows through `futures::Stream` items across `.await`
1097    // points and threads; Send + Sync is a structural contract.
1098
1099    assert_impl_all!(PersistedEnvelope: Send, Sync, Clone, std::fmt::Debug);
1100
1101    #[test]
1102    fn persisted_envelope_clones_across_thread_boundary() {
1103        let event = build(
1104            v(3),
1105            sv(2),
1106            b"MoneyDeposited",
1107            b"payload-bytes",
1108            Some(b"meta"),
1109        );
1110        let moved = event.clone();
1111        // `std::thread::scope` (not the banned free `std::thread::spawn`) moves
1112        // the clone to another thread and joins it — proving `Send` in practice.
1113        let (ver, schema, etype, payload, meta) = std::thread::scope(|scope| {
1114            scope
1115                .spawn(move || {
1116                    (
1117                        moved.version(),
1118                        moved.schema_version(),
1119                        moved.event_type().to_owned(),
1120                        moved.payload().to_vec(),
1121                        moved.metadata().map(<[u8]>::to_vec),
1122                    )
1123                })
1124                .join()
1125                .expect("worker thread must not panic")
1126        });
1127
1128        assert_eq!(ver, v(3));
1129        assert_eq!(schema, 2);
1130        assert_eq!(etype, "MoneyDeposited");
1131        assert_eq!(payload, b"payload-bytes");
1132        assert_eq!(meta, Some(b"meta".to_vec()));
1133        // Original still usable after the clone left for another thread.
1134        assert_eq!(event.event_type(), "MoneyDeposited");
1135    }
1136
1137    // ── Category 1: sequence / protocol ─────────────────────────────────────
1138    // Multi-step interaction on ONE object: repeated and interleaved accessor
1139    // calls must be deterministic and mutually consistent.
1140
1141    #[test]
1142    fn repeated_accessor_calls_are_consistent() {
1143        let event = build(v(9), sv(4), b"TYPE", b"payload", Some(b"meta"));
1144
1145        // Same accessor twice → identical.
1146        assert_eq!(event.event_type(), event.event_type());
1147        assert_eq!(event.payload(), event.payload());
1148        assert_eq!(event.metadata(), event.metadata());
1149
1150        // Interleave borrowed and owned views → still consistent.
1151        assert_eq!(
1152            event.event_type().as_bytes(),
1153            event.event_type_bytes().as_ref()
1154        );
1155        assert_eq!(event.payload(), event.payload_bytes().as_ref());
1156        assert_eq!(event.metadata(), event.metadata_bytes().as_deref());
1157
1158        // Borrowed vs validated-newtype views agree.
1159        assert_eq!(event.event_type(), event.event_type_value().as_str());
1160        assert_eq!(event.payload(), event.payload_value().as_slice());
1161        assert_eq!(
1162            event.metadata(),
1163            event.metadata_value().as_ref().map(Metadata::as_slice),
1164        );
1165
1166        // Scalars are stable across repeated reads.
1167        assert_eq!(event.version(), v(9));
1168        assert_eq!(event.version(), event.version());
1169        assert_eq!(event.schema_version(), 4);
1170        assert_eq!(event.schema_version_value(), sv(4));
1171    }
1172
1173    // ── Category 2: lifecycle (build / clone / access-after-clone) ───────────
1174
1175    #[test]
1176    fn clone_is_field_for_field_equal_view() {
1177        let original = build(
1178            v(42),
1179            sv(7),
1180            b"OrderPlaced",
1181            b"the-payload",
1182            Some(b"the-meta"),
1183        );
1184        let cloned = original.clone();
1185
1186        assert_eq!(cloned.version(), original.version());
1187        assert_eq!(cloned.schema_version(), original.schema_version());
1188        assert_eq!(
1189            cloned.schema_version_value(),
1190            original.schema_version_value()
1191        );
1192        assert_eq!(cloned.event_type(), original.event_type());
1193        assert_eq!(cloned.payload(), original.payload());
1194        assert_eq!(cloned.metadata(), original.metadata());
1195        assert_eq!(
1196            cloned.metadata_value().map(|m| m.as_slice().to_vec()),
1197            original.metadata_value().map(|m| m.as_slice().to_vec()),
1198        );
1199    }
1200
1201    #[test]
1202    fn clone_shares_the_same_backing_buffer() {
1203        // A clone is one Arc refcount inc — owned views from both clones must
1204        // point at the SAME allocation (zero-copy), not a deep copy.
1205        let original = build(v(1), sv(1), b"TYPE", b"payload", None);
1206        let cloned = original.clone();
1207
1208        let from_original = original.payload_bytes();
1209        let from_clone = cloned.payload_bytes();
1210        assert!(
1211            std::ptr::eq(from_original.as_ptr(), from_clone.as_ptr()),
1212            "clone must share the parent buffer, not deep-copy it",
1213        );
1214        assert_eq!(from_original.as_ref(), from_clone.as_ref());
1215    }
1216
1217    // ── Category 3: defensive boundary (reject upstream-guarantee violations) ─
1218
1219    #[test]
1220    fn rejects_inverted_range_start_after_end() {
1221        let value = Bytes::from_static(b"TYPEpayload");
1222        // Built from variables so the inverted range is not a compile-time
1223        // literal (which `reversed_empty_ranges` would reject before runtime).
1224        let (bad_start, bad_end) = (7u32, 2u32);
1225        let err = PersistedEnvelope::try_new(
1226            Version::INITIAL,
1227            value,
1228            SchemaVersion::INITIAL,
1229            0..4,
1230            bad_start..bad_end,
1231            None,
1232        )
1233        .expect_err("start > end must be rejected");
1234        assert!(matches!(
1235            err,
1236            EnvelopeError::RangeOutOfBounds {
1237                start: 7,
1238                end: 2,
1239                ..
1240            }
1241        ));
1242    }
1243
1244    #[test]
1245    fn event_type_cap_uses_range_length_not_endpoint_sum() {
1246        // Mutation-testing pin (cargo-mutants surfaced this gap): the cap check
1247        // is on `event_type_range.end - event_type_range.start`, NOT `+`. Every
1248        // other test starts the event_type range at 0, where `end - 0 == end +
1249        // 0`, so none can tell the operators apart. Here start > 0 and the
1250        // length (end - start = 30_000) is WITHIN the cap, while the endpoint
1251        // sum (end + start = 110_000) EXCEEDS it — so a `- → +` mutant would
1252        // wrongly reject this valid event, and the `.expect` below catches it.
1253        let start: u32 = 40_000;
1254        let end: u32 = 70_000;
1255        let buf_len = usize::try_from(end).expect("70_000 fits usize");
1256        let value = Bytes::from(vec![b'A'; buf_len]);
1257        let event = PersistedEnvelope::try_new(
1258            Version::INITIAL,
1259            value,
1260            SchemaVersion::INITIAL,
1261            start..end,
1262            0..0,
1263            None,
1264        )
1265        .expect("event_type length 30_000 is within MAX_EVENT_TYPE_LEN (65_535)");
1266        assert_eq!(event.event_type().len(), 30_000);
1267    }
1268
1269    // Each of the three ranges is bounds-checked independently. The payload
1270    // path is covered in the outer module; these pin the event_type and
1271    // metadata paths so a regression deleting either check is caught.
1272
1273    #[test]
1274    fn rejects_event_type_range_past_buffer_end() {
1275        let value = Bytes::from_static(b"short");
1276        let err = PersistedEnvelope::try_new(
1277            Version::INITIAL,
1278            value,
1279            SchemaVersion::INITIAL,
1280            // event_type is the FIRST range checked; end 100 > len 5.
1281            0..100,
1282            0..0,
1283            None,
1284        )
1285        .expect_err("event_type range past buffer must be rejected");
1286        assert!(matches!(
1287            err,
1288            EnvelopeError::RangeOutOfBounds {
1289                start: 0,
1290                end: 100,
1291                len: 5
1292            }
1293        ));
1294    }
1295
1296    #[test]
1297    fn rejects_metadata_range_past_buffer_end() {
1298        let value = Bytes::from_static(b"TYPEpayload");
1299        let err = PersistedEnvelope::try_new(
1300            Version::INITIAL,
1301            value,
1302            SchemaVersion::INITIAL,
1303            // event_type + payload in-bounds so the metadata check is reached.
1304            0..4,
1305            4..11,
1306            Some(11..100),
1307        )
1308        .expect_err("metadata range past buffer must be rejected");
1309        assert!(matches!(
1310            err,
1311            EnvelopeError::RangeOutOfBounds {
1312                start: 11,
1313                end: 100,
1314                len: 11
1315            }
1316        ));
1317    }
1318
1319    #[test]
1320    fn rejects_inverted_metadata_range() {
1321        let value = Bytes::from_static(b"TYPEpayload");
1322        let (bad_start, bad_end) = (9u32, 5u32);
1323        let err = PersistedEnvelope::try_new(
1324            Version::INITIAL,
1325            value,
1326            SchemaVersion::INITIAL,
1327            0..4,
1328            4..11,
1329            Some(bad_start..bad_end),
1330        )
1331        .expect_err("metadata start > end must be rejected");
1332        assert!(matches!(
1333            err,
1334            EnvelopeError::RangeOutOfBounds {
1335                start: 9,
1336                end: 5,
1337                ..
1338            }
1339        ));
1340    }
1341
1342    // ── absence-of-invariant pins: independent (possibly overlapping) ranges ─
1343
1344    #[test]
1345    fn ranges_are_independent_and_may_overlap() {
1346        // try_new enforces NO disjointness/contiguity. All three windows can
1347        // point at the SAME 4 bytes. A future change adding a disjointness
1348        // invariant is a conscious break, caught here.
1349        let value = Bytes::from_static(b"TYPE");
1350        let event = PersistedEnvelope::try_new(
1351            Version::INITIAL,
1352            value,
1353            SchemaVersion::INITIAL,
1354            0..4,
1355            0..4,
1356            Some(0..4),
1357        )
1358        .expect("overlapping ranges are structurally permitted");
1359        assert_eq!(event.event_type(), "TYPE");
1360        assert_eq!(event.payload(), b"TYPE");
1361        assert_eq!(event.metadata(), Some(b"TYPE".as_slice()));
1362    }
1363
1364    #[test]
1365    fn empty_payload_range_is_accepted_unlike_empty_metadata() {
1366        // Deliberate asymmetry mirrored from the value newtypes: an empty
1367        // payload is a legal event shape (marker event), an empty metadata
1368        // Some is not (use None).
1369        let value = Bytes::from_static(b"TYPE");
1370        let event = PersistedEnvelope::try_new(
1371            Version::INITIAL,
1372            value,
1373            SchemaVersion::INITIAL,
1374            0..4,
1375            4..4,
1376            None,
1377        )
1378        .expect("empty payload range is accepted");
1379        assert_eq!(event.payload(), b"");
1380        assert_eq!(event.payload_value().as_slice(), b"");
1381    }
1382
1383    // ── metadata None vs Some across every metadata accessor ────────────────
1384
1385    #[test]
1386    fn metadata_absent_is_none_everywhere() {
1387        let event = build(v(1), sv(1), b"TYPE", b"payload", None);
1388        assert!(event.metadata().is_none());
1389        assert!(event.metadata_bytes().is_none());
1390        assert!(event.metadata_value().is_none());
1391    }
1392
1393    #[test]
1394    fn metadata_present_threads_through_every_accessor() {
1395        let event = build(v(1), sv(1), b"TYPE", b"payload", Some(b"META"));
1396        assert_eq!(event.metadata(), Some(b"META".as_slice()));
1397        assert_eq!(event.metadata_bytes().expect("some").as_ref(), b"META");
1398        assert_eq!(event.metadata_value().expect("some").as_slice(), b"META");
1399    }
1400
1401    // ── empty event_type accepted (no minimum-length invariant) ─────────────
1402
1403    #[test]
1404    fn empty_event_type_is_accepted() {
1405        let event = build(v(1), sv(1), b"", b"payload", None);
1406        assert_eq!(event.event_type(), "");
1407        assert_eq!(event.event_type_value().as_str(), "");
1408        assert_eq!(event.payload(), b"payload");
1409    }
1410
1411    // ── boundary scalars carried verbatim (version, schema) ─────────────────
1412
1413    #[test]
1414    fn boundary_version_and_schema_version_carried_verbatim() {
1415        let event = build(v(u64::MAX), sv(u32::MAX), b"TYPE", b"payload", None);
1416        assert_eq!(event.version().as_u64(), u64::MAX);
1417        assert_eq!(event.schema_version(), u32::MAX);
1418        assert_eq!(event.schema_version_value().get(), u32::MAX);
1419    }
1420
1421    // ── zero-copy aliasing: owned views share the one Arc buffer ────────────
1422
1423    #[test]
1424    fn owned_views_alias_the_single_backing_buffer() {
1425        let (value, et, pl, meta) = assemble(b"TYPE", b"payload", Some(b"meta"));
1426        let base = value.clone(); // shares the same allocation as `value`
1427        let event = PersistedEnvelope::try_new(
1428            Version::INITIAL,
1429            value,
1430            SchemaVersion::INITIAL,
1431            et,
1432            pl,
1433            meta,
1434        )
1435        .expect("valid");
1436
1437        let et_bytes = event.event_type_bytes();
1438        let pl_bytes = event.payload_bytes();
1439        let meta_bytes = event.metadata_bytes().expect("metadata present");
1440
1441        // Each owned view points into `base` at exactly its range offset →
1442        // proves no copy AND that all three share the one buffer.
1443        assert!(
1444            std::ptr::eq(et_bytes.as_ptr(), base.as_ptr().wrapping_add(0)),
1445            "event_type_bytes must alias buffer offset 0",
1446        );
1447        assert!(
1448            std::ptr::eq(pl_bytes.as_ptr(), base.as_ptr().wrapping_add(4)),
1449            "payload_bytes must alias buffer offset 4",
1450        );
1451        assert!(
1452            std::ptr::eq(meta_bytes.as_ptr(), base.as_ptr().wrapping_add(11)),
1453            "metadata_bytes must alias buffer offset 11",
1454        );
1455        // The value-newtype path shares the same buffer too.
1456        assert!(std::ptr::eq(
1457            event.event_type_value().as_bytes().as_ptr(),
1458            base.as_ptr().wrapping_add(0),
1459        ));
1460        assert!(std::ptr::eq(
1461            event.payload_value().as_slice().as_ptr(),
1462            base.as_ptr().wrapping_add(4),
1463        ));
1464    }
1465
1466    // ── Property: random valid components round-trip through every accessor ──
1467
1468    fn boundary_u64() -> impl Strategy<Value = u64> {
1469        prop_oneof![
1470            Just(1u64),
1471            Just(2u64),
1472            Just(u64::MAX - 1),
1473            Just(u64::MAX),
1474            1u64..=u64::MAX,
1475        ]
1476    }
1477
1478    fn boundary_u32_nonzero() -> impl Strategy<Value = u32> {
1479        prop_oneof![
1480            Just(1u32),
1481            Just(2u32),
1482            Just(u32::MAX - 1),
1483            Just(u32::MAX),
1484            1u32..=u32::MAX,
1485        ]
1486    }
1487
1488    fn event_type_bytes_strategy() -> impl Strategy<Value = Vec<u8>> {
1489        // ASCII 'A'..='Z' guarantees valid UTF-8; lengths include the empty,
1490        // unit, interior, and exact-cap boundaries.
1491        prop_oneof![
1492            Just(0usize),
1493            Just(1usize),
1494            Just(255usize),
1495            Just(MAX_EVENT_TYPE_LEN),
1496        ]
1497        .prop_flat_map(|n| proptest::collection::vec(b'A'..=b'Z', n))
1498    }
1499
1500    fn payload_bytes_strategy() -> impl Strategy<Value = Vec<u8>> {
1501        prop_oneof![Just(0usize), Just(1usize), Just(1024usize)]
1502            .prop_flat_map(|n| proptest::collection::vec(any::<u8>(), n))
1503    }
1504
1505    fn metadata_bytes_strategy() -> impl Strategy<Value = Option<Vec<u8>>> {
1506        prop_oneof![
1507            Just(None),
1508            Just(Some(vec![0u8])),
1509            (1usize..=2048)
1510                .prop_flat_map(|n| proptest::collection::vec(any::<u8>(), n))
1511                .prop_map(Some),
1512        ]
1513    }
1514
1515    proptest! {
1516        #[test]
1517        fn persisted_envelope_roundtrips_every_accessor(
1518            event_type in event_type_bytes_strategy(),
1519            payload in payload_bytes_strategy(),
1520            metadata in metadata_bytes_strategy(),
1521            version_raw in boundary_u64(),
1522            schema_raw in boundary_u32_nonzero(),
1523        ) {
1524            let version = v(version_raw);
1525            let schema = sv(schema_raw);
1526            let event = build(
1527                version,
1528                schema,
1529                &event_type,
1530                &payload,
1531                metadata.as_deref(),
1532            );
1533
1534            // Scalars verbatim.
1535            prop_assert_eq!(event.version(), version);
1536            prop_assert_eq!(event.version().as_u64(), version_raw);
1537            prop_assert_eq!(event.schema_version(), schema_raw);
1538            prop_assert_eq!(event.schema_version_value(), schema);
1539
1540            // event_type across all three views. Owned views bound to locals
1541            // first so the backing Bytes/newtype outlives the borrow.
1542            let et_bytes = event.event_type_bytes();
1543            let et_value = event.event_type_value();
1544            prop_assert_eq!(event.event_type().as_bytes(), event_type.as_slice());
1545            prop_assert_eq!(et_bytes.as_ref(), event_type.as_slice());
1546            prop_assert_eq!(et_value.as_bytes(), event_type.as_slice());
1547
1548            // payload across all three views.
1549            let pl_bytes = event.payload_bytes();
1550            let pl_value = event.payload_value();
1551            prop_assert_eq!(event.payload(), payload.as_slice());
1552            prop_assert_eq!(pl_bytes.as_ref(), payload.as_slice());
1553            prop_assert_eq!(pl_value.as_slice(), payload.as_slice());
1554
1555            // metadata: presence and bytes consistent across all three views.
1556            if let Some(ref m) = metadata {
1557                prop_assert_eq!(event.metadata(), Some(m.as_slice()));
1558                let meta_bytes = event.metadata_bytes().expect("some");
1559                prop_assert_eq!(meta_bytes.as_ref(), m.as_slice());
1560                let meta_value = event.metadata_value().expect("some");
1561                prop_assert_eq!(meta_value.as_slice(), m.as_slice());
1562            } else {
1563                prop_assert!(event.metadata().is_none());
1564                prop_assert!(event.metadata_bytes().is_none());
1565                prop_assert!(event.metadata_value().is_none());
1566            }
1567        }
1568    }
1569
1570    // ── Adversarial robustness: try_new is fed UNTRUSTED decoder output ──────
1571    //
1572    // On the read path a corrupt/adversarial row reaches try_new. It must be
1573    // panic-free for ANY (value, ranges) and either return Ok or a KNOWN
1574    // EnvelopeError — never a panic, and never an Ok whose unsafe accessors are
1575    // unsound. The exhaustive Err match has no catch-all, so a new variant is a
1576    // compile error here — pinning the rejection surface. Run under Miri to
1577    // prove the unsafe accessors' preconditions hold for fuzzed input.
1578
1579    fn arbitrary_range() -> impl Strategy<Value = Range<u32>> {
1580        (0u32..=260, 0u32..=260).prop_map(|(start, end)| start..end)
1581    }
1582
1583    fn arbitrary_optional_range() -> impl Strategy<Value = Option<Range<u32>>> {
1584        prop_oneof![Just(None), arbitrary_range().prop_map(Some)]
1585    }
1586
1587    proptest! {
1588        #[test]
1589        fn try_new_never_panics_and_accepted_events_have_sound_accessors(
1590            value_bytes in proptest::collection::vec(any::<u8>(), 0..256),
1591            event_type_range in arbitrary_range(),
1592            payload_range in arbitrary_range(),
1593            metadata_range in arbitrary_optional_range(),
1594            version_raw in boundary_u64(),
1595            schema_raw in boundary_u32_nonzero(),
1596        ) {
1597            let result = PersistedEnvelope::try_new(
1598                v(version_raw),
1599                Bytes::from(value_bytes),
1600                sv(schema_raw),
1601                event_type_range,
1602                payload_range,
1603                metadata_range,
1604            );
1605
1606            match result {
1607                Ok(event) => {
1608                    // Every accessor must be callable without panic/UB. The
1609                    // event_type() / *_value() paths run `unsafe`
1610                    // from_utf8_unchecked / from_validated_bytes — under Miri
1611                    // this proves their preconditions held for fuzzed input.
1612                    let et = event.event_type();
1613                    prop_assert!(std::str::from_utf8(et.as_bytes()).is_ok());
1614                    let _ = event.payload();
1615                    let _ = event.metadata();
1616                    let _ = event.event_type_value();
1617                    let _ = event.payload_value();
1618                    let _ = event.metadata_value();
1619                    let _ = event.event_type_bytes();
1620                    let _ = event.payload_bytes();
1621                    let _ = event.metadata_bytes();
1622                }
1623                Err(err) => {
1624                    // Exhaustive — no `_` arm. A new EnvelopeError variant must
1625                    // be classified here, not silently absorbed.
1626                    match err {
1627                        EnvelopeError::RangeOutOfBounds { .. }
1628                        | EnvelopeError::InvalidUtf8 { .. }
1629                        | EnvelopeError::EventTypeRangeTooLong { .. }
1630                        | EnvelopeError::MetadataRangeTooLong { .. }
1631                        | EnvelopeError::MetadataRangeEmpty
1632                        | EnvelopeError::Value(_) => {}
1633                    }
1634                }
1635            }
1636        }
1637    }
1638}
1639
1640#[cfg(test)]
1641#[allow(
1642    clippy::unwrap_used,
1643    clippy::expect_used,
1644    clippy::panic,
1645    reason = "test code asserts exact values"
1646)]
1647mod pending_batch_tests {
1648    use super::{PendingBatch, PendingEnvelope, pending_envelope};
1649    use bytes::Bytes;
1650    use mnesis::Version;
1651
1652    fn env(version: u64) -> PendingEnvelope {
1653        pending_envelope(Version::new(version).expect("version is non-zero"))
1654            .event_type("E")
1655            .payload(Bytes::from_static(b"p"))
1656            .build()
1657            .expect("valid envelope")
1658    }
1659
1660    #[test]
1661    fn empty_slice_yields_no_batch() {
1662        assert!(PendingBatch::new(&[]).is_none());
1663    }
1664
1665    #[test]
1666    fn batch_keeps_every_envelope_in_order() {
1667        let envs = [env(1), env(2), env(3)];
1668        let batch = PendingBatch::new(&envs).expect("three envelopes are non-empty");
1669
1670        assert_eq!(batch.len().get(), 3);
1671        let mut versions = batch.iter().map(|e| e.version().as_u64());
1672        assert_eq!(versions.next(), Some(1));
1673        assert_eq!(versions.next(), Some(2));
1674        assert_eq!(versions.next(), Some(3));
1675        assert_eq!(versions.next(), None);
1676    }
1677
1678    #[test]
1679    fn first_and_last_are_the_boundary_envelopes() {
1680        let envs = [env(7), env(8), env(9)];
1681        let batch = PendingBatch::new(&envs).expect("non-empty");
1682
1683        assert_eq!(batch.first().version().as_u64(), 7);
1684        assert_eq!(batch.last().version().as_u64(), 9);
1685    }
1686
1687    #[test]
1688    fn single_envelope_batch_is_infallible() {
1689        let only = env(4);
1690        let batch = PendingBatch::of(&only);
1691
1692        assert_eq!(batch.len().get(), 1);
1693        assert_eq!(batch.first().version().as_u64(), 4);
1694        assert_eq!(batch.last().version().as_u64(), 4);
1695    }
1696
1697    #[test]
1698    fn from_parts_agrees_with_new_over_the_same_envelopes() {
1699        let envs = [env(1), env(2)];
1700        let (first, rest) = envs.split_first().expect("non-empty");
1701
1702        let parts = PendingBatch::from_parts(first, rest);
1703        let whole = PendingBatch::new(&envs).expect("non-empty");
1704
1705        assert_eq!(parts.len(), whole.len());
1706        let mut a = parts.iter().map(|e| e.version().as_u64());
1707        let mut b = whole.iter().map(|e| e.version().as_u64());
1708        assert_eq!(a.next(), b.next());
1709        assert_eq!(a.next(), b.next());
1710        assert_eq!(a.next(), None);
1711        assert_eq!(b.next(), None);
1712    }
1713}