surrealdb-core 3.2.3

use std::collections::HashMap;
use std::fmt::{self, Debug};
#[cfg(storage)]
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;

#[cfg(feature = "surrealism")]
use anyhow::Context as _;
use anyhow::{Result, bail};
use surrealdb_strand::Strand;
#[cfg(feature = "surrealism")]
use surrealism_runtime::package::{SurrealismPackage, UnpackOptions};
#[cfg(feature = "surrealism")]
use surrealism_runtime::runtime::Runtime;
#[cfg(feature = "http")]
use url::Url;
use uuid::Uuid;
use web_time::Instant;

use crate::buc::manager::BucketsManager;
#[cfg(feature = "surrealism")]
use crate::buc::store::ObjectKey;
use crate::buc::store::ObjectStore;
use crate::catalog::providers::{CatalogProvider, DatabaseProvider, NamespaceProvider};
use crate::catalog::{DatabaseDefinition, DatabaseId, NamespaceId};
use crate::cnf::dynamic::DynamicConfiguration;
use crate::cnf::{CommonConfig, PROTECTED_PARAM_NAMES};
use crate::ctx::cancel::CancelHandle;
use crate::ctx::canceller::Canceller;
use crate::ctx::reason::Reason;
#[cfg(feature = "surrealism")]
use crate::dbs::capabilities::ExperimentalTarget;
#[cfg(feature = "http")]
use crate::dbs::capabilities::NetTarget;
#[cfg(all(feature = "http", feature = "surrealism"))]
use crate::dbs::capabilities::Targets;
use crate::dbs::{
	Capabilities, MessageBroker, NewPlannerStrategy, Options, Session, StatementCounters, Variables,
};
use crate::err::Error;
use crate::exec::function::FunctionRegistry;
use crate::expr::Base;
#[cfg(feature = "http")]
use crate::http::HttpClient;
use crate::iam::{Action, ResourceKind};
use crate::idx::planner::executor::QueryExecutor;
use crate::idx::planner::{IterationStage, QueryPlanner};
use crate::idx::trees::store::IndexStores;
use crate::kvs::Transaction;
use crate::kvs::cache::ds::DatastoreCache;
use crate::kvs::index::IndexBuilder;
use crate::kvs::sequences::Sequences;
use crate::kvs::slowlog::SlowLog;
use crate::mem::ALLOC;
use crate::sql::expression::convert_public_value_to_internal;
#[cfg(feature = "surrealism")]
use crate::surrealism::cache::{SurrealismCache, SurrealismCacheLookup, SurrealismCachedModule};
use crate::types::PublicVariables;
use crate::val::Value;

pub type FrozenContext = Arc<Context>;

/// Ambient state for one query batch: datastore handles, cancellation, capabilities,
/// and **request-wide** configuration (node id, auth enabled, dynamic config, live/broker).
///
/// Stored behind [`FrozenContext`] (`Arc<Context>`) with optional parent links for scoped
/// deadlines, planner state, and parameters. Per-statement toggles live on [`Options`], not here.
pub struct Context {
	// An optional parent context.
	parent: Option<FrozenContext>,
	// An optional deadline.
	deadline: Option<(Instant, Duration)>,
	// An optional slow log configuration used by the executor to log statements
	// that exceed a given duration threshold. This configuration is propagated
	// from the datastore into the context for the lifetime of a request.
	slow_log: Option<SlowLog>,
	// Whether or not this context is cancelled. Fast hot-path view checked
	// by [`Self::done`] at every executor yield.
	cancelled: Arc<AtomicBool>,
	// Awaitable view of the external cancellation signal, when one was
	// installed via [`Self::set_cancellation`]. Used by sites that bare-await
	// an external timer (e.g. `SLEEP`) so they can `select!` against the
	// cancel instead of running to completion. `None` for contexts without
	// an external cancel source (embedded callers, internal use).
	cancel_token: Option<tokio_util::sync::CancellationToken>,
	// A collection of read only values stored in this context.
	values: HashMap<Strand, Arc<Value>>,
	// An optional query planner
	query_planner: Option<Arc<QueryPlanner>>,
	// An optional query executor
	query_executor: Option<QueryExecutor>,
	// An optional iteration stage
	iteration_stage: Option<IterationStage>,
	// An optional datastore cache
	cache: Option<Arc<DatastoreCache>>,
	// The index store
	index_stores: IndexStores,
	// The index concurrent builders
	index_builder: Option<IndexBuilder>,
	// The sequences
	sequences: Option<Sequences>,
	// Capabilities
	capabilities: Arc<Capabilities>,
	#[cfg(storage)]
	// The temporary directory
	temporary_directory: Option<Arc<PathBuf>>,
	// An optional transaction
	transaction: Option<Arc<Transaction>>,
	// Does not read from parent `values`.
	isolated: bool,
	// A map of bucket connections
	buckets: Option<BucketsManager>,
	// The surrealism cache
	#[cfg(feature = "surrealism")]
	surrealism_cache: Option<Arc<SurrealismCache>>,
	// Function registry for built-in and custom functions
	function_registry: Arc<FunctionRegistry>,
	// Strategy for the new streaming planner/executor
	new_planner_strategy: NewPlannerStrategy,
	// When true, EXPLAIN ANALYZE omits elapsed durations for deterministic test output
	redact_volatile_explain_attrs: bool,
	// Per-statement counters, shared with the executor so it can read the
	// number of rows affected by the running DML statement when emitting
	// the corresponding `StatementEvent`. Replaced by the executor before
	// each top-level statement; `None` outside an active statement.
	statement_counters: Option<Arc<StatementCounters>>,
	// Pre-resolved tenant identity (namespace, database, user, session id,
	// client ip) derived from the active session at `attach_session` time.
	// Read by the executor and the transaction layer to populate the
	// `*Ctx` half of every emitted [`crate::observe`] event without
	// re-walking the session value tree on every emit.
	tenant_identity: Option<Arc<crate::observe::TenantIdentity>>,
	// Matches context for index functions (search::highlight, search::score, etc.)
	matches_context: Option<Arc<crate::exec::function::MatchesContext>>,
	// KNN context for index functions (vector::distance::knn)
	knn_context: Option<Arc<crate::exec::function::KnnContext>>,
	/// Client for making http requests.
	#[cfg(feature = "http")]
	http_client: Arc<HttpClient>,
	/// Stable node id for this datastore instance (live routing, queue keys, etc.).
	node_id: Uuid,
	/// Whether authentication is required for non-anonymous access (datastore boot flag).
	pub(crate) auth_enabled: bool,
	/// Runtime-adjustable configuration (e.g. global query timeout).
	dynamic_configuration: DynamicConfiguration,
	/// Whether this session may use realtime (`LIVE` / `KILL LIVE`).
	live: bool,
	/// Optional broker for cross-node live query notifications.
	broker: Option<Arc<dyn MessageBroker>>,
	// Executor config
	pub config: Arc<CommonConfig>,
}

impl Debug for Context {
	fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
		f.debug_struct("Context")
			.field("parent", &self.parent)
			.field("deadline", &self.deadline)
			.field("cancelled", &self.cancelled)
			.field("values", &self.values)
			.finish()
	}
}

impl Context {
	/// Creates a new empty background context.
	pub(crate) fn background(parent: &Context) -> Self {
		Self {
			values: HashMap::default(),
			parent: None,
			deadline: None,
			slow_log: None,
			cancelled: Arc::new(AtomicBool::new(false)),
			cancel_token: None,
			query_planner: None,
			query_executor: None,
			iteration_stage: None,
			capabilities: Arc::clone(&parent.capabilities),
			index_stores: IndexStores::new(
				parent.config.hnsw_cache_size,
				parent.config.diskann_cache_size,
			),
			cache: None,
			index_builder: None,
			sequences: None,
			#[cfg(storage)]
			temporary_directory: None,
			transaction: None,
			isolated: false,
			buckets: None,
			#[cfg(feature = "surrealism")]
			surrealism_cache: None,
			function_registry: Arc::clone(&parent.function_registry),
			new_planner_strategy: NewPlannerStrategy::default(),
			redact_volatile_explain_attrs: false,
			statement_counters: None,
			matches_context: None,
			knn_context: None,
			config: Arc::clone(&parent.config),
			#[cfg(feature = "http")]
			http_client: Arc::clone(&parent.http_client),
			tenant_identity: None,
			node_id: parent.node_id,
			auth_enabled: parent.auth_enabled,
			dynamic_configuration: parent.dynamic_configuration.clone(),
			live: parent.live,
			broker: parent.broker.clone(),
		}
	}

