dataflow_rs/engine/functions/

parse.rs

//! # Parse Function Module
//!
//! Parsing helpers that load payload data into the message's `data` context.
//! Supports JSON (native) and XML (via `serde_json::Value` bridge — XML is the
//! slow path, not worth a dedicated walker).
//!
//! Source paths:
//! - `"payload"` — entire payload
//! - `"payload.<path>"` — a nested field of the payload
//! - `"data.<path>"` — a nested field of the existing data context
//! - `"<path>"` — anything else is resolved against the full context

use crate::engine::error::{DataflowError, Result};
use crate::engine::executor::ArenaContext;
use crate::engine::message::{Change, Message};
use crate::engine::task_outcome::TaskOutcome;
use crate::engine::utils::{get_nested_value, get_nested_value_parts, set_nested_value_parts};
use datavalue::OwnedDataValue;
use log::debug;
use serde::Deserialize;
use serde_json::Value;
use std::sync::Arc;

/// Configuration for parse functions.
#[derive(Debug, Clone, Default, Deserialize)]
pub struct ParseConfig {
    /// Source path to read from.
    pub source: String,

    /// Target field name in `data` (stored at `data.{target}`).
    pub target: String,

    /// Engine-internal: precomputed `"data.{target}"`, populated by
    /// `LogicCompiler` (and eagerly by [`Self::from_json`]). Cloned
    /// (refcount bump) into `Change.path` instead of re-allocating per
    /// call. Not part of the stable API.
    #[doc(hidden)]
    #[serde(skip)]
    pub target_path_arc: Arc<str>,

    /// Engine-internal: pre-split segments of the target path, consumed by
    /// the `*_parts` tree walkers so the hot path never re-splits. Not part
    /// of the stable API.
    #[doc(hidden)]
    #[serde(skip)]
    pub target_path_parts: Arc<[Arc<str>]>,
}

impl ParseConfig {
    pub fn from_json(input: &Value) -> Result<Self> {
        let source = input
            .get("source")
            .and_then(Value::as_str)
            .ok_or_else(|| {
                DataflowError::Validation("Missing 'source' in parse config".to_string())
            })?
            .to_string();

        let target = input
            .get("target")
            .and_then(Value::as_str)
            .ok_or_else(|| {
                DataflowError::Validation("Missing 'target' in parse config".to_string())
            })?
            .to_string();

        let mut config = ParseConfig {
            source,
            target,
            ..Default::default()
        };
        config.precompute_target_path();
        Ok(config)
    }

    /// Populate the precomputed target-path fields from `target`. Called by
    /// `LogicCompiler` for serde-built configs and by `from_json`.
    pub(crate) fn precompute_target_path(&mut self) {
        let path = format!("data.{}", self.target);
        let parts: Vec<Arc<str>> = path.split('.').map(Arc::from).collect();
        self.target_path_parts = parts.into();
        self.target_path_arc = Arc::from(path);
    }

    /// Precomputed `(path, parts)` for `data.{target}` — falls back to
    /// computing on the fly for directly-constructed configs (the test
    /// surface), mirroring the `MapMapping` fallback pattern.
    fn resolve_target_path(&self) -> (Arc<str>, Arc<[Arc<str>]>) {
        if self.target_path_parts.is_empty() {
            let path = format!("data.{}", self.target);
            let parts: Vec<Arc<str>> = path.split('.').map(Arc::from).collect();
            (Arc::from(path), parts.into())
        } else {
            (
                Arc::clone(&self.target_path_arc),
                Arc::clone(&self.target_path_parts),
            )
        }
    }

    /// Extract the source value as an owned `OwnedDataValue`.
    fn extract_source(&self, message: &Message) -> OwnedDataValue {
        if self.source == "payload" {
            (*message.payload).clone()
        } else if let Some(path) = self.source.strip_prefix("payload.") {
            get_nested_value(&message.payload, path)
                .cloned()
                .unwrap_or(OwnedDataValue::Null)
        } else if let Some(path) = self.source.strip_prefix("data.") {
            get_nested_value(message.data(), path)
                .cloned()
                .unwrap_or(OwnedDataValue::Null)
        } else {
            get_nested_value(&message.context, &self.source)
                .cloned()
                .unwrap_or(OwnedDataValue::Null)
        }
    }
}

