reliar_core/
metadata.rs

1//! Canonical, typed framework metadata (SRS §12, ADR 0003, ADR 0004).
2
3use core::fmt;
4
5use crate::{
6    ContentType, ConversationId, CorrelationId, MessageId, RequestId,
7    ids::{IdError, contains_control_char},
8};
9
10/// Canonical, typed framework metadata: the single source of truth. No value here is ever
11/// duplicated into [`Headers`](crate::Headers) (ADR 0004).
12#[derive(Clone, Debug, Default, PartialEq)]
13#[non_exhaustive]
14pub struct Metadata {
15    /// Correlation and conversation identity.
16    pub correlation: CorrelationMetadata,
17    /// W3C Trace Context, carried verbatim.
18    pub trace: TraceContext,
19    /// Transport-independent routing hints.
20    pub routing: RoutingMetadata,
21    /// Serialization and delivery hints.
22    pub delivery: DeliveryMetadata,
23    /// The owning tenant, if this deployment is multi-tenant.
24    pub tenant_id: Option<String>,
25}
26
27/// Correlation and conversation identity for one envelope.
28#[derive(Clone, Debug, PartialEq)]
29#[non_exhaustive]
30pub struct CorrelationMetadata {
31    /// Application/business workflow correlation, set by the caller.
32    pub correlation_id: Option<CorrelationId>,
33    /// Groups every message in one business conversation.
34    pub conversation_id: ConversationId,
35    /// The message that directly caused this one.
36    pub causation_id: Option<MessageId>,
37    /// The inbound request that (transitively) caused this one.
38    pub request_id: Option<RequestId>,
39}
40
41/// Sets `conversation_id` to the [`ConversationId::UNSET`] sentinel (the nil UUID) — **not** a
42/// fresh mint — so [`crate::EnvelopeBuilder::build`] can tell "not yet rooted" from a genuinely
43/// chosen value by comparing it, not by tracking which builder setter was called.
44/// `build` replaces `UNSET` with the envelope's own id, so an un-correlated message is the root
45/// of its own conversation, and leaves any other value alone. A `Metadata` that never passes
46/// through the builder (e.g. read straight off `Default`) keeps the placeholder verbatim.
47impl Default for CorrelationMetadata {
48    fn default() -> Self {
49        Self {
50            correlation_id: None,
51            conversation_id: ConversationId::UNSET,
52            causation_id: None,
53            request_id: None,
54        }
55    }
56}
57
58/// W3C Trace Context, carried verbatim. Reliar never invents or re-derives it (ADR 0004,
59/// ADR 0020): a transport mapper writes these from an active span and reads them back on
60/// decode.
61#[derive(Clone, Debug, Default, PartialEq, Eq)]
62#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
63#[non_exhaustive]
64pub struct TraceContext {
65    /// The W3C `traceparent` header value.
66    pub traceparent: Option<String>,
67    /// The W3C `tracestate` header value.
68    pub tracestate: Option<String>,
69}
70
71/// Transport-independent routing only. Kafka partition keys, `RabbitMQ` exchanges and NATS
72/// subject options are transport concepts and SHALL NOT appear here (§12).
73#[derive(Clone, Debug, Default, PartialEq, Eq)]
74#[non_exhaustive]
75pub struct RoutingMetadata {
76    /// The logical origin of this message.
77    pub source: Option<EndpointAddress>,
78    /// The logical destination of this message.
79    pub destination: Option<EndpointAddress>,
80    /// Where a reply to this message should be sent.
81    pub reply_to: Option<EndpointAddress>,
82}
83
84/// An opaque, transport-interpreted address string (a queue name, a subject, a service name —
85/// Reliar does not care which). Capped at [`Self::MAX_LEN`] bytes.
86#[derive(Clone, Debug, PartialEq, Eq, Hash)]
87pub struct EndpointAddress(String);
88
89impl EndpointAddress {
90    /// Maximum length in bytes.
91    pub const MAX_LEN: usize = 256;
92
93    /// Validates and wraps an endpoint address. Returns `Err` for an empty string, one
94    /// containing a control character (including CR/LF — a header-injection surface), or one
95    /// over [`Self::MAX_LEN`] bytes.
96    ///
97    /// # Errors
98    ///
99    /// Returns [`IdError::Empty`], [`IdError::ControlCharacter`], or [`IdError::TooLong`].
100    pub fn parse(s: impl Into<String>) -> Result<Self, IdError> {
101        let s = s.into();
102        if s.is_empty() {
103            return Err(IdError::Empty);
104        }
105        if contains_control_char(&s) {
106            return Err(IdError::ControlCharacter);
107        }
108        if s.len() > Self::MAX_LEN {
109            return Err(IdError::TooLong {
110                len: s.len(),
111                max: Self::MAX_LEN,
112            });
113        }
114        Ok(Self(s))
115    }
116
117    /// Returns the address as a string slice.
118    #[must_use]
119    pub fn as_str(&self) -> &str {
120        &self.0
121    }
122}
123
124impl fmt::Display for EndpointAddress {
125    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
126        f.write_str(&self.0)
127    }
128}
129
130#[cfg(feature = "serde")]
131impl serde::Serialize for EndpointAddress {
132    fn serialize<S: serde::Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
133        s.collect_str(&self.0)
134    }
135}
136
137#[cfg(feature = "serde")]
138impl<'de> serde::Deserialize<'de> for EndpointAddress {
139    fn deserialize<D: serde::Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
140        let raw = String::deserialize(d)?;
141        Self::parse(raw).map_err(serde::de::Error::custom)
142    }
143}
144
145/// Serialization and delivery hints for one envelope.
146#[derive(Clone, Debug, PartialEq)]
147#[non_exhaustive]
148pub struct DeliveryMetadata {
149    /// **Authoritatively set by the store at enqueue** from `Serializer::content_type()`, and
150    /// read back from the provider's `content_type` column on rehydration. The `Default` value
151    /// below is a placeholder a call site never chooses (ADR 0010).
152    pub content_type: ContentType,
153    /// When the application handed this message to Reliar (app clock; never compared against a
154    /// DB timestamp).
155    pub sent_at: Option<time::OffsetDateTime>,
156    /// The time after which this message SHALL NOT be published. Enforced in DB time by a
157    /// provider's claim predicate; an expired pending row goes dead with
158    /// `DeadReason::Expired` and consumes no retry attempt (§12.2).
159    pub expires_at: Option<time::OffsetDateTime>,
160    /// A transport mapper's broker-specific dedup key (falling back to the message id). Reliar
161    /// never deduplicates on it in the database (§12.3).
162    pub deduplication_id: Option<String>,
163}
164
165impl Default for DeliveryMetadata {
166    fn default() -> Self {
167        Self {
168            content_type: ContentType::JSON,
169            sent_at: None,
170            expires_at: None,
171            deduplication_id: None,
172        }
173    }
174}
175
176#[cfg(feature = "serde")]
177mod serde_impls {
178    //! `Serialize`/`Deserialize` for hosts that want to persist or log `Metadata` themselves.
179    //! Unrelated to `reliar-store-postgres`'s own private JSONB persistence contract (ADR 0012),
180    //! which defines its own `MetadataRest` shape with its own forward-compatibility rules.
181
182    use serde::{Deserialize, Serialize};
183
184    use super::{CorrelationMetadata, DeliveryMetadata, Metadata, RoutingMetadata, TraceContext};
185
186    // `#[serde(default)]` on every field: a persisted blob missing a whole sub-struct (an
187    // 0.2 field addition, or a value written before it existed) still deserializes, falling
188    // back to that sub-struct's own `Default` (§43.A, ADR 0012's sibling contract for
189    // `reliar-core`'s own optional `Metadata` serde).
190    #[derive(Serialize, Deserialize)]
191    #[serde(remote = "Metadata")]
192    struct MetadataDef {
193        #[serde(default)]
194        correlation: CorrelationMetadata,
195        #[serde(default)]
196        trace: TraceContext,
197        #[serde(default)]
198        routing: RoutingMetadata,
199        #[serde(default)]
200        delivery: DeliveryMetadata,
201        #[serde(default)]
202        tenant_id: Option<String>,
203    }
204
205    impl Serialize for Metadata {
206        fn serialize<S: serde::Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
207            MetadataDef::serialize(self, s)
208        }
209    }
210    impl<'de> Deserialize<'de> for Metadata {
211        fn deserialize<D: serde::Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
212            MetadataDef::deserialize(d)
213        }
214    }
215
216    impl Serialize for CorrelationMetadata {
217        fn serialize<S: serde::Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
218            // Field names mirror `CorrelationMetadata` on purpose, for wire compatibility.
219            #[derive(Serialize)]
220            #[allow(clippy::struct_field_names)]
221            struct Def<'a> {
222                correlation_id: &'a Option<super::CorrelationId>,
223                conversation_id: &'a super::ConversationId,
224                causation_id: &'a Option<super::MessageId>,
225                request_id: &'a Option<super::RequestId>,
226            }
227            Def {
228                correlation_id: &self.correlation_id,
229                conversation_id: &self.conversation_id,
230                causation_id: &self.causation_id,
231                request_id: &self.request_id,
232            }
233            .serialize(s)
234        }
235    }
236    impl<'de> Deserialize<'de> for CorrelationMetadata {
237        fn deserialize<D: serde::Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
238            // `ConversationId`'s own `Default` mints a **fresh** `UUIDv7` (it has no notion of
239            // "not yet rooted"), so a bare `#[serde(default)]` here would silently disagree with
240            // `CorrelationMetadata::default()`, which uses the `UNSET` sentinel. A blob missing
241            // `conversation_id` (e.g. written before it existed) must fall back to the same
242            // sentinel, not a random id.
243            fn default_conversation_id() -> super::ConversationId {
244                super::ConversationId::UNSET
245            }
246
247            #[derive(Deserialize)]
248            #[allow(clippy::struct_field_names)]
249            struct Def {
250                #[serde(default)]
251                correlation_id: Option<super::CorrelationId>,
252                #[serde(default = "default_conversation_id")]
253                conversation_id: super::ConversationId,
254                #[serde(default)]
255                causation_id: Option<super::MessageId>,
256                #[serde(default)]
257                request_id: Option<super::RequestId>,
258            }
259            let def = Def::deserialize(d)?;
260            Ok(CorrelationMetadata {
261                correlation_id: def.correlation_id,
262                conversation_id: def.conversation_id,
263                causation_id: def.causation_id,
264                request_id: def.request_id,
265            })
266        }
267    }
268
269    impl Serialize for RoutingMetadata {
270        fn serialize<S: serde::Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
271            #[derive(Serialize)]
272            struct Def<'a> {
273                source: &'a Option<super::EndpointAddress>,
274                destination: &'a Option<super::EndpointAddress>,
275                reply_to: &'a Option<super::EndpointAddress>,
276            }
277            Def {
278                source: &self.source,
279                destination: &self.destination,
280                reply_to: &self.reply_to,
281            }
282            .serialize(s)
283        }
284    }
285    impl<'de> Deserialize<'de> for RoutingMetadata {
286        fn deserialize<D: serde::Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
287            #[derive(Deserialize)]
288            struct Def {
289                #[serde(default)]
290                source: Option<super::EndpointAddress>,
291                #[serde(default)]
292                destination: Option<super::EndpointAddress>,
293                #[serde(default)]
294                reply_to: Option<super::EndpointAddress>,
295            }
296            let def = Def::deserialize(d)?;
297            Ok(RoutingMetadata {
298                source: def.source,
299                destination: def.destination,
300                reply_to: def.reply_to,
301            })
302        }
303    }
304
305    impl Serialize for DeliveryMetadata {
306        fn serialize<S: serde::Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
307            #[derive(Serialize)]
308            struct Def<'a> {
309                content_type: &'a super::ContentType,
310                #[serde(with = "time::serde::rfc3339::option")]
311                sent_at: &'a Option<time::OffsetDateTime>,
312                #[serde(with = "time::serde::rfc3339::option")]
313                expires_at: &'a Option<time::OffsetDateTime>,
314                deduplication_id: &'a Option<String>,
315            }
316            Def {
317                content_type: &self.content_type,
318                sent_at: &self.sent_at,
319                expires_at: &self.expires_at,
320                deduplication_id: &self.deduplication_id,
321            }
322            .serialize(s)
323        }
324    }
325    impl<'de> Deserialize<'de> for DeliveryMetadata {
326        fn deserialize<D: serde::Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
327            // `ContentType` has no public `Default` (ADR 0010: a call site never chooses one),
328            // so its missing-field fallback is this private fn rather than a bare
329            // `#[serde(default)]` — it mirrors `DeliveryMetadata::default()`'s own placeholder.
330            fn default_content_type() -> super::ContentType {
331                super::ContentType::JSON
332            }
333
334            #[derive(Deserialize)]
335            struct Def {
336                #[serde(default = "default_content_type")]
337                content_type: super::ContentType,
338                #[serde(default, with = "time::serde::rfc3339::option")]
339                sent_at: Option<time::OffsetDateTime>,
340                #[serde(default, with = "time::serde::rfc3339::option")]
341                expires_at: Option<time::OffsetDateTime>,
342                #[serde(default)]
343                deduplication_id: Option<String>,
344            }
345            let def = Def::deserialize(d)?;
346            Ok(DeliveryMetadata {
347                content_type: def.content_type,
348                sent_at: def.sent_at,
349                expires_at: def.expires_at,
350                deduplication_id: def.deduplication_id,
351            })
352        }
353    }
354}
reliar_core/metadata.rs

reliar_core/
metadata.rs