	/// Creates a new child context from a frozen parent context.
	pub(crate) fn new_child(parent: &FrozenContext) -> Self {
		Self::new_child_with_capabilities(
			parent,
			Arc::clone(&parent.capabilities),
			#[cfg(feature = "http")]
			Arc::clone(&parent.http_client),
		)
	}

	/// Creates a new context from a frozen parent context with a given capabilities.
	///
	/// Make sure that the capabilities and http_client were created with the same capabilities.
	pub(crate) fn new_child_with_capabilities(
		parent: &FrozenContext,
		cap: Arc<Capabilities>,
		#[cfg(feature = "http")] http_client: Arc<HttpClient>,
	) -> Self {
		Context {
			values: HashMap::default(),
			deadline: parent.deadline,
			slow_log: parent.slow_log.clone(),
			cancelled: Arc::new(AtomicBool::new(false)),
			cancel_token: None,
			query_planner: parent.query_planner.clone(),
			query_executor: parent.query_executor.clone(),
			iteration_stage: parent.iteration_stage.clone(),
			capabilities: cap,
			index_stores: parent.index_stores.clone(),
			cache: parent.cache.clone(),
			index_builder: parent.index_builder.clone(),
			sequences: parent.sequences.clone(),
			#[cfg(storage)]
			temporary_directory: parent.temporary_directory.clone(),
			transaction: parent.transaction.clone(),
			isolated: false,
			parent: Some(Arc::clone(parent)),
			buckets: parent.buckets.clone(),
			#[cfg(feature = "surrealism")]
			surrealism_cache: parent.surrealism_cache.clone(),
			function_registry: Arc::clone(&parent.function_registry),
			new_planner_strategy: parent.new_planner_strategy,
			redact_volatile_explain_attrs: parent.redact_volatile_explain_attrs,
			statement_counters: parent.statement_counters.clone(),
			matches_context: parent.matches_context.clone(),
			knn_context: parent.knn_context.clone(),
			config: Arc::clone(&parent.config),
			#[cfg(feature = "http")]
			http_client,
			tenant_identity: parent.tenant_identity.clone(),
			node_id: parent.node_id,
			auth_enabled: parent.auth_enabled,
			dynamic_configuration: parent.dynamic_configuration.clone(),
			live: parent.live,
			broker: parent.broker.clone(),
		}
	}

	/// Create a new context from a frozen parent context.
	/// This context is isolated, and values specified on
	/// any parent contexts will not be accessible.
	pub(crate) fn new_isolated(parent: &FrozenContext) -> Self {
		Self {
			values: HashMap::default(),
			deadline: parent.deadline,
			slow_log: parent.slow_log.clone(),
			cancelled: Arc::new(AtomicBool::new(false)),
			cancel_token: None,
			query_planner: parent.query_planner.clone(),
			query_executor: parent.query_executor.clone(),
			iteration_stage: parent.iteration_stage.clone(),
			capabilities: Arc::clone(&parent.capabilities),
			index_stores: parent.index_stores.clone(),
			cache: parent.cache.clone(),
			index_builder: parent.index_builder.clone(),
			sequences: parent.sequences.clone(),
			#[cfg(storage)]
			temporary_directory: parent.temporary_directory.clone(),
			transaction: parent.transaction.clone(),
			isolated: true,
			parent: Some(Arc::clone(parent)),
			buckets: parent.buckets.clone(),
			#[cfg(feature = "surrealism")]
			surrealism_cache: parent.surrealism_cache.clone(),
			function_registry: Arc::clone(&parent.function_registry),
			new_planner_strategy: parent.new_planner_strategy,
			redact_volatile_explain_attrs: parent.redact_volatile_explain_attrs,
			statement_counters: parent.statement_counters.clone(),
			matches_context: parent.matches_context.clone(),
			knn_context: parent.knn_context.clone(),
			config: Arc::clone(&parent.config),
			#[cfg(feature = "http")]
			http_client: Arc::clone(&parent.http_client),
			tenant_identity: parent.tenant_identity.clone(),
			node_id: parent.node_id,
			auth_enabled: parent.auth_enabled,
			dynamic_configuration: parent.dynamic_configuration.clone(),
			live: parent.live,
			broker: parent.broker.clone(),
		}
	}

	/// Create an independent snapshot of a frozen context.
	///
	/// Flattens all values from the parent chain into the snapshot's own
	/// `values` map and sets `parent: None`, so the returned context does
	/// **not** hold an `Arc` reference to the original parent.
	///
	/// This is used by the streaming executor to give the operator pipeline
	/// its own `Arc<Context>` that won't interfere with the executor's
	/// `Arc::get_mut` requirements between statements.
	///
	/// The cancellation flag and awaitable token are **cloned** from the
	/// source rather than allocated fresh. Without this, any operator that
	/// bridges back into legacy `Context::done` / `is_done` checks — e.g.
	/// the streaming-exec KNN operator calling into
	/// `idx::trees::hnsw::knn_search` / `idx::trees::diskann::knn_search`,
	/// whose inner loops poll `frozen_ctx.is_done(Some(count))` — would
	/// observe the snapshot's local (never-tripped) flag instead of the
	/// connection-level cancel handle installed on the source root, and
	/// would not abort on WebSocket disconnect.
	pub(crate) fn snapshot(from: &FrozenContext) -> Self {
		Self {
			// Flatten all values from the parent chain into this context
			values: from.collect_values(HashMap::default()),
			deadline: from.deadline,
			slow_log: from.slow_log.clone(),
			cancelled: Arc::clone(&from.cancelled),
			cancel_token: from.cancel_token.clone(),
			query_planner: from.query_planner.clone(),
			query_executor: from.query_executor.clone(),
			iteration_stage: from.iteration_stage.clone(),
			capabilities: Arc::clone(&from.capabilities),
			index_stores: from.index_stores.clone(),
			cache: from.cache.clone(),
			index_builder: from.index_builder.clone(),
			sequences: from.sequences.clone(),
			#[cfg(storage)]
			temporary_directory: from.temporary_directory.clone(),
			transaction: from.transaction.clone(),
			isolated: false,
			parent: None, // No parent reference — fully independent
			buckets: from.buckets.clone(),
			#[cfg(feature = "surrealism")]
			surrealism_cache: from.surrealism_cache.clone(),
			function_registry: Arc::clone(&from.function_registry),
			new_planner_strategy: from.new_planner_strategy,
			redact_volatile_explain_attrs: from.redact_volatile_explain_attrs,
			statement_counters: from.statement_counters.clone(),
			matches_context: from.matches_context.clone(),
			knn_context: from.knn_context.clone(),
			config: Arc::clone(&from.config),
			#[cfg(feature = "http")]
			http_client: Arc::clone(&from.http_client),
			tenant_identity: from.tenant_identity.clone(),
			node_id: from.node_id,
			auth_enabled: from.auth_enabled,
			dynamic_configuration: from.dynamic_configuration.clone(),
			live: from.live,
			broker: from.broker.clone(),
		}
	}

