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        let start = std::time::Instant::now();
60
61        let module_name = self
62            .function
63            .suggest_module_name(request.node.id.as_str(), &content.source);
64
65        let max_duration = self.max_duration.clone();
66        let interrupt_handler = Box::new(move || start.elapsed() > max_duration);
67        self.function
68            .runtime()
69            .set_interrupt_handler(Some(interrupt_handler))
70            .await;
71
72        self.attach_globals()
73            .await
74            .map_err(|e| anyhow!(e.to_string()))?;
75
76        self.function
77            .register_module(&module_name, content.source.as_str())
78            .await
79            .map_err(|e| anyhow!(e.to_string()))?;
80
81        let response_result = self
82            .function
83            .call_handler(&module_name, JsValue(request.input.clone()))
84            .await;
85
86        match response_result {
87            Ok(response) => {
88                self.function.runtime().set_interrupt_handler(None).await;
89
90                Ok(NodeResponse {
91                    output: response.data,
92                    trace_data: self.trace.then(|| json!({ "log": response.logs })),
93                })
94            }
95            Err(e) => {
96                let mut log = self.function.extract_logs().await;
97                log.push(Log {
98                    lines: vec![json!(e.to_string()).to_string()],
99                    ms_since_run: start.elapsed().as_millis() as usize,
100                });
101
102                Err(anyhow!(PartialTraceError {
103                    message: e.to_string(),
104                    trace: Some(json!({ "log": log })),
105                }))
106            }
107        }
108    }
109
110    async fn attach_globals(&self) -> FunctionResult {
111        async_with!(self.function.context() => |ctx| {
112            let config = Object::new(ctx.clone()).catch(&ctx)?;
113
114            config.prop("iteration", self.iteration).catch(&ctx)?;
115            config.prop("maxDepth", self.max_depth).catch(&ctx)?;
116            config.prop("trace", self.trace).catch(&ctx)?;
117
118            ctx.globals().set("config", config).catch(&ctx)?;
119
120            Ok(())
121        })
122        .await
123    }
124}
zen_engine/handler/function/mod.rs

zen_engine/handler/function/
mod.rs
