surrealdb-core 3.2.0

use anyhow::{Result, bail};
use reblessive::tree::Stk;
use surrealdb_types::ToSql;
use uuid::Uuid;

use crate::catalog::providers::TableProvider;
use crate::catalog::{NodeLiveQuery, SubscriptionDefinition, SubscriptionFields};
use crate::ctx::FrozenContext;
use crate::dbs::{Options, ParameterCapturePass, Variables};
use crate::doc::CursorDoc;
use crate::err::Error;
use crate::expr::visit::{Visit, Visitor};
use crate::expr::{Cond, Expr, Fetchs, Fields, FlowResultExt as _, Idiom, Param};
use crate::val::Value;

/// The set of parameter names that carry per-event document data and are therefore
/// only meaningful at notification time, not at LIVE query registration time.
const DOC_PARAMS: &[&str] = &["value", "before", "after", "event", "this"];

/// Visitor that short-circuits with `Err(())` on the first expression node that
/// depends on the document being processed:
///
/// - Any field-access [`Idiom`] (e.g. `name`, `user.age`).
/// - Any of the live-query event params: `$value`, `$before`, `$after`, `$event`, `$this`.
///
/// Session params (`$access`, `$auth`, `$token`, `$session`) and user-defined LET
/// params are captured at registration time and treated as constants.
///
/// If the visitor completes without returning `Err`, the expression is
/// "document-independent": its result is the same for every record that triggers
/// the LIVE query, so it can be evaluated once at registration time to catch
/// errors early.
struct IsDocumentDependentChecker;

impl Visitor for IsDocumentDependentChecker {
	/// `()` is used as an early-termination signal — `Err(())` means "found a
	/// document-dependent node, stop traversal".
	type Error = ();

	fn visit_idiom(&mut self, _idiom: &Idiom) -> Result<(), ()> {
		Err(()) // any field access → document-dependent, stop
	}

	fn visit_param(&mut self, param: &Param) -> Result<(), ()> {
		if DOC_PARAMS.contains(&param.as_str()) {
			Err(()) // document-event param → document-dependent, stop
		} else {
			Ok(())
		}
	}
}

/// Returns `true` if `expr` contains any field-access idiom or document-event
/// param, meaning its result can differ per-document and it cannot be safely
/// evaluated at LIVE query registration time.
fn is_document_dependent(expr: &Expr) -> bool {
	expr.visit(&mut IsDocumentDependentChecker).is_err()
}

#[derive(Clone, Debug, Eq, PartialEq, Hash)]
pub enum LiveFields {
	Diff,
	Select(Fields),
}

#[derive(Clone, Debug, Eq, PartialEq, Hash)]
pub(crate) struct LiveStatement {
	pub id: Uuid,
	pub node: Uuid,
	pub fields: LiveFields,
	pub what: Expr,
	pub cond: Option<Cond>,
	pub fetch: Option<Fetchs>,
}