	/// Create a new context from a frozen parent context.
	/// This context is not linked to the parent context,
	/// and won't be cancelled if the parent is cancelled.
	///
	/// `index_builder` is intentionally cleared: the only caller is the
	/// background index `Building` task, which lives inside the
	/// `IndexBuilder`'s HashMap. Cloning the back-reference would form an
	/// `Arc<RwLock<HashMap<.., Arc<Building>>>>` cycle that pins the
	/// `Datastore` (and its storage handles, ~7 file descriptors per RocksDB
	/// instance, ~3 per SurrealKV) until process exit — see issue
	/// surrealdb/surrealdb#7304.
	pub(crate) fn new_concurrent(from: &FrozenContext) -> Self {
		Self {
			values: HashMap::default(),
			deadline: None,
			slow_log: from.slow_log.clone(),
			cancelled: Arc::new(AtomicBool::new(false)),
			cancel_token: None,
			query_planner: from.query_planner.clone(),
			query_executor: from.query_executor.clone(),
			iteration_stage: from.iteration_stage.clone(),
			capabilities: Arc::clone(&from.capabilities),
			index_stores: from.index_stores.clone(),
			cache: from.cache.clone(),
			index_builder: None,
			sequences: from.sequences.clone(),
			#[cfg(storage)]
			temporary_directory: from.temporary_directory.clone(),
			transaction: None,
			isolated: false,
			parent: None,
			buckets: from.buckets.clone(),
			#[cfg(feature = "surrealism")]
			surrealism_cache: from.surrealism_cache.clone(),
			function_registry: Arc::clone(&from.function_registry),
			new_planner_strategy: from.new_planner_strategy,
			redact_volatile_explain_attrs: from.redact_volatile_explain_attrs,
			statement_counters: from.statement_counters.clone(),
			matches_context: from.matches_context.clone(),
			knn_context: from.knn_context.clone(),
			config: Arc::clone(&from.config),
			#[cfg(feature = "http")]
			http_client: Arc::clone(&from.http_client),
			tenant_identity: from.tenant_identity.clone(),
			node_id: from.node_id,
			auth_enabled: from.auth_enabled,
			dynamic_configuration: from.dynamic_configuration.clone(),
			live: from.live,
			broker: from.broker.clone(),
		}
	}

	/// Creates a new context from a configured datastore.
	#[expect(clippy::too_many_arguments)]
	pub(crate) fn from_ds(
		node_id: Uuid,
		auth_enabled: bool,
		dynamic_configuration: DynamicConfiguration,
		time_out: Option<Duration>,
		slow_log: Option<SlowLog>,
		capabilities: Arc<Capabilities>,
		index_stores: IndexStores,
		index_builder: IndexBuilder,
		sequences: Sequences,
		cache: Arc<DatastoreCache>,
		function_registry: Arc<FunctionRegistry>,
		#[cfg(feature = "http")] http_client: Arc<HttpClient>,
		#[cfg(storage)] temporary_directory: Option<Arc<PathBuf>>,
		buckets: BucketsManager,
		config: Arc<CommonConfig>,
		#[cfg(feature = "surrealism")] surrealism_cache: Arc<SurrealismCache>,
	) -> Result<Context> {
		let planner_strategy = *capabilities.planner_strategy();
		let mut ctx = Self {
			values: HashMap::default(),
			parent: None,
			deadline: None,
			slow_log,
			cancelled: Arc::new(AtomicBool::new(false)),
			cancel_token: None,
			query_planner: None,
			query_executor: None,
			iteration_stage: None,
			capabilities,
			index_stores,
			cache: Some(cache),
			index_builder: Some(index_builder),
			sequences: Some(sequences),
			#[cfg(storage)]
			temporary_directory,
			transaction: None,
			isolated: false,
			buckets: Some(buckets),
			#[cfg(feature = "surrealism")]
			surrealism_cache: Some(surrealism_cache),
			function_registry,
			new_planner_strategy: planner_strategy,
			redact_volatile_explain_attrs: false,
			statement_counters: None,
			matches_context: None,
			knn_context: None,
			config,
			#[cfg(feature = "http")]
			http_client,
			tenant_identity: None,
			node_id,
			auth_enabled,
			dynamic_configuration,
			live: false,
			broker: None,
		};
		if let Some(timeout) = time_out {
			ctx.add_timeout(timeout)?;
		}
		Ok(ctx)
	}

	/// Create a context for tests only
	#[cfg(test)]
	pub(crate) fn new_test() -> Context {
		Self {
			values: HashMap::default(),
			parent: None,
			deadline: None,
			slow_log: None,
			cancelled: Arc::new(AtomicBool::new(false)),
			cancel_token: None,
			query_planner: None,
			query_executor: None,
			iteration_stage: None,
			capabilities: Arc::new(Capabilities::default()),
			index_stores: IndexStores::new(256 * 1024 * 1024, 256 * 1024 * 1024),
			cache: None,
			index_builder: None,
			sequences: None,
			#[cfg(storage)]
			temporary_directory: None,
			transaction: None,
			isolated: false,
			buckets: None,
			#[cfg(feature = "surrealism")]
			surrealism_cache: None,
			function_registry: Arc::new(FunctionRegistry::with_builtins()),
			new_planner_strategy: NewPlannerStrategy::default(),
			redact_volatile_explain_attrs: false,
			statement_counters: None,
			matches_context: None,
			knn_context: None,
			config: Default::default(),
			#[cfg(feature = "http")]
			http_client: Arc::new(
				HttpClient::new(
					crate::dbs::capabilities::Targets::All,
					crate::dbs::capabilities::Targets::None,
					&Default::default(),
				)
				.expect("http client to be created"),
			),
			tenant_identity: None,
			node_id: Uuid::nil(),
			auth_enabled: true,
			dynamic_configuration: DynamicConfiguration::default(),
			live: false,
			broker: None,
		}
	}

	/// Freezes this context, allowing it to be used as a parent context.
	pub(crate) fn freeze(self) -> FrozenContext {
		Arc::new(self)
	}

	/// Unfreezes this context, allowing it to be edited and configured.
	pub(crate) fn unfreeze(ctx: FrozenContext) -> Result<Context> {
		let Some(x) = Arc::into_inner(ctx) else {
			fail!("Tried to unfreeze a Context with multiple references")
		};
		Ok(x)
	}

	#[inline]
	pub fn node_id(&self) -> Uuid {
		self.node_id
	}

	#[inline]
	pub(crate) fn auth_enabled(&self) -> bool {
		self.auth_enabled
	}

	pub(crate) fn dynamic_configuration(&self) -> &DynamicConfiguration {
		&self.dynamic_configuration
	}

	/// Whether this session may open `LIVE` queries or `KILL LIVE` (from [`Session::live`]
	/// applied in [`Context::attach_session`]).
	pub(crate) fn realtime(&self) -> Result<()> {
		if !self.live {
			bail!(Error::RealtimeDisabled);
		}
		Ok(())
	}

	pub(crate) fn broker(&self) -> Option<&Arc<dyn MessageBroker>> {
		self.broker.as_ref()
	}

	pub(crate) fn set_broker(&mut self, broker: Option<Arc<dyn MessageBroker>>) {
		self.broker = broker;
	}

	/// IAM check using auth from [`Options`] and the datastore auth toggle on this context.
	pub fn is_allowed(
		&self,
		opt: &Options,
		action: Action,
		res: ResourceKind,
		base: Base,
	) -> Result<()> {
		let res = match base {
			Base::Root => res.on_root(),
			Base::Ns => res.on_ns(opt.ns()?),
			Base::Db => {
				let (ns, db) = opt.ns_db()?;
				res.on_db(ns, db)
			}
		};

		if !self.auth_enabled && opt.auth.is_anon() {
			return Ok(());
		}

		opt.auth.is_allowed(action, &res)
	}

	/// Table-level permission check frequency (mirrors former [`Options::check_perms`]).
	pub fn check_perms(&self, opt: &Options, action: Action) -> Result<bool> {
		if !opt.perms {
			return Ok(false);
		}
		if !self.auth_enabled && opt.auth.is_anon() {
			return Ok(false);
		}
		match action {
			Action::Edit => {
				let allowed = opt.auth.has_editor_role();
				let (ns, db) = opt.ns_db()?;
				let db_in_actor_level =
					opt.auth.is_root() || opt.auth.is_ns_check(ns) || opt.auth.is_db_check(ns, db);
				Ok(!allowed || !db_in_actor_level)
			}
			Action::View => {
				let allowed = opt.auth.has_viewer_role();
				let (ns, db) = opt.ns_db()?;
				let db_in_actor_level =
					opt.auth.is_root() || opt.auth.is_ns_check(ns) || opt.auth.is_db_check(ns, db);
				Ok(!allowed || !db_in_actor_level)
			}
		}
	}

	/// Get the namespace id for the current context.
	/// If the namespace does not exist, it will be try to be created based on
	/// the `strict` option.
	pub(crate) async fn get_ns_id(&self, opt: &Options) -> Result<NamespaceId> {
		let ns = opt.ns()?;
		let tx = self.tx();
		let ns_def = tx.get_or_add_ns(Some(self), ns).await?;
		Ok(ns_def.namespace_id)
	}

