surrealdb-core 3.2.3

//! GraphQL mutation generation.
//!
//! Generates CREATE, UPDATE, UPSERT, DELETE mutations (single and bulk) for each
//! table in the schema, along with the corresponding input types.
//!
//! - Single mutations: `create{Table}`, `update{Table}`, `upsert{Table}`, `delete{Table}`
//! - Bulk mutations: `create{TablePlural}`, `update{TablePlural}`, `upsert{TablePlural}`,
//!   `delete{TablePlural}` — pluralised via the shared naming helpers, overridden by `GRAPHQL
//!   <alias>` on the table.
//! - For relation tables, `create{Table}` uses RELATE instead of CREATE

use std::collections::{BTreeMap, HashMap};
use std::sync::Arc;

use async_graphql::dynamic::indexmap::IndexMap;
use async_graphql::dynamic::{
	Field, FieldFuture, FieldValue, InputObject, InputValue, Object, Type, TypeRef,
};
use async_graphql::{Name, Value as GraphqlValue};
use surrealdb_strand::Strand;

use super::error::{GraphqlError, resolver_error};
use super::schema::{
	SchemaContext, graphql_to_sql_kind_with_scope, kind_to_type_with_enum_prefix, unwrap_type,
};
use super::tables::{CachedRecord, cond_from_filter, idiom_to_graphql_name};
use super::utils::{GraphqlValueUtils, execute_plan};
use crate::catalog::providers::TableProvider;
use crate::catalog::{FieldDefinition, TableDefinition, TableType};
use crate::dbs::Session;
use crate::expr::part::Part;
use crate::expr::statements::{
	CreateStatement, DeleteStatement, RelateStatement, UpdateStatement, UpsertStatement,
};
use crate::expr::{Cond, Data, Expr, Kind, Literal, LogicalPlan, Output, TopLevelExpr};
use crate::kvs::Datastore;
use crate::val::{Object as SurObject, RecordId, TableName, Value};

/// Capitalize the first character of a string.
fn capitalize_first(s: &str) -> String {
	let mut chars = s.chars();
	match chars.next() {
		None => String::new(),
		Some(c) => c.to_uppercase().to_string() + chars.as_str(),
	}
}

/// Parse a record ID from a table name and user-provided ID string.
///
/// Attempts to parse the full `table:id` string as a proper record ID (handling
/// numeric keys, etc.), falling back to a plain string key.
fn parse_record_id(table_name: &TableName, id: &str) -> Result<RecordId, GraphqlError> {
	let rid_str = format!("{table_name}:{id}");
	match crate::syn::record_id(&rid_str) {
		Ok(x) => Ok(x.into()),
		Err(_) => Ok(RecordId::new(table_name.clone(), id.to_string())),
	}
}

/// Parse a full record ID string (e.g., "person:alice").
fn parse_full_record_id(id_str: &str) -> Result<RecordId, GraphqlError> {
	crate::syn::record_id(id_str)
		.map(|x| x.into())
		.map_err(|e| resolver_error(format!("Invalid record ID: {id_str}: {e}")))
}

/// Convert a GraphQL input object to a SurrealDB Object for use as CONTENT/MERGE data.
///
/// Iterates over the input fields, finds each field's Kind from the field definitions,
/// and converts the GraphQL value to the corresponding SurrealDB value.
fn graphql_input_to_sql_object(
	input: &IndexMap<Name, GraphqlValue>,
	fds: &[FieldDefinition],
	skip_fields: &[&str],
	tb_name: &str,
) -> Result<SurObject, GraphqlError> {
	// Collect into `BTreeMap<Strand, _>` rather than
	// `BTreeMap<String, _>` so the final `SurObject::from(map)`
	// resolves to the zero-cost `From<BTreeMap<Strand, Value>> for
	// Object` impl (`Self(v)`). Going via `String` would allocate a
	// `String` per key here and then reallocate each into a `Strand`
	// again inside `From<BTreeMap<String, Value>>`. See the sibling
	// converters in `graphql/schema.rs` for the same pattern.
	let mut map: BTreeMap<Strand, Value> = BTreeMap::new();
	for (key, val) in input {
		let key_str = key.as_str();
		if skip_fields.contains(&key_str) {
			continue;
		}
		if matches!(val, GraphqlValue::Null) {
			continue;
		}
		// Match by GraphQL alias (#4537) or naming-convention rewrite (#4552)
		// first, then by the raw SurrealQL idiom name. The storage key always
		// comes from the field definition so the underlying record uses the
		// SurrealQL field name regardless of the alias on the input.
		let matched = fds.iter().find(|fd| {
			if fd.name.0.len() != 1 {
				return false;
			}
			if super::naming::field_graphql_name(fd) == key_str {
				return true;
			}
			matches!(&fd.name.0[0], Part::Field(n) if n == key_str)
		});
		let storage_key: String = matched
			.and_then(|fd| match &fd.name.0[0] {
				Part::Field(n) => Some(n.as_str().to_owned()),
				_ => None,
			})
			.unwrap_or_else(|| key_str.to_owned());
		let kind = matched.and_then(|fd| fd.field_kind.clone()).unwrap_or(Kind::Any);
		let enum_scope = format!("{tb_name}_{key_str}");
		let sql_val = graphql_to_sql_kind_with_scope(val, kind, Some(&enum_scope))?;
		map.insert(storage_key.into(), sql_val);
	}
	Ok(SurObject::from(map))
}

