dataflow_rs/engine/functions/
validation.rs

1use crate::engine::error::{DataflowError, Result};
2use crate::engine::functions::FUNCTION_DATA_LOGIC;
3use crate::engine::message::{Change, Message};
4use crate::engine::AsyncFunctionHandler;
5use async_trait::async_trait;
6use serde_json::{json, Value};
7use std::vec;
8/// A validation task function that uses JsonLogic expressions to validate message data.
9///
10/// This function executes validation rules defined in JsonLogic against data in the message,
11/// and reports validation failures by adding errors to the message's errors array.
12pub struct ValidationFunction {
13    // No longer needs data_logic field
14}
15
16// SAFETY: These implementations are technically unsound because DataLogic contains
17// RefCell and Cell which are not thread-safe. In practice, we'll ensure that
18// ValidationTask is only used in a single-threaded context, or we'll use thread-local
19// instances of DataLogic.
20unsafe impl Send for ValidationFunction {}
21unsafe impl Sync for ValidationFunction {}
22
23impl Default for ValidationFunction {
24    fn default() -> Self {
25        Self::new()
26    }
27}
28
29impl ValidationFunction {
30    pub fn new() -> Self {
31        Self { /* no fields */ }
32    }
33}
34
35#[async_trait]
36impl AsyncFunctionHandler for ValidationFunction {
37    async fn execute(&self, message: &mut Message, input: &Value) -> Result<(usize, Vec<Change>)> {
38        // Extract rules from input
39        let rules = input
40            .get("rules")
41            .ok_or_else(|| DataflowError::Validation("Missing rules array".to_string()))?;
42
43        // Use thread-local DataLogic
44        let validation_result = FUNCTION_DATA_LOGIC.with(|data_logic_cell| {
45            let mut data_logic = data_logic_cell.borrow_mut();
46            data_logic.reset_arena();
47
48            if let Some(rules_arr) = rules.as_array() {
49                for rule in rules_arr {
50                    // Extract rule components
51                    let rule_logic = rule.get("logic").ok_or_else(|| {
52                        DataflowError::Validation("Missing logic in rule".to_string())
53                    })?;
54
55                    let rule_path = rule.get("path").and_then(Value::as_str).unwrap_or("data");
56
57                    let data_to_validate = if rule_path == "data" {
58                        &json!({rule_path: message.data})
59                    } else if rule_path == "metadata" {
60                        &json!({rule_path: message.metadata})
61                    } else {
62                        &json!({rule_path: message.data})
63                    };
64
65                    // Evaluate the rule
66                    match data_logic.evaluate_json(rule_logic, data_to_validate, None) {
67                        Ok(v) => {
68                            if !v.as_bool().unwrap_or(false) {
69                                let message_key = rule
70                                    .get("message")
71                                    .and_then(Value::as_str)
72                                    .unwrap_or("Validation failed")
73                                    .to_string();
74
75                                println!("Validation failed: {}", message_key);
76                                return Ok((400, vec![]));
77                            }
78                        }
79                        Err(e) => {
80                            println!("Error evaluating rule: {}", e);
81                            return Err(DataflowError::LogicEvaluation(format!(
82                                "Error evaluating rule: {}",
83                                e
84                            )));
85                        }
86                    }
87                }
88            }
89
90            let changes = vec![Change {
91                path: "temp_data.validation".to_string(),
92                old_value: Value::Null,
93                new_value: message.temp_data["validation"].clone(),
94            }];
95
96            Ok((200, changes))
97        });
98
99        validation_result
100    }
101}
dataflow_rs/engine/functions/validation.rs

dataflow_rs/engine/functions/
validation.rs
