surrealdb-core 3.2.3

//! Database INFO operator - returns database-level metadata.
//!
//! Implements INFO FOR DB [STRUCTURE] [VERSION timestamp] which returns information about:
//! - Database accesses
//! - APIs
//! - Analyzers
//! - Buckets
//! - Functions
//! - Modules
//! - Models
//! - Params
//! - Tables
//! - Users
//! - Configs
//! - Sequences

use std::sync::Arc;

use futures::stream;
use surrealdb_types::ToSql;

use crate::catalog::providers::{
	ApiProvider, AuthorisationProvider, BucketProvider, DatabaseProvider, TableProvider,
	UserProvider,
};
use crate::exec::context::{ContextLevel, ExecutionContext};
use crate::exec::physical_expr::{EvalContext, PhysicalExpr};
use crate::exec::{
	AccessMode, CardinalityHint, ExecOperator, FlowResult, OperatorMetrics, ValueBatch,
	ValueBatchStream,
};
use crate::expr::statements::info::InfoStructure;
use crate::iam::{Action, ResourceKind};
use crate::val::{Datetime, Object, Value};

/// Database INFO operator.
///
/// Returns database-level metadata including accesses, APIs, analyzers,
/// tables, functions, and more.
#[derive(Debug)]
pub struct DatabaseInfoPlan {
	/// Whether to return structured output
	pub structured: bool,
	/// Optional version timestamp to filter schema by
	pub version: Option<Arc<dyn PhysicalExpr>>,
	pub(crate) metrics: Arc<OperatorMetrics>,
}

impl DatabaseInfoPlan {
	pub(crate) fn new(structured: bool, version: Option<Arc<dyn PhysicalExpr>>) -> Self {
		Self {
			structured,
			version,
			metrics: Arc::new(OperatorMetrics::new()),
		}
	}
}
impl ExecOperator for DatabaseInfoPlan {
	fn name(&self) -> &'static str {
		"InfoDatabase"
	}

	fn attrs(&self) -> Vec<(String, String)> {
		let mut attrs = vec![("structured".to_string(), self.structured.to_string())];
		if self.version.is_some() {
			attrs.push(("version".to_string(), "<expr>".to_string()));
		}
		attrs
	}

	fn required_context(&self) -> ContextLevel {
		// Database info needs database context, combined with expression contexts
		let version_ctx =
			self.version.as_ref().map(|e| e.required_context()).unwrap_or(ContextLevel::Root);
		version_ctx.max(ContextLevel::Database)
	}

	fn access_mode(&self) -> AccessMode {
		// Info is inherently read-only, but the version expression
		// could theoretically contain a mutation subquery.
		self.version.as_ref().map(|e| e.access_mode()).unwrap_or(AccessMode::ReadOnly)
	}

	fn cardinality_hint(&self) -> CardinalityHint {
		CardinalityHint::AtMostOne
	}

	fn metrics(&self) -> Option<&OperatorMetrics> {
		Some(self.metrics.as_ref())
	}

	fn expressions(&self) -> Vec<(&str, &Arc<dyn PhysicalExpr>)> {
		if let Some(ref v) = self.version {
			vec![("version", v)]
		} else {
			vec![]
		}
	}

	fn execute(&self, ctx: &ExecutionContext) -> FlowResult<ValueBatchStream> {
		let structured = self.structured;
		let version = self.version.clone();
		let ctx = ctx.clone();

		Ok(Box::pin(stream::once(async move {
			let value = execute_database_info(&ctx, structured, version.as_deref()).await?;
			Ok(ValueBatch {
				values: vec![value],
			})
		})))
	}

	fn is_scalar(&self) -> bool {
		true
	}
}