/// Map a field's `Kind` to a GraphQL `TypeRef` suitable for mutation input types.
///
/// This differs from `kind_to_type(kind, types, true)` in that `record<target>`
/// fields are mapped to `ID` (the user passes a record ID string) rather than
/// the target table's output Object type (which is not a valid input type).
fn kind_to_input_type_ref(
	kind: Kind,
	types: &mut Vec<Type>,
	enum_scope: Option<&str>,
) -> Result<TypeRef, GraphqlError> {
	let optional = kind.can_be_none();

	// Check if the kind (after stripping option) is a record type.
	// Record fields should be represented as ID in input types (the user passes
	// a record ID string, not a nested object).
	match &kind {
		Kind::Record(_) => {
			let ty = TypeRef::named(TypeRef::ID);
			return Ok(if optional {
				ty
			} else {
				TypeRef::NonNull(Box::new(ty))
			});
		}
		// `array<record<T>>` — same reasoning: the inner Object output type is
		// not a valid input. Map to `[ID!]` (or `[ID!]!` when non-optional).
		Kind::Array(inner, _) if matches!(**inner, Kind::Record(_)) => {
			let inner_ty = TypeRef::NonNull(Box::new(TypeRef::named(TypeRef::ID)));
			let list_ty = TypeRef::List(Box::new(inner_ty));
			return Ok(if optional {
				list_ty
			} else {
				TypeRef::NonNull(Box::new(list_ty))
			});
		}
		Kind::Either(ks) => {
			// Strip None/Null variants to see what's underneath
			let non_none: Vec<&Kind> =
				ks.iter().filter(|k| !matches!(k, Kind::None | Kind::Null)).collect();
			if non_none.len() == 1 {
				match non_none[0] {
					// option<record<T>> -> nullable ID
					Kind::Record(_) => return Ok(TypeRef::named(TypeRef::ID)),
					// option<array<record<T>>> -> nullable [ID!]
					Kind::Array(inner, _) if matches!(**inner, Kind::Record(_)) => {
						let inner_ty = TypeRef::NonNull(Box::new(TypeRef::named(TypeRef::ID)));
						return Ok(TypeRef::List(Box::new(inner_ty)));
					}
					_ => {}
				}
			}
		}
		_ => {}
	}

	// For all other kinds, delegate to the standard kind_to_type with is_input=true
	kind_to_type_with_enum_prefix(kind, types, true, enum_scope)
}

