CVE-2025-11060

@@ -296,6 +296,33 @@ impl Document {
 		opt: &Options,
 		permitted: Permitted,
 	) -> Result<bool> {
+		// Check if reduction is required
+		if !self.check_reduction_required(opt)? {
+			return Ok(false);
+		}
+
+		match permitted {
+			Permitted::Initial => {
+				self.initial_reduced =
+					self.compute_reduced_target(stk, ctx, opt, &self.initial).await?;
+			}
+			Permitted::Current => {
+				self.current_reduced =
+					self.compute_reduced_target(stk, ctx, opt, &self.current).await?;
+			}
+			Permitted::Both => {
+				self.initial_reduced =
+					self.compute_reduced_target(stk, ctx, opt, &self.initial).await?;
+				self.current_reduced =
+					self.compute_reduced_target(stk, ctx, opt, &self.current).await?;
+			}
+		}
+
+		// Document has been reduced
+		Ok(true)
+	}
+
+	pub(crate) fn check_reduction_required(&self, opt: &Options) -> Result<bool> {
 		// Check if this record exists
 		if self.id.is_none() {
 			return Ok(false);
@@ -304,58 +331,54 @@ impl Document {
 		if !opt.check_perms(Action::View)? {
 			return Ok(false);
 		}
+
+		// Reduction is required
+		Ok(true)
+	}
+
+	pub(crate) async fn compute_reduced_target(
+		&self,
+		stk: &mut Stk,
+		ctx: &Context,
+		opt: &Options,
+		full: &CursorDoc,
+	) -> Result<CursorDoc> {
 		// Fetch the fields for the table
 		let fds = self.fd(ctx, opt).await?;
-		// Fetch the targets to process
-		let targets = match permitted {
-			Permitted::Initial => vec![(&self.initial, &mut self.initial_reduced)],
-			Permitted::Current => vec![(&self.current, &mut self.current_reduced)],
-			Permitted::Both => vec![
-				(&self.initial, &mut self.initial_reduced),
-				(&self.current, &mut self.current_reduced),
-			],
-		};
-		// Loop over the targets to process
-		for target in targets {
-			// Get the full document
-			let full = target.0;
-			// Process the full document
-			let mut out = (*full.doc).clone();
+		// The document to be reduced
+		let mut reduced = (*full.doc).clone();
+		// Loop over each field in document
+		for fd in fds.iter() {
 			// Loop over each field in document
-			for fd in fds.iter() {
-				// Loop over each field in document
-				for k in out.each(&fd.name).iter() {
-					// Process the field permissions
-					match &fd.permissions.select {
-						Permission::Full => (),
-						Permission::None => out.cut(k),
-						Permission::Specific(e) => {
-							// Disable permissions
-							let opt = &opt.new_with_perms(false);
-							// Get the initial value
-							let val = Arc::new(full.doc.as_ref().pick(k));
-							// Configure the context
-							let mut ctx = MutableContext::new(ctx);
-							ctx.add_value("value", val);
-							let ctx = ctx.freeze();
-							// Process the PERMISSION clause
-							if !stk
-								.run(|stk| e.compute(stk, &ctx, opt, Some(full)))
-								.await
-								.catch_return()?
-								.is_truthy()
-							{
-								out.cut(k);
-							}
+			for k in reduced.each(&fd.name).iter() {
+				// Process the field permissions
+				match &fd.permissions.select {
+					Permission::Full => (),
+					Permission::None => reduced.cut(k),
+					Permission::Specific(e) => {
+						// Disable permissions
+						let opt = &opt.new_with_perms(false);
+						// Get the initial value
+						let val = Arc::new(full.doc.as_ref().pick(k));
+						// Configure the context
+						let mut ctx = MutableContext::new(ctx);
+						ctx.add_value("value", val);
+						let ctx = ctx.freeze();
+						// Process the PERMISSION clause
+						if !stk
+							.run(|stk| e.compute(stk, &ctx, opt, Some(full)))
+							.await
+							.catch_return()?
+							.is_truthy()
+						{
+							reduced.cut(k);
 						}
 					}
 				}
 			}
-			// Update the permitted document
-			target.1.doc = out.into();
 		}
-		// Return the permitted document
-		Ok(true)
+		// Ok
+		Ok(CursorDoc::new(full.rid.clone(), full.ir.clone(), reduced))
 	}
 
 	/// Retrieve the record id for this document

@@ -8,9 +8,9 @@ use crate::ctx::{Context, MutableContext};
 use crate::dbs::{Action, Notification, Options, Statement};
 use crate::doc::{CursorDoc, Document};
 use crate::err::Error;
-use crate::expr::FlowResultExt as _;
 use crate::expr::paths::{AC, META, RD, TK};
 use crate::expr::permission::Permission;
+use crate::expr::{FlowResultExt as _, LiveStatement};
 use crate::val::Value;
 
 impl Document {
@@ -65,14 +65,9 @@ impl Document {
 				})
 				.map_err(anyhow::Error::new)?;
 			// Get the current and initial docs
+			// These are only used for EVENTS, so they should not be reduced
 			let current = self.current.doc.as_arc();
 			let initial = self.initial.doc.as_arc();
-			// Check if this is a delete statement
-			let doc = if stm.is_delete() {
-				&self.initial
-			} else {
-				&self.current
-			};
 			// Ensure that a session exists on the LIVE query
 			let sess = match lv.session.as_ref() {
 				Some(v) => v,
@@ -106,15 +101,29 @@ impl Document {
 			lqctx.add_value("value", current.clone());
 			lqctx.add_value("after", current);
 			lqctx.add_value("before", initial);
+			// Freeze the context
+			let lqctx = lqctx.freeze();
 			// We need to create a new options which we will
 			// use for processing this LIVE query statement.
 			// This ensures that we are using the auth data
 			// of the user who created the LIVE query.
 			let lqopt = opt.new_with_perms(true).with_auth(Arc::from(auth));
+
+			// Get the document to check against and to return based on lq context
+			let doc = match (self.check_reduction_required(&lqopt)?, stm.is_delete()) {
+				(true, true) => {
+					&self.compute_reduced_target(stk, &lqctx, &lqopt, &self.initial).await?
+				}
+				(true, false) => {
+					&self.compute_reduced_target(stk, &lqctx, &lqopt, &self.current).await?
+				}
+				(false, true) => &self.initial,
+				(false, false) => &self.current,
+			};
+
 			// First of all, let's check to see if the WHERE
 			// clause of the LIVE query is matched by this
 			// document. If it is then we can continue.
-			let lqctx = lqctx.freeze();
 			match self.lq_check(stk, &lqctx, &lqopt, &lq, doc).await {
 				Err(IgnoreError::Ignore) => continue,
 				Err(IgnoreError::Error(e)) => return Err(e),
@@ -124,7 +133,7 @@ impl Document {
 			// clause for this table allows this document to
 			// be viewed by the user who created this LIVE
 			// query. If it does, then we can continue.
-			match self.lq_allow(stk, &lqctx, &lqopt, &lq, doc).await {
+			match self.lq_allow(stk, &lqctx, &lqopt, &lq).await {
 				Err(IgnoreError::Ignore) => continue,
 				Err(IgnoreError::Error(e)) => return Err(e),
 				Ok(_) => (),
@@ -151,7 +160,7 @@ impl Document {
 					// An error ignore here is about livequery not the query which invoked the
 					// livequery trigger. So we should catch the ignore and skip this entry in this
 					// case.
-					let result = match self.pluck(stk, &lqctx, &lqopt, &lq).await {
+					let result = match self.lq_pluck(stk, &lqctx, &lqopt, lv, doc).await {
 						Err(IgnoreError::Ignore) => continue,
 						Err(IgnoreError::Error(e)) => return Err(e),
 						Ok(x) => x,
@@ -167,7 +176,7 @@ impl Document {
 					// An error ignore here is about livequery not the query which invoked the
 					// livequery trigger. So we should catch the ignore and skip this entry in this
 					// case.
-					let result = match self.pluck(stk, &lqctx, &lqopt, &lq).await {
+					let result = match self.lq_pluck(stk, &lqctx, &lqopt, lv, doc).await {
 						Err(IgnoreError::Ignore) => continue,
 						Err(IgnoreError::Error(e)) => return Err(e),
 						Ok(x) => x,
@@ -241,7 +250,6 @@ impl Document {
 		ctx: &Context,
 		opt: &Options,
 		stm: &Statement<'_>,
-		doc: &CursorDoc,
 	) -> Result<(), IgnoreError> {
 		// Should we run permissions checks?
 		if opt.check_perms(stm.into())? {
@@ -252,6 +260,13 @@ impl Document {
 				Permission::None => return Err(IgnoreError::Ignore),
 				Permission::Full => return Ok(()),
 				Permission::Specific(e) => {
+					// Retrieve the document to check permissions against
+					let doc = if stm.is_delete() {
+						&self.initial
+					} else {
+						&self.current
+					};
+
 					// Disable permissions
 					let opt = &opt.new_with_perms(false);
 					// Process the PERMISSION clause
@@ -269,4 +284,15 @@ impl Document {
 		// Carry on
 		Ok(())
 	}
+
+	async fn lq_pluck(
+		&self,
+		stk: &mut Stk,
+		ctx: &Context,
+		opt: &Options,
+		stm: &LiveStatement,
+		doc: &CursorDoc,
+	) -> Result<Value, IgnoreError> {
+		stm.expr.compute(stk, ctx, opt, Some(doc), false).await.map_err(IgnoreError::from)
+	}
 }

@@ -77,34 +77,9 @@ impl Document {
 				}
 			},
 			None => match stm {
-				Statement::Live(s) => {
-					// There was a if here which tested if the live statement had no selectors,
-					// which seems like it should never happen so I removed it.
-					/*
-					if s.expr.is_empty() {
-						// Process the permitted documents
-						let (initial, current) = if self.reduced(stk, ctx, opt, Both).await? {
-							(&self.initial_reduced, &self.current_reduced)
-						} else {
-							(&self.initial, &self.current)
-						};
-						// Output a DIFF of any changes applied to the document
-						let ops = initial.doc.as_ref().diff(current.doc.as_ref(), Idiom::default());
-						Ok(Operation::operations_to_value(ops))
-					} else {
-					*/
-					// Process the permitted documents
-					let current = if self.reduced(stk, ctx, opt, Current).await? {
-						&self.current_reduced
-					} else {
-						&self.current
-					};
-					// Process the LIVE SELECT statement fields
-					s.expr
-						.compute(stk, ctx, opt, Some(current), false)
-						.await
-						.map_err(IgnoreError::from)
-				}
+				Statement::Live(_) => Err(IgnoreError::Error(anyhow::anyhow!(
+					".lives() uses .lq_pluck(), not .pluck()"
+				))),
 				Statement::Select(s) => {
 					// Process the permitted documents
 					let current = if self.reduced(stk, ctx, opt, Current).await? {

@@ -1,7 +1,10 @@
 mod helpers;
+use std::time::Duration;
+
 use anyhow::Result;
 use helpers::{new_ds, skip_ok};
-use surrealdb_core::dbs::Session;
+use surrealdb_core::dbs::{Action, Session, Variables};
+use surrealdb_core::expr::Kind;
 use surrealdb_core::val::RecordId;
 use surrealdb_core::{strand, syn};
 
@@ -72,3 +75,182 @@ async fn live_permissions() -> Result<()> {
 	//
 	Ok(())
 }
+
+#[tokio::test]
+async fn live_document_reduction() -> Result<()> {
+	// Create a new datastore with notifications enabled
+	let dbs = new_ds().await?.with_auth_enabled(true).with_notifications();
+	let Some(channel) = dbs.notifications() else {
+		unreachable!("No notification channel");
+	};
+
+	// Create sessions for owner and record user
+	let ses_owner = Session::owner().with_ns("test").with_db("test").with_rt(true);
+	let ses_record = Session::for_record(
+		"test",
+		"test",
+		"test",
+		RecordId::new("user".to_owned(), strand!("test").to_owned()).into(),
+	)
+	.with_rt(true);
+
+	// Setup the scenario
+	let sql = "
+			DEFINE TABLE test SCHEMAFULL PERMISSIONS FULL;
+			DEFINE FIELD visible ON test PERMISSIONS FULL;
+			DEFINE FIELD hidden ON test PERMISSIONS NONE;
+		";
+	let res = &mut dbs.execute(sql, &ses_owner, None).await?;
+	assert_eq!(res.len(), 3);
+	skip_ok(res, 3)?;
+
+	////////////////////////////////////////////////////////////
+
+	// Create a simple live query
+	let sql = "LIVE SELECT * FROM test;";
+	let res = &mut dbs.execute(sql, &ses_record, None).await?;
+	assert_eq!(res.len(), 1);
+	let lqid = res.remove(0).result?;
+	assert_eq!(lqid.kind(), Some(Kind::Uuid));
+
+	////////////////////////////////////////////////////////////
+
+	// Create a record
+	let sql = "CREATE test:1 SET hidden = 123, visible = 'abc';";
+	let res = &mut dbs.execute(sql, &ses_owner, None).await?;
+	assert_eq!(res.len(), 1);
+	//
+	let tmp = res.remove(0).result?;
+	let val = syn::value(
+		"[
+			{
+				id: test:1,
+				visible: 'abc',
+				hidden: 123,
+			},
+		]",
+	)
+	.unwrap();
+	assert_eq!(tmp, val);
+
+	// Receive the notification
+	let tmp = channel.recv().await?;
+	assert_eq!(tmp.action, Action::Create);
+
+	// Check the notification
+	let val = syn::value(
+		"{
+			id: test:1,
+			visible: 'abc',
+		}",
+	)
+	.unwrap();
+	assert_eq!(tmp.result, val);
+
+	////////////////////////////////////////////////////////////
+
+	// Update the record
+	let sql = "UPDATE test:1 SET hidden = 456, visible = 'def';";
+	let res = &mut dbs.execute(sql, &ses_owner, None).await?;
+	assert_eq!(res.len(), 1);
+	//
+	let tmp = res.remove(0).result?;
+	let val = syn::value(
+		"[
+			{
+				id: test:1,
+				visible: 'def',
+				hidden: 456,
+			}
+		]",
+	)
+	.unwrap();
+	assert_eq!(tmp, val);
+
+	// Receive the notification
+	let tmp = channel.recv().await?;
+	assert_eq!(tmp.action, Action::Update);
+
+	// Check the notification
+	let val = syn::value(
+		"{
+			id: test:1,
+			visible: 'def',
+		}",
+	)
+	.unwrap();
+	assert_eq!(tmp.result, val);
+
+	////////////////////////////////////////////////////////////
+
+	// Delete the record
+	let sql = "DELETE test:1;";
+	let res = &mut dbs.execute(sql, &ses_owner, None).await?;
+	assert_eq!(res.len(), 1);
+	skip_ok(res, 1)?;
+
+	// Receive the notification
+	let tmp = channel.recv().await?;
+	assert_eq!(tmp.action, Action::Delete);
+
+	// Check the notification
+	let val = syn::value(
+		"{
+			id: test:1,
+			visible: 'def',
+		}",
+	)
+	.unwrap();
+	assert_eq!(tmp.result, val);
+
+	////////////////////////////////////////////////////////////
+
+	// Kill the live query
+	let sql = "KILL $uuid";
+	let res = &mut dbs
+		.execute(sql, &ses_owner, Some(Variables(map!("uuid".to_string() => lqid))))
+		.await?;
+	assert_eq!(res.len(), 1);
+	skip_ok(res, 1)?;
+
+	// Receive the notification
+	let tmp = channel.recv().await?;
+	assert_eq!(tmp.action, Action::Killed);
+
+	// Create a live query with a WHERE clause
+	let sql = "LIVE SELECT * FROM test WHERE hidden = 123;";
+	let res = &mut dbs.execute(sql, &ses_record, None).await?;
+	assert_eq!(res.len(), 1);
+	let lqid = res.remove(0).result?;
+	assert_eq!(lqid.kind(), Some(Kind::Uuid));
+
+	////////////////////////////////////////////////////////////
+
+	// Create a record
+	let sql = "CREATE test:2 SET hidden = 123, visible = 'abc';";
+	let res = &mut dbs.execute(sql, &ses_owner, None).await?;
+	assert_eq!(res.len(), 1);
+	//
+	let tmp = res.remove(0).result?;
+	let val = syn::value(
+		"[
+			{
+				id: test:2,
+				visible: 'abc',
+				hidden: 123,
+			},
+		]",
+	)
+	.unwrap();
+	assert_eq!(tmp, val);
+
+	// Assert no notification is received
+	tokio::time::sleep(Duration::from_secs(1)).await;
+	let res = channel.try_recv();
+	assert!(res.is_err());
+
+	////////////////////////////////////////////////////////////
+
+	// Test passed!
+	Ok(())
+}