Skip to main content

dataflow_rs/engine/
message.rs

1use crate::engine::error::{DataflowError, ErrorInfo};
2use chrono::{DateTime, Utc};
3use datavalue::OwnedDataValue;
4use serde::{Deserialize, Serialize};
5use serde_json::Value as JsonValue;
6use std::sync::Arc;
7use uuid::Uuid;
8
9/// A message flowing through the dataflow engine.
10///
11/// Construct via [`Message::builder`] for the full API, or use the shortcuts
12/// [`Message::new`] (already-owned `Arc<OwnedDataValue>` payload — the perf
13/// path) and [`Message::from_value`] (bridges from `serde_json::Value`).
14///
15/// `context` is held as an [`OwnedDataValue`] tree (not `serde_json::Value`)
16/// so the JSONLogic evaluator can borrow it into its arena via
17/// `OwnedDataValue::to_arena` with a single deep walk in, and project the
18/// result back via `DataValue::to_owned` with a single deep walk out — no
19/// `serde_json::Value` round-trip in the hot path. The on-the-wire JSON
20/// shape is preserved by datavalue's native `Serialize` / `Deserialize`
21/// impls.
22///
23/// Every other field is encapsulated — read via `id()`, `payload()`,
24/// `audit_trail()`, `errors()`, `capture_changes()`; mutate `errors` via
25/// [`Message::add_error`]; mutate `context` via [`crate::TaskContext::set`].
26/// Direct mutation of `audit_trail` is engine-internal.
27/// Internal id storage for [`Message`]. The default UUID v7 id is written
28/// into a fixed inline buffer (36 hyphenated lowercase ASCII bytes) — no
29/// per-message heap allocation; a caller-supplied id keeps its `String`.
30/// Private repr: every public surface exposes it as `&str` via
31/// [`Message::id`], and the serialized wire shape (a JSON string) is
32/// unchanged.
33#[derive(Clone)]
34pub(crate) enum MessageId {
35    Uuid([u8; 36]),
36    Custom(String),
37}
38
39impl MessageId {
40    /// Generate a fresh UUID v7 id directly into the inline buffer.
41    fn new_uuid_v7() -> Self {
42        let mut buf = [0u8; 36];
43        Uuid::now_v7().hyphenated().encode_lower(&mut buf);
44        Self::Uuid(buf)
45    }
46
47    fn as_str(&self) -> &str {
48        match self {
49            // encode_lower fills all 36 bytes with hyphenated lowercase
50            // ASCII, so the buffer is always valid UTF-8.
51            Self::Uuid(buf) => std::str::from_utf8(buf).expect("uuid ids are ascii"),
52            Self::Custom(s) => s,
53        }
54    }
55}
56
57impl std::fmt::Debug for MessageId {
58    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
59        std::fmt::Debug::fmt(self.as_str(), f)
60    }
61}
62
63#[derive(Debug, Clone)]
64pub struct Message {
65    pub(crate) id: MessageId,
66    pub(crate) payload: Arc<OwnedDataValue>,
67    /// Unified context containing `data`, `metadata`, and `temp_data` keys.
68    /// Always an `OwnedDataValue::Object`; the engine populates the three
69    /// top-level keys at construction. Public for read access (tests do
70    /// `message.context["data"]["x"]` lookups); inside handlers prefer
71    /// [`crate::TaskContext::set`] which records audit-trail changes.
72    pub context: OwnedDataValue,
73    pub(crate) audit_trail: Vec<AuditTrail>,
74    /// Errors that occurred during message processing. Read via
75    /// `errors()`, append via `add_error()`.
76    pub(crate) errors: Vec<ErrorInfo>,
77    /// When `true` (default), built-in functions emit per-write `Change`
78    /// entries into `audit_trail`, capturing `old_value` and `new_value` deep
79    /// clones. When `false`, `AuditTrail` entries are still recorded
80    /// (workflow_id, task_id, status, timestamp) but `changes` is empty —
81    /// the bulk-pipeline fast path. UI debug consumers should leave this at
82    /// `true`. Wire shape is unchanged either way.
83    pub(crate) capture_changes: bool,
84    /// Routing bucket `0..=99` for a workflow traffic split
85    /// ([`crate::Workflow::rollout`]).
86    ///
87    /// Deliberately **not** stored in `context`, so it never appears in `data` /
88    /// `metadata` / `temp_data` and never has to be stripped before the context
89    /// is serialized — that is the point of the field.
90    ///
91    /// Like `capture_changes`, an in-memory hint: the hand-written `Serialize` /
92    /// `Deserialize` impls keep the 5-field wire shape, so the bucket does not
93    /// survive a JSON round trip.
94    pub(crate) routing_bucket: Option<u8>,
95}
96
97// Custom Serialize: stable wire format ({id, payload, context, audit_trail, errors}).
98// `capture_changes` is an in-memory hint only — never serialized.
99impl Serialize for Message {
100    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
101    where
102        S: serde::Serializer,
103    {
104        use serde::ser::SerializeStruct;
105        let mut state = serializer.serialize_struct("Message", 5)?;
106        state.serialize_field("id", &self.id.as_str())?;
107        state.serialize_field("payload", &self.payload)?;
108        state.serialize_field("context", &self.context)?;
109        state.serialize_field("audit_trail", &self.audit_trail)?;
110        state.serialize_field("errors", &self.errors)?;
111        state.end()
112    }
113}
114
115// Custom Deserialize: mirrors the Serialize shape; no cache field to seed.
116// `capture_changes` defaults to `true` for back-compat.
117impl<'de> Deserialize<'de> for Message {
118    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
119    where
120        D: serde::Deserializer<'de>,
121    {
122        #[derive(Deserialize)]
123        struct MessageData {
124            id: String,
125            payload: Arc<OwnedDataValue>,
126            context: OwnedDataValue,
127            audit_trail: Vec<AuditTrail>,
128            errors: Vec<ErrorInfo>,
129        }
130
131        let data = MessageData::deserialize(deserializer)?;
132        Ok(Message {
133            id: MessageId::Custom(data.id),
134            payload: data.payload,
135            context: data.context,
136            audit_trail: data.audit_trail,
137            errors: data.errors,
138            capture_changes: true,
139            routing_bucket: None,
140        })
141    }
142}
143
144impl Message {
145    /// Start building a message. The recommended constructor — chains
146    /// `.id(...)`, `.payload(...)` / `.payload_json(...)`, and
147    /// `.capture_changes(...)` calls, then `.build()`.
148    pub fn builder() -> MessageBuilder {
149        MessageBuilder::new()
150    }
151
152    /// Construct a message from an already-owned payload `Arc`. The perf
153    /// path: zero `serde_json::Value` walk, one Arc refcount bump per
154    /// message. Use this from a hot loop with a payload `Arc` shared across
155    /// messages (e.g. a benchmark harness or an HTTP handler that receives
156    /// already-parsed payloads).
157    pub fn new(payload: Arc<OwnedDataValue>) -> Self {
158        Self {
159            // UUID v7: ms-precision timestamp in the high bits, random tail.
160            // Time-ordered and sortable — better for databases/logs than v4
161            // (random-only) and the same `rng` backend cost. Encoded into the
162            // inline id buffer — no `String` allocation per message.
163            id: MessageId::new_uuid_v7(),
164            payload,
165            context: empty_context(),
166            audit_trail: vec![],
167            errors: vec![],
168            capture_changes: true,
169            routing_bucket: None,
170        }
171    }
172
173    /// Construct a message from a `serde_json::Value` payload. Convenience
174    /// for code that already speaks serde_json; goes through the
175    /// `OwnedDataValue::from(&Value)` bridge (one deep walk).
176    pub fn from_value(payload: &JsonValue) -> Self {
177        Self::new(Arc::new(OwnedDataValue::from(payload)))
178    }
179
180    /// Construct a message from a JSON payload string. Parses with
181    /// `serde_json` and bridges into `OwnedDataValue`. Returns
182    /// `DataflowError::Deserialization` on parse failure.
183    pub fn from_json_str(payload: &str) -> crate::engine::error::Result<Self> {
184        let value: JsonValue = serde_json::from_str(payload).map_err(DataflowError::from_serde)?;
185        Ok(Self::from_value(&value))
186    }
187
188    /// Add an error to the message
189    pub fn add_error(&mut self, error: ErrorInfo) {
190        self.errors.push(error);
191    }
192
193    /// Check if message has errors
194    pub fn has_errors(&self) -> bool {
195        !self.errors.is_empty()
196    }
197
198    /// Message id (UUID v7 string by default; caller-supplied if set via
199    /// [`MessageBuilder::id`]).
200    #[inline]
201    pub fn id(&self) -> &str {
202        self.id.as_str()
203    }
204
205    /// Original payload as the engine received it. Immutable for the
206    /// lifetime of the message — the engine reads it through this Arc and
207    /// copies into `context` only as needed by handlers.
208    #[inline]
209    pub fn payload(&self) -> &OwnedDataValue {
210        &self.payload
211    }
212
213    /// The shared payload `Arc` itself. Useful when forwarding the same
214    /// payload to multiple messages without recloning the underlying
215    /// `OwnedDataValue` tree.
216    #[inline]
217    pub fn payload_arc(&self) -> &Arc<OwnedDataValue> {
218        &self.payload
219    }
220
221    /// Audit-trail entries recorded by the engine, one per task that ran
222    /// (skipped tasks are absent unless `Trace` mode is on).
223    #[inline]
224    pub fn audit_trail(&self) -> &[AuditTrail] {
225        &self.audit_trail
226    }
227
228    /// Errors collected while processing — both validation failures and
229    /// task errors that the workflow swallowed via `continue_on_error`.
230    #[inline]
231    pub fn errors(&self) -> &[ErrorInfo] {
232        &self.errors
233    }
234
235    /// Whether per-write `Change` capture is on. When `false`, audit-trail
236    /// entries are still emitted but their `changes` lists are empty —
237    /// the bulk-pipeline fast path.
238    #[inline]
239    pub fn capture_changes(&self) -> bool {
240        self.capture_changes
241    }
242
243    /// Routing bucket for the traffic-split gate, if one was set.
244    ///
245    /// `None` means every workflow admits this message regardless of its
246    /// [`crate::Workflow::rollout`].
247    #[inline]
248    pub fn routing_bucket(&self) -> Option<u8> {
249        self.routing_bucket
250    }
251
252    /// Get a reference to the `data` field in context. Returns
253    /// `&OwnedDataValue::Null` if missing (matches `serde_json::Value`'s
254    /// `Index` fallback semantics).
255    pub fn data(&self) -> &OwnedDataValue {
256        &self.context["data"]
257    }
258
259    /// Get a reference to the `metadata` field in context.
260    pub fn metadata(&self) -> &OwnedDataValue {
261        &self.context["metadata"]
262    }
263
264    /// Get a reference to the `temp_data` field in context.
265    pub fn temp_data(&self) -> &OwnedDataValue {
266        &self.context["temp_data"]
267    }
268}
269
270/// Builder for [`Message`]. Collapses the historical
271/// `new` / `with_id` / `from_value` / `without_change_capture` four-way
272/// constructor split into a single fluent shape.
273///
274/// ```
275/// use dataflow_rs::Message;
276/// use serde_json::json;
277///
278/// // Minimal: serde_json payload, default UUID id, capture on.
279/// let m = Message::builder()
280///     .payload_json(&json!({"order": {"total": 1500}}))
281///     .build();
282/// assert!(m.id().len() > 0);
283/// assert!(m.capture_changes());
284/// ```
285#[must_use = "MessageBuilder must be `.build()` to produce a Message"]
286#[derive(Default)]
287pub struct MessageBuilder {
288    id: Option<String>,
289    payload: Option<Arc<OwnedDataValue>>,
290    capture_changes: Option<bool>,
291    data: Option<OwnedDataValue>,
292    metadata: Option<OwnedDataValue>,
293    temp_data: Option<OwnedDataValue>,
294    routing_bucket: Option<u8>,
295}
296
297impl MessageBuilder {
298    /// Create an empty builder. Equivalent to [`MessageBuilder::default`].
299    pub fn new() -> Self {
300        Self::default()
301    }
302
303    /// Caller-supplied id (typically a correlation id from upstream).
304    /// Defaults to a freshly-generated UUID v7.
305    pub fn id(mut self, id: impl Into<String>) -> Self {
306        self.id = Some(id.into());
307        self
308    }
309
310    /// Already-owned payload `Arc` — zero serde_json walk, refcount-only
311    /// share. Mutually exclusive with [`Self::payload_json`]; whichever is
312    /// called last wins.
313    pub fn payload(mut self, payload: Arc<OwnedDataValue>) -> Self {
314        self.payload = Some(payload);
315        self
316    }
317
318    /// Construct the payload from a `serde_json::Value`. Goes through the
319    /// `OwnedDataValue::from(&Value)` bridge (one deep walk).
320    pub fn payload_json(mut self, payload: &JsonValue) -> Self {
321        self.payload = Some(Arc::new(OwnedDataValue::from(payload)));
322        self
323    }
324
325    /// Seed `context.data`.
326    ///
327    /// Replaces the empty `Object` that [`Self::build`] would otherwise install;
328    /// the other two root fields are unaffected. Seeding records no audit-trail
329    /// entry and no `Change` — it is initial state, not a mutation.
330    ///
331    /// Keys are taken **literally**: unlike
332    /// [`crate::engine::utils::set_nested_value`], a key containing `.` stays a
333    /// single key and a leading `#` is not stripped.
334    ///
335    /// A non-`Object` value is **ignored**, preserving the crate-wide invariant
336    /// that the three root fields are always objects. Calling this twice keeps
337    /// the last value.
338    ///
339    /// ```
340    /// use dataflow_rs::Message;
341    /// use serde_json::json;
342    ///
343    /// let m = Message::builder().data_json(&json!({"order": {"total": 1500}})).build();
344    /// assert_eq!(m.data()["order"]["total"], json!(1500).into());
345    /// ```
346    pub fn data(mut self, data: OwnedDataValue) -> Self {
347        if data.is_object() {
348            self.data = Some(data);
349        }
350        self
351    }
352
353    /// [`Self::data`] from a `serde_json::Value` (one `OwnedDataValue::from`
354    /// deep walk).
355    pub fn data_json(self, data: &JsonValue) -> Self {
356        self.data(OwnedDataValue::from(data))
357    }
358
359    /// Seed `context.metadata` — request headers, correlation ids, routing
360    /// hints.
361    ///
362    /// `Engine::process_message` adds `processed_at` and `engine_version` on top
363    /// of whatever is seeded here; a seeded `channel` key is overwritten by
364    /// `process_message_for_channel`. Same literal-key and non-object rules as
365    /// [`Self::data`].
366    ///
367    /// ```
368    /// use dataflow_rs::Message;
369    /// use serde_json::json;
370    ///
371    /// let m = Message::builder().metadata_json(&json!({"source": "api"})).build();
372    /// assert_eq!(m.metadata()["source"], json!("api").into());
373    /// ```
374    pub fn metadata(mut self, metadata: OwnedDataValue) -> Self {
375        if metadata.is_object() {
376            self.metadata = Some(metadata);
377        }
378        self
379    }
380
381    /// [`Self::metadata`] from a `serde_json::Value`.
382    pub fn metadata_json(self, metadata: &JsonValue) -> Self {
383        self.metadata(OwnedDataValue::from(metadata))
384    }
385
386    /// Seed `context.temp_data` — scratch space for intermediate task output.
387    ///
388    /// Same literal-key and non-object rules as [`Self::data`].
389    ///
390    /// ```
391    /// use dataflow_rs::Message;
392    /// use serde_json::json;
393    ///
394    /// let m = Message::builder().temp_data_json(&json!({"scratch": 1})).build();
395    /// assert_eq!(m.temp_data()["scratch"], json!(1).into());
396    /// ```
397    pub fn temp_data(mut self, temp_data: OwnedDataValue) -> Self {
398        if temp_data.is_object() {
399            self.temp_data = Some(temp_data);
400        }
401        self
402    }
403
404    /// [`Self::temp_data`] from a `serde_json::Value`.
405    pub fn temp_data_json(self, temp_data: &JsonValue) -> Self {
406        self.temp_data(OwnedDataValue::from(temp_data))
407    }
408
409    /// Routing bucket `0..=99` for [`crate::Workflow::rollout`] matching.
410    ///
411    /// Values `>= 100` are clamped to `99`, keeping the builder infallible like
412    /// the rest of its methods. A message with no bucket is admitted by every
413    /// workflow, split or not.
414    ///
415    /// ```
416    /// use dataflow_rs::Message;
417    ///
418    /// assert_eq!(Message::builder().routing_bucket(7).build().routing_bucket(), Some(7));
419    /// assert_eq!(Message::builder().routing_bucket(200).build().routing_bucket(), Some(99));
420    /// assert_eq!(Message::builder().build().routing_bucket(), None);
421    /// ```
422    pub fn routing_bucket(mut self, bucket: u8) -> Self {
423        self.routing_bucket = Some(bucket.min(99));
424        self
425    }
426
427    /// When `false`, built-in functions skip per-write `Change` capture —
428    /// audit-trail entries are still recorded but their `changes` list is
429    /// empty. Defaults to `true`.
430    pub fn capture_changes(mut self, on: bool) -> Self {
431        self.capture_changes = Some(on);
432        self
433    }
434
435    /// Finalize. Defaults: id = UUID v7, payload = `OwnedDataValue::Null`,
436    /// capture_changes = `true`.
437    pub fn build(self) -> Message {
438        Message {
439            id: self
440                .id
441                .map(MessageId::Custom)
442                .unwrap_or_else(MessageId::new_uuid_v7),
443            payload: self
444                .payload
445                .unwrap_or_else(|| Arc::new(OwnedDataValue::Null)),
446            context: context_from(self.data, self.metadata, self.temp_data),
447            audit_trail: vec![],
448            errors: vec![],
449            capture_changes: self.capture_changes.unwrap_or(true),
450            routing_bucket: self.routing_bucket,
451        }
452    }
453}
454
455/// Build the canonical context object, substituting any caller-seeded root field
456/// for the empty `Object`.
457///
458/// Key order is fixed at `data, metadata, temp_data` so the serialized shape is
459/// identical whether or not a seed was supplied.
460fn context_from(
461    data: Option<OwnedDataValue>,
462    metadata: Option<OwnedDataValue>,
463    temp_data: Option<OwnedDataValue>,
464) -> OwnedDataValue {
465    fn slot(v: Option<OwnedDataValue>) -> OwnedDataValue {
466        v.unwrap_or_else(|| OwnedDataValue::Object(Vec::new()))
467    }
468    OwnedDataValue::Object(vec![
469        ("data".to_string(), slot(data)),
470        ("metadata".to_string(), slot(metadata)),
471        ("temp_data".to_string(), slot(temp_data)),
472    ])
473}
474
475/// Build the canonical empty context shape used by `Message::new`. One
476/// implementation of the shape, shared with the seeded path.
477fn empty_context() -> OwnedDataValue {
478    context_from(None, None, None)
479}
480
481#[derive(Serialize, Deserialize, Debug, Clone)]
482pub struct AuditTrail {
483    pub workflow_id: Arc<str>,
484    pub task_id: Arc<str>,
485    pub timestamp: DateTime<Utc>,
486    pub changes: Vec<Change>,
487    pub status: usize,
488}
489
490/// A single recorded mutation in the audit trail.
491///
492/// `old_value` and `new_value` are owned `OwnedDataValue`s rather than
493/// `Arc<OwnedDataValue>` — eliminates one heap allocation per Change on the
494/// hot path. External consumers that need to share a `Change` across threads
495/// can wrap it themselves; in-process pipelines (audit-on map mappings) don't
496/// pay the Arc cost they were never going to use.
497#[derive(Serialize, Deserialize, Debug, Clone)]
498pub struct Change {
499    pub path: Arc<str>,
500    pub old_value: OwnedDataValue,
501    pub new_value: OwnedDataValue,
502}
503
504#[cfg(test)]
505mod tests {
506    use super::*;
507
508    #[test]
509    fn from_json_str_parses_valid_payload() {
510        let msg =
511            Message::from_json_str(r#"{"order": {"total": 42}}"#).expect("valid JSON should parse");
512        let payload_json = serde_json::to_value(msg.payload()).unwrap();
513        assert_eq!(payload_json, serde_json::json!({"order": {"total": 42}}));
514    }
515
516    #[test]
517    fn from_json_str_rejects_malformed_payload() {
518        let err = Message::from_json_str("{ not json").expect_err("malformed input should fail");
519        assert!(matches!(err, DataflowError::Deserialization(_)));
520    }
521
522    #[test]
523    fn builder_with_no_seed_matches_the_historical_empty_context() {
524        let m = Message::builder().build();
525        let v = serde_json::to_value(&m).unwrap();
526        // Key order is part of the shape, so assert on the serialized object.
527        let ctx = v["context"].as_object().unwrap();
528        assert_eq!(
529            ctx.keys().collect::<Vec<_>>(),
530            vec!["data", "metadata", "temp_data"]
531        );
532        assert_eq!(
533            v["context"],
534            serde_json::json!({
535                "data": {}, "metadata": {}, "temp_data": {}
536            })
537        );
538    }
539
540    #[test]
541    fn each_setter_lands_in_its_own_root_field() {
542        let m = Message::builder()
543            .data_json(&serde_json::json!({"d": 1}))
544            .build();
545        assert_eq!(
546            serde_json::Value::from(m.data()),
547            serde_json::json!({"d": 1})
548        );
549        assert_eq!(serde_json::Value::from(m.metadata()), serde_json::json!({}));
550        assert_eq!(
551            serde_json::Value::from(m.temp_data()),
552            serde_json::json!({})
553        );
554
555        let m = Message::builder()
556            .metadata_json(&serde_json::json!({"m": 1}))
557            .build();
558        assert_eq!(
559            serde_json::Value::from(m.metadata()),
560            serde_json::json!({"m": 1})
561        );
562        assert_eq!(serde_json::Value::from(m.data()), serde_json::json!({}));
563
564        let m = Message::builder()
565            .temp_data_json(&serde_json::json!({"t": 1}))
566            .build();
567        assert_eq!(
568            serde_json::Value::from(m.temp_data()),
569            serde_json::json!({"t": 1})
570        );
571        assert_eq!(serde_json::Value::from(m.data()), serde_json::json!({}));
572    }
573
574    #[test]
575    fn the_owned_and_json_setter_forms_agree() {
576        let v = serde_json::json!({"a": {"b": [1, 2]}});
577        let via_json = Message::builder().data_json(&v).build();
578        let via_owned = Message::builder().data(OwnedDataValue::from(&v)).build();
579        assert_eq!(via_json.context, via_owned.context);
580    }
581
582    #[test]
583    fn seeding_records_no_audit_entry_or_change() {
584        for capture in [true, false] {
585            let m = Message::builder()
586                .capture_changes(capture)
587                .data_json(&serde_json::json!({"d": 1}))
588                .build();
589            assert!(
590                m.audit_trail().is_empty(),
591                "seeding is initial state, not a mutation"
592            );
593            assert_eq!(
594                serde_json::Value::from(m.data()),
595                serde_json::json!({"d": 1})
596            );
597        }
598    }
599
600    #[test]
601    fn calling_a_setter_twice_keeps_the_last_value() {
602        let m = Message::builder()
603            .data_json(&serde_json::json!({"first": 1}))
604            .data_json(&serde_json::json!({"second": 2}))
605            .build();
606        assert_eq!(
607            serde_json::Value::from(m.data()),
608            serde_json::json!({"second": 2})
609        );
610    }
611
612    #[test]
613    fn an_empty_object_seed_is_indistinguishable_from_no_seed() {
614        let seeded = Message::builder().data_json(&serde_json::json!({})).build();
615        let bare = Message::builder().build();
616        assert_eq!(seeded.context, bare.context);
617    }
618
619    #[test]
620    fn seed_keys_are_literal_not_paths() {
621        use crate::engine::utils::get_nested_value;
622
623        let m = Message::builder()
624            .metadata_json(&serde_json::json!({"a.b": 1}))
625            .build();
626        // One literal key, not nested.
627        assert_eq!(
628            serde_json::Value::from(m.metadata()),
629            serde_json::json!({"a.b": 1})
630        );
631        assert!(
632            get_nested_value(&m.context, "metadata.a.b").is_none(),
633            "a dotted key must not become a nested path"
634        );
635
636        // Leading `#` is not stripped.
637        let m = Message::builder()
638            .metadata_json(&serde_json::json!({"#20": 1}))
639            .build();
640        assert_eq!(
641            serde_json::Value::from(m.metadata()),
642            serde_json::json!({"#20": 1})
643        );
644
645        // A numeric-string key stays an object key, not an Array.
646        let m = Message::builder()
647            .data_json(&serde_json::json!({"0": 1}))
648            .build();
649        assert!(matches!(m.data(), OwnedDataValue::Object(_)));
650        assert_eq!(
651            serde_json::Value::from(m.data()),
652            serde_json::json!({"0": 1})
653        );
654    }
655
656    #[test]
657    fn non_ascii_keys_and_values_round_trip() {
658        let v = serde_json::json!({"régión": "東京"});
659        let m = Message::builder().data_json(&v).build();
660        assert_eq!(serde_json::Value::from(m.data()), v);
661        assert_eq!(serde_json::to_value(&m).unwrap()["context"]["data"], v);
662    }
663
664    #[test]
665    fn nested_seeds_resolve_through_the_path_api() {
666        use crate::engine::utils::get_nested_value;
667
668        let m = Message::builder()
669            .data_json(&serde_json::json!({"order": {"items": [1, 2]}}))
670            .build();
671        assert_eq!(
672            get_nested_value(&m.context, "data.order.items.1"),
673            Some(&OwnedDataValue::from(&serde_json::json!(2)))
674        );
675    }
676
677    #[test]
678    fn non_object_seeds_are_ignored_to_preserve_the_context_invariant() {
679        // The recorded decision: the three root fields are always Objects.
680        // Storing a scalar verbatim would make `set_processing_metadata` bail out
681        // and silently drop `processed_at` / `engine_version`.
682        for bad in [
683            serde_json::json!("scalar"),
684            serde_json::json!([1, 2]),
685            serde_json::json!(null),
686            serde_json::json!(7),
687        ] {
688            let m = Message::builder()
689                .data_json(&bad)
690                .metadata_json(&bad)
691                .temp_data_json(&bad)
692                .build();
693            assert_eq!(
694                serde_json::Value::from(&m.context),
695                serde_json::json!({"data": {}, "metadata": {}, "temp_data": {}}),
696                "non-object seed {bad} must be ignored"
697            );
698        }
699    }
700}