surrealdb-core 3.3.1

use std::sync::Arc;

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

use crate::catalog::{LATEST_EDGE_VARIANT, RecordType};
use crate::ctx::{Context, FrozenContext};
use crate::dbs::{Options, Statement};
use crate::doc::{CursorDoc, Document, Error as DocError, Extras};
use crate::exe::FlowResultExt;
use crate::expr::data::Data;
use crate::expr::paths::{ID, IN, OUT};
use crate::expr::{AssignOperator, Error, Idiom, Kind, KindLiteral, Operation, Part};
use crate::iam::AuthLimit;
use crate::val::{Array, RecordId, RecordIdKey, TableName, Value};

impl Document {
	/// Generate (or finalise) a record ID for CREATE, UPSERT, RELATE, and
	/// INSERT statements.
	///
	/// This method handles record ID generation from various sources:
	/// - Existing document IDs
	/// - Data clause specified IDs (including function calls and expressions)
	/// - The `id` field's `DEFAULT` expression when no ID is specified
	/// - Synthesised IDs (UUID / random string / singleton literal) by kind
	/// - Explicit record-id targets (e.g. `CREATE foo:bar`)
	///
	/// In every case the resulting key is validated and coerced against the
	/// declared kind of the `id` field, if one is defined. The method ensures
	/// that all expressions are properly evaluated before being used as record
	/// IDs.
	pub(super) async fn generate_record_id(
		&mut self,
		stk: &mut Stk,
		ctx: &FrozenContext,
		opt: &Options,
	) -> Result<()> {
		// A lightweight relation's edge id is canonical — `[in, out]`, derived
		// from the endpoints, never generated. A record-id target must match
		// it: the id is the edge's entire identity and storage, so an
		// arbitrary id would name a different edge than the one related.
		if let Extras::Relate(l, r, _) = &self.extras
			&& self.table_is_lightweight()?
		{
			let table = self.doc_ctx.tb()?.name.clone();
			let canonical = RecordIdKey::Array(Array(vec![
				Value::RecordId(l.clone()),
				Value::RecordId(r.clone()),
			]));
			if let Some(existing) = self.id.take() {
				ensure!(
					existing.key == canonical,
					crate::exec::Error::Thrown(format!(
						"a LIGHTWEIGHT relation's edge id is canonical: expected {}:{}, found {}",
						table,
						canonical.to_sql(),
						existing.to_sql(),
					))
				);
			}
			self.id = Some(Arc::new(RecordId {
				table,
				key: canonical,
			}));
			return Ok(());
		}

		// This is a CREATE, UPSERT, RELATE, or INSERT statement.
		//
		// Locate the precomputed `id` field definition (if any). Its declared
		// kind constrains the key (so auto-generated keys conform and supplied
		// keys are coerced); its `DEFAULT` supplies the key when none is given.
		// The lookup is O(1) per write — the field index is found once when the
		// table context is built, never rescanned per record.
		let id_field = self.doc_ctx.id_field()?;
		let id_kind = id_field.and_then(|fd| fd.field_kind.as_ref());
		// Evaluate the `id` field's `DEFAULT` with the field's auth-level clamp
		// applied — as the regular field pipeline does for every field — so a
		// lower-privileged definer's DEFAULT cannot run with the caller's
		// broader privileges.
		let id_opt;
		let opt = if let Some(fd) = id_field {
			id_opt = opt.limited_by(&AuthLimit::try_from(&fd.auth_limit)?);
			&id_opt
		} else {
			opt
		};
		// A table target (e.g. `CREATE foo`) means the id may need to be derived
		// from the data clause, produced by the id field's `DEFAULT`, or
		// synthesised; an explicit record-id target (e.g. `CREATE foo:bar`) is
		// already on `self.id` and only needs coercing.
		if let Some(tb) = self.r#gen.clone() {
			// An id already present in the current data takes precedence, then
			// an explicit id in the data clause (e.g. `SET id = …`).
			let supplied = {
				let existing = self.current.doc.as_ref().pick(&ID);
				if existing.is_some() {
					existing
				} else {
					self.input_data
						.as_ref()
						.map(|data| data.pick(ID.as_ref()))
						.unwrap_or(Value::None)
				}
			};
			let id = if supplied.is_some() {
				// A concrete id was supplied; use it.
				supplied.generate(tb, false)?
			} else if let Some(crate::catalog::DefineDefault::Set(expr)) =
				id_field.map(|fd| &fd.default)
			{
				// No id supplied, but the `id` field declares a `DEFAULT`:
				// evaluate it and use the result as the record id. Session
				// params (`$auth`), functions (`time::now()`), and references to
				// the record's own fields all resolve. For INSERT the row data
				// is not yet merged into `self.current` (the id must be known
				// before `process_merge_data` stamps and merges it), so evaluate
				// against the insert payload directly to keep field references
				// consistent with CREATE.
				let insert_doc;
				let doc = if let Extras::Insert(v) = &self.extras {
					insert_doc = CursorDoc::from(v.as_ref().clone());
					Some(&insert_doc)
				} else {
					Some(&self.current)
				};
				let value = stk
					.run(|stk| crate::legacy::expr_compute(expr, stk, ctx, opt, doc))
					.await
					.catch_return()?;
				value.generate(tb, false)?
			} else {
				// No id and no `DEFAULT`: synthesise one for the declared kind.
				Self::generate_typed_id(&tb, id_kind)?
			};
			// The id field can not be a record range
			ensure!(
				!id.key.is_range(),
				Error::IdInvalid {
					value: id.to_sql(),
				}
			);
			// Coerce the key to the declared `id` kind. For generated keys this
			// is a no-op; for supplied or `DEFAULT`-produced keys it validates
			// against the declared type (e.g. rejecting a string key on a
			// `TYPE uuid` id, or a non-int element on a `TYPE array<int>` id).
			let id = Self::coerce_id_key(id, id_kind)?;
			// Set the document id
			self.id = Some(Arc::new(id));
		} else if id_kind.is_some() {
			// An explicit record id was supplied as the statement target (e.g.
			// `CREATE foo:bar`); there is nothing to generate, but the supplied
			// key must still be validated against the declared kind, exactly as
			// a key derived from the data clause would be. Range targets never
			// reach here — they are diverted to range handling and rejected in
			// `Iterator::prepare_range` for create/upsert/relate/insert.
			if let Some(existing) = self.id.take() {
				let id = Self::coerce_id_key(RecordId::clone(&existing), id_kind)?;
				self.id = Some(Arc::new(id));
			}
		}
		Ok(())
	}

	/// Synthesise a fresh record id whose key conforms to the declared `id`
	/// field kind. With no declared kind (or an unconstrained / string kind)
	/// this reproduces the historic behaviour of a random string key.
	fn generate_typed_id(tb: &TableName, kind: Option<&Kind>) -> Result<RecordId> {
		let key = match kind {
			// No declared kind, or an unconstrained / string id: random string.
			None | Some(Kind::Any | Kind::String) => RecordIdKey::rand(),
			// A uuid id: a fresh, time-ordered UUIDv7.
			Some(Kind::Uuid) => RecordIdKey::uuid(),
			// A singleton scalar literal id (e.g. `TYPE 123`, `TYPE 'foo'`) has
			// exactly one valid value, so synthesise it directly.
			Some(Kind::Literal(KindLiteral::Integer(i))) => RecordIdKey::Number(*i),
			Some(Kind::Literal(KindLiteral::String(s))) => RecordIdKey::String(s.clone()),
			// Anything else (int, number, array, object, a union, or a literal
			// that is not a single concrete scalar) cannot be synthesised
			// without an explicit value or sequence.
			Some(other) => bail!(DocError::IdFieldGenerateUnsupported {
				table: tb.to_string(),
				kind: other.to_sql(),
			}),
		};
		Ok(RecordId {
			table: tb.clone(),
			key,
		})
	}

	/// Coerce a record id key to the declared `id` field kind. Returns the id
	/// unchanged when there is no declared kind or the kind constrains the
	/// outer record rather than the key. Otherwise the key is converted to a
	/// value, coerced to the kind, and converted back via the same path used
	/// for user-supplied ids.
	fn coerce_id_key(id: RecordId, kind: Option<&Kind>) -> Result<RecordId> {
		let Some(kind) = kind else {
			return Ok(id);
		};
		// Record-typed id kinds constrain the outer record, not the key value.
		if kind.is_record() {
			return Ok(id);
		}
		// Coerce a clone of the key so the original `id` stays intact for the
		// error context, which is formatted (`to_sql`) only on failure rather
		// than eagerly on every typed-id write.
		match id.key.clone().into_value().coerce_to_kind(kind) {
			// Rebuild the record id from the coerced value using the robust
			// conversion that handles ints, uuids, arrays, and objects.
			Ok(coerced) => coerced.generate(id.table, false),
			Err(error) => Err(DocError::FieldCoerce {
				record: id.to_sql(),
				field_name: "id".to_string(),
				error: Box::new(error),
			}
			.into()),
		}
	}

	/// Clears all of the content of this document.
	/// This is used to empty the current content
	/// of the document within a `DELETE` statement.
	/// This function only clears the document in
	/// memory, and does not store this on disk.
	pub(super) fn clear_record_data(&mut self) {
		*self.current.doc = Default::default();
	}

	/// Sets the default field data that should be
	/// present on this document. For normal records
	/// the `id` field is always specified, and for
	/// relation records, the `in`, `out`, and the
	/// hidden `edge` field are always present. This
	/// ensures that any user modifications of these
	/// fields are reset back to the original state.
	pub(super) fn default_record_data(&mut self) -> Result<()> {
		// Get the record id
		let rid = self.id()?;
		// Set default field values
		self.current.doc.to_mut().def(RecordId::clone(&rid));
		// This is a RELATE statement, so reset fields
		if let Extras::Relate(l, r, _) = &self.extras {
			// Stamp the record-type marker to the current adjacency-key
			// generation. This runs before `store_record_data` writes the
			// record to disk, so the on-disk metadata always matches the
			// keys `store_edges_data` will emit later in the pipeline.
			//
			// Three cases are folded together by this single condition:
			//   * Brand-new edge (`current` not yet an edge): stamp it.
			//   * Re-RELATE of a stale-variant edge (e.g. legacy variant 1 being migrated to 2):
			//     advance the stamp so the post-migration record reflects the upgraded layout.
			//   * Re-RELATE of a current-variant edge: no-op skip, avoiding an `Arc::make_mut`
			//     clone for nothing.
			//
			// `current` starts as a clone of `initial`, and users can't
			// address `metadata` themselves, so the only way `current`
			// can already carry the current variant here is if `initial`
			// did — i.e. there's no risk of a stale write masking a
			// genuine migration.
			if self.current.doc.edge_variant() != Some(LATEST_EDGE_VARIANT) {
				self.current.doc.set_record_type(RecordType::Edge {
					variant: LATEST_EDGE_VARIANT,
				});
			}
			// If this document existed before, check the `in` field
			match (self.initial.doc.as_ref().pick(&IN), self.is_new()) {
				// If the document id matches, then all good
				(Value::RecordId(id), false) if id == *l => {
					self.current.doc.to_mut().put(&IN, l.clone().into());
				}
				// If the document is new then all good
				(_, true) => {
					self.current.doc.to_mut().put(&IN, l.clone().into());
				}
				// Otherwise this is attempting to override the `in` field
				(v, _) => {
					bail!(DocError::InOverride {
						value: v.to_sql(),
					})
				}
			}
			// If this document existed before, check the `out` field
			match (self.initial.doc.as_ref().pick(&OUT), self.is_new()) {
				// If the document id matches, then all good
				(Value::RecordId(id), false) if id == *r => {
					self.current.doc.to_mut().put(&OUT, r.clone().into());
				}
				// If the document is new then all good
				(_, true) => {
					self.current.doc.to_mut().put(&OUT, r.clone().into());
				}
				// Otherwise this is attempting to override the `in` field
				(v, _) => {
					bail!(DocError::OutOverride {
						value: v.to_sql(),
					})
				}
			}
		}
		// This is an UPDATE of a graph edge, so reset its `in` / `out`
		// fields to whatever the prior record held. The edge marker
		// itself doesn't need to be re-stamped: `current` is a clone
		// of `initial`, and only `data` (not `metadata`) is reachable
		// through `to_mut()`, so the variant on the prior edge flows
		// through to the new write untouched.
		if self.initial.doc.is_edge() {
			self.current.doc.to_mut().put(&IN, self.initial.doc.as_ref().pick(&IN));
			self.current.doc.to_mut().put(&OUT, self.initial.doc.as_ref().pick(&OUT));
		}
		// Carry on
		Ok(())
	}

	/// Updates the current document using the data
	/// passed in to each document. This is relevant
	/// for INSERT and RELATE queries where each
	/// document has its own data block. This
	/// function also ensures that standard default
	/// fields are set and reset before and after the
	/// document data is modified.
	pub(super) fn process_merge_data(&mut self) -> Result<()> {
		// Get the record id
		let rid = self.id()?;
		// Set default field values
		self.current.doc.to_mut().def(RecordId::clone(&rid));
		// This is an INSERT statement
		if let Extras::Insert(v) = &self.extras {
			self.current.doc.to_mut().merge(Value::clone(v))?;
		}
		// This is an INSERT RELATION statement
		if let Extras::Relate(_, _, Some(v)) = &self.extras {
			self.current.doc.to_mut().merge(Value::clone(v))?;
		}
		// Carry on
		Ok(())
	}

	/// Reject a `PATCH` whose `copy`/`move` `from` or `test` `path` pointer
	/// reads a field the caller may not select.
	///
	/// `Value::patch` applies its operations to the full record loaded from
	/// storage and resolves those pointers against it, so without this gate a
	/// caller holding table-level write permission reads a field denied by
	/// `DEFINE FIELD ... PERMISSIONS FOR select`: `copy`/`move` land the value
	/// in a destination of the caller's choosing, and `test` reports it in the
	/// mismatch error. The `SET`/`CONTENT`/`MERGE`/`REPLACE` clauses are not
	/// affected because their right-hand sides are computed against the reduced
	/// view and already observe `NONE`.
	///
	/// A pointer is judged by comparing what it resolves to on the record with
	/// what it resolves to on the permission-reduced view of that record. Any
	/// difference means the pointer reaches data the reduction removed, whether
	/// the pointer names the restricted field directly or an ancestor object or
	/// array that contains it. Delegating to the reduction rather than
	/// re-deriving which paths are restricted keeps this in step with the
	/// field-permission kernel, including its wildcard and array-element
	/// handling.
	///
	/// The operations are walked in order against a copy of the record that
	/// each one is applied to in turn, and every pointer is judged against that
	/// copy as it stands when its own operation runs. A patch is a sequence,
	/// not a set: an earlier operation that inserts into or removes from an
	/// array moves every later element by one, so a pointer judged against the
	/// record as it was before the patch names a different value by the time it
	/// is read — which is enough to walk a restricted field into range of a
	/// pointer that resolved to nothing when the patch arrived. The reduced
	/// view is recomputed alongside for the same reason: a `PERMISSIONS FOR
	/// select` predicate reads the record, so what it permits moves with it.
	///
	/// `add` and `change` also read a pointer, but each writes its result back
	/// to the same path it read, so neither carries a value out of the field it
	/// came from.
	async fn check_patch_read_pointers(
		&mut self,
		stk: &mut Stk,
		ctx: &FrozenContext,
		opt: &Options,
		ops: &Value,
	) -> Result<()> {
		/// The pointer an operation reads and then writes somewhere other than
		/// where it read it, mapped by `Operation::path_to_parts` — the mapping
		/// `Value::patch` itself uses, so that the location judged here is the
		/// location the patch reaches. Re-deriving it would let the two drift,
		/// and a pointer judged at a location the patch does not read enforces
		/// nothing.
		///
		/// Matched exhaustively on purpose: an operation added without being
		/// classified here would otherwise default to carrying nothing out.
		fn exposed_read_pointer(op: &Operation) -> Option<Vec<Part>> {
			let path = match op {
				Operation::Copy {
					from,
					..
				}
				| Operation::Move {
					from,
					..
				} => from,
				Operation::Test {
					path,
					..
				} => path,
				Operation::Add {
					..
				}
				| Operation::Remove {
					..
				}
				| Operation::Replace {
					..
				}
				| Operation::Change {
					..
				} => return None,
			};
			Some(Operation::path_to_parts(path))
		}
		// Nothing to enforce when field permissions do not apply to this actor.
		if !self.reduction_required(ctx, opt)? {
			return Ok(());
		}
		let operations = Operation::value_to_operations(ops.clone())
			.map_err(Error::InvalidPatch)
			.map_err(anyhow::Error::new)?;
		// Nothing to enforce when no operation carries a value out of the
		// location it read, which spares the reduction on the common patch.
		if !operations.iter().any(|op| exposed_read_pointer(op).is_some()) {
			return Ok(());
		}
		// A copy of the record, advanced one operation at a time so that each
		// pointer is judged against what its own operation will read.
		let mut working = self.current.clone();
		for op in operations {
			if let Some(pointer) = exposed_read_pointer(&op) {
				let reduced = self.reduce_to_owned(stk, ctx, opt, &working).await?;
				if working.doc.as_ref().pick(&pointer) != reduced.doc.as_ref().pick(&pointer) {
					let field = Idiom(pointer);
					bail!(DocError::PatchPointerNotAllowed(field.to_sql()));
				}
			}
			// An operation that fails here fails the statement when the patch
			// is applied for real, so the walk stops and leaves it to report.
			if working.doc.to_mut().patch(Operation::operations_to_value(vec![op])).is_err() {
				return Ok(());
			}
		}
		Ok(())
	}

	/// Updates the current document using the data
	/// clause present on the statement. This can be
	/// one of CONTENT, REPLACE, MERGE, PATCH, SET,
	/// UNSET, or ON DUPLICATE KEY UPDATE. This
	/// function also ensures that standard default
	/// fields are set and reset before and after the
	/// document data is modified.
	pub(super) async fn process_record_data(
		&mut self,
		stk: &mut Stk,
		ctx: &FrozenContext,
		opt: &Options,
	) -> Result<()> {
		// The statement has a data clause
		if let Some(v) = self.input_data.clone() {
			match v {
				ComputedData::Patch(data) => {
					self.check_patch_read_pointers(stk, ctx, opt, data.as_ref()).await?;
					self.current.doc.to_mut().patch(data.as_ref().clone())?
				}
				ComputedData::Merge(data) => {
					self.current.doc.to_mut().merge(data.as_ref().clone())?
				}
				ComputedData::Replace(data) => {
					self.current.doc.to_mut().replace(data.as_ref().clone())?
				}
				ComputedData::Content(data) => {
					self.current.doc.to_mut().replace(data.as_ref().clone())?
				}
				ComputedData::Unset(i) => {
					for i in i.iter() {
						self.current.doc.to_mut().cut(i);
					}
				}
				ComputedData::Set(x) => {
					// The assignment right-hand sides were already evaluated
					// against the reduced view of `current` in
					// `compute_input_data`, so here we just write the
					// pre-computed values into `self.current` — the actual
					// storage-bound document. Writing to the reduced view
					// would leave `self.current` unchanged and the mutation
					// would be invisible to subsequent pipeline steps that
					// re-reduce from `current` (e.g. `output_after`).
					apply_assignments(stk, ctx, opt, self.current.doc.to_mut(), &x).await?;
				}
			};
			// Every arm mutates `self.current.doc`, so the pre-mutation
			// snapshot is now stale in both halves. Invalidate the reduced
			// view so any downstream caller that re-reduces sees the new
			// field values rather than relying on `output_*` to do this
			// implicitly, and clear `fields_computed` so the computed fields
			// are re-derived from those new values instead of the snapshot
			// ones — otherwise a computed field that depends on an assigned
			// field keeps its pre-mutation value, or disappears from the
			// returned record once `process_table_fields` strips it.
			self.current_reduced = None;
			self.current.fields_computed = false;
		};
		// Carry on
		Ok(())
	}

	/// Evaluate the statement's data clause once and cache the result.
	///
	/// The expressions inside `SET`/`CONTENT`/`MERGE`/`PATCH`/`REPLACE`/`UNSET`
	/// can reference `$input` (for `INSERT … ON DUPLICATE KEY UPDATE` and
	/// `RELATE`) and the current document fields, so the data clause is
	/// evaluated against the pre-mutation snapshot built by
	/// [`Self::materialise_current_snapshot`]: field-level permissions
	/// applied, `COMPUTED` fields populated, and none of this statement's own
	/// assignments visible. Every expression is evaluated against that one
	/// image before any of them is applied, which is what makes
	/// `SET a = a + 1, b = a + 1` assign `b` from the old `a`, and what makes
	/// a computed field read the same in the data clause as in the `WHERE`
	/// condition.
	///
	/// The first call computes each expression and stores the resulting
	/// [`ComputedData`] on `self`. Subsequent calls in the same pipeline
	/// (e.g. the UPSERT retry path, or `process_record_data` reusing the
	/// value computed by an earlier permission check) are no-ops and return
	/// the cached value without re-evaluating any user expression.
	///
	/// Returns `Ok(None)` when the statement has no data clause.
	pub(super) async fn compute_input_data(
		&mut self,
		stk: &mut Stk,
		ctx: &FrozenContext,
		opt: &Options,
		stm: &Statement<'_>,
	) -> Result<Option<&ComputedData>> {
		// Check if the input data has been computed
		if self.input_data.is_some() {
			return Ok(self.input_data.as_ref());
		}
		// Check if there is a data clause on the statement
		if let Some(data) = stm.data() {
			// Snapshot `$input` before reduce_current takes &mut self
			let input_value: Option<Arc<Value>> = match &self.extras {
				Extras::Insert(value) => Some(Arc::clone(value)),
				Extras::Relate(_, _, Some(value)) => Some(Arc::clone(value)),
				_ => None,
			};
			// Evaluate against the pre-mutation snapshot, so the data clause
			// reads exactly what the `WHERE` condition read
			let doc = self.materialise_current_snapshot(stk, ctx, opt).await?;
			// Compote the input data from the statement
			self.input_data = Some(match data {
				// This is a UNSET expression
				Data::UnsetExpression(data) => ComputedData::Unset(data.clone()),
				// This is a PATCH expression
				Data::PatchExpression(data) => ComputedData::Patch(Arc::new(
					crate::legacy::expr_compute(data, stk, ctx, opt, Some(doc))
						.await
						.catch_return()?,
				)),
				// This is a MERGE expression
				Data::MergeExpression(data) => ComputedData::Merge(Arc::new(
					crate::legacy::expr_compute(data, stk, ctx, opt, Some(doc))
						.await
						.catch_return()?,
				)),
				// This is a REPLACE expression
				Data::ReplaceExpression(data) => ComputedData::Replace(Arc::new(
					crate::legacy::expr_compute(data, stk, ctx, opt, Some(doc))
						.await
						.catch_return()?,
				)),
				// This is a CONTENT expression
				Data::ContentExpression(data) => ComputedData::Content(Arc::new(
					crate::legacy::expr_compute(data, stk, ctx, opt, Some(doc))
						.await
						.catch_return()?,
				)),
				// This is a SET or ON DUPLICATE KEY UPDATE expression
				x @ Data::SetExpression(data) | x @ Data::UpdateExpression(data) => {
					let ctx = if matches!(x, Data::UpdateExpression(_)) {
						// Duplicate context
						let mut ctx = Context::new_child(ctx);
						// Add insertable value
						if let Some(value) = input_value {
							ctx.add_value("input", value);
						}
						// Freeze the context
						ctx.freeze()
					} else {
						Arc::clone(ctx)
					};

					let mut assignments = Vec::with_capacity(data.len());
					for x in data.iter() {
						assignments.push(ComputedAssignment {
							place: x.place.clone(),
							operator: x.operator.clone(),
							value: crate::legacy::expr_compute(&x.value, stk, &ctx, opt, Some(doc))
								.await
								.catch_return()?,
						});
					}

					ComputedData::Set(assignments)
				}
				x => bail!("Unexpected data clause type: {x:?}"),
			});
		}

		Ok(self.input_data.as_ref())
	}

	/// Compute the statement's data clause (if needed) and materialise it as a
	/// single synthetic [`Value`] suitable for binding to `$input` or feeding
	/// into permission checks.
	///
	/// This is a thin wrapper around [`Self::compute_input_data`] followed by
	/// [`ComputedData::materialize`]: the data clause is evaluated lazily on
	/// the first call and cached, then projected into a concrete object/array
	/// value. Use this when you need the user-supplied data as a `Value`;
	/// reach for [`Self::compute_input_data`] directly when you only need to
	/// dispatch on the data variant (PATCH/MERGE/SET/…).
	///
	/// Returns `Ok(None)` when the statement has no data clause.
	pub(super) async fn compute_input_value(
		&mut self,
		stk: &mut Stk,
		ctx: &FrozenContext,
		opt: &Options,
		stm: &Statement<'_>,
	) -> Result<Option<Arc<Value>>> {
		// Make sure the input data clause has been computed.
		if self.compute_input_data(stk, ctx, opt, stm).await?.is_none() {
			return Ok(None);
		}
		// Re-borrow self.input_data so the &mut self borrow from
		// compute_input_data is released before the await below.
		let data = self.input_data.as_ref().expect("just verified Some above");
		Ok(Some(data.materialize(stk, ctx, opt).await?))
	}

	/// Materialize the synthetic input value from the already-computed
	/// `input_data`. Unlike [`Self::compute_input_value`] this never falls
	/// back to evaluating the statement's data clause: callers must have
	/// already run [`Self::compute_input_data`] (typically via
	/// `process_record_data`) before getting here.
	pub(super) async fn materialize_input_value(
		&self,
		stk: &mut Stk,
		ctx: &FrozenContext,
		opt: &Options,
	) -> Result<Option<Arc<Value>>> {
		match self.input_data.as_ref() {
			Some(data) => Ok(Some(data.materialize(stk, ctx, opt).await?)),
			None => Ok(None),
		}
	}
}

