surrealdb-core 3.3.1

//! Closure expressions - closure creation and invocation.

use std::collections::HashMap;
use std::sync::Arc;

use surrealdb_strand::Strand;
use surrealdb_types::{SqlFormat, ToSql};

use super::helpers::{args_access_mode, args_required_context, evaluate_args, validate_return};
use crate::exec::physical_expr::{BlockPhysicalExpr, EvalContext, PhysicalExpr};
use crate::exec::{AccessMode, BoxFut, Error as ExecError};
use crate::expr::{ControlFlow, Error as ExprError, FlowResult};
use crate::val::Value;

// =============================================================================
// ClosureExec - closure value creation
// =============================================================================

/// Closure expression - |$x| $x * 2
#[derive(Debug, Clone)]
pub struct ClosureExec {
	pub(crate) closure: crate::expr::ClosureExpr,
}
impl PhysicalExpr for ClosureExec {
	fn name(&self) -> &'static str {
		"Closure"
	}

	fn as_any(&self) -> &dyn std::any::Any {
		self
	}

	fn required_context(&self) -> crate::exec::ContextLevel {
		// Closures are just values, they don't need any special context
		crate::exec::ContextLevel::Root
	}

	fn evaluate<'a>(&'a self, ctx: EvalContext<'a>) -> BoxFut<'a, FlowResult<Value>> {
		Box::pin(async move {
			use crate::dbs::ParameterCapturePass;
			use crate::val::Closure;

			// Capture parameters from the context that are referenced in the closure body
			let frozen_ctx = ctx.exec_ctx.ctx();
			let captures = ParameterCapturePass::capture(frozen_ctx, &self.closure.body);

			// Create a Value::Closure with the captured variables
			Ok(Value::Closure(Box::new(Closure::Expr {
				args: self.closure.args.clone(),
				returns: self.closure.returns.clone(),
				body: self.closure.body.clone(),
				captures,
			})))
		})
	}

	fn access_mode(&self) -> AccessMode {
		// A closure literal is inert as a value, but wherever it can flow it can
		// also be invoked — the call operator, or a closure-taking builtin such
		// as `array::map` that runs the body per element under this statement's
		// options. A writing body therefore makes the expression writable, so
		// anything combining this access mode (e.g. `args_access_mode` on the
		// builtin holding it) is conservative. Mirrors the expr layer's
		// `Expr::Closure(c) => c.body.read_only()`.
		if self.closure.body.read_only() {
			AccessMode::ReadOnly
		} else {
			AccessMode::ReadWrite
		}
	}
}

impl ToSql for ClosureExec {
	fn fmt_sql(&self, f: &mut String, fmt: SqlFormat) {
		self.closure.fmt_sql(f, fmt);
	}
}

// =============================================================================
// ClosureCallExec - for invoking closures stored in parameters
// =============================================================================

/// Closure call expression - $closure(args...)
///
/// Invokes a closure value with the provided arguments. The target expression
/// must evaluate to a `Value::Closure`.
#[derive(Debug, Clone)]
pub struct ClosureCallExec {
	/// The expression that evaluates to a closure value
	pub(crate) target: Arc<dyn PhysicalExpr>,
	/// The argument expressions to pass to the closure
	pub(crate) arguments: Vec<Arc<dyn PhysicalExpr>>,
}
impl PhysicalExpr for ClosureCallExec {
	fn name(&self) -> &'static str {
		"ClosureCall"
	}

	fn as_any(&self) -> &dyn std::any::Any {
		self
	}

	fn required_context(&self) -> crate::exec::ContextLevel {
		// The closure body is a runtime value, invisible at plan time, so
		// this can only declare what the target and argument expressions
		// need. The executor materializes the highest context level the
		// session's ns/db selection supports rather than just this minimum,
		// which is what lets a database-touching body run. Declaring
		// `Database` here instead would wrongly reject pure closures in
		// sessions with no database selected.
		let target_ctx = self.target.required_context();
		let args_ctx = args_required_context(&self.arguments);
		target_ctx.max(args_ctx)
	}

	fn evaluate<'a>(&'a self, ctx: EvalContext<'a>) -> BoxFut<'a, FlowResult<Value>> {
		Box::pin(async move {
			use crate::val::Closure;

			// 1. Evaluate the target expression to get the closure
			let target_value = self.target.evaluate(ctx.clone()).await?;

			let closure = match target_value {
				Value::Closure(c) => c,
				other => {
					return Err(ExecError::InvalidFunction {
						name: "ANONYMOUS".to_string(),
						message: format!("'{}' is not a function", other.kind_of()),
					}
					.into());
				}
			};

			// 2. Evaluate all argument expressions
			let evaluated_args = evaluate_args(&self.arguments, ctx.clone()).await?;

			// 3. Invoke the closure based on its type
			match closure.as_ref() {
				Closure::Expr {
					args: arg_spec,
					returns,
					body,
					captures,
				} => {
					// Create isolated execution context with captured variables
					let mut isolated_ctx = ctx.exec_ctx.clone();
					for (name, value) in captures.clone() {
						isolated_ctx = isolated_ctx.with_param(name, value);
					}

					// Check for missing required arguments
					if arg_spec.len() > evaluated_args.len()
						&& let Some((param, kind)) =
							arg_spec[evaluated_args.len()..].iter().find(|(_, k)| !k.can_be_none())
					{
						return Err(ExprError::InvalidFunctionArguments {
							name: "ANONYMOUS".to_string(),
							message: format!(
								"Expected a value of type '{}' for argument {}",
								kind.to_sql(),
								param.to_sql()
							),
						}
						.into());
					}

					// Bind arguments to parameter names with type coercion
					let mut local_params: HashMap<Strand, Value> = HashMap::new();
					for ((param, kind), arg_value) in arg_spec.iter().zip(evaluated_args) {
						let coerced = arg_value.coerce_to_kind(kind).map_err(|_| {
							ExprError::InvalidFunctionArguments {
								name: "ANONYMOUS".to_string(),
								message: format!(
									"Expected a value of type '{}' for argument {}",
									kind.to_sql(),
									param.to_sql()
								),
							}
						})?;
						local_params.insert(param.clone().into_strand(), coerced);
					}

					// Add parameters to the execution context
					for (name, value) in &local_params {
						isolated_ctx = isolated_ctx.with_param(name.clone(), value.clone());
					}

					// Execute the closure body
					let block_expr = BlockPhysicalExpr {
						block: crate::expr::Block(vec![body.clone()]),
						// A closure body is evaluated wherever the closure is
						// called, which need not be the statement that wrote
						// it, so it inherits no MATCHES registrations.
						matches_scope: None,
					};
					let eval_ctx = EvalContext {
						exec_ctx: &isolated_ctx,
						current_value: ctx.current_value,
						local_params: Some(&local_params),
						recursion_ctx: None,
						document_root: None,
						skip_fetch_perms: ctx.skip_fetch_perms,
						computing_record: ctx.computing_record.clone(),
						// The closure body is one re-entry deeper — continue the
						// depth count so nested recursion stays bounded.
						plan_depth: ctx.plan_depth + 1,
					};

					let result = match block_expr.evaluate(eval_ctx).await {
						Ok(v) => v,
						Err(ControlFlow::Return(v)) => v,
						Err(ControlFlow::Break) | Err(ControlFlow::Continue) => {
							// BREAK/CONTINUE inside a closure (outside of loop) is an error
							return Err(ExecError::InvalidControlFlow.into());
						}
						Err(e) => return Err(e),
					};

					// Coerce return value to declared type if specified
					Ok(validate_return("ANONYMOUS", returns.as_ref(), result)?)
				}
				Closure::Builtin(_) => {
					// Builtin closures are not yet supported in the streaming executor
					// They require the legacy compute path with Stk
					Err(anyhow::anyhow!(
						"Builtin closures are not yet supported in the streaming executor"
					)
					.into())
				}
			}
		})
	}

	fn access_mode(&self) -> AccessMode {
		// Closures can potentially do anything, so be conservative
		AccessMode::ReadWrite
			.combine(self.target.access_mode())
			.combine(args_access_mode(&self.arguments))
	}
}

