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}