	/// Get the namespace id for the current context.
	/// If the namespace does not exist, it will return an error.
	pub(crate) async fn expect_ns_id(&self, opt: &Options) -> Result<NamespaceId> {
		let ns = opt.ns()?;
		let Some(ns_def) = self.tx().get_ns_by_name(ns, None).await? else {
			return Err(Error::NsNotFound {
				name: ns.to_string(),
			}
			.into());
		};
		Ok(ns_def.namespace_id)
	}

	/// Get the namespace and database ids for the current context.
	/// If the namespace or database does not exist, it will be try to be
	/// created based on the `strict` option.
	pub(crate) async fn get_ns_db_ids(&self, opt: &Options) -> Result<(NamespaceId, DatabaseId)> {
		let (ns, db) = opt.ns_db()?;
		let db_def = self.tx().ensure_ns_db(Some(self), ns, db).await?;
		Ok((db_def.namespace_id, db_def.database_id))
	}

	/// Get the namespace and database ids for the current context.
	/// If the namespace or database does not exist, it will be try to be
	/// created based on the `strict` option.
	pub(crate) async fn try_ns_db_ids(
		&self,
		opt: &Options,
	) -> Result<Option<(NamespaceId, DatabaseId)>> {
		let (ns, db) = opt.ns_db()?;
		let Some(db_def) = self.tx().get_db_by_name(ns, db, None).await? else {
			return Ok(None);
		};
		Ok(Some((db_def.namespace_id, db_def.database_id)))
	}

	/// Get the namespace and database ids for the current context.
	/// If the namespace or database does not exist, it will return an error.
	pub(crate) async fn expect_ns_db_ids(
		&self,
		opt: &Options,
	) -> Result<(NamespaceId, DatabaseId)> {
		let (ns, db) = opt.ns_db()?;
		let Some(db_def) = self.tx().get_db_by_name(ns, db, None).await? else {
			return Err(Error::DbNotFound {
				name: db.to_string(),
			}
			.into());
		};
		Ok((db_def.namespace_id, db_def.database_id))
	}

	pub(crate) async fn get_db(&self, opt: &Options) -> Result<Arc<DatabaseDefinition>> {
		let (ns, db) = opt.ns_db()?;
		let db_def = self.tx().ensure_ns_db(Some(self), ns, db).await?;
		Ok(db_def)
	}

	/// Add a value to the context. It overwrites any previously set values
	/// with the same key.
	pub(crate) fn add_value<K>(&mut self, key: K, value: Arc<Value>)
	where
		K: Into<Strand>,
	{
		self.values.insert(key.into(), value);
	}

	/// Add a value to the context. It overwrites any previously set values
	/// with the same key.
	pub(crate) fn add_values<T, K, V>(&mut self, iter: T)
	where
		T: IntoIterator<Item = (K, V)>,
		K: Into<Strand>,
		V: Into<Arc<Value>>,
	{
		self.values.extend(iter.into_iter().map(|(k, v)| (k.into(), v.into())))
	}

	/// Add cancellation to the context. The value that is returned will cancel
	/// the context and it's children once called.
	pub(crate) fn add_cancel(&mut self) -> Canceller {
		let cancelled = Arc::clone(&self.cancelled);
		Canceller::new(cancelled)
	}

	/// Install an externally-owned cancellation handle on this context.
	/// Child contexts allocate their own `cancelled` flag but walk up the
	/// parent chain in [`Self::done`], so installing the handle on the
	/// root is enough for the entire execution tree to observe it.
	///
	/// Stores both views of the handle:
	///
	/// * The `AtomicBool` flag is moved into `self.cancelled` so the executor's hot-path `done`
	///   walk fires `Reason::Canceled` at the next yield point.
	/// * The `CancellationToken` is retained on `self.cancel_token` so bare-await sites (e.g.
	///   `SLEEP`) can `select!` against it via [`Self::cancel_token`].
	///
	/// Used by the RPC layer so that a WebSocket disconnect cancels
	/// in-flight queries cleanly, including ones blocked inside an
	/// external timer.
	pub(crate) fn set_cancellation(&mut self, handle: &CancelHandle) {
		self.cancelled = handle.flag();
		self.cancel_token = Some(handle.token());
	}

	/// Awaitable cancellation token installed by [`Self::set_cancellation`].
	/// Walks up the parent chain so any child context can observe a
	/// connection-level cancel without needing its own copy.
	///
	/// Returns `None` for contexts without an external cancel source
	/// (embedded callers, internal use). Callers SHOULD treat `None` as
	/// "no awaitable cancel" rather than failing — the `AtomicBool` view
	/// is the primary cancel mechanism; this is the awaitable companion
	/// for `select!`-based races.
	pub(crate) fn cancel_token(&self) -> Option<tokio_util::sync::CancellationToken> {
		if let Some(token) = &self.cancel_token {
			return Some(token.clone());
		}
		self.parent.as_ref().and_then(|p| p.cancel_token())
	}

	/// Add a deadline to the context. If the current deadline is sooner than
	/// the provided deadline, this method does nothing.
	pub(crate) fn add_deadline(&mut self, deadline: Instant, duration: Duration) {
		match self.deadline {
			Some((current, _)) if current < deadline => (),
			_ => self.deadline = Some((deadline, duration)),
		}
	}

	/// Add a timeout to the context. If the current timeout is sooner than
	/// the provided timeout, this method does nothing. If the result of the
	/// addition causes an overflow, this method returns an error.
	pub(crate) fn add_timeout(&mut self, timeout: Duration) -> Result<(), Error> {
		match Instant::now().checked_add(timeout) {
			Some(deadline) => {
				self.add_deadline(deadline, timeout);
				Ok(())
			}
			None => Err(Error::InvalidTimeout(timeout.as_secs())),
		}
	}

	pub(crate) fn set_query_planner(&mut self, qp: QueryPlanner) {
		self.query_planner = Some(Arc::new(qp));
	}

	/// Cache a table-specific QueryExecutor in the Context.
	///
	/// This is set by the collector/processor when iterating over a specific
	/// table or index so that downstream per-record operations can access the
	/// executor without repeatedly looking it up from the QueryPlanner.
	pub(crate) fn set_query_executor(&mut self, qe: QueryExecutor) {
		self.query_executor = Some(qe);
	}

	pub(crate) fn set_iteration_stage(&mut self, is: IterationStage) {
		self.iteration_stage = Some(is);
	}

	pub(crate) fn set_transaction(&mut self, txn: Arc<Transaction>) {
		self.transaction = Some(txn);
	}

	/// Install the per-statement counter set on this context. Called by the
	/// executor before each top-level statement so the iterator can record
	/// the actual number of records affected -- including for DML
	/// statements with `RETURN NONE`, where the post-RETURN value would
	/// otherwise be empty.
	pub(crate) fn set_statement_counters(&mut self, counters: Option<Arc<StatementCounters>>) {
		self.statement_counters = counters;
	}

	/// The per-statement counter set, if one is currently installed. Read
	/// from the iterator's record-result path.
	pub(crate) fn statement_counters(&self) -> Option<&Arc<StatementCounters>> {
		self.statement_counters.as_ref()
	}

	pub(crate) fn tx(&self) -> Arc<Transaction> {
		self.transaction
			.clone()
			.unwrap_or_else(|| unreachable!("The context was not associated with a transaction"))
	}

	/// Returns the transaction if one is associated with this context.
	pub(crate) fn try_tx(&self) -> Option<&Arc<Transaction>> {
		self.transaction.as_ref()
	}

	/// Get the timeout for this operation, if any. This is useful for
	/// checking if a long job should be started or not.
	pub(crate) fn timeout(&self) -> Option<Duration> {
		self.deadline.map(|(v, _)| v.saturating_duration_since(Instant::now()))
	}

	/// Returns the slow log configuration, if any, attached to this context.
	/// The executor consults this to decide whether to emit slow-query log lines.
	pub(crate) fn slow_log(&self) -> Option<&SlowLog> {
		self.slow_log.as_ref()
	}

	pub(crate) fn get_query_planner(&self) -> Option<&QueryPlanner> {
		self.query_planner.as_ref().map(|qp| qp.as_ref())
	}