impl LiveStatement {
	/// Process this type returning a computed simple Value
	#[instrument(level = "trace", name = "LiveStatement::compute", skip_all)]
	pub(crate) async fn compute(
		&self,
		stk: &mut Stk,
		ctx: &FrozenContext,
		opt: &Options,
		doc: Option<&CursorDoc>,
	) -> Result<Value> {
		// Is realtime enabled?
		ctx.realtime()?;
		// Valid options?
		opt.valid_for_db()?;
		// Get the Node ID
		let nid = ctx.node_id();

		let mut vars = Variables::new();
		let mut pass = ParameterCapturePass {
			context: ctx,
			captures: &mut vars,
		};
		if let LiveFields::Select(x) = &self.fields {
			let _ = x.visit(&mut pass);
		};
		let _ = self.what.visit(&mut pass);
		if let Some(cond) = &self.cond {
			let _ = cond.0.visit(&mut pass);
		}
		if let Some(fetch) = &self.fetch {
			for i in fetch.iter() {
				let _ = i.0.visit(&mut pass);
			}
		}

		let fields = match &self.fields {
			LiveFields::Diff => SubscriptionFields::Diff,
			LiveFields::Select(x) => SubscriptionFields::Select(x.clone()),
		};

		// Check that auth has been set
		let mut subscription_definition = SubscriptionDefinition {
			id: self.id,
			node: self.node,
			fields,
			what: self.what.clone(),
			cond: self.cond.clone().map(|c| c.0),
			fetch: self.fetch.clone(),

			// Use the current session authentication
			// for when we store the LIVE Statement
			auth: Some(opt.auth.as_ref().clone()),
			// Use the current session authentication
			// for when we store the LIVE Statement
			session: ctx.value("session").cloned(),
			// Add the variables to the subscription definition. Keys are
			// copied out of `Strand` into owned `String` here because
			// `SubscriptionDefinition` is persisted in the catalog and
			// stores `BTreeMap<String, Value>`.
			vars: vars.0.into_iter().map(|(k, v)| (k.into_string(), v)).collect(),
		};
		// Get the id
		let live_query_id = subscription_definition.id;
		// Process the live query table
		match stk
			.run(|stk| subscription_definition.what.compute(stk, ctx, opt, doc))
			.await
			.catch_return()?
		{
			Value::Table(tb) => {
				// Store the current Node ID
				subscription_definition.node = nid;
				// Get the NS and DB
				let (ns, db) = ctx.expect_ns_db_ids(opt).await?;
				// Get the transaction
				let txn = ctx.tx();
				// Ensure that the table definition exists
				{
					let (ns, db) = opt.ns_db()?;
					txn.expect_tb_by_name(ns, db, &tb).await?;
				}
				// If the WHERE clause is document-independent (no field references or
				// document-event params), evaluate it now while we still have a full
				// execution context. This catches obviously broken expressions such as
				// `WHERE string::len(NONE)` at registration time instead of silently
				// suppressing every future notification.
				if let Some(cond) = &self.cond
					&& !is_document_dependent(&cond.0)
					&& let Err(e) =
						stk.run(|stk| cond.0.compute(stk, ctx, opt, None)).await.catch_return()
				{
					bail!("LIVE query WHERE clause is invalid and will never match: {e}");
				}
				// Insert the node live query
				let key = crate::key::node::lq::new(nid, live_query_id);
				txn.replace(
					&key,
					&NodeLiveQuery {
						ns,
						db,
						tb: tb.clone(),
					},
				)
				.await?;
				// Insert the table live query
				let key = crate::key::table::lq::new(ns, db, &tb, live_query_id);
				txn.replace(&key, &subscription_definition).await?;
				// Refresh the table cache for lives
				if let Some(cache) = ctx.get_cache() {
					cache.set_live_queries_version(ns, db, &tb);
				}
				// Clear the cache
				txn.clear_cache();
			}
			v => {
				bail!(Error::LiveStatement {
					value: v.to_sql(),
				});
			}
		};
		// Return the query id
		Ok(crate::val::Uuid(live_query_id).into())
	}
}

