platform_core/
envelope.rs

1//
2// Copyright 2018-2026 Accenture Technology
3//
4// Licensed under the Apache License, Version 2.0 (the "License");
5// you may not use this file except in compliance with the License.
6// You may obtain a copy of the License at
7//
8//     http://www.apache.org/licenses/LICENSE-2.0
9//
10// Unless required by applicable law or agreed to in writing, software
11// distributed under the License is distributed on an "AS IS" BASIS,
12// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13// See the License for the specific language governing permissions and
14// limitations under the License.
15//
16
17//! Rust port of the Java `EventEnvelope`
18//! (`org.platformlambda.core.models.EventEnvelope`) — the immutable message
19//! container between composable functions.
20//!
21//! Three parts, as in Java: **metadata** (routing, correlation, tracing, status,
22//! timing), **headers** (`String → String`), and a dynamic **body**
23//! (`rmpv::Value` — the analog of Java's untyped `Object` payload).
24//!
25//! Wire format: **idiomatic serde MsgPack** (design D4) — deliberately *not*
26//! byte-compatible with Java's compact flag-keyed encoding, since cross-JVM
27//! interop is out of scope. Later fields (`tags`, `annotations`, `span_id`,
28//! serialized exceptions) arrive with the increments that need them.
29
30use std::collections::HashMap;
31
32use serde::de::DeserializeOwned;
33use serde::{Deserialize, Serialize};
34
35use crate::function::AppError;
36
37/// The immutable message container between functions. Build with the fluent
38/// setters (`EventEnvelope::new().set_to("v1.echo").set_body(...)`).
39/// Wire format (increment 59): the **standard event envelope wire format** —
40/// one MsgPack map with these descriptive string keys, shared verbatim with
41/// the Java engine for Event over HTTP (normative spec:
42/// `docs/guides/event-envelope-wire-format.md` in the Java repo; golden
43/// vectors under `tests/resources/envelope-vectors/`). Encoders emit `id` and
44/// `headers` always and other fields only when set; decoders treat absent and
45/// nil identically and ignore unknown keys (Java may add `tags`, `stack`,
46/// `obj_type`, `exception`).
47#[derive(Clone, Debug, Serialize, Deserialize)]
48pub struct EventEnvelope {
49    id: String,
50    #[serde(default, skip_serializing_if = "Option::is_none")]
51    to: Option<String>,
52    #[serde(default, skip_serializing_if = "Option::is_none")]
53    from: Option<String>,
54    #[serde(default, skip_serializing_if = "Option::is_none")]
55    reply_to: Option<String>,
56    #[serde(default, skip_serializing_if = "Option::is_none")]
57    cid: Option<String>,
58    #[serde(default, skip_serializing_if = "Option::is_none")]
59    trace_id: Option<String>,
60    #[serde(default, skip_serializing_if = "Option::is_none")]
61    trace_path: Option<String>,
62    /// The sender's OTel span id, carried so the receiver knows its own
63    /// parent span (Java parity — the `s` flag on the wire).
64    #[serde(default, skip_serializing_if = "Option::is_none")]
65    span_id: Option<String>,
66    #[serde(default, skip_serializing_if = "Option::is_none")]
67    status: Option<i32>,
68    headers: HashMap<String, String>,
69    /// Java omits an unset body on the wire, so absent decodes as `Nil`.
70    #[serde(default = "nil_value", skip_serializing_if = "is_nil")]
71    body: rmpv::Value,
72    #[serde(default, skip_serializing_if = "Option::is_none")]
73    exec_time: Option<f32>,
74    /// RPC round-trip milliseconds (Java `roundTrip`) — carried for the wire
75    /// format; stamped by callers that measure it.
76    #[serde(default, skip_serializing_if = "Option::is_none")]
77    round_trip: Option<f32>,
78    /// Trace annotations riding a REPLY envelope (Java `annotations`): a
79    /// worker attaches the function's `annotate_trace` key-values to its
80    /// response, and the RPC caller folds them into the `round_trip` trace
81    /// record (then strips them — user code never sees them). Same key on the
82    /// wire as the Java standard format, so annotations survive an
83    /// Event-over-HTTP hop in either language direction.
84    #[serde(default, skip_serializing_if = "HashMap::is_empty")]
85    annotations: HashMap<String, rmpv::Value>,
86    /// Engine-managed tags (Java `tags`): reserved key-values visible to the
87    /// engine only — the worker scrubs them from the function's view at
88    /// delivery. Carries e.g. the business correlation-id
89    /// ([`crate::post_office::BUSINESS_CID_TAG`]) across touch points and the
90    /// Event-over-HTTP wire. Same key on the wire as the Java standard
91    /// format — metadata is never transported as envelope headers.
92    #[serde(default, skip_serializing_if = "HashMap::is_empty")]
93    tags: HashMap<String, String>,
94}
95
96fn nil_value() -> rmpv::Value {
97    rmpv::Value::Nil
98}
99
100fn is_nil(value: &rmpv::Value) -> bool {
101    matches!(value, rmpv::Value::Nil)
102}
103
104impl Default for EventEnvelope {
105    fn default() -> Self {
106        EventEnvelope {
107            id: uuid::Uuid::new_v4().simple().to_string(),
108            to: None,
109            from: None,
110            reply_to: None,
111            cid: None,
112            trace_id: None,
113            trace_path: None,
114            span_id: None,
115            status: None,
116            headers: HashMap::new(),
117            body: rmpv::Value::Nil,
118            exec_time: None,
119            round_trip: None,
120            annotations: HashMap::new(),
121            tags: HashMap::new(),
122        }
123    }
124}
125
126impl EventEnvelope {
127    pub fn new() -> Self {
128        Self::default()
129    }
130
131    // ---- fluent builders (Java setX chaining) ----
132
133    pub fn set_to(mut self, route: &str) -> Self {
134        self.to = Some(route.to_string());
135        self
136    }
137
138    pub fn set_from(mut self, route: &str) -> Self {
139        self.from = Some(route.to_string());
140        self
141    }
142
143    pub fn set_reply_to(mut self, route: &str) -> Self {
144        self.reply_to = Some(route.to_string());
145        self
146    }
147
148    /// Remove the reply-to address (Java `setReplyTo(null)`) — used when an
149    /// event is re-targeted, e.g. the declarative Event-over-HTTP forward
150    /// nulls the local callback route before the envelope crosses the wire.
151    pub fn clear_reply_to(mut self) -> Self {
152        self.reply_to = None;
153        self
154    }
155
156    /// Remove the routing address (Java `setTo(null)`) — used when an envelope
157    /// is serialized for a wire that rewrites addressing on the consuming side,
158    /// e.g. the envelope-mode streaming frames, so a server-internal route name
159    /// never leaks.
160    pub fn clear_to(mut self) -> Self {
161        self.to = None;
162        self
163    }
164
165    pub fn set_correlation_id(mut self, cid: &str) -> Self {
166        self.cid = Some(cid.to_string());
167        self
168    }
169
170    pub fn set_trace(mut self, trace_id: &str, trace_path: &str) -> Self {
171        self.trace_id = Some(trace_id.to_string());
172        self.trace_path = Some(trace_path.to_string());
173        self
174    }
175
176    /// Carry a span id on the envelope — the sender's span, which the receiver
177    /// adopts as its `parent_span_id` (OTel lineage).
178    pub fn set_span_id(mut self, span_id: &str) -> Self {
179        self.span_id = Some(span_id.to_string());
180        self
181    }
182
183    pub fn set_status(mut self, status: i32) -> Self {
184        self.status = Some(status);
185        self
186    }
187
188    pub fn set_header(mut self, key: &str, value: &str) -> Self {
189        // Java setHeader guarantees CR/LF never enter a header value
190        // (header-injection guard); same filter here
191        let value: String = value.chars().filter(|c| *c != '\r' && *c != '\n').collect();
192        self.headers.insert(key.to_string(), value);
193        self
194    }
195
196    /// Serialize any `Serialize` value into the dynamic body
197    /// (the analog of Java's `setBody(Object)`).
198    pub fn set_body<T: Serialize>(mut self, value: T) -> Result<Self, AppError> {
199        self.body = rmpv::ext::to_value(value)
200            .map_err(|e| AppError::new(500, format!("unable to serialize body: {e}")))?;
201        Ok(self)
202    }
203
204    /// Set the body from an already-dynamic value.
205    pub fn set_raw_body(mut self, value: rmpv::Value) -> Self {
206        self.body = value;
207        self
208    }
209
210    // ---- getters ----
211
212    pub fn id(&self) -> &str {
213        &self.id
214    }
215
216    pub fn to(&self) -> Option<&str> {
217        self.to.as_deref()
218    }
219
220    pub fn from(&self) -> Option<&str> {
221        self.from.as_deref()
222    }
223
224    pub fn reply_to(&self) -> Option<&str> {
225        self.reply_to.as_deref()
226    }
227
228    pub fn correlation_id(&self) -> Option<&str> {
229        self.cid.as_deref()
230    }
231
232    pub fn trace_id(&self) -> Option<&str> {
233        self.trace_id.as_deref()
234    }
235
236    pub fn trace_path(&self) -> Option<&str> {
237        self.trace_path.as_deref()
238    }
239
240    /// The sender's span id (the receiver's parent span).
241    pub fn span_id(&self) -> Option<&str> {
242        self.span_id.as_deref()
243    }
244
245    /// HTTP-style status; unset means 200 (Java `getStatus`).
246    pub fn status(&self) -> i32 {
247        self.status.unwrap_or(200)
248    }
249
250    /// An error condition is a status code >= 400 (Java `hasError`).
251    pub fn has_error(&self) -> bool {
252        self.status() >= 400
253    }
254
255    pub fn headers(&self) -> &HashMap<String, String> {
256        &self.headers
257    }
258
259    pub fn header(&self, key: &str) -> Option<&str> {
260        // Java getHeader falls back to a case-insensitive scan when the
261        // exact key is absent
262        if let Some(value) = self.headers.get(key) {
263            return Some(value.as_str());
264        }
265        self.headers
266            .iter()
267            .find(|(name, _)| name.eq_ignore_ascii_case(key))
268            .map(|(_, value)| value.as_str())
269    }
270
271    pub fn body(&self) -> &rmpv::Value {
272        &self.body
273    }
274
275    /// Deserialize the dynamic body into a concrete type
276    /// (the analog of Java's `getBody(Class)`).
277    pub fn body_as<T: DeserializeOwned>(&self) -> Result<T, AppError> {
278        rmpv::ext::from_value(self.body.clone())
279            .map_err(|e| AppError::new(500, format!("unable to deserialize body: {e}")))
280    }
281
282    /// Function execution time in milliseconds, when stamped by a worker.
283    pub fn exec_time(&self) -> Option<f32> {
284        self.exec_time
285    }
286
287    /// RPC round-trip milliseconds (Java `getRoundTrip`), when measured.
288    pub fn round_trip(&self) -> Option<f32> {
289        self.round_trip
290    }
291
292    /// Stamp the RPC round-trip time (Java parity: the requester measures
293    /// the full request/response cycle).
294    pub fn set_round_trip(mut self, ms: f32) -> Self {
295        self.round_trip = Some(ms);
296        self
297    }
298
299    /// Trace annotations riding this (reply) envelope (Java `getAnnotations`).
300    pub fn annotations(&self) -> &HashMap<String, rmpv::Value> {
301        &self.annotations
302    }
303
304    /// Remove all annotations (Java `clearAnnotations`) — the RPC caller
305    /// strips them after folding them into the trace record.
306    pub fn clear_annotations(mut self) -> Self {
307        self.annotations.clear();
308        self
309    }
310
311    /// An engine-managed tag value (Java `getTag`).
312    pub fn tag(&self, key: &str) -> Option<&str> {
313        self.tags.get(key).map(String::as_str)
314    }
315
316    /// Attach an engine-managed tag (Java `addTag`). Reserved for the engine:
317    /// tags never reach a user function's view — the worker scrubs them at
318    /// delivery after extracting what it injects into the header copy.
319    pub fn add_tag(mut self, key: &str, value: &str) -> Self {
320        self.tags.insert(key.to_string(), value.to_string());
321        self
322    }
323
324    // ---- crate-internal mutators (worker bookkeeping) ----
325
326    pub(crate) fn set_body_internal(&mut self, body: rmpv::Value) {
327        self.body = body;
328    }
329
330    pub(crate) fn set_cid_internal(&mut self, cid: Option<String>) {
331        self.cid = cid;
332    }
333
334    pub(crate) fn set_from_internal(&mut self, from: &str) {
335        self.from = Some(from.to_string());
336    }
337
338    pub(crate) fn set_to_internal(&mut self, to: &str) {
339        self.to = Some(to.to_string());
340    }
341
342    pub(crate) fn set_exec_time_internal(&mut self, ms: f32) {
343        self.exec_time = Some(ms);
344    }
345
346    pub(crate) fn set_trace_internal(&mut self, trace_id: &str, trace_path: &str) {
347        self.trace_id = Some(trace_id.to_string());
348        self.trace_path = Some(trace_path.to_string());
349    }
350
351    pub(crate) fn set_span_id_internal(&mut self, span_id: &str) {
352        self.span_id = Some(span_id.to_string());
353    }
354
355    pub(crate) fn clear_span_id_internal(&mut self) {
356        self.span_id = None;
357    }
358
359    pub(crate) fn set_annotations_internal(&mut self, annotations: HashMap<String, rmpv::Value>) {
360        self.annotations = annotations;
361    }
362
363    pub(crate) fn clear_annotations_internal(&mut self) {
364        self.annotations.clear();
365    }
366
367    pub(crate) fn clear_tags_internal(&mut self) {
368        self.tags.clear();
369    }
370
371    /// Remove a header, returning its value (worker-entry scrubbing of
372    /// engine-internal / legacy metadata keys from the function's view).
373    pub(crate) fn remove_header_internal(&mut self, key: &str) -> Option<String> {
374        self.headers.remove(key)
375    }
376
377    // ---- wire format ----
378
379    /// Encode the envelope as MsgPack bytes (idiomatic serde — design D4).
380    ///
381    /// The body's `Nil` map entries are omitted unless `serializer.null.transport`
382    /// is `true` — the Rust mirror of Java `MsgPack.packMap`'s null-skip. Since
383    /// increment 58 (the F2 decision) the same strip also runs explicitly on the
384    /// in-memory fast path (`platform::normalize_null_transport`), so delivery
385    /// semantics are deterministic on every hop — here it is normally a no-op.
386    /// The clone + strip runs **only** when the body actually carries a
387    /// strippable `Nil` (`has_nil_map_entry`); otherwise — a scalar body, a
388    /// structured body with no nulls, or transport on — `self` encodes directly
389    /// with no extra allocation, so the common case pays only a read-only scan.
390    pub fn to_bytes(&self) -> Result<Vec<u8>, AppError> {
391        if crate::serializer::null_transport() || !crate::serializer::has_nil_map_entry(&self.body)
392        {
393            return rmp_serde::to_vec_named(self)
394                .map_err(|e| AppError::new(500, format!("unable to encode envelope: {e}")));
395        }
396        let mut stripped = self.clone();
397        stripped.body = crate::serializer::strip_nulls_always(&self.body);
398        rmp_serde::to_vec_named(&stripped)
399            .map_err(|e| AppError::new(500, format!("unable to encode envelope: {e}")))
400    }
401
402    /// Decode an envelope from MsgPack bytes.
403    pub fn from_bytes(bytes: &[u8]) -> Result<Self, AppError> {
404        rmp_serde::from_slice(bytes)
405            .map_err(|e| AppError::new(500, format!("unable to decode envelope: {e}")))
406    }
407}
platform_core/envelope.rs

platform_core/
envelope.rs
