surrealdb-core 3.2.1

use std::ops::Deref;
use std::sync::Arc;

use futures::future::try_join_all;
use reblessive::tree::Stk;
use surrealdb_types::ToSql;

use crate::ctx::{Context, FrozenContext};
use crate::dbs::Options;
use crate::doc::CursorDoc;
use crate::err::Error;
use crate::exe::try_join_all_buffered;
use crate::expr::field::Fields;
use crate::expr::idiom::recursion::{Recursion, compute_idiom_recursion};
use crate::expr::part::{FindRecursionPlan, Next, NextMethod, Part, Recurse, SplitByRepeatRecurse};
use crate::expr::statements::select::SelectStatement;
use crate::expr::{ControlFlow, Expr, FlowResult, FlowResultExt as _, Idiom, Literal, Lookup};
use crate::fnc::idiom;
use crate::val::{Object, RecordIdKey, Value};

macro_rules! fallback_function {
	(if $first:expr => InvalidFunction($e:ident) then $second:expr) => {
		match $first {
			Err(e) if matches!(e.downcast_ref(), Some(Error::InvalidFunction { .. })) => {
				let $e = e;
				$second
			}
			x => x,
		}
	};
}

/// Returns true if the expression references the `$parent` parameter.
/// Used to avoid allocating a child context in Part::Where when the
/// predicate does not need `$parent`.
fn expr_references_parent(expr: &Expr) -> bool {
	use crate::expr::visit::{Visit, Visitor};
	struct Check(bool);
	impl Visitor for Check {
		type Error = std::convert::Infallible;
		fn visit_expr(&mut self, e: &Expr) -> Result<(), Self::Error> {
			if let Expr::Param(p) = e
				&& p.as_str() == "parent"
			{
				self.0 = true;
			}
			if self.0 {
				return Ok(());
			}
			e.visit(self)
		}
		fn visit_select(&mut self, _: &crate::expr::SelectStatement) -> Result<(), Self::Error> {
			Ok(())
		}
	}
	let mut c = Check(false);
	let _ = c.visit_expr(expr);
	c.0
}