	/// Get the cached QueryExecutor (if any) attached by the current iteration
	/// context.
	pub(crate) fn get_query_executor(&self) -> Option<&QueryExecutor> {
		self.query_executor.as_ref()
	}

	pub(crate) fn get_iteration_stage(&self) -> Option<&IterationStage> {
		self.iteration_stage.as_ref()
	}

	/// Get the index_store for this context/ds
	pub(crate) fn get_index_stores(&self) -> &IndexStores {
		&self.index_stores
	}

	/// Get the index_builder for this context/ds
	pub(crate) fn get_index_builder(&self) -> Option<&IndexBuilder> {
		self.index_builder.as_ref()
	}

	/// Return the sequences manager
	pub(crate) fn get_sequences(&self) -> Option<&Sequences> {
		self.sequences.as_ref()
	}

	pub(crate) fn try_get_sequences(&self) -> Result<&Sequences> {
		if let Some(sqs) = self.get_sequences() {
			Ok(sqs)
		} else {
			bail!(Error::Internal("Sequences are not supported in this context.".to_string(),))
		}
	}

	// Get the current datastore cache
	pub(crate) fn get_cache(&self) -> Option<Arc<DatastoreCache>> {
		self.cache.clone()
	}

	/// Check if the context is done. If it returns `None` the operation may
	/// proceed, otherwise the operation should be stopped.
	///
	/// # Check Priority Order
	/// The checks are performed in the following order, with earlier checks taking priority:
	/// 1. **Cancellation** (always checked): Fast atomic flag check via `self.cancelled`
	/// 2. **Memory threshold** (only if `deep_check = true`): Expensive check via
	///    `ALLOC.is_beyond_threshold()`
	/// 3. **Deadline** (only if `deep_check = true`): Moderately expensive check via
	///    `Instant::now()`
	///
	/// # Parameters
	/// - `deep_check`: When `true`, performs all checks (cancellation, memory, deadline). When
	///   `false`, only checks the cancellation flag (fast atomic operation).
	///
	/// # Performance Note
	/// - Checking an `AtomicBool` (cancellation): single-digit nanoseconds
	/// - Checking `Instant::now()` (deadline): tens to hundreds of nanoseconds
	/// - Checking `ALLOC.is_beyond_threshold()` (memory): hundreds of nanoseconds (requires lock +
	///   traversal)
	///
	/// Use `deep_check = false` in hot loops to minimize overhead while still allowing
	/// cancellation.
	pub(crate) fn done(&self, deep_check: bool) -> Result<Option<Reason>> {
		// Check cancellation FIRST (fast atomic operation)
		if self.cancelled.load(Ordering::Relaxed) {
			return Ok(Some(Reason::Canceled));
		}
		if deep_check {
			if ALLOC.is_beyond_threshold() {
				bail!(Error::QueryBeyondMemoryThreshold);
			}
			let now = Instant::now();
			if let Some((deadline, timeout)) = self.deadline
				&& deadline <= now
			{
				return Ok(Some(Reason::Timedout(timeout.into())));
			}
		}
		if let Some(ctx) = &self.parent {
			return ctx.done(deep_check);
		}
		Ok(None)
	}

	/// Check if there is some reason to stop processing the current query.
	///
	/// Returns `true` when the query should be stopped (cancelled, timed out, or exceeded memory
	/// threshold).
	///
	/// # Parameters
	/// - `count`: Optional iteration count for optimization. Pass:
	///   - `Some(count)` when called in a loop - enables adaptive checking to balance
	///     responsiveness with performance. The method will:
	///     - Yield every 32 iterations to allow other tasks to run
	///     - Perform deep checks (memory/deadline) at iterations 1, 2, 4, 8, 16, 32, then every 64
	///   - `None` when called outside a loop (e.g., single operations) - always performs a deep
	///     check for immediate cancellation/timeout detection
	///
	/// # Performance
	/// The adaptive checking strategy ("jitter-based back-off") minimizes overhead in hot loops
	/// while maintaining reasonable responsiveness to cancellation and timeout events.
	pub(crate) async fn is_done(&self, count: Option<usize>) -> Result<bool> {
		let deep_check = if let Some(count) = count {
			// We yield every 32 iterations
			if count % 32 == 0 {
				yield_now!();
			}
			// Adaptive back-off strategy for deep checks based on iteration number:
			// Check frequently early (powers of 2), then settle into every 64 iterations
			match count {
				1 | 2 | 4 | 8 | 16 | 32 => true,
				_ => count % 64 == 0,
			}
		} else {
			// No count provided - perform a deep check immediately (single operation context)
			true
		};
		Ok(self.done(deep_check)?.is_some())
	}

	/// Check if the context is not ok to continue, because it timed out.
	pub(crate) async fn is_timedout(&self) -> Result<Option<Duration>> {
		yield_now!();
		if let Some(Reason::Timedout(d)) = self.done(true)? {
			Ok(Some(d.0))
		} else {
			Ok(None)
		}
	}

	pub(crate) async fn expect_not_timedout(&self) -> Result<()> {
		if let Some(d) = self.is_timedout().await? {
			bail!(Error::QueryTimedout(d.into()))
		} else {
			Ok(())
		}
	}

	#[cfg(storage)]
	/// Return the location of the temporary directory if any
	pub(crate) fn temporary_directory(&self) -> Option<&Arc<PathBuf>> {
		self.temporary_directory.as_ref()
	}

	/// Get a value from the context. If no value is stored under the
	/// provided key, then this will return None.
	pub(crate) fn value(&self, key: &str) -> Option<&Value> {
		match self.values.get(key) {
			Some(v) => Some(v.as_ref()),
			None if PROTECTED_PARAM_NAMES.contains(&key) || !self.isolated => match &self.parent {
				Some(p) => p.value(key),
				_ => None,
			},
			None => None,
		}
	}

	/// Collect context values into the provided map, walking up parent contexts
	/// unless this context is isolated.
	pub(crate) fn collect_values(
		&self,
		map: HashMap<Strand, Arc<Value>>,
	) -> HashMap<Strand, Arc<Value>> {
		let mut map = if !self.isolated
			&& let Some(p) = &self.parent
		{
			p.collect_values(map)
		} else {
			map
		};
		self.values.iter().for_each(|(k, v)| {
			map.insert(k.clone(), Arc::clone(v));
		});
		map
	}

	/// Get a 'static view into the cancellation status.
	#[cfg(feature = "scripting")]
	pub(crate) fn cancellation(&self) -> crate::ctx::cancellation::Cancellation {
		crate::ctx::cancellation::Cancellation::new(
			self.deadline.map(|(deadline, _)| deadline),
			std::iter::successors(Some(self), |ctx| ctx.parent.as_ref().map(|c| c.as_ref()))
				.map(|ctx| Arc::clone(&ctx.cancelled))
				.collect(),
		)
	}

	/// Attach a session to the context and add any session variables to the
	/// context.
	pub(crate) fn attach_session(&mut self, session: &Session) -> Result<(), Error> {
		self.live = session.live();
		self.add_values(session.values());
		// Only override the planner strategy if the session explicitly sets a
		// non-default value (e.g. language tests). Otherwise the capability-level
		// strategy (set via from_ds) is preserved.
		if session.new_planner_strategy != NewPlannerStrategy::default() {
			self.new_planner_strategy = session.new_planner_strategy;
		}
		// Propagate duration redaction flag from session.
		if session.redact_volatile_explain_attrs {
			self.redact_volatile_explain_attrs = true;
		}
		if !session.variables.is_empty() {
			self.attach_variables(session.variables.clone().into())?;
		}
		// Pre-resolve the tenant identity so emit sites do not need to
		// re-walk the session value tree on every event dispatch.
		self.tenant_identity =
			Some(Arc::new(crate::observe::TenantIdentity::from_session(session)));
		Ok(())
	}

	/// Pre-resolved tenant identity for the active session.
	pub(crate) fn tenant_identity(&self) -> Option<&Arc<crate::observe::TenantIdentity>> {
		self.tenant_identity.as_ref()
	}

	/// Attach variables to the context.
	pub(crate) fn attach_variables(&mut self, vars: Variables) -> Result<(), Error> {
		for (name, val) in vars {
			if PROTECTED_PARAM_NAMES.contains(&name.as_str()) {
				return Err(Error::InvalidParam {
					name: name.into_string(),
				});
			}
			self.add_value(name, Arc::new(val));
		}
		Ok(())
	}

