reliar_core/
envelope.rs

1//! `Envelope<T>` / `SerializedEnvelope` and their builder (ADR 0003, ADR 0011).
2
3use core::fmt;
4
5use bytes::Bytes;
6
7use crate::{
8    ConversationId, CorrelationId, CorrelationMetadata, HeaderError, Headers, Message, MessageId,
9    MessageType, Metadata,
10};
11
12/// An envelope: a typed or serialized body plus the metadata Reliar understands and the custom
13/// headers it does not. `Envelope != OutboxRecord != InboxRecord` — nothing here
14/// carries delivery state (attempts, leases, dead-letter bookkeeping).
15///
16/// ```
17/// use reliar_core::Envelope;
18///
19/// #[derive(serde::Serialize, serde::Deserialize)]
20/// struct Ping;
21/// impl reliar_core::Message for Ping {
22///     const TYPE: &'static str = "ping";
23///     const VERSION: u16 = 1;
24/// }
25///
26/// let envelope = Envelope::builder(Ping).build();
27/// assert_eq!(envelope.message_type.to_string(), "ping.v1");
28/// ```
29#[non_exhaustive]
30pub struct Envelope<T> {
31    /// The envelope's own identity.
32    pub id: MessageId,
33
34    /// The message's stable contract identity — `T::TYPE`/`T::VERSION`, never chosen ad hoc.
35    pub message_type: MessageType,
36
37    /// The message body: typed on the application side, `bytes::Bytes` once serialized.
38    pub body: T,
39
40    /// Canonical, typed framework metadata — the single source of truth (ADR 0004).
41    pub metadata: Metadata,
42
43    /// Private: preserves [`Headers`]' validation invariants — mutate only through
44    /// [`Self::headers_mut`]/[`Self::set_headers`].
45    pub(crate) headers: Option<Headers>,
46}
47
48/// The persistence/transport form: an envelope whose body has already been serialized to bytes.
49///
50/// ```
51/// use bytes::Bytes;
52/// use reliar_core::{Envelope, SerializedEnvelope};
53/// # #[derive(serde::Serialize, serde::Deserialize)]
54/// # struct Ping;
55/// # impl reliar_core::Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
56///
57/// let typed = Envelope::builder(Ping).build();
58/// let wire: SerializedEnvelope = typed.map_body(|_| Bytes::from_static(b"{}"));
59/// assert_eq!(wire.body.as_ref(), b"{}");
60/// ```
61pub type SerializedEnvelope = Envelope<Bytes>;
62
63impl<T> Envelope<T> {
64    /// The envelope's custom headers, if any were set.
65    ///
66    /// ```
67    /// use reliar_core::Envelope;
68    /// # #[derive(serde::Serialize, serde::Deserialize)]
69    /// # struct Ping;
70    /// # impl reliar_core::Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
71    /// let envelope = Envelope::builder(Ping).build();
72    /// assert!(envelope.headers().is_none());
73    /// ```
74    #[must_use]
75    pub fn headers(&self) -> Option<&Headers> {
76        self.headers.as_ref()
77    }
78
79    /// Mutably accesses the envelope's custom headers, lazily allocating an empty [`Headers`]
80    /// the first time this is called.
81    ///
82    /// ```
83    /// use reliar_core::Envelope;
84    /// # #[derive(serde::Serialize, serde::Deserialize)]
85    /// # struct Ping;
86    /// # impl reliar_core::Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
87    /// let mut envelope = Envelope::builder(Ping).build();
88    /// envelope.headers_mut().insert("x-a", "1")?;
89    /// assert_eq!(envelope.headers().unwrap().get("x-a"), Some("1"));
90    /// # Ok::<(), reliar_core::HeaderError>(())
91    /// ```
92    pub fn headers_mut(&mut self) -> &mut Headers {
93        self.headers.get_or_insert_with(Headers::default)
94    }
95
96    /// Replaces the whole header map. The rehydration path for providers and transport mappers,
97    /// which read back an already-validated map rather than inserting key by key.
98    ///
99    /// ```
100    /// use reliar_core::{Envelope, Headers};
101    /// # #[derive(serde::Serialize, serde::Deserialize)]
102    /// # struct Ping;
103    /// # impl reliar_core::Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
104    /// let mut envelope = Envelope::builder(Ping).build();
105    /// let mut headers = Headers::default();
106    /// headers.insert("x-a", "1")?;
107    /// envelope.set_headers(Some(headers));
108    /// assert_eq!(envelope.headers().unwrap().get("x-a"), Some("1"));
109    /// # Ok::<(), reliar_core::HeaderError>(())
110    /// ```
111    pub fn set_headers(&mut self, headers: Option<Headers>) {
112        self.headers = headers;
113    }
114
115    /// Converts the body, keeping every other field. The only conversion between typed and
116    /// serialized envelopes — no field is ever re-declared, so none can be dropped in the
117    /// process (ADR 0003).
118    ///
119    /// ```
120    /// use bytes::Bytes;
121    /// use reliar_core::Envelope;
122    /// # #[derive(serde::Serialize, serde::Deserialize)]
123    /// # struct Ping;
124    /// # impl reliar_core::Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
125    /// let envelope = Envelope::builder(Ping).build();
126    /// let serialized: Envelope<Bytes> = envelope.map_body(|_| Bytes::from_static(b"{}"));
127    /// assert_eq!(serialized.body.as_ref(), b"{}");
128    /// ```
129    #[must_use]
130    pub fn map_body<U>(self, f: impl FnOnce(T) -> U) -> Envelope<U> {
131        Envelope {
132            id: self.id,
133            message_type: self.message_type,
134            body: f(self.body),
135            metadata: self.metadata,
136            headers: self.headers,
137        }
138    }
139
140    /// Fallible variant of [`Self::map_body`], for `SerializedEnvelope -> Envelope<T>` via a
141    /// [`Serializer`](crate::Serializer).
142    ///
143    /// # Errors
144    ///
145    /// Returns whatever error `f` returns, unchanged.
146    ///
147    /// ```
148    /// use bytes::Bytes;
149    /// use reliar_core::Envelope;
150    /// # #[derive(serde::Serialize, serde::Deserialize)]
151    /// # struct Ping;
152    /// # impl reliar_core::Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
153    /// let wire = Envelope::builder(Ping)
154    ///     .build()
155    ///     .map_body(|_| Bytes::from_static(b"{}"));
156    ///
157    /// let typed: Envelope<Ping> = wire.try_map_body(|_body| Ok::<_, std::convert::Infallible>(Ping))?;
158    /// assert_eq!(typed.message_type.to_string(), "ping.v1");
159    /// # Ok::<(), std::convert::Infallible>(())
160    /// ```
161    pub fn try_map_body<U, E>(self, f: impl FnOnce(T) -> Result<U, E>) -> Result<Envelope<U>, E> {
162        Ok(Envelope {
163            id: self.id,
164            message_type: self.message_type,
165            body: f(self.body)?,
166            metadata: self.metadata,
167            headers: self.headers,
168        })
169    }
170}
171
172impl<T: Message> Envelope<T> {
173    /// Starts building an envelope for `body`. `message_type` is derived from `T::TYPE`/
174    /// `T::VERSION` and cannot be passed in (ADR 0010).
175    ///
176    /// ```
177    /// use reliar_core::Envelope;
178    ///
179    /// #[derive(serde::Serialize, serde::Deserialize)]
180    /// struct OrderCreated { order_id: u64 }
181    ///
182    /// impl reliar_core::Message for OrderCreated {
183    ///     const TYPE: &'static str = "orders.created";
184    ///     const VERSION: u16 = 1;
185    /// }
186    ///
187    /// let envelope = Envelope::builder(OrderCreated { order_id: 42 })
188    ///     .tenant("acme")
189    ///     .header("x-import-batch", "2026-09-04")?
190    ///     .build();
191    ///
192    /// assert_eq!(envelope.message_type.to_string(), "orders.created.v1");
193    /// assert_eq!(envelope.metadata.tenant_id.as_deref(), Some("acme"));
194    /// # Ok::<(), reliar_core::HeaderError>(())
195    /// ```
196    pub fn builder(body: T) -> EnvelopeBuilder<T> {
197        EnvelopeBuilder::new(body)
198    }
199}
200
201impl SerializedEnvelope {
202    /// Rehydration entry point for providers and transport mappers, which have a `MessageType`
203    /// read from storage or the wire rather than from a Rust type (ADR 0011).
204    ///
205    /// ```
206    /// use bytes::Bytes;
207    /// use reliar_core::{Metadata, MessageId, MessageType, SerializedEnvelope};
208    ///
209    /// let envelope = SerializedEnvelope::from_parts(
210    ///     MessageId::new(),
211    ///     MessageType::from_parts("orders.created".to_string(), 1),
212    ///     Bytes::from_static(b"{}"),
213    ///     Metadata::default(),
214    ///     None,
215    /// );
216    /// assert_eq!(envelope.message_type.name(), "orders.created");
217    /// ```
218    #[must_use]
219    pub fn from_parts(
220        id: MessageId,
221        message_type: MessageType,
222        body: Bytes,
223        metadata: Metadata,
224        headers: Option<Headers>,
225    ) -> Self {
226        Self {
227            id,
228            message_type,
229            body,
230            metadata,
231            headers,
232        }
233    }
234}
235
236/// Elides the body unconditionally: a typed body may be arbitrary application data and a
237/// serialized one is raw payload bytes, and neither belongs in a log line (ADR 0003).
238impl<T> fmt::Debug for Envelope<T> {
239    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
240        f.debug_struct("Envelope")
241            .field("id", &self.id)
242            .field("message_type", &self.message_type)
243            .field("body", &"<elided>")
244            .field("metadata", &self.metadata)
245            .field("headers", &self.headers)
246            .finish()
247    }
248}
249
250impl<T: PartialEq> PartialEq for Envelope<T> {
251    fn eq(&self, other: &Self) -> bool {
252        self.id == other.id
253            && self.message_type == other.message_type
254            && self.body == other.body
255            && self.metadata == other.metadata
256            && self.headers == other.headers
257    }
258}
259
260/// `Clone` only where `T: Clone` — nothing in Reliar requires it, since a dispatcher moves owned
261/// records into publish tasks rather than cloning them; the impl exists for tests and host code.
262impl<T: Clone> Clone for Envelope<T> {
263    fn clone(&self) -> Self {
264        Self {
265            id: self.id,
266            message_type: self.message_type.clone(),
267            body: self.body.clone(),
268            metadata: self.metadata.clone(),
269            headers: self.headers.clone(),
270        }
271    }
272}
273
274/// Builds an [`Envelope<T>`]. Obtained from [`Envelope::builder`].
275///
276/// ```
277/// use reliar_core::Envelope;
278///
279/// #[derive(serde::Serialize, serde::Deserialize)]
280/// struct OrderCreated { order_id: u64 }
281/// impl reliar_core::Message for OrderCreated {
282///     const TYPE: &'static str = "orders.created";
283///     const VERSION: u16 = 1;
284/// }
285///
286/// let envelope = Envelope::builder(OrderCreated { order_id: 42 })
287///     .tenant("acme")
288///     .build();
289/// assert_eq!(envelope.metadata.tenant_id.as_deref(), Some("acme"));
290/// ```
291#[must_use]
292pub struct EnvelopeBuilder<T> {
293    id: Option<MessageId>,
294
295    body: T,
296
297    metadata: Metadata,
298
299    headers: Option<Headers>,
300}
301
302/// Elides the body: an in-progress envelope's body is arbitrary application data.
303impl<T> fmt::Debug for EnvelopeBuilder<T> {
304    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
305        f.debug_struct("EnvelopeBuilder")
306            .field("id", &self.id)
307            .field("body", &"<elided>")
308            .field("metadata", &self.metadata)
309            .field("headers", &self.headers)
310            .finish()
311    }
312}
313
314impl<T: Message> EnvelopeBuilder<T> {
315    fn new(body: T) -> Self {
316        Self {
317            id: None,
318            body,
319            metadata: Metadata::default(),
320            headers: None,
321        }
322    }
323
324    /// Overrides the generated id. Defaults to a fresh `UUIDv7`.
325    ///
326    /// ```
327    /// use reliar_core::{Envelope, MessageId};
328    /// # #[derive(serde::Serialize, serde::Deserialize)]
329    /// # struct Ping;
330    /// # impl reliar_core::Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
331    /// let id = MessageId::new();
332    /// let envelope = Envelope::builder(Ping).id(id).build();
333    /// assert_eq!(envelope.id, id);
334    /// ```
335    pub fn id(mut self, id: MessageId) -> Self {
336        self.id = Some(id);
337
338        self
339    }
340
341    /// Replaces the whole metadata struct, including its correlation metadata. Conversation
342    /// rooting is decided by *value*, not by call order: if the replacement's `conversation_id`
343    /// is still [`crate::ConversationId::UNSET`], [`Self::build`] roots it at the envelope's own
344    /// id regardless of an earlier [`Self::conversation`] call; a non-`UNSET` value (including
345    /// one copied from a causing message) is kept.
346    ///
347    /// ```
348    /// use reliar_core::{Envelope, Metadata};
349    /// # #[derive(serde::Serialize, serde::Deserialize)]
350    /// # struct Ping;
351    /// # impl reliar_core::Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
352    /// let mut metadata = Metadata::default();
353    /// metadata.tenant_id = Some("acme".to_string());
354    /// let envelope = Envelope::builder(Ping).metadata(metadata).build();
355    /// assert_eq!(envelope.metadata.tenant_id.as_deref(), Some("acme"));
356    /// ```
357    pub fn metadata(mut self, metadata: Metadata) -> Self {
358        self.metadata = metadata;
359
360        self
361    }
362
363    /// Replaces the correlation metadata (correlation id, conversation id, causation, request
364    /// id) as a group. Same value-decides-rooting rule as [`Self::metadata`].
365    ///
366    /// ```
367    /// use reliar_core::{CorrelationMetadata, Envelope};
368    /// # #[derive(serde::Serialize, serde::Deserialize)]
369    /// # struct Ping;
370    /// # impl reliar_core::Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
371    /// let mut correlation = CorrelationMetadata::default();
372    /// correlation.causation_id = Some(reliar_core::MessageId::new());
373    /// let envelope = Envelope::builder(Ping).correlation(correlation).build();
374    /// assert!(envelope.metadata.correlation.causation_id.is_some());
375    /// ```
376    pub fn correlation(mut self, correlation: CorrelationMetadata) -> Self {
377        self.metadata.correlation = correlation;
378
379        self
380    }
381
382    /// Sets the business correlation id.
383    ///
384    /// ```
385    /// use reliar_core::{CorrelationId, Envelope};
386    /// # #[derive(serde::Serialize, serde::Deserialize)]
387    /// # struct Ping;
388    /// # impl reliar_core::Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
389    /// let envelope = Envelope::builder(Ping)
390    ///     .correlation_id(CorrelationId::parse("checkout-42")?)
391    ///     .build();
392    /// assert_eq!(envelope.metadata.correlation.correlation_id.unwrap().as_str(), "checkout-42");
393    /// # Ok::<(), reliar_core::IdError>(())
394    /// ```
395    pub fn correlation_id(mut self, id: CorrelationId) -> Self {
396        self.metadata.correlation.correlation_id = Some(id);
397
398        self
399    }
400
401    /// Joins an existing conversation — typically the causing message's own `conversation_id`.
402    /// [`Self::build`] keeps this value as long as nothing later replaces it with
403    /// [`Self::metadata`] or [`Self::correlation`] (setter order matters only in that sense: the
404    /// last write to `conversation_id` wins, same as any other field).
405    ///
406    /// ```
407    /// use reliar_core::{ConversationId, Envelope};
408    /// use uuid::Uuid;
409    /// # #[derive(serde::Serialize, serde::Deserialize)]
410    /// # struct Ping;
411    /// # impl reliar_core::Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
412    /// let parent_conversation = ConversationId::from_uuid(Uuid::now_v7());
413    /// let envelope = Envelope::builder(Ping).conversation(parent_conversation).build();
414    /// assert_eq!(envelope.metadata.correlation.conversation_id, parent_conversation);
415    /// ```
416    pub fn conversation(mut self, id: ConversationId) -> Self {
417        self.metadata.correlation.conversation_id = id;
418
419        self
420    }
421
422    /// Records the message that caused this one.
423    ///
424    /// ```
425    /// use reliar_core::{Envelope, MessageId};
426    /// # #[derive(serde::Serialize, serde::Deserialize)]
427    /// # struct Ping;
428    /// # impl reliar_core::Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
429    /// let parent = MessageId::new();
430    /// let envelope = Envelope::builder(Ping).causation(parent).build();
431    /// assert_eq!(envelope.metadata.correlation.causation_id, Some(parent));
432    /// ```
433    pub fn causation(mut self, parent: MessageId) -> Self {
434        self.metadata.correlation.causation_id = Some(parent);
435
436        self
437    }
438
439    /// Sets the owning tenant.
440    ///
441    /// ```
442    /// use reliar_core::Envelope;
443    /// # #[derive(serde::Serialize, serde::Deserialize)]
444    /// # struct Ping;
445    /// # impl reliar_core::Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
446    /// let envelope = Envelope::builder(Ping).tenant("acme").build();
447    /// assert_eq!(envelope.metadata.tenant_id.as_deref(), Some("acme"));
448    /// ```
449    pub fn tenant(mut self, tenant_id: impl Into<String>) -> Self {
450        self.metadata.tenant_id = Some(tenant_id.into());
451
452        self
453    }
454
455    /// Sets the time after which the message must not be published.
456    ///
457    /// ```
458    /// use reliar_core::Envelope;
459    /// # #[derive(serde::Serialize, serde::Deserialize)]
460    /// # struct Ping;
461    /// # impl reliar_core::Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
462    /// let one_day = time::Duration::days(1);
463    /// let envelope = Envelope::builder(Ping)
464    ///     .expires_at(time::OffsetDateTime::now_utc() + one_day)
465    ///     .build();
466    /// assert!(envelope.metadata.delivery.expires_at.is_some());
467    /// ```
468    pub fn expires_at(mut self, at: time::OffsetDateTime) -> Self {
469        self.metadata.delivery.expires_at = Some(at);
470
471        self
472    }
473
474    /// Sets the W3C Trace Context to carry verbatim. Reliar never invents or re-derives it
475    /// (ADR 0004, ADR 0020).
476    ///
477    /// ```
478    /// use reliar_core::Envelope;
479    /// # #[derive(serde::Serialize, serde::Deserialize)]
480    /// # struct Ping;
481    /// # impl reliar_core::Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
482    /// let envelope = Envelope::builder(Ping)
483    ///     .trace("00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01", None)
484    ///     .build();
485    /// assert!(envelope.metadata.trace.traceparent.is_some());
486    /// ```
487    pub fn trace(mut self, traceparent: impl Into<String>, tracestate: Option<String>) -> Self {
488        self.metadata.trace.traceparent = Some(traceparent.into());
489        self.metadata.trace.tracestate = tracestate;
490
491        self
492    }
493
494    /// Sets one custom header. Returns `Err` if `k` uses the reserved `reliar-` prefix or
495    /// breaches a cap (see [`Headers::insert`]).
496    ///
497    /// # Errors
498    ///
499    /// Returns [`HeaderError`] under the same conditions as [`Headers::insert`].
500    ///
501    /// ```
502    /// use reliar_core::Envelope;
503    /// # #[derive(serde::Serialize, serde::Deserialize)]
504    /// # struct Ping;
505    /// # impl reliar_core::Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
506    /// let envelope = Envelope::builder(Ping).header("x-import-batch", "2026-09-04")?.build();
507    /// assert_eq!(envelope.headers().unwrap().get("x-import-batch"), Some("2026-09-04"));
508    /// # Ok::<(), reliar_core::HeaderError>(())
509    /// ```
510    pub fn header(
511        mut self,
512        k: impl Into<String>,
513        v: impl Into<String>,
514    ) -> Result<Self, HeaderError> {
515        self.headers
516            .get_or_insert_with(Headers::default)
517            .insert(k, v)?;
518
519        Ok(self)
520    }
521
522    /// Builds the envelope. `message_type` is `MessageType::of::<T>()`. `conversation_id`:
523    /// **iff** it is still [`crate::ConversationId::UNSET`], it becomes this envelope's own id
524    /// (an un-correlated message roots its own conversation); any other value — set via
525    /// [`Self::conversation`], [`Self::correlation`], or [`Self::metadata`] — is kept verbatim.
526    /// Rooting is decided by the value alone, never by which setter was called or in what order
527    /// (ADR 0011).
528    ///
529    /// ```
530    /// use reliar_core::Envelope;
531    /// # #[derive(serde::Serialize, serde::Deserialize)]
532    /// # struct Ping;
533    /// # impl reliar_core::Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
534    /// let envelope = Envelope::builder(Ping).build();
535    /// // An un-correlated message roots its own conversation.
536    /// assert_eq!(envelope.metadata.correlation.conversation_id.as_uuid(), envelope.id.as_uuid());
537    /// ```
538    #[must_use]
539    pub fn build(mut self) -> Envelope<T> {
540        let id = self.id.unwrap_or_default();
541
542        if self.metadata.correlation.conversation_id.is_unset() {
543            self.metadata.correlation.conversation_id = ConversationId::from_uuid(id.as_uuid());
544        }
545
546        Envelope {
547            id,
548            message_type: MessageType::of::<T>(),
549            body: self.body,
550            metadata: self.metadata,
551            headers: self.headers,
552        }
553    }
554}
reliar_core/envelope.rs

reliar_core/
envelope.rs