/// Generate Create/Update/Upsert input types for a table and return their names.
fn generate_input_types(
	tb_name: &str,
	fds: &[FieldDefinition],
	is_relation: bool,
	types: &mut Vec<Type>,
) -> Result<(String, String, String), GraphqlError> {
	let cap_name = capitalize_first(tb_name);
	let create_name = format!("Create{cap_name}Input");
	let update_name = format!("Update{cap_name}Input");
	let upsert_name = format!("Upsert{cap_name}Input");

	let mut create_input = InputObject::new(&create_name)
		.description(format!("Input for creating a `{tb_name}` record"));
	let mut update_input = InputObject::new(&update_name)
		.description(format!("Input for updating a `{tb_name}` record"));
	let mut upsert_input = InputObject::new(&upsert_name)
		.description(format!("Input for upserting a `{tb_name}` record"));

	// Add optional `id` field for create and upsert
	create_input = create_input.field(InputValue::new("id", TypeRef::named(TypeRef::ID)));
	upsert_input = upsert_input.field(InputValue::new("id", TypeRef::named(TypeRef::ID)));

	// For relation tables, add required `in` and `out` ID fields
	if is_relation {
		create_input = create_input
			.field(InputValue::new("in", TypeRef::named_nn(TypeRef::ID)))
			.field(InputValue::new("out", TypeRef::named_nn(TypeRef::ID)));
		upsert_input = upsert_input
			.field(InputValue::new("in", TypeRef::named_nn(TypeRef::ID)))
			.field(InputValue::new("out", TypeRef::named_nn(TypeRef::ID)));
	}

	// Add fields from field definitions
	for fd in fds.iter() {
		let Some(ref kind) = fd.field_kind else {
			continue;
		};
		if fd.name.is_id() {
			continue;
		}
		// Skip nested fields (multi-part idioms)
		if fd.name.0.len() > 1 {
			continue;
		}
		// Skip `in` and `out` for relation tables (already added above). Use
		// the raw idiom name for this check since both the alias and the
		// idiom must avoid clashing with the relation-fixed fields.
		let raw_name = idiom_to_graphql_name(&fd.name);
		if is_relation && (raw_name == "in" || raw_name == "out") {
			continue;
		}
		// Skip server-managed fields — `COMPUTED` is re-evaluated on every
		// write and `READONLY` rejects writes entirely. Advertising them as
		// settable in `Create*Input` / `Update*Input` / `Upsert*Input` would
		// mislead schema introspection (the user-supplied value is silently
		// dropped at SurrealQL evaluation time).
		if fd.computed.is_some() || fd.readonly {
			continue;
		}
		// GraphQL-safe input field name — honours `GRAPHQL <alias>` (#4537)
		// and the active naming convention (#4552). The mutation resolver
		// translates the input key back to the SurrealQL field name when
		// building the SET clause (see `graphql_input_to_sql_object`).
		let fd_name = super::naming::field_graphql_name(fd);

		// Carry the field's `COMMENT` plus `GRAPHQL_DEPRECATED "reason"` into
		// the description so introspection surfaces deprecation on input
		// fields too. async-graphql 7.2.1 doesn't expose the `@deprecated`
		// directive setter on dynamic `InputValue`, so we encode the reason
		// inline (see `naming::description_with_deprecation`).
		let fd_desc = super::naming::description_with_deprecation(
			fd.comment.as_deref(),
			fd.graphql_deprecated.as_deref(),
		);

		// For create and upsert: use the field's input type (preserving non-null)
		let enum_scope = format!("{}_{}", tb_name, fd_name);
		let create_type = kind_to_input_type_ref(kind.clone(), types, Some(&enum_scope))?;
		let mut create_iv = InputValue::new(&fd_name, create_type.clone());
		let mut upsert_iv = InputValue::new(&fd_name, create_type);
		if let Some(ref d) = fd_desc {
			create_iv = create_iv.description(d);
			upsert_iv = upsert_iv.description(d);
		}
		create_input = create_input.field(create_iv);
		upsert_input = upsert_input.field(upsert_iv);

		// For update: all fields are optional (strip NonNull)
		let update_type =
			unwrap_type(kind_to_input_type_ref(kind.clone(), types, Some(&enum_scope))?);
		let mut update_iv = InputValue::new(&fd_name, update_type);
		if let Some(ref d) = fd_desc {
			update_iv = update_iv.description(d);
		}
		update_input = update_input.field(update_iv);
	}

	types.push(Type::InputObject(create_input));
	types.push(Type::InputObject(update_input));
	types.push(Type::InputObject(upsert_input));

	Ok((create_name, update_name, upsert_name))
}

/// Shared context for generating mutation fields for a single table.
///
/// Groups the per-table parameters that are passed to every mutation field
/// builder, replacing scattered arguments with a single reference.
struct MutationTableContext {
	/// PascalCased base used in mutation field names (e.g., "Store" →
	/// `createStore` / `updateStore`). Apollo-style; aliases override.
	cap_name: String,
	/// Pluralised PascalCased base used in bulk mutation field names
	/// (e.g., "Stores" → `createStores`).
	cap_plural_name: String,
	/// The table name as a string (e.g., "person").
	tb_name_str: String,
	/// The table name.
	tb_name: TableName,
	/// Whether the table is a relation table.
	is_relation: bool,
	/// Field definitions for this table.
	fds: Arc<[FieldDefinition]>,
	/// The datastore.
	kvs: Arc<Datastore>,
	/// The filter type name for this table (e.g., "_filter_person").
	table_filter_name: String,
	/// Optional `GRAPHQL_DEPRECATED "reason"` set on the table — prepended to
	/// every mutation description for this table so introspection surfaces
	/// the deprecation alongside the auto-generated text. async-graphql 7.2.1
	/// doesn't expose a public setter for the `@deprecated` directive on
	/// dynamic fields; switch to that once it does.
	graphql_deprecated: Option<String>,
}

impl MutationTableContext {
	fn describe(&self, base: String) -> String {
		match self.graphql_deprecated.as_deref() {
			Some(reason) => format!("[Deprecated: {reason}]\n\n{base}"),
			None => base,
		}
	}
}