/// The result of evaluating a statement's data clause once, cached on the
/// [`Document`] so that downstream pipeline steps (permission checks,
/// `process_record_data`, `$input` materialisation, the UPSERT retry path)
/// can reuse it without re-running any user-supplied expression.
///
/// Each variant mirrors a SurrealQL data-clause form:
/// - [`ComputedData::Patch`] — `PATCH [{op: "...", path: "...", value: ...}, …]`
/// - [`ComputedData::Merge`] — `MERGE {…}`
/// - [`ComputedData::Replace`] — `REPLACE {…}`
/// - [`ComputedData::Content`] — `CONTENT {…}`
/// - [`ComputedData::Unset`] — `UNSET field, …`
/// - [`ComputedData::Set`] — `SET …` / `ON DUPLICATE KEY UPDATE …`
///
/// The materialised value variants store an `Arc<Value>` so the payload can
/// be cheaply shared between the cached entry and the consumers that need it
/// by value (e.g. `merge`/`replace`/`patch` on `current.doc`).
#[derive(Clone, Debug, Eq, PartialEq, Hash)]
pub(super) enum ComputedData {
	Patch(Arc<Value>),
	Merge(Arc<Value>),
	Replace(Arc<Value>),
	Content(Arc<Value>),
	Unset(Vec<Idiom>),
	Set(Vec<ComputedAssignment>),
}