async fn execute_database_info(
	ctx: &ExecutionContext,
	structured: bool,
	version: Option<&dyn PhysicalExpr>,
) -> crate::expr::FlowResult<Value> {
	// Allowed to run?
	ctx.is_allowed(Action::View, ResourceKind::Any, crate::expr::Base::Db)?;

	// Get database context
	let db_ctx = ctx.database()?;
	let ns = db_ctx.ns_ctx.ns.namespace_id;
	let db = db_ctx.db.database_id;

	// Convert the version to u64 if present
	let version = match version {
		Some(v) => {
			let eval_ctx = EvalContext::from_exec_ctx(ctx);
			let value = v.evaluate(eval_ctx).await?;
			Some(
				value
					.cast_to::<Datetime>()
					.map_err(|e| anyhow::anyhow!("{e}"))?
					.to_version_stamp(ctx.txn().timestamp_impl().as_ref())?,
			)
		}
		None => None,
	};

	// Get the transaction
	let txn = ctx.txn();

	// Create the result set
	if structured {
		let object = map! {
			"accesses" => process(&txn.all_db_accesses(ns, db, version).await?),
			"apis" => process(&txn.all_db_apis(ns, db, version).await?),
			"analyzers" => process(&txn.all_db_analyzers(ns, db, version).await?),
			"buckets" => process(&txn.all_db_buckets(ns, db, version).await?),
			"functions" => process(&txn.all_db_functions(ns, db, version).await?),
			"modules" => crate::expr::statements::info::process_modules(ctx.ctx(), ns, db, txn.all_db_modules(ns, db, version).await?).await,
			"models" => process(&txn.all_db_models(ns, db, version).await?),
			"params" => process(&txn.all_db_params(ns, db, version).await?),
			"tables" => process(&txn.all_tb(ns, db, version).await?),
			"users" => process(&txn.all_db_users(ns, db, version).await?),
			"configs" => process(&txn.all_db_configs(ns, db, version).await?),
			"sequences" => process(&txn.all_db_sequences(ns, db, version).await?),
		};
		Ok(Value::Object(Object::from(object)))
	} else {
		let object = map! {
			"accesses" => {
				let mut out = Object::default();
				for v in txn.all_db_accesses(ns, db, version).await?.iter() {
					out.insert(v.name.clone(), v.to_sql().into());
				}
				out.into()
			},
			"apis" => {
				let mut out = Object::default();
				for v in txn.all_db_apis(ns, db, version).await?.iter() {
					out.insert(v.path.to_string(), v.to_sql().into());
				}
				out.into()
			},
			"analyzers" => {
				let mut out = Object::default();
				for v in txn.all_db_analyzers(ns, db, version).await?.iter() {
					out.insert(v.name.clone(), v.to_sql().into());
				}
				out.into()
			},
			"buckets" => {
				let mut out = Object::default();
				for v in txn.all_db_buckets(ns, db, version).await?.iter() {
					out.insert(v.name.clone(), v.to_sql().into());
				}
				out.into()
			},
			"functions" => {
				let mut out = Object::default();
				for v in txn.all_db_functions(ns, db, version).await?.iter() {
					out.insert(v.name.clone(), v.to_sql().into());
				}
				out.into()
			},
			"modules" => {
				let mut out = Object::default();
				for v in txn.all_db_modules(ns, db, version).await?.iter() {
					out.insert(v.get_storage_name()?, v.to_sql().into());
				}
				out.into()
			},
			"models" => {
				let mut out = Object::default();
				for v in txn.all_db_models(ns, db, version).await?.iter() {
					out.insert(v.name.clone(), v.to_sql().into());
				}
				out.into()
			},
			"params" => {
				let mut out = Object::default();
				for v in txn.all_db_params(ns, db, version).await?.iter() {
					out.insert(v.name.clone(), v.to_sql().into());
				}
				out.into()
			},
			"tables" => {
				let mut out = Object::default();
				for v in txn.all_tb(ns, db, version).await?.iter() {
					out.insert(v.name.clone(), v.to_sql().into());
				}
				out.into()
			},
			"users" => {
				let mut out = Object::default();
				for v in txn.all_db_users(ns, db, version).await?.iter() {
					out.insert(v.name.clone(), v.to_sql().into());
				}
				out.into()
			},
			"configs" => {
				let mut out = Object::default();
				for v in txn.all_db_configs(ns, db, version).await?.iter() {
					out.insert(v.name(), v.to_sql().into());
				}
				out.into()
			},
			"sequences" => {
				let mut out = Object::default();
				for v in txn.all_db_sequences(ns, db, version).await?.iter() {
					out.insert(v.name.clone(), v.to_sql().into());
				}
				out.into()
			},
		};
		Ok(Value::Object(Object::from(object)))
	}
}

fn process<T>(a: &Arc<[T]>) -> Value
where
	T: InfoStructure + Clone,
{
	Value::Array(a.iter().cloned().map(InfoStructure::structure).collect())
}