/// Process all tables and generate the Mutation root object with mutation fields.
pub async fn process_mutations(
	tbs: Arc<[TableDefinition]>,
	types: &mut Vec<Type>,
	schema_ctx: &SchemaContext<'_>,
) -> Result<Object, GraphqlError> {
	let mut mutation = Object::new("Mutation");

	// Collision check: aliases (#4537) + Apollo capitalisation must produce
	// unique mutation field names for every (verb, table) pair. Two tables
	// aliased to the same name would otherwise blow up async-graphql at
	// build time with an opaque error.
	{
		let mut seen: HashMap<String, String> = HashMap::new();
		for tb in tbs.iter() {
			if tb.view.is_some() {
				continue;
			}
			let cap = super::naming::mutation_cap_name(tb);
			let cap_plural = super::naming::mutation_cap_plural_name(tb);
			let mut names = Vec::with_capacity(8);
			for verb in ["create", "update", "upsert", "delete"] {
				names.push(format!("{verb}{cap}"));
				names.push(format!("{verb}{cap_plural}"));
			}
			for name in names {
				if let Some(prior) = seen.get(&name) {
					return Err(super::error::schema_error(format!(
						"GraphQL naming collision on mutation `{}` — both `{}` and `{}` \
						 produce the same mutation field. Set an explicit `GRAPHQL_ALIAS` \
						 on one of them.",
						name, prior, tb.name
					)));
				}
				seen.insert(name, tb.name.as_str().to_owned());
			}
		}
	}

	for tb in tbs.iter() {
		// Pre-computed table views (`DEFINE TABLE x AS SELECT ... FROM y`) are
		// materialised by the SurrealQL engine from upstream tables — direct
		// writes are rejected at execution time. Skip mutation generation so
		// the GraphQL surface doesn't advertise nonsense create/update/delete
		// fields for views.
		if tb.view.is_some() {
			continue;
		}

		let tb_name = tb.name.clone();
		let tb_name_str = tb_name.as_str().to_string();
		let is_relation = matches!(tb.table_type, TableType::Relation(_));

		let fds = schema_ctx.tx.all_tb_fields(schema_ctx.ns, schema_ctx.db, &tb.name, None).await?;

		// Generate input types — fields are camelCased / aliased via the
		// shared `naming` helpers (see GitHub issues #4537 and #4552).
		let (create_input_name, update_input_name, upsert_input_name) =
			generate_input_types(&tb_name_str, &fds, is_relation, types)?;
		let ctx = MutationTableContext {
			cap_name: super::naming::mutation_cap_name(tb),
			cap_plural_name: super::naming::mutation_cap_plural_name(tb),
			table_filter_name: format!("_filter_{tb_name_str}"),
			tb_name_str,
			tb_name,
			is_relation,
			fds,
			kvs: Arc::clone(schema_ctx.datastore),
			graphql_deprecated: tb.graphql_deprecated.clone(),
		};

		// --- Single-record mutations ---
		mutation = add_create_field(mutation, &ctx, &create_input_name);
		mutation = add_update_field(mutation, &ctx, &update_input_name);
		mutation = add_upsert_field(mutation, &ctx, &upsert_input_name);
		mutation = add_delete_field(mutation, &ctx);

		// --- Bulk mutations ---
		mutation = add_create_many_field(mutation, &ctx, &create_input_name);
		mutation = add_update_many_field(mutation, &ctx, &update_input_name);
		mutation = add_upsert_many_field(mutation, &ctx, &upsert_input_name);
		mutation = add_delete_many_field(mutation, &ctx);
	}

	Ok(mutation)
}

// ---------------------------------------------------------------------------
// Single-record mutation field builders
// ---------------------------------------------------------------------------

fn add_create_field(mutation: Object, tc: &MutationTableContext, input_name: &str) -> Object {
	let fds = Arc::clone(&tc.fds);
	let kvs = Arc::clone(&tc.kvs);
	let tb_name = tc.tb_name.clone();
	let is_relation = tc.is_relation;
	mutation.field(
		Field::new(
			format!("create{}", tc.cap_name),
			TypeRef::named(tc.tb_name_str.as_str()),
			move |ctx| {
				let fds = Arc::clone(&fds);
				let kvs = Arc::clone(&kvs);
				let tb_name = tb_name.clone();
				FieldFuture::new(async move {
					let sess = ctx.data::<Arc<Session>>()?;
					let args = ctx.args.as_index_map();
					let data_obj = get_data_object(args)?;
					let id_opt = data_obj.get("id").and_then(GraphqlValueUtils::as_string);

					if is_relation {
						execute_relate_create(&kvs, sess, &tb_name, data_obj, &fds, id_opt).await
					} else {
						execute_normal_create(&kvs, sess, &tb_name, data_obj, &fds, id_opt).await
					}
				})
			},
		)
		.description(tc.describe(format!("Create a new `{}` record", tc.tb_name_str)))
		.argument(InputValue::new("data", TypeRef::named_nn(input_name))),
	)
}