impl Value {
	/// Asynchronous method for getting a local or remote field from a `Value`
	pub(crate) async fn get(
		&self,
		stk: &mut Stk,
		ctx: &FrozenContext,
		opt: &Options,
		doc: Option<&CursorDoc>,
		path: &[Part],
	) -> FlowResult<Self> {
		// Limit recursion depth.
		if path.len() > ctx.config.max_computation_depth as usize {
			return Err(ControlFlow::from(anyhow::Error::new(Error::ComputationDepthExceeded)));
		}

		let Some(first) = path.first() else {
			return Ok(self.clone());
		};

		match first {
			// The knowledge of the current value is not relevant to Part::Recurse
			Part::Recurse(recurse, inner_path, instruction) => {
				// Find the path to recurse and what path to process after the recursion is
				// finished
				let (path, after) = match inner_path {
					Some(p) => (p.0.as_slice(), path.next().to_vec()),
					_ => (path.next(), vec![]),
				};

				// We first try to split out a root-level repeat-recurse symbol
				// By doing so, we can eliminate un-needed recursion, as we can
				// simply loop.
				let (path, plan, after) = match path.split_by_repeat_recurse() {
					Some((path, local_after)) => (path, None, [local_after, &after].concat()),

					// If we do not find a root-level repeat-recurse symbol, we
					// can scan for a nested one. We only ever allow for a single
					// repeat recurse symbol, hence the separate check.
					_ => {
						// If the user already specified a recursion instruction,
						// we will not process any recursion plans.
						if instruction.is_some() {
							match path.find_recursion_plan() {
								Some(_) => {
									return Err(ControlFlow::Err(anyhow::Error::new(
										Error::RecursionInstructionPlanConflict,
									)));
								}
								_ => (path, None, after),
							}
						} else {
							match path.find_recursion_plan() {
								Some((path, plan, local_after)) => {
									(path, Some(plan), [local_after, &after].concat())
								}
								_ => (path, None, after),
							}
						}
					}
				};

				// Collect the min & max for the recursion context
				let (min, max) = match recurse {
					Recurse::Fixed(n) => (*n, Some(*n)),
					Recurse::Range(min, max) => (min.unwrap_or(1), *max),
				};

				if min < 1 {
					return Err(ControlFlow::Err(anyhow::Error::new(Error::InvalidBound {
						found: min.to_string(),
						expected: "at least 1".into(),
					})));
				}

				if let Some(max) = max
					&& max > ctx.config.idiom_recursion_limit
				{
					return Err(ControlFlow::Err(anyhow::Error::new(Error::InvalidBound {
						found: max.to_string(),
						expected: format!("{} at most", ctx.config.idiom_recursion_limit),
					})));
				}

				// Construct the recursion context
				let rec = Recursion {
					min,
					max,
					iterated: 0,
					current: self,
					path,
					plan: plan.as_ref(),
					instruction: instruction.as_ref(),
				};

				// Compute the recursion
				let v = compute_idiom_recursion(stk, ctx, opt, doc, rec).await?;

				// If we have a leftover path, process it
				if !after.is_empty() {
					stk.run(|stk| v.get(stk, ctx, opt, doc, after.as_slice())).await
				} else {
					Ok(v)
				}
			}
			// We only support repeat recurse symbol in certain scenarios, to
			// ensure we can process them efficiently. When encountering a
			// recursion part, it will find the repeat recurse part and handle
			// it. If we find one in any unsupported scenario, we throw an error.
			Part::RepeatRecurse => {
				Err(ControlFlow::Err(anyhow::Error::new(Error::UnsupportedRepeatRecurse)))
			}
			Part::Doc => {
				// Try to obtain a Record ID from the document, otherwise we'll operate on NONE
				let v = match doc {
					Some(doc) => match &doc.rid {
						Some(id) => Value::RecordId(id.deref().to_owned()),
						_ => Value::None,
					},
					None => Value::None,
				};

				stk.run(|stk| v.get(stk, ctx, opt, doc, path.next())).await
			}
			// Get the current value at the path
			p => match self {
				// Current value at path is a geometry
				Value::Geometry(v) => match p {
					// If this is the 'type' field then continue
					Part::Field(f) if f == "type" => {
						let v = Value::from(v.as_type());
						stk.run(|stk| v.get(stk, ctx, opt, doc, path.next())).await
					}
					// If this is the 'coordinates' field then continue
					Part::Field(f) if f == "coordinates" && v.is_geometry() => {
						let v = v.as_coordinates();
						stk.run(|stk| v.get(stk, ctx, opt, doc, path.next())).await
					}
					// If this is the 'geometries' field then continue
					Part::Field(f) if f == "geometries" && v.is_collection() => {
						let v = v.as_coordinates();
						stk.run(|stk| v.get(stk, ctx, opt, doc, path.next())).await
					}
					Part::Destructure(_) => {
						let obj = Value::Object(v.as_object());
						stk.run(|stk| obj.get(stk, ctx, opt, doc, path)).await
					}
					Part::Method(name, args) => {
						let a = stk
							.scope(|scope| {
								try_join_all(
									args.iter()
										.map(|v| scope.run(|stk| v.compute(stk, ctx, opt, doc))),
								)
							})
							.await?;
						let v = stk
							.run(|stk| idiom(stk, ctx, opt, doc, v.clone().into(), name, a))
							.await?;
						stk.run(|stk| v.get(stk, ctx, opt, doc, path.next())).await
					}
					Part::Optional => {
						stk.run(|stk| self.get(stk, ctx, opt, doc, path.next())).await
					}
					// Otherwise return none
					_ => Ok(Value::None),
				},
				// Current value at path is an object
				Value::Object(v) => match p {
					Part::Lookup(_) => match v.rid() {
						Some(v) => {
							let v = Value::RecordId(v);
							stk.run(|stk| v.get(stk, ctx, opt, doc, path)).await
						}
						None => {
							stk.run(|stk| Value::None.get(stk, ctx, opt, doc, path.next())).await
						}
					},
					Part::Field(f) => match v.get(f.as_str()) {
						Some(v) => stk.run(|stk| v.get(stk, ctx, opt, doc, path.next())).await,
						None => {
							stk.run(|stk| Value::None.get(stk, ctx, opt, doc, path.next())).await
						}
					},
					Part::Value(x) => match stk
						.run(|stk| x.compute(stk, ctx, opt, doc))
						.await
						.catch_return()?
					{
						Value::Number(n) => match v.get(&n.to_sql()) {
							Some(v) => stk.run(|stk| v.get(stk, ctx, opt, doc, path.next())).await,
							None => {
								stk.run(|stk| Value::None.get(stk, ctx, opt, doc, path.next()))
									.await
							}
						},
						Value::String(f) => match v.get(f.as_str()) {
							Some(v) => stk.run(|stk| v.get(stk, ctx, opt, doc, path.next())).await,
							None => Ok(Value::None),
						},
						Value::RecordId(t) => match v.get(&t.to_sql()) {
							Some(v) => stk.run(|stk| v.get(stk, ctx, opt, doc, path.next())).await,
							None => Ok(Value::None),
						},
						_ => stk.run(|stk| Value::None.get(stk, ctx, opt, doc, path.next())).await,
					},
					Part::All => stk.run(|stk| self.get(stk, ctx, opt, doc, path.next())).await,
					Part::Destructure(p) => {
						let cur_doc = CursorDoc::from(self.clone());
						let mut obj = Object::default();
						for p in p.iter() {
							let idiom = p.idiom();
							obj.insert(
								p.field(),
								stk.run(|stk| idiom.compute(stk, ctx, opt, Some(&cur_doc))).await?,
							);
						}

						let obj = Value::Object(obj);
						stk.run(|stk| obj.get(stk, ctx, opt, doc, path.next())).await
					}
					Part::Method(name, args) => {
						let args = stk
							.scope(|scope| {
								try_join_all(
									args.iter()
										.map(|v| scope.run(|stk| v.compute(stk, ctx, opt, doc))),
								)
							})
							.await?;

						let res = stk
							.run(|stk| {
								idiom(stk, ctx, opt, doc, v.clone().into(), name, args.clone())
							})
							.await;

						let res = fallback_function! {
							if res => InvalidFunction(e) then {
								if let Some(Value::Closure(x)) = v.get(name) {
									x.invoke(stk, ctx, opt, doc, args).await
								} else {
									Err(e)
								}
							}
						}?;

						stk.run(|stk| res.get(stk, ctx, opt, doc, path.next())).await
					}
					Part::Optional => {
						stk.run(|stk| self.get(stk, ctx, opt, doc, path.next())).await
					}
					_ => Ok(Value::None),
				},
				// Current value at path is an array
				Value::Array(v) => match p {
					// Current path is an `*` part
					Part::All => {
						stk.scope(|scope| {
							let futs = v.iter().map(|v| {
								scope.run(|stk| {
									let path = if v.is_record() {
										path
									} else {
										// .* applies to the elements of the array it was applied
										// to, not recursively if one of the values is an
										// array, we skip the .* part, as it implied that
										// the user collected all values of the nested array. See
										// `array_range.surql`.
										path.next()
									};

									v.get(stk, ctx, opt, doc, path)
								})
							});
							try_join_all_buffered(futs, ctx.config.max_concurrent_tasks)
						})
						.await
						.map(Into::into)
					}
					Part::Flatten => {
						let path = path.next();
						stk.scope(|scope| {
							let futs =
								v.iter().map(|v| scope.run(|stk| v.get(stk, ctx, opt, doc, path)));
							try_join_all_buffered(futs, ctx.config.max_concurrent_tasks)
						})
						.await
						.map(Into::into)
					}
					Part::First => match v.first() {
						Some(v) => stk.run(|stk| v.get(stk, ctx, opt, doc, path.next())).await,
						None => {
							stk.run(|stk| Value::None.get(stk, ctx, opt, doc, path.next())).await
						}
					},
					Part::Last => match v.last() {
						Some(v) => stk.run(|stk| v.get(stk, ctx, opt, doc, path.next())).await,
						None => {
							stk.run(|stk| Value::None.get(stk, ctx, opt, doc, path.next())).await
						}
					},
					Part::Where(w) => {
						// Bind $parent to the enclosing document when the
						// predicate references it. Always overrides any existing
						// binding (e.g. from an outer subquery) so that $parent
						// in a graph [WHERE] refers to the current SELECT's row.
						let parent_ctx = match doc {
							Some(d) if expr_references_parent(w) => {
								let mut child = Context::new_child(ctx);
								child.add_value("parent", Arc::new(d.doc.as_ref().clone()));
								Some(child.freeze())
							}
							_ => None,
						};
						let ctx = parent_ctx.as_ref().unwrap_or(ctx);
						let mut a = Vec::new();
						for v in v.iter() {
							let cur = v.clone().into();
							if stk
								.run(|stk| w.compute(stk, ctx, opt, Some(&cur)))
								.await
								.catch_return()?
								.is_truthy()
							{
								a.push(v.clone());
							}
						}
						let v = Value::from(a);
						stk.run(|stk| v.get(stk, ctx, opt, doc, path.next())).await
					}
					Part::Value(x) => match stk
						.run(|stk| x.compute(stk, ctx, opt, doc))
						.await
						.catch_return()?
					{
						Value::Number(i) => match i.as_array_index().and_then(|i| v.get(i)) {
							Some(v) => stk.run(|stk| v.get(stk, ctx, opt, doc, path.next())).await,
							None => Ok(Value::None),
						},
						Value::Range(r) => {
							let v = r
								.coerce_to_typed::<i64>()
								.map_err(Error::from)
								.map_err(anyhow::Error::new)
								.map_err(ControlFlow::Err)?
								.slice(v.as_slice())
								.map(|v| Value::from(v.to_vec()))
								.unwrap_or_default();

							stk.run(|stk| v.get(stk, ctx, opt, doc, path.next())).await
						}
						_ => stk.run(|stk| Value::None.get(stk, ctx, opt, doc, path.next())).await,
					},
					Part::Method(name, args) => {
						let a = stk
							.scope(|scope| {
								try_join_all(
									args.iter()
										.map(|v| scope.run(|stk| v.compute(stk, ctx, opt, doc))),
								)
							})
							.await?;
						let v = stk
							.run(|stk| idiom(stk, ctx, opt, doc, v.clone().into(), name, a))
							.await?;
						stk.run(|stk| v.get(stk, ctx, opt, doc, path.next())).await
					}
					Part::Optional => {
						stk.run(|stk| self.get(stk, ctx, opt, doc, path.next())).await
					}
					_ => {
						let res = stk
							.scope(|scope| {
								let futs = v
									.iter()
									.map(|v| scope.run(|stk| v.get(stk, ctx, opt, doc, path)));
								try_join_all_buffered(futs, ctx.config.max_concurrent_tasks)
							})
							.await
							.map(Value::from)?;

						Ok(res)
					}
				},
				// Current value at path is a set
				Value::Set(v) => match p {
					Part::All => stk
						.scope(|scope| {
							let futs = v.iter().map(|v| {
								scope.run(|stk| {
									let path = if v.is_record() {
										path
									} else {
										path.next()
									};

									v.get(stk, ctx, opt, doc, path)
								})
							});
							try_join_all_buffered(futs, ctx.config.max_concurrent_tasks)
						})
						.await
						.map(|vals| Value::Set(vals.into_iter().collect())),
					Part::Flatten => {
						let path = path.next();
						stk.scope(|scope| {
							let futs =
								v.iter().map(|v| scope.run(|stk| v.get(stk, ctx, opt, doc, path)));
							try_join_all_buffered(futs, ctx.config.max_concurrent_tasks)
						})
						.await
						.map(|vals| Value::Set(vals.into_iter().collect()))
					}
					Part::First => match v.first() {
						Some(v) => stk.run(|stk| v.get(stk, ctx, opt, doc, path.next())).await,
						None => {
							stk.run(|stk| Value::None.get(stk, ctx, opt, doc, path.next())).await
						}
					},
					Part::Last => match v.last() {
						Some(v) => stk.run(|stk| v.get(stk, ctx, opt, doc, path.next())).await,
						None => {
							stk.run(|stk| Value::None.get(stk, ctx, opt, doc, path.next())).await
						}
					},
					Part::Where(w) => {
						let parent_ctx = match doc {
							Some(d) if expr_references_parent(w) => {
								let mut child = Context::new_child(ctx);
								child.add_value("parent", Arc::new(d.doc.as_ref().clone()));
								Some(child.freeze())
							}
							_ => None,
						};
						let ctx = parent_ctx.as_ref().unwrap_or(ctx);
						let mut a = Vec::new();
						for v in v.iter() {
							let cur = v.clone().into();
							if stk
								.run(|stk| w.compute(stk, ctx, opt, Some(&cur)))
								.await
								.catch_return()?
								.is_truthy()
							{
								a.push(v.clone());
							}
						}
						let v = Value::Set(a.into_iter().collect());
						stk.run(|stk| v.get(stk, ctx, opt, doc, path.next())).await
					}
					Part::Value(x) => match stk
						.run(|stk| x.compute(stk, ctx, opt, doc))
						.await
						.catch_return()?
					{
						Value::Number(i) => match i.as_array_index().and_then(|i| v.nth(i)) {
							Some(v) => stk.run(|stk| v.get(stk, ctx, opt, doc, path.next())).await,
							None => Ok(Value::None),
						},
						Value::Range(r) => {
							let values: Vec<_> = v.iter().cloned().collect();
							let v = r
								.coerce_to_typed::<i64>()
								.map_err(Error::from)
								.map_err(anyhow::Error::new)
								.map_err(ControlFlow::Err)?
								.slice(values.as_slice())
								.map(|v| Value::Set(v.iter().cloned().collect()))
								.unwrap_or_default();

							stk.run(|stk| v.get(stk, ctx, opt, doc, path.next())).await
						}
						_ => stk.run(|stk| Value::None.get(stk, ctx, opt, doc, path.next())).await,
					},
					Part::Method(name, args) => {
						let a = stk
							.scope(|scope| {
								try_join_all(
									args.iter()
										.map(|v| scope.run(|stk| v.compute(stk, ctx, opt, doc))),
								)
							})
							.await?;
						let v = stk
							.run(|stk| idiom(stk, ctx, opt, doc, v.clone().into(), name, a))
							.await?;
						stk.run(|stk| v.get(stk, ctx, opt, doc, path.next())).await
					}
					Part::Optional => {
						stk.run(|stk| self.get(stk, ctx, opt, doc, path.next())).await
					}
					_ => {
						let res = stk
							.scope(|scope| {
								let futs = v
									.iter()
									.map(|v| scope.run(|stk| v.get(stk, ctx, opt, doc, path)));
								try_join_all_buffered(futs, ctx.config.max_concurrent_tasks)
							})
							.await
							.map(|vals| Value::Set(vals.into_iter().collect()))?;

						Ok(res)
					}
				},
				// Current value at path is a record
				Value::RecordId(v) => {
					// Clone the record
					let val = v.clone();
					// Check how many path parts are remaining
					if path.is_empty() {
						return Ok(Value::RecordId(val));
					}

					match p {
						// This is a graph traversal expression
						Part::Lookup(g) => {
							let last_part = path.len() == 1;
							let fields = g.expr.clone().unwrap_or(Fields::value_id());
							let what = Expr::Idiom(Idiom(vec![
								Part::Start(Expr::Literal(Literal::RecordId(val.into_literal()))),
								Part::Lookup(Box::new(Lookup {
									what: g.what.clone(),
									kind: g.kind.clone(),
									..Default::default()
								})),
							]));

							let stm = SelectStatement {
								fields,
								what: vec![what],
								cond: g.cond.clone(),
								limit: g.limit.clone(),
								order: g.order.clone(),
								split: g.split.clone(),
								group: g.group.clone(),
								start: g.start.clone(),
								omit: vec![],
								only: false,
								with: None,
								fetch: None,
								version: Expr::Literal(Literal::None),
								timeout: Expr::Literal(Literal::None),
								explain: None,
								tempfiles: false,
							};

							let res = stk.run(|stk| stm.compute(stk, ctx, opt, doc)).await?.all();

							let res = if g.only {
								match res {
									Value::Array(arr) if arr.is_empty() => Value::None,
									Value::Array(mut arr) if arr.len() == 1 => {
										arr.0.pop().expect("Exactly one item in this array")
									}
									Value::Array(_) => {
										return Err(crate::expr::ControlFlow::Err(
											anyhow::anyhow!(crate::err::Error::SingleOnlyOutput),
										));
									}
									other => other,
								}
							} else {
								res
							};

							if last_part {
								Ok(res)
							} else {
								let res =
									stk.run(|stk| res.get(stk, ctx, opt, doc, path.next())).await?;

								match path.get(1) {
									Some(Part::Lookup(_)) => Ok(res.flatten()),
									Some(Part::Where(_)) => Ok(res.flatten()),
									_ => Ok(res),
								}
							}
						}
						Part::Method(name, args) => {
							let a = stk
								.scope(|scope| {
									try_join_all(
										args.iter().map(|v| {
											scope.run(|stk| v.compute(stk, ctx, opt, doc))
										}),
									)
								})
								.await?;

							let res = stk
								.run(|stk| {
									idiom(stk, ctx, opt, doc, v.clone().into(), name, a.clone())
								})
								.await?;

							stk.run(|stk| res.get(stk, ctx, opt, doc, path.next())).await
						}
						Part::Optional => {
							stk.run(|stk| self.get(stk, ctx, opt, doc, path.next())).await
						}
						// This is a remote field expression with two exceptions
						// covering composite RecordId keys. SECURITY: without these
						// special cases, `id.<field>` on a complex record id falls
						// through to `select_document`, which:
						//   - runs `SELECT *` with permissions disabled, exposing mutable document
						//     fields where an immutable id component was intended (Codex finding
						//     c67c7232 — `PERMISSIONS WHERE id.tenant = $token.tenant`); and
						//   - during UPDATE returns the same already-stored value for both
						//     `self.initial` and `self.current`, so `store_index_data`'s `o != n`
						//     guard suppresses index maintenance and leaves stale entries (Codex
						//     finding 9c442c96).
						// Resolve the path against the RecordId key directly when it
						// is an Object/Array so the semantics match the immutable
						// key component the user actually wrote.
						p => {
							// Discover what the path is that we need to continue with
							let next = match (p, &val.key) {
								// `id.field` on an Object-keyed record id reads
								// the key component directly when that name is
								// present in the key, without going through a
								// `SELECT *`. Names that aren't key components
								// (e.g. `record:foo.id`) keep the remote-fetch
								// semantics below.
								(Part::Field(name), RecordIdKey::Object(obj))
									if obj.contains_key(name.as_str()) =>
								{
									let v = obj.get(name.as_str()).cloned().unwrap_or(Value::None);
									return stk
										.run(|stk| v.get(stk, ctx, opt, doc, path.next()))
										.await;
								}
								// `id["name"]` on an Object-keyed record id is the
								// same thing — `Part::Value` of a string literal
								// is the bracket-access form of a field.
								(Part::Value(x), RecordIdKey::Object(obj)) => {
									match stk
										.run(|stk| x.compute(stk, ctx, opt, doc))
										.await
										.catch_return()?
									{
										Value::String(s) if obj.contains_key(s.as_str()) => {
											let v =
												obj.get(s.as_str()).cloned().unwrap_or(Value::None);
											return stk
												.run(|stk| v.get(stk, ctx, opt, doc, path.next()))
												.await;
										}
										// Other key types fall through to the
										// remote-fetch path with the computed
										// value re-substituted into the path.
										x => &[&[Part::Value(x.into_literal())], path.next()]
											.concat(),
									}
								}
								// If the computed value is a number, and the RecordIdKey is an
								// array, then we return the value at the index of the
								// array. Otherwise, we return the computed value and the
								// next path part.
								(Part::Value(x), RecordIdKey::Array(arr)) => {
									match stk
										.run(|stk| x.compute(stk, ctx, opt, doc))
										.await
										.catch_return()?
									{
										Value::Number(n) => {
											return match n.as_array_index().and_then(|i| arr.get(i))
											{
												Some(v) => {
													stk.run(|stk| {
														v.get(stk, ctx, opt, doc, path.next())
													})
													.await
												}
												None => Ok(Value::None),
											};
										}
										x => &[&[Part::Value(x.into_literal())], path.next()]
											.concat(),
									}
								}

								// No special case, fetch the document and continue processing the
								// path
								_ => path,
							};

							// Self-reference cycle guard: if the record id being dereferenced
							// is the same record currently being computed (e.g. a COMPUTED
							// field like `$this.id.prop`), re-running `select_document` would
							// re-enter `computed_fields_inner` for the same record and
							// infinitely.
							if let Some(cur) = doc
								&& cur.rid.as_deref() == Some(&val)
							{
								let current = cur.doc.as_ref().clone();
								return stk.run(|stk| current.get(stk, ctx, opt, None, next)).await;
							}

							// Fetch the record id's contents
							let v = val
								.select_document(stk, ctx, opt, doc)
								.await?
								.map(Value::Object)
								.unwrap_or(Value::None);

							// Continue processing the path on the now fetched record
							stk.run(|stk| v.get(stk, ctx, opt, None, next)).await
						}
					}
				}
				v => {
					match p {
						Part::Optional => match &self {
							Value::None => Ok(Value::None),
							v => stk.run(|stk| v.get(stk, ctx, opt, doc, path.next())).await,
						},
						Part::Flatten => {
							stk.run(|stk| v.get(stk, ctx, opt, None, path.next())).await
						}
						Part::Method(name, args) => {
							let a = stk
								.scope(|scope| {
									try_join_all(
										args.iter().map(|v| {
											scope.run(|stk| v.compute(stk, ctx, opt, doc))
										}),
									)
								})
								.await?;
							let v = stk
								.run(|stk| idiom(stk, ctx, opt, doc, v.clone(), name, a))
								.await?;
							stk.run(|stk| v.get(stk, ctx, opt, doc, path.next())).await
						}
						// Only continue processing the path from the point that it contains a
						// method
						_ => {
							stk.run(|stk| Value::None.get(stk, ctx, opt, doc, path.next_method()))
								.await
						}
					}
				}
			},
		}
	}
}

