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 data_logic = data_logic_cell.borrow_mut();
46
47            if let Some(rules_arr) = rules.as_array() {
48                for rule in rules_arr {
49                    // Extract rule components
50                    let rule_logic = rule.get("logic").ok_or_else(|| {
51                        DataflowError::Validation("Missing logic in rule".to_string())
52                    })?;
53
54                    let rule_path = rule.get("path").and_then(Value::as_str).unwrap_or("data");
55
56                    let data_to_validate = if rule_path == "data" {
57                        &json!({rule_path: message.data})
58                    } else if rule_path == "metadata" {
59                        &json!({rule_path: message.metadata})
60                    } else {
61                        &json!({rule_path: message.data})
62                    };
63
64                    // Evaluate the rule
65                    match data_logic.evaluate_json(rule_logic, data_to_validate, None) {
66                        Ok(v) => {
67                            if !v.as_bool().unwrap_or(false) {
68                                let message_key = rule
69                                    .get("message")
70                                    .and_then(Value::as_str)
71                                    .unwrap_or("Validation failed")
72                                    .to_string();
73
74                                println!("Validation failed: {}", message_key);
75                                return Ok((400, vec![]));
76                            }
77                        }
78                        Err(e) => {
79                            println!("Error evaluating rule: {}", e);
80                            return Err(DataflowError::LogicEvaluation(format!(
81                                "Error evaluating rule: {}",
82                                e
83                            )));
84                        }
85                    }
86                }
87            }
88
89            let changes = vec![Change {
90                path: "temp_data.validation".to_string(),
91                old_value: Value::Null,
92                new_value: message.temp_data["validation"].clone(),
93            }];
94
95            Ok((200, changes))
96        });
97
98        validation_result
99    }
100}
dataflow_rs/engine/functions/validation.rs

dataflow_rs/engine/functions/
validation.rs