impl ComputedData {
	/// Returns `true` when this data clause is a `PATCH` expression.
	///
	/// Used by the pipeline to special-case PATCH semantics (which operate on
	/// the full document via JSON-Patch ops) where the other variants are
	/// treated more uniformly as object overlays/replacements.
	pub(super) fn is_patch(&self) -> bool {
		matches!(self, ComputedData::Patch(_))
	}

	/// Synchronously pick a value at the given path from the user-supplied
	/// data clause. For SET this scans the assignments for a matching
	/// `Assign` operator at `path`; it deliberately ignores compound
	/// operators (+=, -=, +?) because they need the existing field value
	/// to evaluate and so cannot be resolved without the initial document.
	pub(super) fn pick(&self, path: &[Part]) -> Value {
		match self {
			ComputedData::Patch(v) => v.pick(path),
			ComputedData::Merge(v) => v.pick(path),
			ComputedData::Replace(v) => v.pick(path),
			ComputedData::Content(v) => v.pick(path),
			ComputedData::Unset(_) => Value::None,
			ComputedData::Set(assignments) => {
				for a in assignments {
					if a.operator == AssignOperator::Assign && a.place.0.as_slice() == path {
						return a.value.clone();
					}
				}
				Value::None
			}
		}
	}

	/// Asynchronously materialize the synthetic input value used by `$input`
	/// in DEFINE EVENT and DEFINE FIELD VALUE / ASSERT expressions. For SET
	/// this applies the assignments to an empty object so compound operators
	/// resolve against `Value::None` — the same semantics as on a freshly
	/// created record. Other data clauses already carry the materialized
	/// value and clone the `Arc`.
	pub(super) async fn materialize(
		&self,
		stk: &mut Stk,
		ctx: &FrozenContext,
		opt: &Options,
	) -> Result<Arc<Value>> {
		match self {
			ComputedData::Patch(v) => Ok(Arc::clone(v)),
			ComputedData::Merge(v) => Ok(Arc::clone(v)),
			ComputedData::Replace(v) => Ok(Arc::clone(v)),
			ComputedData::Content(v) => Ok(Arc::clone(v)),
			ComputedData::Unset(_) => Ok(Arc::new(Value::None)),
			ComputedData::Set(assignments) => {
				let mut input = Value::Object(Default::default());
				apply_assignments(stk, ctx, opt, &mut input, assignments).await?;
				Ok(Arc::new(input))
			}
		}
	}
}