fn add_update_field(mutation: Object, tc: &MutationTableContext, input_name: &str) -> Object {
	let fds = Arc::clone(&tc.fds);
	let kvs = Arc::clone(&tc.kvs);
	let tb_name = tc.tb_name.clone();
	mutation.field(
		Field::new(
			format!("update{}", tc.cap_name),
			TypeRef::named(tc.tb_name_str.as_str()),
			move |ctx| {
				let fds = Arc::clone(&fds);
				let kvs = Arc::clone(&kvs);
				let tb_name = tb_name.clone();
				FieldFuture::new(async move {
					let sess = ctx.data::<Arc<Session>>()?;
					let args = ctx.args.as_index_map();
					let id_str = get_required_id(args)?;
					let data_obj = get_data_object(args)?;

					let rid = parse_record_id(&tb_name, &id_str)?;
					let content =
						graphql_input_to_sql_object(data_obj, &fds, &["id"], tb_name.as_str())?;

					let data = if content.0.is_empty() {
						None
					} else {
						Some(Data::MergeExpression(Value::Object(content).into_literal()))
					};

					let stmt = UpdateStatement {
						only: true,
						what: vec![Value::RecordId(rid).into_literal()],
						data,
						cond: None,
						output: None,
						timeout: Expr::Literal(Literal::None),
						..Default::default()
					};

					let plan = LogicalPlan {
						expressions: vec![TopLevelExpr::Expr(Expr::Update(Box::new(stmt)))],
					};

					let res = execute_plan(&kvs, sess, plan).await?;
					extract_single_record(res)
				})
			},
		)
		.description(tc.describe(format!("Update an existing `{}` record", tc.tb_name_str)))
		.argument(InputValue::new("id", TypeRef::named_nn(TypeRef::ID)))
		.argument(InputValue::new("data", TypeRef::named_nn(input_name))),
	)
}

fn add_upsert_field(mutation: Object, tc: &MutationTableContext, input_name: &str) -> Object {
	let fds = Arc::clone(&tc.fds);
	let kvs = Arc::clone(&tc.kvs);
	let tb_name = tc.tb_name.clone();
	mutation.field(
		Field::new(
			format!("upsert{}", tc.cap_name),
			TypeRef::named(tc.tb_name_str.as_str()),
			move |ctx| {
				let fds = Arc::clone(&fds);
				let kvs = Arc::clone(&kvs);
				let tb_name = tb_name.clone();
				FieldFuture::new(async move {
					let sess = ctx.data::<Arc<Session>>()?;
					let args = ctx.args.as_index_map();
					let id_str = get_required_id(args)?;
					let data_obj = get_data_object(args)?;

					let rid = parse_record_id(&tb_name, &id_str)?;
					let content =
						graphql_input_to_sql_object(data_obj, &fds, &["id"], tb_name.as_str())?;

					let data = if content.0.is_empty() {
						None
					} else {
						Some(Data::ContentExpression(Value::Object(content).into_literal()))
					};

					let stmt = UpsertStatement {
						only: true,
						what: vec![Value::RecordId(rid).into_literal()],
						data,
						cond: None,
						output: None,
						timeout: Expr::Literal(Literal::None),
						..Default::default()
					};

					let plan = LogicalPlan {
						expressions: vec![TopLevelExpr::Expr(Expr::Upsert(Box::new(stmt)))],
					};

					let res = execute_plan(&kvs, sess, plan).await?;
					extract_single_record(res)
				})
			},
		)
		.description(
			tc.describe(format!("Upsert a `{}` record (create or update)", tc.tb_name_str)),
		)
		.argument(InputValue::new("id", TypeRef::named_nn(TypeRef::ID)))
		.argument(InputValue::new("data", TypeRef::named_nn(input_name))),
	)
}

fn add_delete_field(mutation: Object, tc: &MutationTableContext) -> Object {
	let kvs = Arc::clone(&tc.kvs);
	let tb_name = tc.tb_name.clone();
	mutation.field(
		Field::new(
			format!("delete{}", tc.cap_name),
			TypeRef::named_nn(TypeRef::BOOLEAN),
			move |ctx| {
				let kvs = Arc::clone(&kvs);
				let tb_name = tb_name.clone();
				FieldFuture::new(async move {
					let sess = ctx.data::<Arc<Session>>()?;
					let args = ctx.args.as_index_map();
					let id_str = get_required_id(args)?;

					let rid = parse_record_id(&tb_name, &id_str)?;

					let stmt = DeleteStatement {
						only: false,
						what: vec![Value::RecordId(rid).into_literal()],
						cond: None,
						output: None,
						timeout: Expr::Literal(Literal::None),
						..Default::default()
					};

					let plan = LogicalPlan {
						expressions: vec![TopLevelExpr::Expr(Expr::Delete(Box::new(stmt)))],
					};

					let _res = execute_plan(&kvs, sess, plan).await?;
					Ok(Some(FieldValue::value(GraphqlValue::Boolean(true))))
				})
			},
		)
		.description(tc.describe(format!("Delete a `{}` record by ID", tc.tb_name_str)))
		.argument(InputValue::new("id", TypeRef::named_nn(TypeRef::ID))),
	)
}