#[cfg(test)]
mod tests {

	use surrealdb_strand::Strand;

	use super::*;
	use crate::dbs::test::mock;
	use crate::expr::idiom::Idiom;
	use crate::sql::idiom::Idiom as SqlIdiom;
	use crate::syn;
	use crate::val::RecordId;

	macro_rules! parse_val {
		($input:expr) => {
			crate::val::convert_public_value_to_internal(syn::value($input).unwrap())
		};
	}

	#[tokio::test]
	async fn get_none() {
		let (ctx, opt) = mock().await;
		let idi: Idiom = SqlIdiom::default().into();
		let val: Value = parse_val!("{ test: { other: null, something: 123 } }");
		let mut stack = reblessive::tree::TreeStack::new();
		let res = stack.enter(|stk| val.get(stk, &ctx, &opt, None, &idi)).finish().await.unwrap();
		assert_eq!(res, val);
	}

	#[tokio::test]
	async fn get_basic() {
		let (ctx, opt) = mock().await;
		let idi: Idiom = syn::idiom("test.something").unwrap().into();
		let val: Value = parse_val!("{ test: { other: null, something: 123 } }");
		let mut stack = reblessive::tree::TreeStack::new();
		let res = stack.enter(|stk| val.get(stk, &ctx, &opt, None, &idi)).finish().await.unwrap();
		assert_eq!(res, Value::from(123));
	}

