zen_engine/handler/function/
mod.rs

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

zen_engine/handler/function/
mod.rs