#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod tests {
	use anyhow::Result;

	use crate::catalog::providers::{CatalogProvider, TableProvider};
	use crate::channel::Receiver;
	use crate::dbs::{Capabilities, Session};
	use crate::kvs::Datastore;
	use crate::kvs::LockType::Optimistic;
	use crate::kvs::TransactionType::Write;
	use crate::syn;
	use crate::types::{
		PublicAction, PublicNotification, PublicRecordId, PublicRecordIdKey, PublicValue,
	};

	pub async fn new_ds() -> Result<(Receiver<PublicNotification>, Datastore)> {
		let (send, recv) = crate::channel::bounded(1000);
		let ds = Datastore::builder()
			.with_capabilities(Capabilities::all())
			.with_notify(send)
			.build_with_path("memory")
			.await?;
		Ok((recv, ds))
	}

	#[tokio::test]
	async fn test_table_definition_is_created_for_live_query() {
		let (recv, dbs) = new_ds().await.unwrap();
		let (ns, db, tb) = ("test", "test", "person");
		let ses = Session::owner().with_ns(ns).with_db(db).with_rt(true);

		let tx = dbs.transaction(Write, Optimistic).await.unwrap();
		let db = tx.ensure_ns_db(None, ns, db).await.unwrap();
		tx.commit().await.unwrap();

		// Create a new transaction and verify that there are no tables defined.
		let tx = dbs.transaction(Write, Optimistic).await.unwrap();
		let table_occurrences = &*(tx.all_tb(db.namespace_id, db.database_id, None).await.unwrap());
		assert!(table_occurrences.is_empty());
		tx.cancel().await.unwrap();

		// Define the table
		let define_statement = format!("DEFINE TABLE {tb};");
		dbs.execute(&define_statement, &ses, None).await.unwrap();

		// Initiate a live query statement
		let lq_stmt = format!("LIVE SELECT * FROM {}", tb);
		let live_query_response = &mut dbs.execute(&lq_stmt, &ses, None).await.unwrap();

		let live_id = live_query_response.remove(0).result.unwrap();
		let live_id = match live_id {
			PublicValue::Uuid(id) => id,
			_ => panic!("expected uuid"),
		};

		// Verify that the table definition has been created.
		let tx = dbs.transaction(Write, Optimistic).await.unwrap();
		let table_occurrences = &*(tx.all_tb(db.namespace_id, db.database_id, None).await.unwrap());
		assert_eq!(table_occurrences.len(), 1);
		assert_eq!(table_occurrences[0].name, tb);
		tx.cancel().await.unwrap();

		// Initiate a Create record
		let create_statement = format!("CREATE {tb}:test_true SET condition = true");
		let create_response = &mut dbs.execute(&create_statement, &ses, None).await.unwrap();
		assert_eq!(create_response.len(), 1);
		let expected_record: PublicValue = syn::value(&format!(
			"[{{
				id: {tb}:test_true,
				condition: true,
			}}]"
		))
		.unwrap();

		let tmp = create_response.remove(0).result.unwrap();
		assert_eq!(tmp, expected_record);

		// Create a new transaction to verify that the same table was used.
		let tx = dbs.transaction(Write, Optimistic).await.unwrap();
		let table_occurrences = &*(tx.all_tb(db.namespace_id, db.database_id, None).await.unwrap());
		assert_eq!(table_occurrences.len(), 1);
		assert_eq!(table_occurrences[0].name, tb);
		tx.cancel().await.unwrap();

		// Validate notification
		let notification = recv.recv().await.unwrap();
		assert_eq!(
			notification,
			PublicNotification::new(
				live_id,
				None,
				PublicAction::Create,
				PublicValue::RecordId(PublicRecordId {
					table: tb.into(),
					key: PublicRecordIdKey::String("test_true".to_owned())
				}),
				syn::value(&format!(
					"{{
						id: {tb}:test_true,
						condition: true,
					}}"
				))
				.unwrap(),
			)
		);
	}

	#[tokio::test]
	async fn test_table_exists_for_live_query() {
		let (_, dbs) = new_ds().await.unwrap();
		let (ns, db, tb) = ("test", "test", "person");
		let ses = Session::owner().with_ns(ns).with_db(db).with_rt(true);

		let tx = dbs.transaction(Write, Optimistic).await.unwrap();
		let db = tx.ensure_ns_db(None, ns, db).await.unwrap();
		tx.commit().await.unwrap();

		// Create a new transaction and verify that there are no tables defined.
		let tx = dbs.transaction(Write, Optimistic).await.unwrap();
		let table_occurrences = &*(tx.all_tb(db.namespace_id, db.database_id, None).await.unwrap());
		assert!(table_occurrences.is_empty());
		tx.cancel().await.unwrap();

		// Initiate a Create record
		let create_statement = format!("CREATE {}:test_true SET condition = true", tb);
		dbs.execute(&create_statement, &ses, None).await.unwrap();

		// Create a new transaction and confirm that a new table is created.
		let tx = dbs.transaction(Write, Optimistic).await.unwrap();
		let table_occurrences = &*(tx.all_tb(db.namespace_id, db.database_id, None).await.unwrap());
		assert_eq!(table_occurrences.len(), 1);
		assert_eq!(table_occurrences[0].name, tb);
		tx.cancel().await.unwrap();

		// Initiate a live query statement
		let lq_stmt = format!("LIVE SELECT * FROM {}", tb);
		dbs.execute(&lq_stmt, &ses, None).await.unwrap();

		// Verify that the old table definition was used.
		let tx = dbs.transaction(Write, Optimistic).await.unwrap();
		let table_occurrences = &*(tx.all_tb(db.namespace_id, db.database_id, None).await.unwrap());
		assert_eq!(table_occurrences.len(), 1);
		assert_eq!(table_occurrences[0].name, tb);
		tx.cancel().await.unwrap();
	}
}