// ---------------------------------------------------------------------------
// Bulk mutation field builders
// ---------------------------------------------------------------------------

fn add_create_many_field(mutation: Object, tc: &MutationTableContext, input_name: &str) -> Object {
	let fds = Arc::clone(&tc.fds);
	let kvs = Arc::clone(&tc.kvs);
	let tb_name = tc.tb_name.clone();
	let is_relation = tc.is_relation;
	mutation.field(
		Field::new(
			format!("create{}", tc.cap_plural_name),
			TypeRef::named_nn_list_nn(tc.tb_name_str.as_str()),
			move |ctx| {
				let fds = Arc::clone(&fds);
				let kvs = Arc::clone(&kvs);
				let tb_name = tb_name.clone();
				FieldFuture::new(async move {
					let sess = ctx.data::<Arc<Session>>()?;
					let args = ctx.args.as_index_map();
					let data_list =
						args.get("data").and_then(GraphqlValueUtils::as_list).ok_or_else(|| {
							resolver_error("Missing required 'data' argument (must be a list)")
						})?;

					let mut results = Vec::with_capacity(data_list.len());

					for item in data_list {
						let data_obj = item.as_object().ok_or_else(|| {
							resolver_error("Each item in 'data' must be an object")
						})?;
						let id_opt = data_obj.get("id").and_then(GraphqlValueUtils::as_string);

						let res = if is_relation {
							execute_relate_create(&kvs, sess, &tb_name, data_obj, &fds, id_opt)
								.await
						} else {
							execute_normal_create(&kvs, sess, &tb_name, data_obj, &fds, id_opt)
								.await
						};

						match res? {
							Some(fv) => results.push(fv),
							None => {
								return Err(resolver_error(
									"Create returned no result for bulk create",
								)
								.into());
							}
						}
					}

					Ok(Some(FieldValue::list(results)))
				})
			},
		)
		.description(tc.describe(format!("Create multiple `{}` records", tc.tb_name_str)))
		.argument(InputValue::new("data", TypeRef::named_nn_list_nn(input_name))),
	)
}

fn add_update_many_field(mutation: Object, tc: &MutationTableContext, input_name: &str) -> Object {
	let fds = Arc::clone(&tc.fds);
	let kvs = Arc::clone(&tc.kvs);
	let tb_name = tc.tb_name.clone();
	mutation.field(
		Field::new(
			format!("update{}", tc.cap_plural_name),
			TypeRef::named_nn_list_nn(tc.tb_name_str.as_str()),
			move |ctx| {
				let fds = Arc::clone(&fds);
				let kvs = Arc::clone(&kvs);
				let tb_name = tb_name.clone();
				FieldFuture::new(async move {
					let sess = ctx.data::<Arc<Session>>()?;
					let args = ctx.args.as_index_map();

					let data_obj = get_data_object(args)?;
					let content =
						graphql_input_to_sql_object(data_obj, &fds, &["id"], tb_name.as_str())?;
					let data = if content.0.is_empty() {
						None
					} else {
						Some(Data::MergeExpression(Value::Object(content).into_literal()))
					};

					let cond = parse_where_arg(args, &fds, tb_name.as_str())?;

					let stmt = UpdateStatement {
						only: false,
						what: vec![Expr::Table(tb_name)],
						data,
						cond,
						output: None,
						timeout: Expr::Literal(Literal::None),
						..Default::default()
					};

					let plan = LogicalPlan {
						expressions: vec![TopLevelExpr::Expr(Expr::Update(Box::new(stmt)))],
					};

					let res = execute_plan(&kvs, sess, plan).await?;
					extract_record_list(res)
				})
			},
		)
		.description(
			tc.describe(format!("Update multiple `{}` records matching a filter", tc.tb_name_str)),
		)
		.argument(InputValue::new("where", TypeRef::named(tc.table_filter_name.as_str())))
		.argument(InputValue::new("data", TypeRef::named_nn(input_name))),
	)
}