impl ToSql for ClosureCallExec {
	fn fmt_sql(&self, f: &mut String, fmt: SqlFormat) {
		self.target.fmt_sql(f, fmt);
		f.push_str("(...)");
	}
}

#[cfg(all(test, feature = "kv-mem"))]
#[allow(clippy::unwrap_used)]
mod tests {
	use crate::exec::AccessMode;
	use crate::exec::operators::test_util::TestDb;
	use crate::exec::planner::Planner;

	/// Plan `expression` against a txn-backed planner over `db` and return the
	/// resolved access mode of the resulting physical expression tree.
	async fn access_mode_of(db: &TestDb, expression: &str) -> AccessMode {
		let ctx = db.exec_ctx().await;
		let crate::exec::ExecutionContext::Database(db_ctx) = &ctx else {
			panic!("exec_ctx builds a Database context");
		};
		let txn = ctx.txn();
		let planner = Planner::with_txn(
			ctx.ctx(),
			&db_ctx.ns_ctx.root.function_registry,
			txn,
			Some("test".to_owned()),
			Some("test".to_owned()),
		);
		let expr: crate::expr::Expr = crate::syn::expr(expression).unwrap().into();
		planner.physical_expr(expr).await.unwrap().access_mode()
	}

	#[tokio::test]
	async fn a_pure_closure_literal_is_read_only() {
		let db = TestDb::new("").await;
		assert_eq!(access_mode_of(&db, "|$x: any| { RETURN $x; }").await, AccessMode::ReadOnly);
	}

	#[tokio::test]
	async fn a_writing_closure_literal_is_read_write() {
		// A closure does not write when it is created, but it can be invoked
		// wherever the value flows, so a writing body makes the expression
		// writable — matching the expr layer's `Expr::Closure(c) => c.body.read_only()`.
		let db = TestDb::new("DEFINE TABLE log SCHEMALESS;").await;
		assert_eq!(
			access_mode_of(&db, "|$x: any| { CREATE log SET v = $x; }").await,
			AccessMode::ReadWrite
		);
	}

	#[tokio::test]
	async fn a_writing_closure_passed_to_a_builtin_is_read_write() {
		// `array::map` runs the closure per element under this statement's
		// options, so a writing closure argument makes the whole call writable.
		// Feeding the closure body into the builtin's access mode is what closes
		// the gap: otherwise the call would resolve read-only and could be
		// overlapped or spawned as if it never wrote.
		let db = TestDb::new("DEFINE TABLE log SCHEMALESS;").await;
		assert_eq!(
			access_mode_of(&db, "array::map([1, 2], |$x: any| { CREATE log SET v = $x; })").await,
			AccessMode::ReadWrite
		);
		// A pure closure argument keeps the call read-only — no regression for
		// the common case.
		assert_eq!(
			access_mode_of(&db, "array::map([1, 2], |$x: any| { RETURN $x; })").await,
			AccessMode::ReadOnly
		);
	}
}