	pub(crate) fn attach_public_variables(&mut self, vars: PublicVariables) -> Result<(), Error> {
		for (name, val) in vars {
			if PROTECTED_PARAM_NAMES.contains(&name.as_str()) {
				return Err(Error::InvalidParam {
					name,
				});
			}
			self.add_value(name, Arc::new(convert_public_value_to_internal(val)));
		}
		Ok(())
	}

	//
	// Capabilities
	//

	/// Get the capabilities for this context
	pub(crate) fn get_capabilities(&self) -> Arc<Capabilities> {
		Arc::clone(&self.capabilities)
	}

	/// Get the function registry for this context
	pub(crate) fn function_registry(&self) -> &Arc<FunctionRegistry> {
		&self.function_registry
	}

	/// Set the matches context for index functions (search::highlight, etc.)
	pub(crate) fn set_matches_context(&mut self, ctx: crate::exec::function::MatchesContext) {
		self.matches_context = Some(Arc::new(ctx));
	}

	/// Get the matches context for index functions
	pub(crate) fn get_matches_context(
		&self,
	) -> Option<&Arc<crate::exec::function::MatchesContext>> {
		self.matches_context.as_ref()
	}

	/// Set the KNN context for index functions (vector::distance::knn)
	pub(crate) fn set_knn_context(&mut self, ctx: Arc<crate::exec::function::KnnContext>) {
		self.knn_context = Some(ctx);
	}

	/// Get the KNN context for index functions
	pub(crate) fn get_knn_context(&self) -> Option<&Arc<crate::exec::function::KnnContext>> {
		self.knn_context.as_ref()
	}

	/// Get the new planner strategy for this context
	pub(crate) fn new_planner_strategy(&self) -> &NewPlannerStrategy {
		&self.new_planner_strategy
	}

	/// Whether EXPLAIN ANALYZE should redact elapsed durations.
	pub(crate) fn redact_volatile_explain_attrs(&self) -> bool {
		self.redact_volatile_explain_attrs
	}

	/// Check if scripting is allowed
	#[cfg_attr(not(feature = "scripting"), expect(dead_code))]
	pub(crate) fn check_allowed_scripting(&self) -> Result<()> {
		if !self.capabilities.allows_scripting() {
			warn!("Capabilities denied scripting attempt");
			bail!(Error::ScriptingNotAllowed);
		}
		trace!("Capabilities allowed scripting");
		Ok(())
	}

	/// Check if a function is allowed
	pub(crate) fn check_allowed_function(&self, target: &str) -> Result<()> {
		if !self.capabilities.allows_function_name(target) {
			warn!("Capabilities denied function execution attempt, target: '{target}'");
			bail!(Error::FunctionNotAllowed(target.to_string()));
		}
		trace!("Capabilities allowed function execution, target: '{target}'");
		Ok(())
	}

	/// Checks if the provided URL's network target is allowed based on current
	/// capabilities.
	///
	/// This function performs a validation to ensure that the outgoing network
	/// connection specified by the provided `url` is permitted. It checks the
	/// resolved network targets associated with the URL and ensures that all
	/// targets adhere to the configured capabilities.
	///
	/// # Features
	/// The function is only available if the `http` feature is enabled.
	///
	/// # Parameters
	/// - `url`: A reference to a [`Url`] object representing the target endpoint to check.
	///
	/// # Returns
	/// This function returns a [`Result<()>`]:
	/// - On success, it returns `Ok(())` indicating the network target is allowed.
	/// - On failure, it returns an error wrapped in the [`Error`] type:
	///   - `NetTargetNotAllowed` if the target is not permitted.
	///   - `InvalidUrl` if the provided URL is invalid.
	///
	/// # Behavior
	/// 1. Extracts the host and port information from the URL.
	/// 2. Constructs a [`NetTarget`] object and checks if it is allowed by the current network
	///    capabilities.
	/// 3. If the network target resolves to multiple targets (e.g., DNS resolution), each target is
	///    validated individually.
	/// 4. Logs a warning and prevents the connection if the target is denied by the capabilities.
	///
	/// # Logging
	/// - Logs a warning message if the network target is denied.
	/// - Logs a trace message if the network target is permitted.
	///
	/// # Errors
	/// - `NetTargetNotAllowed`: Returned if any of the resolved targets are not allowed.
	/// - `InvalidUrl`: Returned if the URL does not have a valid host.
	#[cfg(feature = "http")]
	pub(crate) async fn check_allowed_net(&self, url: &Url) -> Result<()> {
		let match_any_deny_net = |t| {
			if self.capabilities.matches_any_deny_net(t) {
				warn!("Capabilities denied outgoing network connection attempt, target: '{t}'");
				bail!(Error::NetTargetNotAllowed(t.to_string()));
			}
			Ok(())
		};
		match url.host() {
			Some(host) => {
				let target = NetTarget::Host(host.to_owned(), url.port_or_known_default());
				// Check the domain name (if any) matches the allow list
				let host_allowed = self.capabilities.matches_any_allow_net(&target);
				if !host_allowed {
					warn!(
						"Capabilities denied outgoing network connection attempt, target: '{target}'"
					);
					bail!(Error::NetTargetNotAllowed(target.to_string()));
				}
				// Check against the deny list
				match_any_deny_net(&target)?;
				// Resolve the domain name to a vector of IP addresses
				#[cfg(not(target_family = "wasm"))]
				let targets = target.resolve().await?;
				#[cfg(target_family = "wasm")]
				let targets = target.resolve()?;
				for t in &targets {
					match_any_deny_net(t)?;
				}
				trace!("Capabilities allowed outgoing network connection, target: '{target}'");
				Ok(())
			}
			_ => bail!(Error::InvalidUrl(url.to_string())),
		}
	}

	pub(crate) fn get_buckets(&self) -> Option<&BucketsManager> {
		self.buckets.as_ref()
	}

	/// Obtain the connection for a bucket
	pub(crate) async fn get_bucket_store(
		&self,
		ns: NamespaceId,
		db: DatabaseId,
		bu: &str,
	) -> Result<Arc<dyn ObjectStore>> {
		// Do we have a buckets context?
		if let Some(buckets) = &self.buckets {
			buckets.get_bucket_store(&self.tx(), ns, db, bu).await
		} else {
			bail!(Error::BucketUnavailable(bu.into()))
		}
	}

	#[cfg(feature = "surrealism")]
	pub(crate) fn get_surrealism_cache(&self) -> Option<Arc<SurrealismCache>> {
		self.surrealism_cache.as_ref().map(Arc::clone)
	}