fn add_upsert_many_field(mutation: Object, tc: &MutationTableContext, input_name: &str) -> Object {
	let fds = Arc::clone(&tc.fds);
	let kvs = Arc::clone(&tc.kvs);
	let tb_name = tc.tb_name.clone();
	mutation.field(
		Field::new(
			format!("upsert{}", tc.cap_plural_name),
			TypeRef::named_nn_list_nn(tc.tb_name_str.as_str()),
			move |ctx| {
				let fds = Arc::clone(&fds);
				let kvs = Arc::clone(&kvs);
				let tb_name = tb_name.clone();
				FieldFuture::new(async move {
					let sess = ctx.data::<Arc<Session>>()?;
					let args = ctx.args.as_index_map();

					let data_obj = get_data_object(args)?;
					let content =
						graphql_input_to_sql_object(data_obj, &fds, &["id"], tb_name.as_str())?;
					let data = if content.0.is_empty() {
						None
					} else {
						Some(Data::ContentExpression(Value::Object(content).into_literal()))
					};

					let cond = parse_where_arg(args, &fds, tb_name.as_str())?;

					let stmt = UpsertStatement {
						only: false,
						what: vec![Expr::Table(tb_name)],
						data,
						cond,
						output: None,
						timeout: Expr::Literal(Literal::None),
						..Default::default()
					};

					let plan = LogicalPlan {
						expressions: vec![TopLevelExpr::Expr(Expr::Upsert(Box::new(stmt)))],
					};

					let res = execute_plan(&kvs, sess, plan).await?;
					extract_record_list(res)
				})
			},
		)
		.description(
			tc.describe(format!("Upsert multiple `{}` records matching a filter", tc.tb_name_str)),
		)
		.argument(InputValue::new("where", TypeRef::named(tc.table_filter_name.as_str())))
		.argument(InputValue::new("data", TypeRef::named_nn(input_name))),
	)
}

fn add_delete_many_field(mutation: Object, tc: &MutationTableContext) -> Object {
	let fds = Arc::clone(&tc.fds);
	let kvs = Arc::clone(&tc.kvs);
	let tb_name = tc.tb_name.clone();
	mutation.field(
		Field::new(
			format!("delete{}", tc.cap_plural_name),
			TypeRef::named_nn(TypeRef::INT),
			move |ctx| {
				let fds = Arc::clone(&fds);
				let kvs = Arc::clone(&kvs);
				let tb_name = tb_name.clone();
				FieldFuture::new(async move {
					let sess = ctx.data::<Arc<Session>>()?;
					let args = ctx.args.as_index_map();

					let cond = parse_where_arg(args, &fds, tb_name.as_str())?;

					// Use RETURN BEFORE to count deleted records
					let stmt = DeleteStatement {
						only: false,
						what: vec![Expr::Table(tb_name)],
						cond,
						output: Some(Output::Before),
						timeout: Expr::Literal(Literal::None),
						..Default::default()
					};

					let plan = LogicalPlan {
						expressions: vec![TopLevelExpr::Expr(Expr::Delete(Box::new(stmt)))],
					};

					let res = execute_plan(&kvs, sess, plan).await?;
					let count = match res {
						Value::Array(a) => a.len() as i64,
						_ => 0,
					};
					Ok(Some(FieldValue::value(GraphqlValue::Number(count.into()))))
				})
			},
		)
		.description(tc.describe(format!(
			"Delete multiple `{}` records matching a filter, returns count",
			tc.tb_name_str
		)))
		.argument(InputValue::new("where", TypeRef::named(tc.table_filter_name.as_str()))),
	)
}

// ---------------------------------------------------------------------------
// Shared helpers for mutation resolvers
// ---------------------------------------------------------------------------

/// Extract the required `data` argument as an object from the args map.
fn get_data_object(
	args: &IndexMap<Name, GraphqlValue>,
) -> Result<&IndexMap<Name, GraphqlValue>, GraphqlError> {
	args.get("data")
		.ok_or_else(|| resolver_error("Missing required 'data' argument"))
		.and_then(|v| v.as_object().ok_or_else(|| resolver_error("'data' must be an object")))
}

/// Extract the required `id` argument as a string from the args map.
fn get_required_id(args: &IndexMap<Name, GraphqlValue>) -> Result<String, GraphqlError> {
	args.get("id")
		.and_then(GraphqlValueUtils::as_string)
		.ok_or_else(|| resolver_error("Missing required 'id' argument"))
}

/// Parse the optional `where` argument into a `Cond`.
fn parse_where_arg(
	args: &IndexMap<Name, GraphqlValue>,
	fds: &[FieldDefinition],
	tb_name: &str,
) -> Result<Option<Cond>, GraphqlError> {
	match args.get("where") {
		Some(GraphqlValue::Object(o)) if !o.is_empty() => {
			Ok(Some(cond_from_filter(o, fds, tb_name, &[])?))
		}
		_ => Ok(None),
	}
}