	#[tokio::test]
	async fn get_basic_deep_ok() {
		let (ctx, opt) = mock().await;
		let depth = 20;
		let idi: Idiom = syn::idiom(&format!("{}something", "test.".repeat(depth))).unwrap().into();
		let val: Value = parse_val!(&format!(
			"{} {{ other: null, something: 123 {} }}",
			"{ test: ".repeat(depth),
			"}".repeat(depth)
		));
		let mut stack = reblessive::tree::TreeStack::new();
		let res = stack.enter(|stk| val.get(stk, &ctx, &opt, None, &idi)).finish().await.unwrap();
		assert_eq!(res, Value::from(123));
	}

	#[tokio::test]
	async fn get_basic_deep_ko() {
		let (ctx, opt) = mock().await;
		let depth = 2000;
		let idi: Idiom = syn::idiom(&format!("{}something", "test.".repeat(depth))).unwrap().into();
		let val: Value = parse_val!("{}"); // A deep enough object cannot be parsed.
		let mut stack = reblessive::tree::TreeStack::new();
		let err = stack
			.enter(|stk| val.get(stk, &ctx, &opt, None, &idi))
			.finish()
			.await
			.catch_return()
			.unwrap_err();

		assert!(
			matches!(err.downcast_ref(), Some(Error::ComputationDepthExceeded)),
			"expected computation depth exceeded, got {:?}",
			err
		);
	}

