liminal/
envelope.rs

1use chrono::{DateTime, Utc};
2
3use crate::causal::{CausalContext, MessageId};
4use crate::channel::SchemaId;
5
6/// Identity of the publisher that submitted a message.
7#[derive(Clone, Debug, PartialEq, Eq, Hash)]
8pub struct PublisherId(String);
9
10impl PublisherId {
11    /// Creates a publisher identifier from a string-like value.
12    #[must_use]
13    pub fn new(value: impl Into<String>) -> Self {
14        Self(value.into())
15    }
16
17    /// Returns the publisher identifier as a string slice.
18    #[must_use]
19    pub fn as_str(&self) -> &str {
20        &self.0
21    }
22}
23
24impl From<&str> for PublisherId {
25    fn from(value: &str) -> Self {
26        Self::new(value)
27    }
28}
29
30impl From<String> for PublisherId {
31    fn from(value: String) -> Self {
32        Self::new(value)
33    }
34}
35
36impl Default for PublisherId {
37    fn default() -> Self {
38        Self::new("anonymous")
39    }
40}
41
42/// Message envelope used as the delivery unit inside the core bus.
43#[derive(Clone, Debug, PartialEq, Eq)]
44pub struct Envelope {
45    /// Live transport provenance; it expires with the transport incarnation.
46    pub delivery_id: Option<crate::protocol::DeliveredId>,
47    /// UUID-based unique identifier assigned when the message is published.
48    pub message_id: MessageId,
49    /// Validated payload bytes, normalized by the schema when defaults are applied.
50    pub payload: Vec<u8>,
51    /// Optional parent reference for causal-chain metadata.
52    pub causal_context: Option<CausalContext>,
53    /// Schema version that validated this payload.
54    pub schema_id: SchemaId,
55    /// Publisher identity attached at publish time.
56    pub publisher_id: PublisherId,
57    /// UTC timestamp captured when the message was published.
58    pub timestamp: DateTime<Utc>,
59}
60
61impl Envelope {
62    /// Creates an envelope with the current UTC publish timestamp.
63    #[must_use]
64    pub fn new(
65        payload: Vec<u8>,
66        causal_context: Option<CausalContext>,
67        schema_id: SchemaId,
68        publisher_id: PublisherId,
69    ) -> Self {
70        Self::with_timestamp(payload, causal_context, schema_id, publisher_id, Utc::now())
71    }
72
73    /// Creates an envelope with an explicit timestamp.
74    #[must_use]
75    pub fn with_timestamp(
76        payload: Vec<u8>,
77        causal_context: Option<CausalContext>,
78        schema_id: SchemaId,
79        publisher_id: PublisherId,
80        timestamp: DateTime<Utc>,
81    ) -> Self {
82        Self::with_message_id_and_timestamp(
83            MessageId::new(),
84            payload,
85            causal_context,
86            schema_id,
87            publisher_id,
88            timestamp,
89        )
90    }
91
92    /// Creates an envelope with an explicit message identifier and current UTC timestamp.
93    #[must_use]
94    pub fn with_message_id(
95        message_id: MessageId,
96        payload: Vec<u8>,
97        causal_context: Option<CausalContext>,
98        schema_id: SchemaId,
99        publisher_id: PublisherId,
100    ) -> Self {
101        Self::with_message_id_and_timestamp(
102            message_id,
103            payload,
104            causal_context,
105            schema_id,
106            publisher_id,
107            Utc::now(),
108        )
109    }
110
111    /// Creates an envelope with explicit message identifier and timestamp.
112    #[must_use]
113    pub const fn with_message_id_and_timestamp(
114        message_id: MessageId,
115        payload: Vec<u8>,
116        causal_context: Option<CausalContext>,
117        schema_id: SchemaId,
118        publisher_id: PublisherId,
119        timestamp: DateTime<Utc>,
120    ) -> Self {
121        Self {
122            delivery_id: None,
123            message_id,
124            payload,
125            causal_context,
126            schema_id,
127            publisher_id,
128            timestamp,
129        }
130    }
131}
132
133#[cfg(test)]
134mod tests {
135    use chrono::Utc;
136
137    use super::{Envelope, PublisherId};
138    use crate::causal::{CausalContext, MessageId};
139    use crate::channel::SchemaId;
140
141    #[test]
142    fn envelope_carries_required_fields() {
143        let schema_id = SchemaId::new();
144        let publisher_id = PublisherId::from("publisher-1");
145        let parent = MessageId::new();
146        let causal_context = Some(CausalContext::child_of(parent));
147        let timestamp = fixed_timestamp();
148        let message_id = MessageId::new();
149
150        let envelope = Envelope::with_message_id_and_timestamp(
151            message_id,
152            b"{}".to_vec(),
153            causal_context.clone(),
154            schema_id,
155            publisher_id.clone(),
156            timestamp,
157        );
158
159        assert_eq!(envelope.message_id, message_id);
160        assert_eq!(envelope.payload, b"{}".to_vec());
161        assert_eq!(envelope.causal_context, causal_context);
162        assert_eq!(envelope.schema_id, schema_id);
163        assert_eq!(envelope.publisher_id, publisher_id);
164        assert_eq!(envelope.timestamp, timestamp);
165    }
166
167    #[test]
168    fn envelope_assigns_unique_message_ids() {
169        let first = Envelope::new(vec![], None, SchemaId::new(), PublisherId::default());
170        let second = Envelope::new(vec![], None, SchemaId::new(), PublisherId::default());
171
172        assert_ne!(first.message_id, second.message_id);
173    }
174
175    fn fixed_timestamp() -> chrono::DateTime<Utc> {
176        Utc::now()
177    }
178}
liminal/envelope.rs

liminal/
envelope.rs
