reliar_core/envelope.rs
1//! `Envelope<T>` / `SerializedEnvelope` and their builder (SRS §9, §9.1, 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 (§9). `Envelope != OutboxRecord != InboxRecord` (§17) — nothing here
14/// carries delivery state (attempts, leases, dead-letter bookkeeping).
15#[non_exhaustive]
16pub struct Envelope<T> {
17 /// The envelope's own identity.
18 pub id: MessageId,
19 /// The message's stable contract identity — `T::TYPE`/`T::VERSION`, never chosen ad hoc.
20 pub message_type: MessageType,
21 /// The message body: typed on the application side, `bytes::Bytes` once serialized.
22 pub body: T,
23 /// Canonical, typed framework metadata — the single source of truth (ADR 0004).
24 pub metadata: Metadata,
25 /// Private: preserves [`Headers`]' validation invariants — mutate only through
26 /// [`Self::headers_mut`]/[`Self::set_headers`].
27 pub(crate) headers: Option<Headers>,
28}
29
30/// The persistence/transport form: an envelope whose body has already been serialized to bytes.
31pub type SerializedEnvelope = Envelope<Bytes>;
32
33impl<T> Envelope<T> {
34 /// The envelope's custom headers, if any were set.
35 #[must_use]
36 pub fn headers(&self) -> Option<&Headers> {
37 self.headers.as_ref()
38 }
39
40 /// Mutably accesses the envelope's custom headers, lazily allocating an empty [`Headers`]
41 /// the first time this is called.
42 pub fn headers_mut(&mut self) -> &mut Headers {
43 self.headers.get_or_insert_with(Headers::default)
44 }
45
46 /// Replaces the whole header map. The rehydration path for providers and transport mappers,
47 /// which read back an already-validated map rather than inserting key by key.
48 pub fn set_headers(&mut self, headers: Option<Headers>) {
49 self.headers = headers;
50 }
51
52 /// Converts the body, keeping every other field. The only conversion between typed and
53 /// serialized envelopes — no field is ever re-declared, so none can be dropped in the
54 /// process (ADR 0003).
55 ///
56 /// ```
57 /// use bytes::Bytes;
58 /// use reliar_core::Envelope;
59 /// # #[derive(serde::Serialize, serde::Deserialize)]
60 /// # struct Ping;
61 /// # impl reliar_core::Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
62 /// let envelope = Envelope::builder(Ping).build();
63 /// let serialized: Envelope<Bytes> = envelope.map_body(|_| Bytes::from_static(b"{}"));
64 /// assert_eq!(serialized.body.as_ref(), b"{}");
65 /// ```
66 #[must_use]
67 pub fn map_body<U>(self, f: impl FnOnce(T) -> U) -> Envelope<U> {
68 Envelope {
69 id: self.id,
70 message_type: self.message_type,
71 body: f(self.body),
72 metadata: self.metadata,
73 headers: self.headers,
74 }
75 }
76
77 /// Fallible variant of [`Self::map_body`], for `SerializedEnvelope -> Envelope<T>` via a
78 /// [`Serializer`](crate::Serializer).
79 ///
80 /// # Errors
81 ///
82 /// Returns whatever error `f` returns, unchanged.
83 pub fn try_map_body<U, E>(self, f: impl FnOnce(T) -> Result<U, E>) -> Result<Envelope<U>, E> {
84 Ok(Envelope {
85 id: self.id,
86 message_type: self.message_type,
87 body: f(self.body)?,
88 metadata: self.metadata,
89 headers: self.headers,
90 })
91 }
92}
93
94impl<T: Message> Envelope<T> {
95 /// Starts building an envelope for `body`. `message_type` is derived from `T::TYPE`/
96 /// `T::VERSION` and cannot be passed in (ADR 0010).
97 ///
98 /// ```
99 /// use reliar_core::Envelope;
100 ///
101 /// #[derive(serde::Serialize, serde::Deserialize)]
102 /// struct OrderCreated { order_id: u64 }
103 ///
104 /// impl reliar_core::Message for OrderCreated {
105 /// const TYPE: &'static str = "orders.created";
106 /// const VERSION: u16 = 1;
107 /// }
108 ///
109 /// let envelope = Envelope::builder(OrderCreated { order_id: 42 })
110 /// .tenant("acme")
111 /// .header("x-import-batch", "2026-09-04")?
112 /// .build();
113 ///
114 /// assert_eq!(envelope.message_type.to_string(), "orders.created.v1");
115 /// assert_eq!(envelope.metadata.tenant_id.as_deref(), Some("acme"));
116 /// # Ok::<(), reliar_core::HeaderError>(())
117 /// ```
118 pub fn builder(body: T) -> EnvelopeBuilder<T> {
119 EnvelopeBuilder::new(body)
120 }
121}
122
123impl SerializedEnvelope {
124 /// Rehydration entry point for providers and transport mappers, which have a `MessageType`
125 /// read from storage or the wire rather than from a Rust type (ADR 0011).
126 #[must_use]
127 pub fn from_parts(
128 id: MessageId,
129 message_type: MessageType,
130 body: Bytes,
131 metadata: Metadata,
132 headers: Option<Headers>,
133 ) -> Self {
134 Self {
135 id,
136 message_type,
137 body,
138 metadata,
139 headers,
140 }
141 }
142}
143
144/// Elides the body unconditionally: a typed body may be arbitrary application data and a
145/// serialized one is raw payload bytes, and neither belongs in a log line (§33, ADR 0003).
146impl<T> fmt::Debug for Envelope<T> {
147 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
148 f.debug_struct("Envelope")
149 .field("id", &self.id)
150 .field("message_type", &self.message_type)
151 .field("body", &"<elided>")
152 .field("metadata", &self.metadata)
153 .field("headers", &self.headers)
154 .finish()
155 }
156}
157
158impl<T: PartialEq> PartialEq for Envelope<T> {
159 fn eq(&self, other: &Self) -> bool {
160 self.id == other.id
161 && self.message_type == other.message_type
162 && self.body == other.body
163 && self.metadata == other.metadata
164 && self.headers == other.headers
165 }
166}
167
168/// `Clone` only where `T: Clone` — nothing in Reliar requires it (the dispatcher moves owned
169/// records into publish tasks, SRS §9.1); the impl exists for tests and host code.
170impl<T: Clone> Clone for Envelope<T> {
171 fn clone(&self) -> Self {
172 Self {
173 id: self.id,
174 message_type: self.message_type.clone(),
175 body: self.body.clone(),
176 metadata: self.metadata.clone(),
177 headers: self.headers.clone(),
178 }
179 }
180}
181
182/// Builds an [`Envelope<T>`]. Obtained from [`Envelope::builder`].
183#[must_use]
184pub struct EnvelopeBuilder<T> {
185 id: Option<MessageId>,
186 body: T,
187 metadata: Metadata,
188 headers: Option<Headers>,
189}
190
191/// Elides the body: an in-progress envelope's body is arbitrary application data (§33).
192impl<T> fmt::Debug for EnvelopeBuilder<T> {
193 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
194 f.debug_struct("EnvelopeBuilder")
195 .field("id", &self.id)
196 .field("body", &"<elided>")
197 .field("metadata", &self.metadata)
198 .field("headers", &self.headers)
199 .finish()
200 }
201}
202
203impl<T: Message> EnvelopeBuilder<T> {
204 fn new(body: T) -> Self {
205 Self {
206 id: None,
207 body,
208 metadata: Metadata::default(),
209 headers: None,
210 }
211 }
212
213 /// Overrides the generated id. Defaults to a fresh `UUIDv7`.
214 pub fn id(mut self, id: MessageId) -> Self {
215 self.id = Some(id);
216 self
217 }
218
219 /// Replaces the whole metadata struct, including its correlation metadata. Conversation
220 /// rooting is decided by *value*, not by call order: if the replacement's `conversation_id`
221 /// is still [`crate::ConversationId::UNSET`], [`Self::build`] roots it at the envelope's own
222 /// id regardless of an earlier [`Self::conversation`] call; a non-`UNSET` value (including
223 /// one copied from a causing message) is kept.
224 pub fn metadata(mut self, metadata: Metadata) -> Self {
225 self.metadata = metadata;
226 self
227 }
228
229 /// Replaces the correlation metadata (correlation id, conversation id, causation, request
230 /// id) as a group. Same value-decides-rooting rule as [`Self::metadata`].
231 pub fn correlation(mut self, correlation: CorrelationMetadata) -> Self {
232 self.metadata.correlation = correlation;
233 self
234 }
235
236 /// Sets the business correlation id.
237 pub fn correlation_id(mut self, id: CorrelationId) -> Self {
238 self.metadata.correlation.correlation_id = Some(id);
239 self
240 }
241
242 /// Joins an existing conversation — typically the causing message's own `conversation_id`.
243 /// [`Self::build`] keeps this value as long as nothing later replaces it with
244 /// [`Self::metadata`] or [`Self::correlation`] (setter order matters only in that sense: the
245 /// last write to `conversation_id` wins, same as any other field).
246 pub fn conversation(mut self, id: ConversationId) -> Self {
247 self.metadata.correlation.conversation_id = id;
248 self
249 }
250
251 /// Records the message that caused this one.
252 pub fn causation(mut self, parent: MessageId) -> Self {
253 self.metadata.correlation.causation_id = Some(parent);
254 self
255 }
256
257 /// Sets the owning tenant.
258 pub fn tenant(mut self, tenant_id: impl Into<String>) -> Self {
259 self.metadata.tenant_id = Some(tenant_id.into());
260 self
261 }
262
263 /// Sets the time after which the message SHALL NOT be published (§12.2).
264 pub fn expires_at(mut self, at: time::OffsetDateTime) -> Self {
265 self.metadata.delivery.expires_at = Some(at);
266 self
267 }
268
269 /// Sets the W3C Trace Context to carry verbatim. Reliar never invents or re-derives it
270 /// (ADR 0004, ADR 0020).
271 pub fn trace(mut self, traceparent: impl Into<String>, tracestate: Option<String>) -> Self {
272 self.metadata.trace.traceparent = Some(traceparent.into());
273 self.metadata.trace.tracestate = tracestate;
274 self
275 }
276
277 /// Sets one custom header. Returns `Err` if `k` uses the reserved `reliar-` prefix or
278 /// breaches a cap (see [`Headers::insert`]).
279 ///
280 /// # Errors
281 ///
282 /// Returns [`HeaderError`] under the same conditions as [`Headers::insert`].
283 pub fn header(
284 mut self,
285 k: impl Into<String>,
286 v: impl Into<String>,
287 ) -> Result<Self, HeaderError> {
288 self.headers
289 .get_or_insert_with(Headers::default)
290 .insert(k, v)?;
291 Ok(self)
292 }
293
294 /// Builds the envelope. `message_type` is `MessageType::of::<T>()`. `conversation_id`:
295 /// **iff** it is still [`crate::ConversationId::UNSET`], it becomes this envelope's own id
296 /// (an un-correlated message roots its own conversation); any other value — set via
297 /// [`Self::conversation`], [`Self::correlation`], or [`Self::metadata`] — is kept verbatim.
298 /// Rooting is decided by the value alone, never by which setter was called or in what order
299 /// (ADR 0011).
300 #[must_use]
301 pub fn build(mut self) -> Envelope<T> {
302 let id = self.id.unwrap_or_default();
303 if self.metadata.correlation.conversation_id.is_unset() {
304 self.metadata.correlation.conversation_id = ConversationId::from_uuid(id.as_uuid());
305 }
306 Envelope {
307 id,
308 message_type: MessageType::of::<T>(),
309 body: self.body,
310 metadata: self.metadata,
311 headers: self.headers,
312 }
313 }
314}