	#[tokio::test]
	async fn get_thing() {
		let (ctx, opt) = mock().await;
		let idi: Idiom = syn::idiom("test.other").unwrap().into();
		let val: Value = parse_val!("{ test: { other: test:tobie, something: 123 } }");
		let mut stack = reblessive::tree::TreeStack::new();
		let res = stack.enter(|stk| val.get(stk, &ctx, &opt, None, &idi)).finish().await.unwrap();
		assert_eq!(
			res,
			Value::from(RecordId {
				table: "test".into(),
				key: RecordIdKey::String(Strand::new_static("tobie"))
			})
		);
	}

	#[tokio::test]
	async fn get_array() {
		let (ctx, opt) = mock().await;
		let idi: Idiom = syn::idiom("test.something[1]").unwrap().into();
		let val: Value = parse_val!("{ test: { something: [123, 456, 789] } }");
		let mut stack = reblessive::tree::TreeStack::new();
		let res = stack.enter(|stk| val.get(stk, &ctx, &opt, None, &idi)).finish().await.unwrap();
		assert_eq!(res, Value::from(456));
	}

	#[tokio::test]
	async fn get_array_thing() {
		let (ctx, opt) = mock().await;
		let idi: Idiom = syn::idiom("test.something[1]").unwrap().into();
		let val: Value = parse_val!("{ test: { something: [test:tobie, test:jaime] } }");
		let mut stack = reblessive::tree::TreeStack::new();
		let res = stack.enter(|stk| val.get(stk, &ctx, &opt, None, &idi)).finish().await.unwrap();
		assert_eq!(
			res,
			Value::from(RecordId {
				table: "test".into(),
				key: RecordIdKey::String(Strand::new_static("jaime"))
			})
		);
	}

