zen_engine/handler/function/
mod.rs

1use std::rc::Rc;
2use std::sync::atomic::Ordering;
3use std::time::Duration;
4
5use crate::handler::function::error::FunctionResult;
6use crate::handler::function::function::{Function, HandlerResponse};
7use crate::handler::function::module::console::Log;
8use crate::handler::function::serde::JsValue;
9use crate::handler::node::{NodeError, NodeRequest, NodeResponse, NodeResult};
10use crate::model::{DecisionNodeKind, FunctionNodeContent};
11use crate::ZEN_CONFIG;
12use ::serde::{Deserialize, Serialize};
13use anyhow::anyhow;
14use rquickjs::{async_with, CatchResultExt, Object};
15use serde_json::json;
16use zen_expression::variable::ToVariable;
17
18pub(crate) mod error;
19pub(crate) mod function;
20pub(crate) mod listener;
21pub(crate) mod module;
22pub(crate) mod serde;
23
24#[derive(Serialize, Deserialize)]
25pub struct FunctionResponse {
26    performance: String,
27    data: Option<HandlerResponse>,
28}
29
30pub struct FunctionHandler {
31    function: Rc<Function>,
32    trace: bool,
33    iteration: u8,
34    max_depth: u8,
35    max_duration: Duration,
36}
37
38#[derive(ToVariable)]
39#[serde(rename_all = "camelCase")]
40struct FunctionTrace {
41    pub log: Vec<Log>,
42}
43
44impl FunctionHandler {
45    pub fn new(function: Rc<Function>, trace: bool, iteration: u8, max_depth: u8) -> Self {
46        let max_duration_millis = ZEN_CONFIG.function_timeout_millis.load(Ordering::Relaxed);
47
48        Self {
49            function,
50            trace,
51            iteration,
52            max_depth,
53            max_duration: Duration::from_millis(max_duration_millis),
54        }
55    }
56
57    pub async fn handle(&self, request: NodeRequest) -> NodeResult {
58        let content = match &request.node.kind {
59            DecisionNodeKind::FunctionNode { content } => match content {
60                FunctionNodeContent::Version2(content) => Ok(content),
61                _ => Err(anyhow!("Unexpected node type")),
62            },
63            _ => Err(anyhow!("Unexpected node type")),
64        }?;
65
66        let start = std::time::Instant::now();
67        if content.omit_nodes {
68            request.input.dot_remove("$nodes");
69        }
70
71        let module_name = self
72            .function
73            .suggest_module_name(request.node.id.as_str(), &content.source);
74
75        let max_duration = self.max_duration.clone();
76        let interrupt_handler = Box::new(move || start.elapsed() > max_duration);
77        self.function
78            .runtime()
79            .set_interrupt_handler(Some(interrupt_handler))
80            .await;
81
82        self.attach_globals()
83            .await
84            .map_err(|e| anyhow!(e.to_string()))?;
85
86        self.function
87            .register_module(&module_name, content.source.as_str())
88            .await
89            .map_err(|e| anyhow!(e.to_string()))?;
90
91        let response_result = self
92            .function
93            .call_handler(&module_name, JsValue(request.input.clone()))
94            .await;
95
96        match response_result {
97            Ok(response) => {
98                self.function.runtime().set_interrupt_handler(None).await;
99
100                Ok(NodeResponse {
101                    output: response.data,
102                    trace_data: self.trace.then(|| {
103                        FunctionTrace {
104                            log: response.logs.clone(),
105                        }
106                        .to_variable()
107                    }),
108                })
109            }
110            Err(e) => {
111                let mut log = self.function.extract_logs().await;
112                log.push(Log {
113                    lines: vec![json!(e.to_string()).to_string()],
114                    ms_since_run: start.elapsed().as_millis() as usize,
115                });
116
117                Err(NodeError::PartialTrace {
118                    message: e.to_string(),
119                    trace: Some(FunctionTrace { log }.to_variable()),
120                })
121            }
122        }
123    }
124
125    async fn attach_globals(&self) -> FunctionResult {
126        async_with!(self.function.context() => |ctx| {
127            let config = Object::new(ctx.clone()).catch(&ctx)?;
128
129            config.prop("iteration", self.iteration).catch(&ctx)?;
130            config.prop("maxDepth", self.max_depth).catch(&ctx)?;
131            config.prop("trace", self.trace).catch(&ctx)?;
132
133            ctx.globals().set("config", config).catch(&ctx)?;
134
135            Ok(())
136        })
137        .await
138    }
139}
zen_engine/handler/function/mod.rs

zen_engine/handler/function/
mod.rs