/// Execute a CREATE for a normal (non-relation) table.
async fn execute_normal_create(
	kvs: &Arc<Datastore>,
	sess: &Arc<Session>,
	tb_name: &TableName,
	data_obj: &IndexMap<Name, GraphqlValue>,
	fds: &[FieldDefinition],
	id_opt: Option<String>,
) -> Result<Option<FieldValue<'static>>, async_graphql::Error> {
	let content = graphql_input_to_sql_object(data_obj, fds, &["id"], tb_name.as_str())?;

	let what = match id_opt {
		Some(id_str) => {
			let rid = parse_record_id(tb_name, &id_str)?;
			vec![Value::RecordId(rid).into_literal()]
		}
		None => vec![Expr::Table(tb_name.clone())],
	};

	let data = if content.0.is_empty() {
		None
	} else {
		Some(Data::ContentExpression(Value::Object(content).into_literal()))
	};

	let stmt = CreateStatement {
		only: true,
		what,
		data,
		output: None,
		timeout: Expr::Literal(Literal::None),
	};

	let plan = LogicalPlan {
		expressions: vec![TopLevelExpr::Expr(Expr::Create(Box::new(stmt)))],
	};

	let res = execute_plan(kvs, sess, plan).await?;
	extract_single_record(res)
}

/// Execute a RELATE for a relation table creation.
async fn execute_relate_create(
	kvs: &Arc<Datastore>,
	sess: &Arc<Session>,
	tb_name: &TableName,
	data_obj: &IndexMap<Name, GraphqlValue>,
	fds: &[FieldDefinition],
	id_opt: Option<String>,
) -> Result<Option<FieldValue<'static>>, async_graphql::Error> {
	let in_str = data_obj
		.get("in")
		.and_then(GraphqlValueUtils::as_string)
		.ok_or_else(|| resolver_error("Relation create requires 'in' field"))?;
	let out_str = data_obj
		.get("out")
		.and_then(GraphqlValueUtils::as_string)
		.ok_or_else(|| resolver_error("Relation create requires 'out' field"))?;

	let from_rid = parse_full_record_id(&in_str)?;
	let to_rid = parse_full_record_id(&out_str)?;

	let content =
		graphql_input_to_sql_object(data_obj, fds, &["id", "in", "out"], tb_name.as_str())?;

	let through = match id_opt {
		Some(id_str) => {
			let rid = parse_record_id(tb_name, &id_str)?;
			Value::RecordId(rid).into_literal()
		}
		None => Expr::Table(tb_name.clone()),
	};

	let data = if content.0.is_empty() {
		None
	} else {
		Some(Data::ContentExpression(Value::Object(content).into_literal()))
	};

	let stmt = RelateStatement {
		only: true,
		or_update: false,
		through,
		from: Value::RecordId(from_rid).into_literal(),
		to: Value::RecordId(to_rid).into_literal(),
		data,
		output: None,
		timeout: Expr::Literal(Literal::None),
	};

	let plan = LogicalPlan {
		expressions: vec![TopLevelExpr::Expr(Expr::Relate(Box::new(stmt)))],
	};

	let res = execute_plan(kvs, sess, plan).await?;
	extract_single_record(res)
}

/// Extract a single record from a mutation result and return it as a FieldValue.
///
/// The full result object is cached in a [`CachedRecord`] so that field
/// resolvers can extract values directly without additional database queries.
fn extract_single_record(val: Value) -> Result<Option<FieldValue<'static>>, async_graphql::Error> {
	match val {
		Value::Object(obj) => {
			let rid = match obj.get("id") {
				Some(Value::RecordId(rid)) => rid.clone(),
				_ => return Err(resolver_error("Mutation result missing 'id' field").into()),
			};
			Ok(Some(FieldValue::owned_any(CachedRecord {
				rid,
				version: None,
				data: obj,
			})))
		}
		Value::None | Value::Null => Ok(None),
		_ => {
			error!("Unexpected mutation result type: {val:?}");
			Err(resolver_error("Unexpected mutation result").into())
		}
	}
}

/// Extract a list of records from a bulk mutation result.
///
/// Each result object is cached in a [`CachedRecord`] for efficient field
/// resolution.
fn extract_record_list(val: Value) -> Result<Option<FieldValue<'static>>, async_graphql::Error> {
	match val {
		Value::Array(arr) => {
			let items: Result<Vec<FieldValue>, GraphqlError> = arr
				.0
				.into_iter()
				.map(|v| match v {
					Value::Object(obj) => {
						let rid = match obj.get("id") {
							Some(Value::RecordId(rid)) => rid.clone(),
							_ => return Err(resolver_error("Mutation result missing 'id' field")),
						};
						Ok(FieldValue::owned_any(CachedRecord {
							rid,
							version: None,
							data: obj,
						}))
					}
					_ => {
						error!("Expected object in mutation result, found: {v:?}");
						Err(resolver_error("Unexpected mutation result format"))
					}
				})
				.collect();
			Ok(Some(FieldValue::list(items?)))
		}
		_ => {
			error!("Expected array result for bulk mutation, found: {val:?}");
			Err(resolver_error("Unexpected bulk mutation result format").into())
		}
	}
}