http_nu/
engine.rs

1use std::path::Path;
2use std::sync::{atomic::AtomicBool, Arc};
3
4use tokio_util::sync::CancellationToken;
5
6use nu_cli::{add_cli_context, gather_parent_env_vars};
7use nu_cmd_lang::create_default_context;
8use nu_command::add_shell_command_context;
9use nu_engine::eval_block_with_early_return;
10use nu_parser::parse;
11use nu_plugin_engine::{GetPlugin, PluginDeclaration};
12use nu_protocol::engine::Command;
13use nu_protocol::format_cli_error;
14use nu_protocol::{
15    debugger::WithoutDebug,
16    engine::{Closure, EngineState, Redirection, Stack, StateWorkingSet},
17    OutDest, PipelineData, PluginIdentity, RegisteredPlugin, ShellError, Signals, Span, Type,
18    Value,
19};
20
21use crate::commands::{
22    HighlightCommand, HighlightLangCommand, HighlightThemeCommand, MdCommand, MjCommand,
23    MjCompileCommand, MjRenderCommand, PrintCommand, ReverseProxyCommand, StaticCommand, ToSse,
24};
25use crate::logging::log_error;
26use crate::stdlib::load_http_nu_stdlib;
27use crate::Error;
28
29/// CLI options exposed to scripts as the `$HTTP_NU` const
30#[derive(Clone, Default)]
31pub struct HttpNuOptions {
32    pub dev: bool,
33    pub datastar: bool,
34    pub watch: bool,
35    pub store: Option<String>,
36    pub topic: Option<String>,
37    pub expose: Option<String>,
38    pub tls: Option<String>,
39    pub services: bool,
40}
41
42#[derive(Clone)]
43pub struct Engine {
44    pub state: EngineState,
45    pub closure: Option<Closure>,
46    /// Cancellation token for SSE streams
47    pub sse_cancel_token: CancellationToken,
48}
49
50impl Engine {
51    pub fn new() -> Result<Self, Error> {
52        let mut engine_state = create_default_context();
53
54        engine_state = add_shell_command_context(engine_state);
55        engine_state = add_cli_context(engine_state);
56        engine_state = nu_cmd_extra::extra::add_extra_command_context(engine_state);
57
58        load_http_nu_stdlib(&mut engine_state)?;
59        nu_std::load_standard_library(&mut engine_state)?;
60
61        let init_cwd = std::env::current_dir()?;
62        gather_parent_env_vars(&mut engine_state, init_cwd.as_ref());
63
64        Ok(Self {
65            state: engine_state,
66            closure: None,
67            sse_cancel_token: CancellationToken::new(),
68        })
69    }
70
71    /// Sets `$HTTP_NU` const with server configuration for stdlib modules
72    pub fn set_http_nu_const(&mut self, options: &HttpNuOptions) -> Result<(), Error> {
73        let span = Span::unknown();
74        let opt_str = |v: &Option<String>| match v {
75            Some(s) => Value::string(s, span),
76            None => Value::nothing(span),
77        };
78        let record = Value::record(
79            nu_protocol::record! {
80                "dev" => Value::bool(options.dev, span),
81                "datastar" => Value::bool(options.datastar, span),
82                "watch" => Value::bool(options.watch, span),
83                "store" => opt_str(&options.store),
84                "topic" => opt_str(&options.topic),
85                "expose" => opt_str(&options.expose),
86                "tls" => opt_str(&options.tls),
87                "services" => Value::bool(options.services, span),
88            },
89            span,
90        );
91        let mut working_set = StateWorkingSet::new(&self.state);
92        let var_id = working_set.add_variable(b"$HTTP_NU".into(), span, Type::record(), false);
93        working_set.set_variable_const_val(var_id, record);
94        self.state.merge_delta(working_set.render())?;
95        Ok(())
96    }
97
98    pub fn add_commands(&mut self, commands: Vec<Box<dyn Command>>) -> Result<(), Error> {
99        let mut working_set = StateWorkingSet::new(&self.state);
100        for command in commands {
101            working_set.add_decl(command);
102        }
103        self.state.merge_delta(working_set.render())?;
104        Ok(())
105    }
106
107    /// Load a Nushell plugin from the given path
108    pub fn load_plugin(&mut self, path: &Path) -> Result<(), Error> {
109        // Canonicalize the path
110        let path = path.canonicalize().map_err(|e| {
111            Error::from(format!("Failed to canonicalize plugin path {path:?}: {e}"))
112        })?;
113
114        // Create the plugin identity
115        let identity = PluginIdentity::new(&path, None).map_err(|_| {
116            Error::from(format!(
117                "Invalid plugin path {path:?}: must be named nu_plugin_*"
118            ))
119        })?;
120
121        let mut working_set = StateWorkingSet::new(&self.state);
122
123        // Add plugin to working set and get handle
124        let plugin = nu_plugin_engine::add_plugin_to_working_set(&mut working_set, &identity)?;
125
126        // Merge working set to make plugin available
127        self.state.merge_delta(working_set.render())?;
128
129        // Spawn the plugin to get its signatures
130        let interface = plugin.clone().get_plugin(None)?;
131
132        // Set plugin metadata
133        plugin.set_metadata(Some(interface.get_metadata()?));
134
135        // Add command declarations from plugin signatures
136        let mut working_set = StateWorkingSet::new(&self.state);
137        for signature in interface.get_signature()? {
138            let decl = PluginDeclaration::new(plugin.clone(), signature);
139            working_set.add_decl(Box::new(decl));
140        }
141        self.state.merge_delta(working_set.render())?;
142
143        Ok(())
144    }
145
146    pub fn parse_closure(&mut self, script: &str, file: Option<&Path>) -> Result<(), Error> {
147        self.state.file = file.map(|p| p.to_path_buf());
148        let fname = file.map(|p| p.to_string_lossy().into_owned());
149        let mut working_set = StateWorkingSet::new(&self.state);
150        let block = parse(&mut working_set, fname.as_deref(), script.as_bytes(), false);
151
152        // Handle parse errors
153        if let Some(err) = working_set.parse_errors.first() {
154            let shell_error = ShellError::GenericError {
155                error: "Parse error".into(),
156                msg: format!("{err:?}"),
157                span: Some(err.span()),
158                help: None,
159                inner: vec![],
160            };
161            return Err(Error::from(format_cli_error(
162                None,
163                &working_set,
164                &shell_error,
165                None,
166            )));
167        }
168
169        // Handle compile errors
170        if let Some(err) = working_set.compile_errors.first() {
171            let shell_error = ShellError::GenericError {
172                error: format!("Compile error {err}"),
173                msg: "".into(),
174                span: None,
175                help: None,
176                inner: vec![],
177            };
178            return Err(Error::from(format_cli_error(
179                None,
180                &working_set,
181                &shell_error,
182                None,
183            )));
184        }
185
186        self.state.merge_delta(working_set.render())?;
187
188        let mut stack = Stack::new();
189        let result = eval_block_with_early_return::<WithoutDebug>(
190            &self.state,
191            &mut stack,
192            &block,
193            PipelineData::empty(),
194        )
195        .map_err(|err| {
196            let working_set = StateWorkingSet::new(&self.state);
197            Error::from(format_cli_error(None, &working_set, &err, None))
198        })?;
199
200        let closure = result
201            .body
202            .into_value(Span::unknown())
203            .map_err(|err| {
204                let working_set = StateWorkingSet::new(&self.state);
205                Error::from(format_cli_error(None, &working_set, &err, None))
206            })?
207            .into_closure()
208            .map_err(|err| {
209                let working_set = StateWorkingSet::new(&self.state);
210                Error::from(format_cli_error(None, &working_set, &err, None))
211            })?;
212
213        // Verify closure accepts exactly one argument
214        let block = self.state.get_block(closure.block_id);
215        if block.signature.required_positional.len() != 1 {
216            return Err(format!(
217                "Closure must accept exactly one request argument, found {}",
218                block.signature.required_positional.len()
219            )
220            .into());
221        }
222
223        self.state.merge_env(&mut stack)?;
224
225        self.closure = Some(closure);
226        Ok(())
227    }
228
229    /// Sets the interrupt signal for the engine
230    pub fn set_signals(&mut self, interrupt: Arc<AtomicBool>) {
231        self.state.set_signals(Signals::new(interrupt));
232    }
233
234    /// Sets NU_LIB_DIRS const for module resolution
235    pub fn set_lib_dirs(&mut self, paths: &[std::path::PathBuf]) -> Result<(), Error> {
236        if paths.is_empty() {
237            return Ok(());
238        }
239        let span = Span::unknown();
240        let vals: Vec<Value> = paths
241            .iter()
242            .map(|p| Value::string(p.to_string_lossy(), span))
243            .collect();
244
245        let mut working_set = StateWorkingSet::new(&self.state);
246        let var_id = working_set.add_variable(
247            b"$NU_LIB_DIRS".into(),
248            span,
249            Type::List(Box::new(Type::String)),
250            false,
251        );
252        working_set.set_variable_const_val(var_id, Value::list(vals, span));
253        self.state.merge_delta(working_set.render())?;
254        Ok(())
255    }
256
257    /// Evaluate a script string and return the result value
258    pub fn eval(&mut self, script: &str, file: Option<&Path>) -> Result<Value, Error> {
259        self.state.file = file.map(|p| p.to_path_buf());
260        let fname = file.map(|p| p.to_string_lossy().into_owned());
261        let mut working_set = StateWorkingSet::new(&self.state);
262        let block = parse(&mut working_set, fname.as_deref(), script.as_bytes(), false);
263
264        if let Some(err) = working_set.parse_errors.first() {
265            let shell_error = ShellError::GenericError {
266                error: "Parse error".into(),
267                msg: format!("{err:?}"),
268                span: Some(err.span()),
269                help: None,
270                inner: vec![],
271            };
272            return Err(Error::from(format_cli_error(
273                None,
274                &working_set,
275                &shell_error,
276                None,
277            )));
278        }
279
280        if let Some(err) = working_set.compile_errors.first() {
281            let shell_error = ShellError::GenericError {
282                error: format!("Compile error {err}"),
283                msg: "".into(),
284                span: None,
285                help: None,
286                inner: vec![],
287            };
288            return Err(Error::from(format_cli_error(
289                None,
290                &working_set,
291                &shell_error,
292                None,
293            )));
294        }
295
296        // Clone engine state and merge the parsed block
297        let mut engine_state = self.state.clone();
298        engine_state.merge_delta(working_set.render())?;
299
300        let mut stack = Stack::new();
301        let result = eval_block_with_early_return::<WithoutDebug>(
302            &engine_state,
303            &mut stack,
304            &block,
305            PipelineData::empty(),
306        )
307        .map_err(|err| {
308            let working_set = StateWorkingSet::new(&engine_state);
309            Error::from(format_cli_error(None, &working_set, &err, None))
310        })?;
311
312        result.body.into_value(Span::unknown()).map_err(|err| {
313            let working_set = StateWorkingSet::new(&engine_state);
314            Error::from(format_cli_error(None, &working_set, &err, None))
315        })
316    }
317
318    /// Run the parsed closure with input value and pipeline data
319    pub fn run_closure(
320        &self,
321        input: Value,
322        pipeline_data: PipelineData,
323    ) -> Result<PipelineData, Error> {
324        let closure = self.closure.as_ref().ok_or("Closure not parsed")?;
325
326        let mut stack = Stack::new().captures_to_stack(closure.captures.clone());
327        let mut stack =
328            stack.push_redirection(Some(Redirection::Pipe(OutDest::PipeSeparate)), None);
329        let block = self.state.get_block(closure.block_id);
330
331        stack.add_var(
332            block.signature.required_positional[0].var_id.unwrap(),
333            input,
334        );
335
336        eval_block_with_early_return::<WithoutDebug>(&self.state, &mut stack, block, pipeline_data)
337            .map(|exec_data| exec_data.body)
338            .map_err(|err| {
339                let working_set = StateWorkingSet::new(&self.state);
340                Error::from(format_cli_error(None, &working_set, &err, None))
341            })
342    }
343
344    /// Adds http-nu custom commands to the engine
345    pub fn add_custom_commands(&mut self) -> Result<(), Error> {
346        self.add_commands(vec![
347            Box::new(ReverseProxyCommand::new()),
348            Box::new(StaticCommand::new()),
349            Box::new(ToSse {}),
350            Box::new(MjCommand::new()),
351            Box::new(MjCompileCommand::new()),
352            Box::new(MjRenderCommand::new()),
353            Box::new(HighlightCommand::new()),
354            Box::new(HighlightThemeCommand::new()),
355            Box::new(HighlightLangCommand::new()),
356            Box::new(MdCommand::new()),
357            Box::new(PrintCommand::new()),
358        ])
359    }
360
361    /// Adds cross.stream store commands (.cat, .append, .cas, .last) to the engine
362    #[cfg(feature = "cross-stream")]
363    pub fn add_store_commands(&mut self, store: &xs::store::Store) -> Result<(), Error> {
364        self.add_commands(vec![
365            Box::new(xs::nu::commands::cat_stream_command::CatStreamCommand::new(
366                store.clone(),
367            )),
368            Box::new(xs::nu::commands::append_command::AppendCommand::new(
369                store.clone(),
370                serde_json::json!({}),
371            )),
372            Box::new(xs::nu::commands::cas_command::CasCommand::new(
373                store.clone(),
374            )),
375            Box::new(xs::nu::commands::last_stream_command::LastStreamCommand::new(store.clone())),
376            Box::new(xs::nu::commands::get_command::GetCommand::new(
377                store.clone(),
378            )),
379            Box::new(xs::nu::commands::remove_command::RemoveCommand::new(
380                store.clone(),
381            )),
382            Box::new(xs::nu::commands::scru128_command::Scru128Command::new()),
383        ])
384    }
385
386    /// Re-registers .mj commands with store access for stream-backed template loading
387    #[cfg(feature = "cross-stream")]
388    pub fn add_store_mj_commands(&mut self, store: &xs::store::Store) -> Result<(), Error> {
389        self.add_commands(vec![
390            Box::new(MjCommand::with_store(store.clone())),
391            Box::new(MjCompileCommand::with_store(store.clone())),
392        ])
393    }
394}
395
396/// Creates an engine from a script by cloning a base engine and parsing the closure.
397/// On error, prints to stderr and emits JSON to stdout, returning None.
398pub fn script_to_engine(base: &Engine, script: &str, file: Option<&Path>) -> Option<Engine> {
399    let mut engine = base.clone();
400    // Fresh cancellation token for this engine instance
401    engine.sse_cancel_token = CancellationToken::new();
402
403    if let Err(e) = engine.parse_closure(script, file) {
404        log_error(&nu_utils::strip_ansi_string_likely(e.to_string()));
405        return None;
406    }
407
408    Some(engine)
409}
http_nu/engine.rs

http_nu/
engine.rs