	#[tokio::test]
	async fn get_set_last() {
		let (ctx, opt) = mock().await;
		let idi: Idiom = syn::idiom("tags[$]").unwrap().into();
		let val: Value = parse_val!("{ tags: {1, 3, 5} }");
		let mut stack = reblessive::tree::TreeStack::new();
		let res = stack.enter(|stk| val.get(stk, &ctx, &opt, None, &idi)).finish().await.unwrap();
		assert_eq!(res, Value::from(5));
	}

	#[tokio::test]
	async fn get_array_field() {
		let (ctx, opt) = mock().await;
		let idi: Idiom = syn::idiom("test.something[1].age").unwrap().into();
		let val: Value = parse_val!("{ test: { something: [{ age: 34 }, { age: 36 }] } }");
		let mut stack = reblessive::tree::TreeStack::new();
		let res = stack.enter(|stk| val.get(stk, &ctx, &opt, None, &idi)).finish().await.unwrap();
		assert_eq!(res, Value::from(36));
	}

	#[tokio::test]
	async fn get_array_fields() {
		let (ctx, opt) = mock().await;
		let idi: Idiom = syn::idiom("test.something[*].age").unwrap().into();
		let val: Value = parse_val!("{ test: { something: [{ age: 34 }, { age: 36 }] } }");
		let mut stack = reblessive::tree::TreeStack::new();
		let res = stack.enter(|stk| val.get(stk, &ctx, &opt, None, &idi)).finish().await.unwrap();
		assert_eq!(res, [Value::from(34i64), Value::from(36i64)].into_iter().collect::<Value>());
	}

