featherbit/plugins/native/
serverless_pre_function.rs

1//! The `serverless-pre-function` node — runs one or more inline Lua functions
2//! against the `Context`, threading it through each in sequence.
3//!
4//! Port of APISIX's `serverless-pre-function` plugin. In APISIX the two
5//! serverless plugins (`serverless-pre-function` / `serverless-post-function`)
6//! are identical except for the phase they run in; here they share all logic
7//! ([`ServerlessRunner`]) and differ only in their registered node-type name.
8//!
9//! ## Deviations from APISIX
10//!
11//! - **Function contract.** APISIX functions are `return function(conf, ctx)
12//!   ... end` chunks invoked with `(conf, ctx)`. featherbit reuses the
13//!   `script` plugin's Lua runtime, so each function is a script that defines a
14//!   global `function execute(ctx) ... return ctx end`, receiving and returning
15//!   the marshalled Context table (see the `script` plugin docs for the table
16//!   shape). This is the same contract as the `script` node.
17//! - **Phase by graph position.** APISIX's `phase` field selects a request
18//!   lifecycle phase. featherbit expresses phase through *placement in the
19//!   policy graph*: a `serverless-pre-function` node is wired before the
20//!   `upstream` node, a `serverless-post-function` node after it. The `phase`
21//!   config key is accepted for compatibility but is inert.
22
23use async_trait::async_trait;
24use std::collections::HashMap;
25use std::path::PathBuf;
26
27use crate::context::Context;
28use crate::plugins::script::lua_runtime::LuaRuntime;
29use crate::plugins::{Plugin, PluginExecutionError, PluginOutput, PluginResult};
30
31/// Shared engine behind both serverless nodes: a compiled list of Lua
32/// functions run in order against the Context.
33///
34/// Each entry is a fully validated [`LuaRuntime`] (compiled at config load, so
35/// bad Lua fails policy compilation, not a live request). At request time the
36/// Context is threaded through every function: the table returned by one
37/// function's `execute` becomes the input to the next. If any function errors,
38/// that error is propagated (routing the Context through the node's `error`
39/// port); otherwise the final Context flows through the `success` port.
40pub struct ServerlessRunner {
41    functions: Vec<LuaRuntime>,
42    plugin_type: &'static str,
43}
44
45impl ServerlessRunner {
46    /// Builds the runner from node config, compiling every function up front.
47    ///
48    /// Accepted keys:
49    /// - `functions` (array of strings, **required**, ≥1): each string is Lua
50    ///   source defining a global `execute(ctx)` function. Each is compiled and
51    ///   validated here; an empty array, a non-string entry, a Lua syntax
52    ///   error, or a missing `execute` all fail at config load.
53    /// - `phase` (string, optional): accepted for APISIX compatibility but
54    ///   **inert** — placement in the graph determines phase.
55    /// - `timeout_ms` (integer, default `5000`): per-function execution timeout
56    ///   passed to the Lua runtime (stored, not yet enforced by the VM).
57    pub fn from_config(
58        config: &HashMap<String, serde_json::Value>,
59        plugin_type: &'static str,
60    ) -> Result<Self, String> {
61        let raw = config
62            .get("functions")
63            .ok_or_else(|| format!("{}: 'functions' is required", plugin_type))?;
64
65        let arr = raw.as_array().ok_or_else(|| {
66            format!(
67                "{}: 'functions' must be an array of Lua strings",
68                plugin_type
69            )
70        })?;
71
72        if arr.is_empty() {
73            return Err(format!(
74                "{}: 'functions' must contain at least one function",
75                plugin_type
76            ));
77        }
78
79        let timeout_ms = config
80            .get("timeout_ms")
81            .and_then(|v| v.as_u64())
82            .unwrap_or(5000);
83
84        let modules_path: Option<PathBuf> = config
85            .get("modules_path")
86            .and_then(|v| v.as_str())
87            .map(PathBuf::from);
88
89        let mut functions = Vec::with_capacity(arr.len());
90        for (i, item) in arr.iter().enumerate() {
91            let src = item.as_str().ok_or_else(|| {
92                format!(
93                    "{}: functions[{}] must be a Lua source string",
94                    plugin_type, i
95                )
96            })?;
97            let rt = LuaRuntime::new(src, timeout_ms, modules_path.clone()).map_err(|e| {
98                format!("{}: functions[{}] failed to compile: {}", plugin_type, i, e)
99            })?;
100            functions.push(rt);
101        }
102
103        Ok(Self {
104            functions,
105            plugin_type,
106        })
107    }
108
109    /// Runs each function in order, threading the Context. Propagates the first
110    /// error encountered.
111    pub fn run(&self, mut ctx: Context) -> Result<Context, PluginExecutionError> {
112        for func in &self.functions {
113            ctx = func.execute(ctx)?;
114        }
115        Ok(ctx)
116    }
117}
118
119/// The `serverless-pre-function` node. Runs its Lua functions before the
120/// upstream call (by convention of its graph placement).
121pub struct ServerlessPreFunctionPlugin {
122    runner: ServerlessRunner,
123}
124
125impl ServerlessPreFunctionPlugin {
126    /// Builds the plugin from node config.
127    ///
128    /// ```yaml
129    /// type: serverless-pre-function
130    /// config:
131    ///   phase: access          # accepted for compatibility; inert
132    ///   timeout_ms: 2000
133    ///   functions:
134    ///     - |
135    ///       function execute(ctx)
136    ///         ctx.request.headers["x-serverless"] = {"pre"}
137    ///         return ctx
138    ///       end
139    /// ```
140    pub fn from_config(config: &HashMap<String, serde_json::Value>) -> Result<Self, String> {
141        Ok(Self {
142            runner: ServerlessRunner::from_config(config, "serverless-pre-function")?,
143        })
144    }
145}
146
147#[async_trait]
148impl Plugin for ServerlessPreFunctionPlugin {
149    fn plugin_type(&self) -> &str {
150        self.runner.plugin_type
151    }
152
153    async fn execute(
154        &self,
155        ctx: Context,
156        _named_inputs: &HashMap<String, serde_json::Value>,
157    ) -> PluginResult {
158        let ctx = self.runner.run(ctx)?;
159        Ok(PluginOutput {
160            context: ctx,
161            named_outputs: HashMap::new(),
162        })
163    }
164}
165
166#[cfg(test)]
167mod tests {
168    use super::*;
169    use crate::context::{GatewayRequest, GatewayResponse, Protocol};
170
171    fn test_context() -> Context {
172        Context {
173            request: GatewayRequest {
174                method: "GET".to_string(),
175                path: "/test".to_string(),
176                host: "localhost".to_string(),
177                scheme: "http".to_string(),
178                headers: HashMap::new(),
179                query_params: HashMap::new(),
180                body: bytes::Bytes::new(),
181                remote_addr: "127.0.0.1:1234".to_string(),
182                protocol: Protocol::Http1,
183            },
184            response: GatewayResponse {
185                status_code: 0,
186                headers: HashMap::new(),
187                body: bytes::Bytes::new(),
188            },
189            message: HashMap::new(),
190            errors: Vec::new(),
191        }
192    }
193
194    fn cfg(functions: serde_json::Value) -> HashMap<String, serde_json::Value> {
195        let mut c = HashMap::new();
196        c.insert("functions".to_string(), functions);
197        c
198    }
199
200    #[tokio::test]
201    async fn test_serverless_pre_function_mutates_and_threads() {
202        // Two functions: first sets a header, second reads it into message.
203        let p = ServerlessPreFunctionPlugin::from_config(&cfg(serde_json::json!([
204            "function execute(ctx)\n  ctx.request.headers[\"x-step\"] = {\"one\"}\n  return ctx\nend",
205            "function execute(ctx)\n  ctx.message.seen = ctx.request.headers[\"x-step\"][1]\n  return ctx\nend"
206        ])))
207        .unwrap();
208
209        let out = p.execute(test_context(), &HashMap::new()).await.unwrap();
210        assert_eq!(
211            out.context.request.headers.get("x-step"),
212            Some(&vec!["one".to_string()])
213        );
214        // ordering: second function observed the first's mutation
215        assert_eq!(
216            out.context.message.get("seen"),
217            Some(&serde_json::json!("one"))
218        );
219    }
220
221    #[tokio::test]
222    async fn test_serverless_pre_function_order() {
223        // Each function appends to a message array; order must be preserved.
224        let p = ServerlessPreFunctionPlugin::from_config(&cfg(serde_json::json!([
225            "function execute(ctx)\n  ctx.message.trail = {\"a\"}\n  return ctx\nend",
226            "function execute(ctx)\n  local t = ctx.message.trail\n  t[#t+1] = \"b\"\n  ctx.message.trail = t\n  return ctx\nend"
227        ])))
228        .unwrap();
229
230        let out = p.execute(test_context(), &HashMap::new()).await.unwrap();
231        assert_eq!(
232            out.context.message.get("trail"),
233            Some(&serde_json::json!(["a", "b"]))
234        );
235    }
236
237    #[test]
238    fn test_serverless_pre_function_compile_error_fails_config() {
239        // Syntax error in Lua fails from_config.
240        let err = ServerlessPreFunctionPlugin::from_config(&cfg(serde_json::json!([
241            "function execute(ctx) this is not lua"
242        ])));
243        assert!(err.is_err());
244
245        // Missing execute function fails too.
246        let err =
247            ServerlessPreFunctionPlugin::from_config(&cfg(serde_json::json!(["local x = 1"])));
248        assert!(err.is_err());
249    }
250
251    #[test]
252    fn test_serverless_pre_function_empty_functions_rejected() {
253        assert!(
254            ServerlessPreFunctionPlugin::from_config(&cfg(serde_json::json!([])).clone()).is_err()
255        );
256        assert!(ServerlessPreFunctionPlugin::from_config(&HashMap::new()).is_err());
257        // non-string entry
258        assert!(ServerlessPreFunctionPlugin::from_config(&cfg(serde_json::json!([123]))).is_err());
259    }
260
261    #[tokio::test]
262    async fn test_serverless_pre_function_error_propagates() {
263        let p = ServerlessPreFunctionPlugin::from_config(&cfg(serde_json::json!([
264            "function execute(ctx)\n  error(\"boom\")\n  return ctx\nend"
265        ])))
266        .unwrap();
267        let err = p
268            .execute(test_context(), &HashMap::new())
269            .await
270            .unwrap_err();
271        assert_eq!(err.error.code, "LUA_EXECUTION_ERROR");
272    }
273}
featherbit/plugins/native/serverless_pre_function.rs

featherbit/plugins/native/
serverless_pre_function.rs