	#[cfg(feature = "surrealism")]
	pub(crate) async fn get_surrealism_module(
		&self,
		lookup: SurrealismCacheLookup<'_>,
	) -> Result<SurrealismCachedModule> {
		if !self.get_capabilities().allows_experimental(&ExperimentalTarget::Surrealism) {
			bail!(
				"Failed to get surrealism runtime: Experimental capability `surrealism` is not enabled"
			);
		}

		let Some(cache) = self.get_surrealism_cache() else {
			bail!("Surrealism cache is not available");
		};
		let max_pool_size = self.config.surrealism_max_pool_size;
		let max_memory = self.config.surrealism_max_memory;
		let max_execution_time =
			self.config.surrealism_max_execution_time.map(Duration::from_millis);
		let max_kv_entries = self.config.surrealism_max_kv_entries;
		let max_kv_value_bytes = self.config.surrealism_max_kv_value_bytes;
		#[cfg(feature = "http")]
		let config = Arc::clone(&self.config);

		cache
			.get_or_insert_with(&lookup, async || {
				let SurrealismCacheLookup::File(ns, db, bucket, key) = lookup else {
					bail!("silo lookups are not supported yet");
				};

				let bucket = self.get_bucket_store(*ns, *db, bucket).await?;
				let key = ObjectKey::new(key);
				let surli = bucket
					.get(&key)
					.await
					.map_err(|e| anyhow::anyhow!("failed to get file: {}", e))?;

				let Some(surli) = surli else {
					bail!("file not found");
				};

				let safe_key = key.to_string().replace(['/', '\\'], "_");
				let temp_prefix = format!("SURREAL_MODFS_{ns}_{db}_{safe_key}_");
				let unpack_opts = UnpackOptions {
					#[cfg(storage)]
					temp_base: self.temporary_directory().map(|p| p.as_path()),
					#[cfg(not(storage))]
					temp_base: None,
					temp_prefix: &temp_prefix,
					max_fs_bytes: self.config.surrealism_max_fs_bytes,
				};
				let package =
					SurrealismPackage::from_reader(std::io::Cursor::new(surli), &unpack_opts)?;

				self.get_capabilities()
					.validate_surrealism_capabilities(&package.config.capabilities)?;

				let org = package.config.meta.organisation.clone();
				let name = package.config.meta.name.clone();

				#[cfg(feature = "http")]
				let module_net_targets =
					crate::surrealism::host::module_allow_net_targets(&package.config.capabilities);

				let runtime = tokio::task::spawn_blocking(move || {
					Runtime::new(
						package,
						max_pool_size,
						max_memory,
						max_execution_time,
						max_kv_entries,
						max_kv_value_bytes,
					)
				})
				.await
				.context("WASM compile task aborted")??;
				let runtime = Arc::new(runtime);

				let module_display_name: Arc<str> = format!("{org}::{name}").into();

				#[cfg(feature = "http")]
				let client = if module_net_targets.is_empty() {
					Arc::new(
						HttpClient::new(Targets::None, Targets::All, &config)
							.context("Failed to create http client for WASM module")?,
					)
				} else {
					let allow = Targets::Some(module_net_targets);
					Arc::new(
						HttpClient::new(
							allow,
							self.capabilities.denied_network_targets_ref().clone(),
							&config,
						)
						.context("Failed to create http client for WASM module")?,
					)
				};

				Ok(SurrealismCachedModule {
					runtime,
					module_display_name,
					#[cfg(feature = "http")]
					client,
				})
			})
			.await
	}

	#[cfg(feature = "http")]
	pub(crate) fn http_client(&self) -> Arc<HttpClient> {
		Arc::clone(&self.http_client)
	}

	#[cfg(feature = "surrealism")]
	pub(crate) async fn get_surrealism_runtime(
		&self,
		lookup: SurrealismCacheLookup<'_>,
	) -> Result<Arc<Runtime>> {
		Ok(self.get_surrealism_module(lookup).await?.runtime)
	}
}

#[cfg(test)]
mod tests {
	#[cfg(feature = "http")]
	use std::str::FromStr;
	use std::time::Duration;

	#[cfg(feature = "http")]
	use url::Url;

	use crate::cnf::CommonConfig;
	#[cfg(all(feature = "allocation-tracking", feature = "allocator"))]
	use crate::cnf::MEMORY_THRESHOLD;
	use crate::ctx::Context;
	use crate::ctx::reason::Reason;
	#[cfg(feature = "http")]
	use crate::dbs::Capabilities;
	use crate::dbs::Options;
	#[cfg(feature = "http")]
	use crate::dbs::capabilities::{NetTarget, Targets};
	use crate::expr::Base;
	use crate::iam::{Action, Auth, ResourceKind, Role};

	#[test]
	fn is_allowed_respects_context_auth_toggle_and_base() {
		let config = CommonConfig::default();

		// Auth disabled: anonymous allowed without IAM; still needs valid NS/DB for bases.
		{
			let mut ctx = Context::new_test();
			ctx.auth_enabled = false;

			let empty = Options::new(&config);
			ctx.is_allowed(&empty, Action::View, ResourceKind::Any, Base::Ns).unwrap_err();
			ctx.is_allowed(&empty, Action::View, ResourceKind::Any, Base::Db).unwrap_err();
			let db_only = Options::new(&config).with_db(Some("db".into()));
			ctx.is_allowed(&db_only, Action::View, ResourceKind::Any, Base::Db).unwrap_err();

			ctx.is_allowed(&empty, Action::View, ResourceKind::Any, Base::Root).unwrap();
			let ns = Options::new(&config).with_ns(Some("ns".into()));
			ctx.is_allowed(&ns, Action::View, ResourceKind::Any, Base::Ns).unwrap();
			let ns_db = Options::new(&config).with_ns(Some("ns".into())).with_db(Some("db".into()));
			ctx.is_allowed(&ns_db, Action::View, ResourceKind::Any, Base::Db).unwrap();
		}

		// Auth enabled: root owner still needs NS/DB set for NS/Db bases.
		{
			let mut ctx = Context::new_test();
			ctx.auth_enabled = true;

			let opts = Options::new(&config).with_auth(Auth::for_root(Role::Owner).into());
			ctx.is_allowed(&opts, Action::View, ResourceKind::Any, Base::Ns).unwrap_err();
			ctx.is_allowed(&opts, Action::View, ResourceKind::Any, Base::Db).unwrap_err();
			let db_only = opts.clone().with_db(Some("db".into()));
			ctx.is_allowed(&db_only, Action::View, ResourceKind::Any, Base::Db).unwrap_err();

			ctx.is_allowed(&opts, Action::View, ResourceKind::Any, Base::Root).unwrap();
			let ns = opts.with_ns(Some("ns".into()));
			ctx.is_allowed(&ns, Action::View, ResourceKind::Any, Base::Ns).unwrap();
			let ns_db = ns.with_db(Some("db".into()));
			ctx.is_allowed(&ns_db, Action::View, ResourceKind::Any, Base::Db).unwrap();
		}
	}

	#[cfg(feature = "http")]
	#[tokio::test]
	async fn test_context_check_allowed_net() {
		let cap = Capabilities::all().without_network_targets(Targets::Some(
			[NetTarget::from_str("127.0.0.1").unwrap()].into(),
		));
		let mut ctx = Context::new_test();
		ctx.capabilities = cap.into();
		let ctx = ctx.freeze();
		let r = ctx.check_allowed_net(&Url::parse("http://localhost").unwrap()).await;
		assert_eq!(
			r.err().unwrap().to_string(),
			"Access to network target '127.0.0.1/32' is not allowed"
		);
	}

	#[tokio::test]
	async fn test_context_cancellation_priority() {
		// Test that cancellation is detected even when a deadline is set and exceeded
		let mut ctx = Context::new_test();

		// Set a deadline in the past (already exceeded)
		ctx.add_timeout(Duration::from_nanos(1)).unwrap();
		// Give time for the deadline to pass
		tokio::time::sleep(Duration::from_millis(10)).await;

		// Cancel the context
		let canceller = ctx.add_cancel();
		canceller.cancel();

		let ctx = ctx.freeze();

		// Cancellation should be detected first, not timeout
		let result = ctx.done(true);
		assert!(result.is_ok());
		assert_eq!(result.unwrap(), Some(Reason::Canceled));
	}

	#[tokio::test]
	async fn test_context_deadline_detection() {
		// Test that deadline timeout is detected when context is not cancelled
		let mut ctx = Context::new_test();

		// Set a very short timeout
		ctx.add_timeout(Duration::from_nanos(1)).unwrap();
		// Give time for the deadline to pass
		tokio::time::sleep(Duration::from_millis(10)).await;

		let ctx = ctx.freeze();

		// Should detect timeout
		let result = ctx.done(true);
		assert!(result.is_ok());
		assert!(matches!(result.unwrap(), Some(Reason::Timedout(_))));
	}

	#[tokio::test]
	async fn test_context_no_deadline() {
		// Test that a context without deadline or cancellation returns None
		let ctx = Context::new_test();
		let ctx = ctx.freeze();

		// Should return None (ok to continue)
		let result = ctx.done(true);
		assert!(result.is_ok());
		assert_eq!(result.unwrap(), None);
	}

	#[tokio::test]
	async fn test_context_is_done_adaptive_backoff() {
		// Test the adaptive back-off strategy in is_done()
		let ctx = Context::new_test();
		let ctx = ctx.freeze();

		// Test that early iterations trigger deep checks (1, 2, 4, 8, 16, 32)
		for count in [1, 2, 4, 8, 16, 32] {
			let result = ctx.is_done(Some(count)).await;
			assert!(result.is_ok());
			assert_eq!(result.unwrap(), false, "Count {} should not be done", count);
		}

		// Test that later iterations only check every 64
		// Count 33-63 should not trigger deep checks (except at 64)
		for count in 33..64 {
			let result = ctx.is_done(Some(count)).await;
			assert!(result.is_ok());
			assert_eq!(result.unwrap(), false, "Count {} should not be done", count);
		}

		// Count 64 should trigger a deep check (64 % 64 == 0)
		let result = ctx.is_done(Some(64)).await;
		assert!(result.is_ok());
		assert_eq!(result.unwrap(), false);
	}