	#[tokio::test]
	async fn get_array_fields_flat() {
		let (ctx, opt) = mock().await;
		let idi: Idiom = syn::idiom("test.something.age").unwrap().into();
		let val: Value = parse_val!("{ test: { something: [{ age: 34 }, { age: 36 }] } }");
		let mut stack = reblessive::tree::TreeStack::new();
		let res = stack.enter(|stk| val.get(stk, &ctx, &opt, None, &idi)).finish().await.unwrap();
		assert_eq!(res, [Value::from(34i64), Value::from(36i64)].into_iter().collect::<Value>());
	}

	#[tokio::test]
	async fn get_array_where_field() {
		let (ctx, opt) = mock().await;
		let idi: Idiom = syn::idiom("test.something[WHERE age > 35].age").unwrap().into();
		let val: Value = parse_val!("{ test: { something: [{ age: 34 }, { age: 36 }] } }");
		let mut stack = reblessive::tree::TreeStack::new();
		let res = stack.enter(|stk| val.get(stk, &ctx, &opt, None, &idi)).finish().await.unwrap();
		assert_eq!(res, [Value::from(36i64)].into_iter().collect::<Value>());
	}

	#[tokio::test]
	async fn get_array_where_fields() {
		let (ctx, opt) = mock().await;
		let idi: Idiom = syn::idiom("test.something[WHERE age > 35]").unwrap().into();
		let val: Value = parse_val!("{ test: { something: [{ age: 34 }, { age: 36 }] } }");
		let mut stack = reblessive::tree::TreeStack::new();
		let res = stack.enter(|stk| val.get(stk, &ctx, &opt, None, &idi)).finish().await.unwrap();
		assert_eq!(
			res,
			Value::from(vec![Value::from(map! {
				"age" => Value::from(36),
			})])
		);
	}

	#[tokio::test]
	async fn get_array_where_fields_array_index() {
		let (ctx, opt) = mock().await;
		let idi: Idiom = syn::idiom("test.something[WHERE age > 30][0]").unwrap().into();
		let val: Value = parse_val!("{ test: { something: [{ age: 34 }, { age: 36 }] } }");
		let mut stack = reblessive::tree::TreeStack::new();
		let res = stack.enter(|stk| val.get(stk, &ctx, &opt, None, &idi)).finish().await.unwrap();
		assert_eq!(
			res,
			Value::from(map! {
				"age" => Value::from(34),
			})
		);
	}

	#[tokio::test]
	async fn get_object_with_thing_based_key() {
		let (ctx, opt) = mock().await;
		let idi: Idiom = syn::idiom("test[city:london]").unwrap().into();
		let val: Value =
			parse_val!("{ test: { 'city:london': true, other: test:tobie, something: 123 } }");
		let mut stack = reblessive::tree::TreeStack::new();
		let res = stack.enter(|stk| val.get(stk, &ctx, &opt, None, &idi)).finish().await.unwrap();
		assert_eq!(res, Value::from(true));
	}
}