/// Execute `parse_json`: read the source value and store it under `data.{target}`.
/// If the source is a JSON string, attempt to parse it; on failure, store the
/// string as-is (matches prior behaviour).
pub fn execute_parse_json(
    message: &mut Message,
    config: &ParseConfig,
) -> Result<(TaskOutcome, Vec<Change>)> {
    debug!(
        "ParseJson: Extracting from '{}' to 'data.{}'",
        config.source, config.target
    );

    let (target_path_arc, target_parts) = config.resolve_target_path();

    // Hot path: source == "payload" and not a JSON-string payload. The
    // payload Arc is already on the message; clone-into-context once, reuse
    // the Arc for the audit entry (refcount bump). This is the realistic
    // benchmark's exact shape.
    let payload_fast_path =
        config.source == "payload" && !matches!(*message.payload, OwnedDataValue::String(_));

    if message.capture_changes {
        let old_value = get_nested_value_parts(&message.context, &target_parts)
            .cloned()
            .unwrap_or(OwnedDataValue::Null);

        // Resolve the source value once. For the payload fast-path we clone
        // out of the shared `Arc<OwnedDataValue>` payload; for the slow path
        // we extract from a sub-tree and re-parse JSON-string payloads.
        let source_data = if payload_fast_path {
            (*message.payload).clone()
        } else {
            let raw = config.extract_source(message);
            match &raw {
                OwnedDataValue::String(s) => {
                    OwnedDataValue::from_json(s).unwrap_or_else(|_| raw.clone())
                }
                _ => raw,
            }
        };

        // Clone the source value once for the audit `new_value`; the original
        // is moved into the context below. (No `Arc` wrapping in the audit
        // entry — `Change` owns its values directly.)
        let new_value = source_data.clone();

        set_nested_value_parts(&mut message.context, &target_parts, source_data);
        debug!(
            "ParseJson: Successfully stored data to 'data.{}'",
            config.target
        );
        return Ok((
            TaskOutcome::Success,
            vec![Change {
                path: target_path_arc,
                old_value,
                new_value,
            }],
        ));
    }

    // Audit-off fast path: only the deep clone into the context survives.
    let source_data_for_context: OwnedDataValue = if payload_fast_path {
        (*message.payload).clone()
    } else {
        let raw = config.extract_source(message);
        match &raw {
            OwnedDataValue::String(s) => {
                OwnedDataValue::from_json(s).unwrap_or_else(|_| raw.clone())
            }
            _ => raw,
        }
    };
    set_nested_value_parts(&mut message.context, &target_parts, source_data_for_context);

    debug!(
        "ParseJson: Successfully stored data to 'data.{}'",
        config.target
    );

    Ok((TaskOutcome::Success, Vec::new()))
}

/// Same as `execute_parse_json` but also refreshes the supplied
/// `ArenaContext` so subsequent sync tasks in the same workflow stretch see
/// the written `data.<target>` slot without rebuilding the whole arena form.
pub(crate) fn execute_parse_json_in_arena(
    message: &mut Message,
    config: &ParseConfig,
    arena_ctx: &mut ArenaContext<'_>,
) -> Result<(TaskOutcome, Vec<Change>)> {
    let result = execute_parse_json(message, config)?;
    // Refresh ONLY the affected depth-2 slot in the arena cache. For
    // source == "payload" target = "input", this is `data.input` — the
    // heavy slot — but it's re-arena'd exactly once per workflow stretch
    // here, not once per subsequent map mapping.
    let (_, target_parts) = config.resolve_target_path();
    arena_ctx.refresh_for_path_parts(&message.context, &target_parts);
    Ok(result)
}

/// Execute `parse_xml`: read the source string, parse XML into a
/// `serde_json::Value` (existing quick-xml path), convert to `OwnedDataValue`,
/// store under `data.{target}`.
pub fn execute_parse_xml(
    message: &mut Message,
    config: &ParseConfig,
) -> Result<(TaskOutcome, Vec<Change>)> {
    debug!(
        "ParseXml: Extracting from '{}' to 'data.{}'",
        config.source, config.target
    );

    let source_data = config.extract_source(message);

    let xml_string = match &source_data {
        OwnedDataValue::String(s) => s.clone(),
        _ => {
            return Err(DataflowError::Validation(format!(
                "ParseXml: Source '{}' is not a string",
                config.source
            )));
        }
    };

    let parsed_json = xml_to_json(&xml_string)?;
    let parsed_owned = OwnedDataValue::from(&parsed_json);

    let (target_path_arc, target_parts) = config.resolve_target_path();
    let old_value = get_nested_value_parts(&message.context, &target_parts)
        .cloned()
        .unwrap_or(OwnedDataValue::Null);

    set_nested_value_parts(&mut message.context, &target_parts, parsed_owned.clone());

    debug!(
        "ParseXml: Successfully parsed and stored XML to 'data.{}'",
        config.target
    );

    Ok((
        TaskOutcome::Success,
        vec![Change {
            path: target_path_arc,
            old_value,
            new_value: parsed_owned,
        }],
    ))
}