	#[tokio::test]
	async fn test_context_is_done_with_none() {
		// Test that is_done(None) always performs a deep check
		let ctx = Context::new_test();
		let ctx = ctx.freeze();

		// Should perform deep check and return Ok(false) since no cancellation/timeout
		let result = ctx.is_done(None).await;
		assert!(result.is_ok());
		assert_eq!(result.unwrap(), false);
	}

	#[tokio::test]
	async fn test_context_is_done_detects_cancellation() {
		// Test that is_done detects cancellation
		let mut ctx = Context::new_test();
		let canceller = ctx.add_cancel();
		canceller.cancel();
		let ctx = ctx.freeze();

		// Should detect cancellation
		let result = ctx.is_done(None).await;
		assert!(result.is_ok());
		assert_eq!(result.unwrap(), true);

		// Should also detect with count
		let result = ctx.is_done(Some(1)).await;
		assert!(result.is_ok());
		assert_eq!(result.unwrap(), true);
	}

	/// `Context::snapshot` is called by the streaming executor at
	/// `dbs::Executor::execute_operator_plan` and produces a ctx with
	/// `parent: None`. Operators that bridge back into legacy
	/// `Context::done` / `is_done` checks (HNSW/DiskANN KNN search inner
	/// loops, etc.) read the snapshot's own `cancelled` flag — without
	/// parent walk to fall back to. If `snapshot` allocated a fresh
	/// `Arc<AtomicBool>` instead of sharing the source's, those legacy
	/// checks would never observe a connection-level cancel installed
	/// via `set_cancellation`, and a WS disconnect during e.g. a KNN
	/// search would not abort the operation — the read loop's drain
	/// would block until the search finished.
	#[tokio::test]
	async fn test_snapshot_preserves_external_cancellation() {
		use crate::ctx::CancelHandle;

		let handle = CancelHandle::new();
		let mut ctx = Context::new_test();
		ctx.set_cancellation(&handle);
		let root = ctx.freeze();
		let snap = Context::snapshot(&root).freeze();
		assert_eq!(
			snap.is_done(Some(1)).await.unwrap(),
			false,
			"snapshot reports done before cancel was tripped"
		);

		handle.trip();
		assert_eq!(
			snap.is_done(Some(1)).await.unwrap(),
			true,
			"snapshot did not observe external cancel after trip — legacy `is_done` on \
			 streaming-exec snapshot would silently miss WebSocket disconnect"
		);
		assert!(
			snap.cancel_token().is_some(),
			"snapshot lost the awaitable cancel token — bare-await sites reached via \
			 the snapshot (legacy SLEEP fallback, etc.) could not `select!` against cancel"
		);
	}

	/// Test documenting the expected behavior when memory threshold is exceeded.
	///
	/// Note: This test documents the expected behavior but cannot easily test actual
	/// memory threshold violations without:
	/// 1. Setting MEMORY_THRESHOLD configuration (via cnf::MEMORY_THRESHOLD)
	/// 2. Actually allocating enough memory to exceed it
	/// 3. Having the "allocation-tracking" feature enabled
	///
	/// The key behavior tested elsewhere is that when is_beyond_threshold() returns true,
	/// it takes priority over deadline timeout, which prevents OOM crashes from being
	/// masked by timeout errors.
	#[tokio::test]
	async fn test_context_memory_threshold_priority_documentation() {
		// This test documents that the priority order in done() is:
		// 1. Cancellation (always checked, fast atomic operation)
		// 2. Memory threshold (checked when deep_check=true, if beyond threshold returns Error)
		// 3. Deadline (checked when deep_check=true, returns Reason::Timedout)

		// When ALLOC.is_beyond_threshold() returns true, done() will bail with
		// Error::QueryBeyondMemoryThreshold before checking the deadline.
		// This ensures memory violations are always detected before timeout errors.

		let ctx = Context::new_test();
		let ctx = ctx.freeze();

		// With no memory pressure, deadline not set, and no cancellation:
		let result = ctx.done(true);
		assert!(result.is_ok());
		assert_eq!(result.unwrap(), None);
	}

	/// Integration test that actually tests memory threshold detection.
	///
	/// This test requires:
	/// 1. The "allocation-tracking" feature to be enabled
	/// 2. The "allocator" feature to be enabled (for tracking to work)
	/// 3. Running with #[serial] to avoid interference from other tests
	///
	/// The test sets SURREAL_MEMORY_THRESHOLD environment variable, allocates memory
	/// to exceed the threshold, and verifies that context.done(true) detects the violation.
	#[tokio::test]
	#[cfg(all(feature = "allocation-tracking", feature = "allocator"))]
	#[serial_test::serial]
	async fn test_context_memory_threshold_integration() {
		use crate::err::Error;
		use crate::str::ParseBytes;

		// Set a low memory threshold (1MB) before MEMORY_THRESHOLD is accessed
		// This must happen before any code accesses cnf::MEMORY_THRESHOLD
		// Safety: This test runs with #[serial] ensuring no other tests run concurrently,
		// so there's no risk of data races when modifying the environment variable.
		unsafe {
			std::env::set_var(
				"SURREAL_MEMORY_THRESHOLD",
				"1MB".parse_bytes::<u64>().unwrap().to_string(),
			);
		}
		// Assert that SURREAL_MEMORY_THRESHOLD shows up as MEMORY_THRESHOLD with the expected value
		assert_eq!(*MEMORY_THRESHOLD, 1048576);

		// Force reinitialization by dropping and recreating (this won't work with LazyLock)
		// Instead, we rely on this test running in isolation with #[serial]
		// and being run in a fresh process where MEMORY_THRESHOLD hasn't been accessed yet

		// Note: This test may not work reliably if MEMORY_THRESHOLD was already accessed
		// elsewhere in the test suite. The #[serial] attribute ensures tests run one at a time,
		// but doesn't guarantee a fresh process. For reliable testing, this should be run
		// as a separate integration test binary.

		// Allocate a large vector (10MB) to exceed the threshold
		// Using Vec::with_capacity to ensure the memory is actually allocated
		let _large_allocation: Vec<u8> = Vec::with_capacity(20 * 1024 * 1024);

		// Give the allocator tracking time to register the allocation
		tokio::time::sleep(Duration::from_millis(10)).await;

		let ctx = Context::new_test();
		let ctx = ctx.freeze();

		// The memory threshold check should detect that we've exceeded the limit
		let result = ctx.done(true);

		// We expect either:
		// 1. An error if memory tracking properly detected the threshold violation
		// 2. Ok(None) if MEMORY_THRESHOLD was already initialized with default (0) before we set
		//    the environment variable
		match result {
			Err(e) => {
				// Verify it's the correct error type
				match e.downcast_ref::<Error>() {
					Some(Error::QueryBeyondMemoryThreshold) => {
						// Success! Memory threshold was properly detected
						println!("✓ Memory threshold violation detected as expected");
					}
					other => {
						panic!("Expected QueryBeyondMemoryThreshold error, got: {:?}", other);
					}
				}
			}
			Ok(None) => {
				// This means MEMORY_THRESHOLD was already initialized before we set the env var
				// This is expected behavior in the test suite - document it
				println!(
					"⚠ Memory threshold not enforced - MEMORY_THRESHOLD was already initialized"
				);
				println!("  This is expected when running as part of the full test suite.");
				println!(
					"  To properly test memory threshold enforcement, run this test in isolation:"
				);
				println!(
					"  cargo test --package surrealdb-core --features allocation-tracking,allocator test_context_memory_threshold_integration"
				);
				panic!("MEMORY_THRESHOLD was already initialized")
			}
			Ok(Some(reason)) => {
				panic!("Unexpected reason returned: {:?}", reason);
			}
		}

		// Clean up the environment variable
		// Safety: Same as above - #[serial] ensures no concurrent access
		unsafe {
			std::env::remove_var("SURREAL_MEMORY_THRESHOLD");
		}
	}
}