zen_engine/nodes/
context.rs

1use crate::nodes::definition::{NodeDataType, TraceDataType};
2use crate::nodes::extensions::NodeHandlerExtensions;
3use crate::nodes::function::v2::function::Function;
4use crate::nodes::result::{NodeResponse, NodeResult};
5use crate::nodes::variable_json::{Guards, VariableNode};
6use crate::nodes::NodeError;
7use crate::ZEN_CONFIG;
8use ahash::AHasher;
9use jsonschema::ValidationError;
10use serde::Serialize;
11use serde_json::Value;
12use std::cell::RefCell;
13use std::fmt::{Display, Formatter};
14use std::hash::Hasher;
15use std::sync::atomic::Ordering;
16use std::sync::Arc;
17use thiserror::Error;
18use zen_expression::Isolate;
19use zen_types::variable::{ToVariable, Variable};
20
21#[derive(Clone)]
22pub struct NodeContext<NodeData, TraceData>
23where
24    NodeData: NodeDataType,
25    TraceData: TraceDataType,
26{
27    pub id: Arc<str>,
28    pub name: Arc<str>,
29    pub node: NodeData,
30    pub input: Variable,
31    pub nodes: Option<Variable>,
32    pub trace: Option<RefCell<TraceData>>,
33    pub extensions: NodeHandlerExtensions,
34    pub iteration: u8,
35    pub config: NodeContextConfig,
36}
37
38impl<NodeData, TraceData> NodeContext<NodeData, TraceData>
39where
40    NodeData: NodeDataType,
41    TraceData: TraceDataType,
42{
43    pub fn input_with_nodes(&self) -> Variable {
44        let Some(nodes) = &self.nodes else {
45            return self.input.shallow_clone();
46        };
47        let Variable::Object(object) = &self.input else {
48            return self.input.shallow_clone();
49        };
50
51        let mut map = object.borrow().clone();
52        map.insert(Variable::nodes_key(), nodes.clone());
53        Variable::from_object(map)
54    }
55
56    pub fn from_base(base: NodeContextBase, data: NodeData) -> Self {
57        Self {
58            id: base.id,
59            name: base.name,
60            input: base.input,
61            nodes: base.nodes,
62            extensions: base.extensions,
63            iteration: base.iteration,
64            trace: base.config.trace.then(|| Default::default()),
65            node: data,
66            config: base.config,
67        }
68    }
69
70    pub fn isolate(&self) -> Isolate {
71        make_isolate(&self.input, self.nodes.as_ref(), &self.extensions)
72    }
73
74    pub fn trace<Function>(&self, mutator: Function)
75    where
76        Function: FnOnce(&mut TraceData),
77    {
78        if let Some(trace) = &self.trace {
79            mutator(&mut *trace.borrow_mut());
80        }
81    }
82
83    pub fn error<Error>(&self, error: Error) -> NodeResult
84    where
85        Error: Into<Box<dyn std::error::Error>>,
86    {
87        Err(self.make_error(error))
88    }
89
90    pub fn success(&self, output: Variable) -> NodeResult {
91        Ok(NodeResponse {
92            output,
93            trace_data: self.trace.as_ref().map(|v| (*v.borrow()).to_variable()),
94        })
95    }
96
97    pub(crate) fn make_error<Error>(&self, error: Error) -> NodeError
98    where
99        Error: Into<Box<dyn std::error::Error>>,
100    {
101        NodeError {
102            node_id: self.id.clone(),
103            trace: self.trace.as_ref().map(|v| (*v.borrow()).to_variable()),
104            source: error.into(),
105        }
106    }
107
108    pub(crate) async fn function_runtime(&self) -> Result<&Function, NodeError> {
109        self.extensions.function_runtime().await.node_context(self)
110    }
111
112    pub fn validate(&self, schema: &Value, value: &Variable) -> Result<(), NodeError> {
113        let validator_cache = self.extensions.validator_cache();
114        let hash = self.hash_node();
115
116        let validator = validator_cache
117            .get_or_insert(hash, schema)
118            .node_context(self)?;
119
120        let guards = Guards::default();
121        validator
122            .validate(VariableNode::new(value, &guards))
123            .map_err(|err| ValidationErrorJson::from(err))
124            .node_context(self)?;
125
126        Ok(())
127    }
128
129    fn hash_node(&self) -> u64 {
130        let mut hasher = AHasher::default();
131        hasher.write(self.id.as_bytes());
132        hasher.write(self.name.as_bytes());
133        hasher.write_u64(self.config.validation_salt);
134        hasher.finish()
135    }
136}
137
138pub trait NodeContextExt<T, Context>: Sized {
139    type Error: Into<Box<dyn std::error::Error>>;
140
141    fn with_node_context<Function, NewError>(
142        self,
143        ctx: &Context,
144        f: Function,
145    ) -> Result<T, NodeError>
146    where
147        Function: FnOnce(Self::Error) -> NewError,
148        NewError: Into<Box<dyn std::error::Error>>;
149
150    fn node_context(self, ctx: &Context) -> Result<T, NodeError> {
151        self.with_node_context(ctx, |e| e.into())
152    }
153
154    fn node_context_message(self, ctx: &Context, message: &str) -> Result<T, NodeError> {
155        self.with_node_context(ctx, |err| format!("{}: {}", message, err.into()))
156    }
157}
158
159impl<T, E, NodeData, TraceData> NodeContextExt<T, NodeContext<NodeData, TraceData>> for Result<T, E>
160where
161    E: Into<Box<dyn std::error::Error>>,
162    NodeData: NodeDataType,
163    TraceData: TraceDataType,
164{
165    type Error = E;
166
167    fn with_node_context<Function, NewError>(
168        self,
169        ctx: &NodeContext<NodeData, TraceData>,
170        f: Function,
171    ) -> Result<T, NodeError>
172    where
173        Function: FnOnce(Self::Error) -> NewError,
174        NewError: Into<Box<dyn std::error::Error>>,
175    {
176        self.map_err(|err| ctx.make_error(f(err)))
177    }
178}
179
180impl<T, NodeData, TraceData> NodeContextExt<T, NodeContext<NodeData, TraceData>> for Option<T>
181where
182    NodeData: NodeDataType,
183    TraceData: TraceDataType,
184{
185    type Error = &'static str;
186
187    fn with_node_context<Function, NewError>(
188        self,
189        ctx: &NodeContext<NodeData, TraceData>,
190        f: Function,
191    ) -> Result<T, NodeError>
192    where
193        Function: FnOnce(Self::Error) -> NewError,
194        NewError: Into<Box<dyn std::error::Error>>,
195    {
196        self.ok_or_else(|| ctx.make_error(f("None")))
197    }
198
199    fn node_context_message(
200        self,
201        ctx: &NodeContext<NodeData, TraceData>,
202        message: &str,
203    ) -> Result<T, NodeError> {
204        self.with_node_context(ctx, |_| message.to_string())
205    }
206}
207
208#[derive(Clone)]
209pub struct NodeContextBase {
210    pub id: Arc<str>,
211    pub name: Arc<str>,
212    pub input: Variable,
213    pub nodes: Option<Variable>,
214    pub iteration: u8,
215    pub extensions: NodeHandlerExtensions,
216    pub config: NodeContextConfig,
217    pub trace: Option<RefCell<Variable>>,
218}
219
220pub(crate) fn make_isolate(
221    input: &Variable,
222    nodes: Option<&Variable>,
223    extensions: &NodeHandlerExtensions,
224) -> Isolate {
225    let mut isolate =
226        Isolate::with_environment(input.clone()).with_cache(extensions.compiled_cache.clone());
227    if let Some(nodes) = nodes {
228        isolate.set_local(Variable::nodes_key(), nodes.clone());
229    }
230
231    isolate
232}
233
234impl NodeContextBase {
235    pub fn isolate(&self) -> Isolate {
236        make_isolate(&self.input, self.nodes.as_ref(), &self.extensions)
237    }
238
239    pub fn error<Error>(&self, error: Error) -> NodeResult
240    where
241        Error: Into<Box<dyn std::error::Error>>,
242    {
243        Err(self.make_error(error))
244    }
245
246    pub fn success(&self, output: Variable) -> NodeResult {
247        Ok(NodeResponse {
248            output,
249            trace_data: self.trace.as_ref().map(|v| v.borrow().to_variable()),
250        })
251    }
252
253    fn make_error<Error>(&self, error: Error) -> NodeError
254    where
255        Error: Into<Box<dyn std::error::Error>>,
256    {
257        NodeError {
258            node_id: self.id.clone(),
259            trace: self.trace.as_ref().map(|t| t.borrow().to_variable()),
260            source: error.into(),
261        }
262    }
263
264    pub fn trace<Function>(&self, mutator: Function)
265    where
266        Function: FnOnce(&mut Variable),
267    {
268        if let Some(trace) = &self.trace {
269            mutator(&mut *trace.borrow_mut());
270        }
271    }
272}
273
274impl<NodeData, TraceData> From<NodeContext<NodeData, TraceData>> for NodeContextBase
275where
276    NodeData: NodeDataType,
277    TraceData: TraceDataType,
278{
279    fn from(value: NodeContext<NodeData, TraceData>) -> Self {
280        let trace = match value.config.trace {
281            true => Some(RefCell::new(Variable::Null)),
282            false => None,
283        };
284
285        Self {
286            id: value.id,
287            name: value.name,
288            input: value.input,
289            nodes: value.nodes,
290            extensions: value.extensions,
291            iteration: value.iteration,
292            config: value.config,
293            trace,
294        }
295    }
296}
297
298impl<T, E> NodeContextExt<T, NodeContextBase> for Result<T, E>
299where
300    E: Into<Box<dyn std::error::Error>>,
301{
302    type Error = E;
303
304    fn with_node_context<Function, NewError>(
305        self,
306        ctx: &NodeContextBase,
307        f: Function,
308    ) -> Result<T, NodeError>
309    where
310        Function: FnOnce(Self::Error) -> NewError,
311        NewError: Into<Box<dyn std::error::Error>>,
312    {
313        self.map_err(|err| ctx.make_error(f(err)))
314    }
315}
316
317impl<T> NodeContextExt<T, NodeContextBase> for Option<T> {
318    type Error = &'static str;
319
320    fn with_node_context<Function, NewError>(
321        self,
322        ctx: &NodeContextBase,
323        f: Function,
324    ) -> Result<T, NodeError>
325    where
326        Function: FnOnce(Self::Error) -> NewError,
327        NewError: Into<Box<dyn std::error::Error>>,
328    {
329        self.ok_or_else(|| ctx.make_error(f("None")))
330    }
331
332    fn node_context_message(self, ctx: &NodeContextBase, message: &str) -> Result<T, NodeError> {
333        self.with_node_context(ctx, |_| message.to_string())
334    }
335}
336
337#[derive(Debug, Serialize, Error)]
338#[serde(rename_all = "camelCase")]
339struct ValidationErrorJson {
340    path: String,
341    message: String,
342}
343
344impl Display for ValidationErrorJson {
345    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
346        write!(f, "{}: {}", self.path, self.message)
347    }
348}
349
350impl<'a> From<ValidationError<'a>> for ValidationErrorJson {
351    fn from(value: ValidationError<'a>) -> Self {
352        ValidationErrorJson {
353            path: value.instance_path().to_string(),
354            message: format!("{}", value),
355        }
356    }
357}
358
359#[derive(Clone)]
360pub struct NodeContextConfig {
361    pub trace: bool,
362    pub nodes_in_context: bool,
363    pub max_depth: u8,
364    pub function_timeout_millis: u64,
365    pub http_auth: bool,
366    pub validation_salt: u64,
367}
368
369impl Default for NodeContextConfig {
370    fn default() -> Self {
371        Self {
372            trace: false,
373            nodes_in_context: ZEN_CONFIG.nodes_in_context.load(Ordering::Relaxed),
374            function_timeout_millis: ZEN_CONFIG.function_timeout_millis.load(Ordering::Relaxed),
375            http_auth: ZEN_CONFIG.http_auth.load(Ordering::Relaxed),
376            max_depth: 5,
377            validation_salt: 0,
378        }
379    }
380}
zen_engine/nodes/context.rs

zen_engine/nodes/
context.rs