/// Convert an XML string to `serde_json::Value` using quick-xml's serde path.
fn xml_to_json(xml: &str) -> Result<Value> {
    use quick_xml::de::from_str;

    let parsed: Value = from_str(xml)
        .map_err(|e| DataflowError::Validation(format!("Failed to parse XML: {}", e)))?;

    Ok(parsed)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::engine::utils::set_nested_value;
    use serde_json::json;

    fn dv(v: serde_json::Value) -> OwnedDataValue {
        OwnedDataValue::from(&v)
    }

    #[test]
    fn test_parse_config_from_json() {
        let input = json!({"source": "payload", "target": "input_data"});
        let config = ParseConfig::from_json(&input).unwrap();
        assert_eq!(config.source, "payload");
        assert_eq!(config.target, "input_data");
    }

    #[test]
    fn test_parse_config_missing_source() {
        assert!(ParseConfig::from_json(&json!({"target": "input_data"})).is_err());
    }

    #[test]
    fn test_parse_config_missing_target() {
        assert!(ParseConfig::from_json(&json!({"source": "payload"})).is_err());
    }

    #[test]
    fn test_execute_parse_json_from_payload() {
        let payload = json!({"name": "John", "age": 30});
        let mut message = Message::from_value(&payload);

        let config = ParseConfig {
            source: "payload".to_string(),
            target: "input".to_string(),
            ..Default::default()
        };

        let result = execute_parse_json(&mut message, &config);
        assert!(result.is_ok());

        let (outcome, changes) = result.unwrap();
        assert_eq!(outcome, TaskOutcome::Success);
        assert_eq!(changes.len(), 1);
        assert_eq!(changes[0].path.as_ref(), "data.input");

        assert_eq!(message.data()["input"]["name"], dv(json!("John")));
        assert_eq!(message.data()["input"]["age"], dv(json!(30)));
    }

    #[test]
    fn test_execute_parse_json_from_nested_payload() {
        let payload = json!({"body": {"user": {"name": "Alice"}}});
        let mut message = Message::from_value(&payload);

        let config = ParseConfig {
            source: "payload.body.user".to_string(),
            target: "user_data".to_string(),
            ..Default::default()
        };

        let result = execute_parse_json(&mut message, &config);
        assert!(result.is_ok());

        let (outcome, _) = result.unwrap();
        assert_eq!(outcome, TaskOutcome::Success);
        assert_eq!(message.data()["user_data"]["name"], dv(json!("Alice")));
    }

    #[test]
    fn test_execute_parse_json_from_data() {
        let mut message = Message::new(Arc::new(dv(json!({}))));
        set_nested_value(
            &mut message.context,
            "data",
            dv(json!({"existing": {"value": 42}})),
        );

        let config = ParseConfig {
            source: "data.existing".to_string(),
            target: "copied".to_string(),
            ..Default::default()
        };

        let result = execute_parse_json(&mut message, &config);
        assert!(result.is_ok());

        assert_eq!(message.data()["copied"]["value"], dv(json!(42)));
    }

    #[test]
    fn test_execute_parse_xml_simple() {
        let xml_payload = json!("<root><name>John</name><age>30</age></root>");
        let mut message = Message::from_value(&xml_payload);

        let config = ParseConfig {
            source: "payload".to_string(),
            target: "parsed".to_string(),
            ..Default::default()
        };

        let result = execute_parse_xml(&mut message, &config);
        assert!(result.is_ok());

        let (outcome, _) = result.unwrap();
        assert_eq!(outcome, TaskOutcome::Success);

        let parsed = &message.data()["parsed"];
        assert!(parsed.is_object());
    }

    #[test]
    fn test_execute_parse_xml_not_string() {
        let payload = json!({"not": "a string"});
        let mut message = Message::from_value(&payload);

        let config = ParseConfig {
            source: "payload".to_string(),
            target: "parsed".to_string(),
            ..Default::default()
        };

        assert!(execute_parse_xml(&mut message, &config).is_err());
    }

    #[test]
    fn test_xml_to_json_simple() {
        let xml = "<root><name>Test</name></root>";
        let result = xml_to_json(xml);
        assert!(result.is_ok());
        let json = result.unwrap();
        assert!(json.is_object());
    }

    #[test]
    fn test_xml_to_json_invalid() {
        let xml = "<root><unclosed>";
        assert!(xml_to_json(xml).is_err());
    }

    #[test]
    fn test_xml_to_json_with_attributes() {
        let xml = r#"<person id="123"><name>John</name></person>"#;
        assert!(xml_to_json(xml).is_ok());
    }

    #[test]
    fn test_xml_to_json_nested() {
        let xml = r#"<root><user><name>Alice</name><email>alice@example.com</email></user></root>"#;
        let result = xml_to_json(xml);
        assert!(result.is_ok());
        let json = result.unwrap();
        assert!(json.is_object());
    }

    #[test]
    fn test_execute_parse_json_from_string_payload() {
        let payload = Value::String(r#"{"name":"John","age":30}"#.to_string());
        let mut message = Message::from_value(&payload);

        let config = ParseConfig {
            source: "payload".to_string(),
            target: "input".to_string(),
            ..Default::default()
        };

        let result = execute_parse_json(&mut message, &config);
        assert!(result.is_ok());

        let (outcome, _) = result.unwrap();
        assert_eq!(outcome, TaskOutcome::Success);

        assert_eq!(message.data()["input"]["name"], dv(json!("John")));
        assert_eq!(message.data()["input"]["age"], dv(json!(30)));
    }
}