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

zen_engine/nodes/
context.rs