/// A single pre-evaluated assignment from a `SET …` / `ON DUPLICATE KEY
/// UPDATE …` data clause, cached as part of [`ComputedData::Set`].
///
/// The right-hand side has already been evaluated against the reduced
/// `current` document by [`Document::compute_input_data`], so re-applying
/// the assignment (e.g. on the UPSERT retry path) does not re-run any
/// user-supplied expression.
///
/// Fields:
/// - `place` — target field path (the left-hand `Idiom`).
/// - `operator` — how the value combines with the existing field (plain `=`, compound
///   `+=`/`-=`/`+?`).
/// - `value` — the already-evaluated right-hand side.
#[derive(Clone, Debug, Eq, PartialEq, Hash)]
pub(super) struct ComputedAssignment {
	pub place: Idiom,
	pub operator: AssignOperator,
	pub value: Value,
}

/// Apply a list of pre-evaluated `SET` assignments to `doc` in order.
///
/// Each assignment dispatches to the matching `Value` mutator based on its
/// operator:
/// - `Assign` (`=`) — `set` for non-`NONE`, `del` when the right-hand side evaluated to `NONE`
///   (treated as field removal).
/// - `Add` (`+=`) — `increment`.
/// - `Subtract` (`-=`) — `decrement`.
/// - `Extend` (`+?`) — `extend` (array/object union).
///
/// Used both by `process_record_data` (to apply the assignments to the
/// current document) and by [`ComputedData::materialize`] (to project the
/// assignments onto an empty object so compound operators resolve against
/// `NONE`, matching freshly-created-record semantics for `$input`).
async fn apply_assignments(
	stk: &mut Stk,
	ctx: &FrozenContext,
	opt: &Options,
	doc: &mut Value,
	assignments: &[ComputedAssignment],
) -> Result<()> {
	for x in assignments {
		match &x.operator {
			AssignOperator::Assign => match &x.value {
				Value::None => crate::legacy::value_del(doc, stk, ctx, opt, &x.place).await?,
				_ => {
					crate::legacy::value_set(doc, stk, ctx, opt, &x.place, x.value.clone()).await?
				}
			},
			AssignOperator::Add => {
				crate::legacy::value_increment(doc, stk, ctx, opt, &x.place, x.value.clone())
					.await?
			}
			AssignOperator::Subtract => {
				crate::legacy::value_decrement(doc, stk, ctx, opt, &x.place, x.value.clone())
					.await?
			}
			AssignOperator::Extend => {
				crate::legacy::value_extend(doc, stk, ctx, opt, &x.place, x.value.clone()).await?
			}
		}
	}
	Ok(())
}