necrassrs 0.1.2

use futures_util::{StreamExt, future::join_all, stream};
use reed_compiler::{
    Name, Node, Schema,
    ast::{Type, Value},
    collections::{HashSet, IndexMap},
    executable::{Field, Selection},
    introspection,
    parser::SourceSpan,
    response::{GraphQLError, JsonMap, JsonValue, ResponseDataPathSegment},
    schema::{ExtendedType, FieldDefinition, ObjectType},
    validation::Valid,
};

use crate::{
    Request, ResolvedObject, ResolvedValue, ResolverError, Response, ResponseStream, ScalarError,
    ScalarLiteralMap, SourceEvent, SourceStream, SubscriptionError,
    input::literal_to_json as input_literal_to_json,
    request::{PreparedRequest, prepare_request},
};

/// Original SDL coordinates of a field to resolve, independent of query aliases.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct FieldCoordinate<'a> {
    /// Name of the object type declaring the field.
    pub parent_type: &'a str,
    /// Field name as written in SDL, not its response alias or Rust identifier.
    pub field: &'a str,
}

/// Connects execution to application resolvers for Context type `C`.
///
/// Usually implemented by the generated `SchemaDispatcher`. Applications can
/// provide a custom dispatcher when integrating directly with the runtime.
pub trait Dispatcher<C> {
    /// Validates one custom scalar value during variable coercion.
    ///
    /// Generated dispatchers call [`crate::Scalar::parse_value`]. The default
    /// accepts the value so existing handwritten dispatchers remain compatible.
    fn validate_custom_scalar_variable(
        &self,
        _scalar_name: &str,
        _value: &JsonValue,
    ) -> Result<(), ScalarError> {
        Ok(())
    }

    /// Validates one custom scalar literal applied through a default value.
    fn validate_custom_scalar_literal(
        &self,
        _scalar_name: &str,
        _value: &Value,
    ) -> Result<(), ScalarError> {
        Ok(())
    }

    /// Resolves a selected field using coerced arguments and request Context.
    ///
    /// Argument keys are original SDL names. The returned `Send` future may
    /// borrow these inputs for the call lifetime. Return [`ResolverError`] for
    /// application failures; execution adds the response path and location and
    /// completes the returned value tree according to the field's declared type.
    /// Embedded conversion errors receive their list-index response paths during
    /// completion; they do not discard successful sibling items.
    fn resolve<'a>(
        &'a self,
        context: &'a C,
        coordinate: FieldCoordinate<'a>,
        arguments: &'a JsonMap,
    ) -> impl Future<Output = Result<ResolvedValue<C>, ResolverError>> + Send + 'a;

    /// Resolves a field while retaining the original kinds of literal arguments.
    fn resolve_with_literals<'a>(
        &'a self,
        context: &'a C,
        coordinate: FieldCoordinate<'a>,
        arguments: &'a JsonMap,
        _argument_literals: &'a ScalarLiteralMap,
    ) -> impl Future<Output = Result<ResolvedValue<C>, ResolverError>> + Send + 'a {
        self.resolve(context, coordinate, arguments)
    }

    /// Creates the source stream for a selected subscription root field.
    fn resolve_event_stream<'stream, 'call>(
        &'stream self,
        _context: &'stream C,
        _coordinate: FieldCoordinate<'call>,
        _arguments: &'call JsonMap,
        _argument_literals: &'call ScalarLiteralMap,
    ) -> impl Future<Output = Result<SourceStream<'stream, ResolvedValue<C>>, ResolverError>>
    + Send
    + 'call
    where
        'stream: 'call,
    {
        async {
            Err(ResolverError::new(
                "Subscription sources are not supported.",
            ))
        }
    }
}

/// Server-controlled execution settings.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct ExecutionOptions {
    /// Whether `__schema` and `__type` may be queried. Enabled by default.
    pub introspection: bool,
}

impl Default for ExecutionOptions {
    fn default() -> Self {
        Self {
            introspection: true,
        }
    }
}

/// Validates and executes a request with introspection enabled.
///
/// Equivalent to [`execute_with_options`] with [`ExecutionOptions::default`].
/// The application owns the validated schema, dispatcher, and Context; the
/// returned future borrows them and is `Send`. Validation failures are returned
/// as request-error [`Response`] values.
///
/// # Scheduling and cancellation
///
/// Query-root fields, fields in nested selection sets, and list items are polled
/// concurrently. Top-level mutation fields are awaited serially in document order;
/// nested selections below each mutation field use normal concurrent execution.
/// A mutation field that propagates null to the operation root prevents later
/// top-level mutation fields from starting.
///
/// Concurrent scheduling does not change response ordering. Object fields retain
/// query order, list items retain input order, and execution errors are merged in
/// the corresponding field or item order. Applications must not rely on sibling
/// resolver start or completion order.
///
/// Execution does not spawn independent field tasks or impose a concurrency limit.
/// Dropping the returned future drops in-progress field futures, but it does not
/// roll back external effects that resolvers have already performed.
///
/// # Example
///
/// ```
/// use necrassrs::{Dispatcher, FieldCoordinate, JsonMap, ResolvedValue, Request,
///     ResolverError, Schema, execute};
///
/// struct Greeting;
/// impl Dispatcher<()> for Greeting {
///     async fn resolve<'a>(
///         &'a self, _context: &'a (), coordinate: FieldCoordinate<'a>,
///         _arguments: &'a JsonMap,
///     ) -> Result<ResolvedValue<()>, ResolverError> {
///         match (coordinate.parent_type, coordinate.field) {
///             ("Query", "hello") => Ok(ResolvedValue::Json("Hello, Sheri".into())),
///             _ => Err(ResolverError::new("Unknown field")),
///         }
///     }
/// }
///
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// let schema = Schema::parse_and_validate(
///     "type Query { hello: String! }", "schema.graphql",
/// ).unwrap();
/// let response = execute(&schema, &Request::new("{ hello }"), &Greeting, &()).await;
/// assert!(!response.is_request_error());
/// # }
/// ```
pub async fn execute<C, D>(
    schema: &Valid<Schema>,
    request: &Request,
    dispatcher: &D,
    context: &C,
) -> Response
where
    C: Sync,
    D: Dispatcher<C> + Sync,
{
    execute_with_options(
        schema,
        request,
        dispatcher,
        context,
        ExecutionOptions::default(),
    )
    .await
}

/// Validates a subscription request and creates its source stream.
pub fn subscribe<'a, C, D>(
    schema: &'a Valid<Schema>,
    request: &Request,
    dispatcher: &'a D,
    context: &'a C,
) -> impl Future<Output = Result<ResponseStream<'a>, Response>> + Send + 'a
where
    C: Sync,
    D: Dispatcher<C> + Sync,
{
    subscribe_with_options(
        schema,
        request,
        dispatcher,
        context,
        ExecutionOptions::default(),
    )
}

/// Validates a subscription request and creates its source stream with server settings.
pub fn subscribe_with_options<'a, C, D>(
    schema: &'a Valid<Schema>,
    request: &Request,
    dispatcher: &'a D,
    context: &'a C,
    _options: ExecutionOptions,
) -> impl Future<Output = Result<ResponseStream<'a>, Response>> + Send + 'a
where
    C: Sync,
    D: Dispatcher<C> + Sync,
{
    let prepared = prepare_request(schema, request).map_err(Response::from);
    async move {
        let prepared = prepared?;
        if !prepared.operation.is_subscription() {
            return Err(Response::request_error(GraphQLError::new(
                "Expected a subscription operation.",
                prepared.operation.location(),
                &prepared.document.sources,
            )));
        }
        if let Err(error) = validate_custom_scalar_variables::<C, D>(schema, &prepared, dispatcher)
        {
            return Err(Response::request_error(*error));
        }

        let fields = collect_fields(schema, &prepared);
        let Some((response_key, selected_fields)) = fields.first() else {
            return Err(Response::request_error(GraphQLError::new(
                "Subscription must select one root field.",
                prepared.operation.location(),
                &prepared.document.sources,
            )));
        };
        let field = selected_fields[0];
        let object_type = schema
            .get_object(prepared.operation.selection_set.ty.as_str())
            .expect("validated subscription root must be an object");
        let definition = schema
            .type_field(object_type.name.as_str(), field.name.as_str())
            .expect("validated subscription selection must have a field definition");
        let path = vec![ResponseDataPathSegment::Field(response_key.clone())];
        let arguments = coerce_argument_values(schema, &prepared, &path, field, definition)
            .map_err(|error| Response::request_error(*error))?;
        let source = dispatcher
            .resolve_event_stream(
                context,
                FieldCoordinate {
                    parent_type: object_type.name.as_str(),
                    field: field.name.as_str(),
                },
                &arguments.values,
                &arguments.literals,
            )
            .await
            .map_err(|error| source_creation_error(&prepared, selected_fields, error))?;

        let state = SubscriptionState {
            schema,
            prepared,
            context,
            source: Some(source),
        };
        Ok(ResponseStream::from_stream(stream::unfold(
            state,
            |mut state| async move {
                let event = state.source.as_mut()?.next().await;
                match event {
                    None => None,
                    Some(SourceEvent::SourceError(error)) => {
                        state.source = None;
                        Some((Err(SubscriptionError::Source(error)), state))
                    }
                    Some(event) => {
                        let response = execute_subscription_event(
                            state.schema,
                            &state.prepared,
                            state.context,
                            event,
                        )
                        .await;
                        Some((Ok(response), state))
                    }
                }
            },
        )))
    }
}

struct SubscriptionState<'a, C> {
    schema: &'a Valid<Schema>,
    prepared: PreparedRequest,
    context: &'a C,
    source: Option<SourceStream<'a, ResolvedValue<C>>>,
}

fn source_creation_error(
    prepared: &PreparedRequest,
    fields: &[&Field],
    error: ResolverError,
) -> Response {
    let ResolverError {
        message,
        extensions,
    } = error;
    let mut error = GraphQLError::new(
        message,
        fields[0].name.location(),
        &prepared.document.sources,
    );
    error
        .locations
        .extend(fields[1..].iter().filter_map(|field| {
            field
                .name
                .location()?
                .line_column(&prepared.document.sources)
        }));
    error.extensions = extensions.unwrap_or_default();
    Response::request_error(error)
}

async fn execute_subscription_event<C>(
    schema: &Valid<Schema>,
    prepared: &PreparedRequest,
    context: &C,
    event: SourceEvent<ResolvedValue<C>>,
) -> Response
where
    C: Sync,
{
    let fields = collect_fields(schema, prepared);
    let (response_key, fields) = fields
        .into_iter()
        .next()
        .expect("validated subscription must select one root field");
    let field = fields[0];
    let object_type = schema
        .get_object(prepared.operation.selection_set.ty.as_str())
        .expect("validated subscription root must be an object");
    let definition = schema
        .type_field(object_type.name.as_str(), field.name.as_str())
        .expect("validated subscription selection must have a field definition");
    let path = vec![ResponseDataPathSegment::Field(response_key.clone())];
    let mut errors = Vec::new();
    let value = match event {
        SourceEvent::Value(value) => value,
        SourceEvent::FieldError(error) => {
            errors.push(*subscription_field_error_to_graphql_error(
                prepared, &fields, &path, error,
            ));
            if definition.ty.is_non_null() {
                return Response::execution(None, errors);
            }
            ResolvedValue::Json(JsonValue::Null)
        }
        SourceEvent::SourceError(_) => unreachable!("source errors terminate before execution"),
    };
    let completed = complete_value(
        schema,
        prepared,
        &fields,
        &definition.ty,
        value,
        context,
        &path,
        &mut errors,
    )
    .await;
    let data = match completed {
        Ok(value) => Some(JsonMap::from_iter([(response_key.as_str().into(), value)])),
        Err(PropagateNull) => None,
    };
    Response::execution(data, errors)
}

fn subscription_field_error_to_graphql_error(
    prepared: &PreparedRequest,
    fields: &[&Field],
    path: &[ResponseDataPathSegment],
    error: ResolverError,
) -> Box<GraphQLError> {
    let ResolverError {
        message,
        extensions,
    } = error;
    let location = |field: &Field| {
        field
            .alias
            .as_ref()
            .and_then(|alias| alias.location())
            .or_else(|| field.name.location())
    };
    let mut error = new_execution_error(prepared, path, message, location(fields[0]));
    error.locations.extend(
        fields[1..]
            .iter()
            .filter_map(|field| location(field)?.line_column(&prepared.document.sources)),
    );
    error.extensions = extensions.unwrap_or_default();
    error
}

/// Executes a request with server-owned settings.
///
/// Performs document validation, operation selection, and variable coercion
/// before invoking the dispatcher. Subscription operations are rejected.
/// Disabling introspection rejects selected `__schema` and `__type` fields
/// before resolver calls; `__typename` remains available.
///
/// Resolver errors become execution errors and follow GraphQL null propagation.
/// Resolver panics are not caught. Applications enforcing termination for
/// unimplemented resolvers must set `panic = "abort"` in their root Cargo profiles.
/// Scheduling and cancellation follow [`execute`]. See that function for a complete
/// example. To disable introspection, pass `ExecutionOptions { introspection: false }`
/// as the final argument here.
pub async fn execute_with_options<C, D>(
    schema: &Valid<Schema>,
    request: &Request,
    dispatcher: &D,
    context: &C,
    options: ExecutionOptions,
) -> Response
where
    C: Sync,
    D: Dispatcher<C> + Sync,
{
    let prepared = match prepare_request(schema, request) {
        Ok(prepared) => prepared,
        Err(error) => return error.into(),
    };

    if let Err(error) = validate_custom_scalar_variables::<C, D>(schema, &prepared, dispatcher) {
        return Response::request_error(*error);
    }

    // TODO: Temporary unsupported-feature contract
    if prepared.operation.is_subscription() {
        return Response::request_error(GraphQLError::new(
            "Subscription operations are not supported.",
            prepared.operation.location(),
            &prepared.document.sources,
        ));
    }

    let object_type = schema
        .get_object(prepared.operation.selection_set.ty.as_str())
        .expect("validated operation root must be an object");

    let fields = collect_fields(schema, &prepared);
    let introspection_field = fields
        .values()
        .map(|fields| fields[0])
        .find(|field| matches!(field.name.as_str(), "__schema" | "__type"));
    if let Some(field) = introspection_field.filter(|_| !options.introspection) {
        return Response::request_error(GraphQLError::new(
            "Schema introspection is disabled.",
            field.name.location(),
            &prepared.document.sources,
        ));
    }

    let mut introspection_data = JsonMap::new();
    let mut errors = Vec::new();
    if introspection_field.is_some() {
        if let Err(error) = introspection::check_max_depth(&prepared.document, &prepared.operation)
        {
            return Response::request_error(error.to_graphql_error(&prepared.document.sources));
        }
        match introspection::partial_execute(
            schema,
            &schema.implementers_map(),
            &prepared.document,
            &prepared.operation,
            &prepared.variables,
        ) {
            Ok(response) => {
                introspection_data = response.data.unwrap_or_default();
                errors = response.errors;
            }
            Err(error) => {
                return Response::request_error(error.to_graphql_error(&prepared.document.sources));
            }
        }
    }

    let mut data = JsonMap::new();
    if prepared.operation.is_mutation() {
        for (response_key, fields) in fields {
            let (response_key, value, mut field_errors) = execute_root_field(
                schema,
                &prepared,
                object_type,
                response_key,
                fields,
                &introspection_data,
                dispatcher,
                context,
            )
            .await;
            errors.append(&mut field_errors);

            match value {
                Some(Ok(value)) => {
                    data.insert(response_key.as_str(), value);
                }
                Some(Err(PropagateNull)) => return Response::execution(None, errors),
                None => {}
            }
        }

        return Response::execution(Some(data), errors);
    }

    let results = join_all(fields.into_iter().map(|(response_key, fields)| {
        execute_root_field(
            schema,
            &prepared,
            object_type,
            response_key,
            fields,
            &introspection_data,
            dispatcher,
            context,
        )
    }))
    .await;
    let mut propagate_null = false;
    for (response_key, value, mut field_errors) in results {
        errors.append(&mut field_errors);

        match value {
            Some(Ok(value)) => {
                data.insert(response_key.as_str(), value);
            }
            Some(Err(PropagateNull)) => propagate_null = true,
            None => {}
        }
    }

    Response::execution((!propagate_null).then_some(data), errors)
}

fn validate_custom_scalar_variables<C, D>(
    schema: &Valid<Schema>,
    prepared: &PreparedRequest,
    dispatcher: &D,
) -> Result<(), Box<GraphQLError>>
where
    D: Dispatcher<C>,
{
    for definition in &prepared.operation.variables {
        let Some(value) = prepared.variables.get(definition.name.as_str()) else {
            continue;
        };

        let literal = variable_literal(schema, prepared, definition, value, &[])?;

        validate_custom_scalar_value::<C, D>(
            schema,
            &definition.ty,
            value,
            Some(&literal),
            dispatcher,
        )
        .map_err(
            |ResolverError {
                 message,
                 extensions,
             }| {
                let mut error =
                    GraphQLError::new(message, definition.location(), &prepared.document.sources);
                error.extensions = extensions.unwrap_or_default();
                Box::new(error)
            },
        )?;
    }

    Ok(())
}

fn validate_custom_scalar_value<C, D>(
    schema: &Valid<Schema>,
    ty: &Type,
    value: &JsonValue,
    literal: Option<&Value>,
    dispatcher: &D,
) -> Result<(), ScalarError>
where
    D: Dispatcher<C>,
{
    if value.is_null() {
        return Ok(());
    }

    let type_name = match ty {
        Type::List(item_type) | Type::NonNullList(item_type) => {
            if let Some(items) = value.as_array() {
                for (index, item_value) in items.iter().enumerate() {
                    let item_literal = match literal {
                        Some(Value::Variable(_)) => literal,
                        Some(Value::List(items)) => items.get(index).map(Node::as_ref),
                        _ => None,
                    };
                    validate_custom_scalar_value::<C, D>(
                        schema,
                        item_type,
                        item_value,
                        item_literal,
                        dispatcher,
                    )?;
                }
            }
            return Ok(());
        }
        Type::Named(name) | Type::NonNullNamed(name) => name,
    };

    match schema.types.get(type_name) {
        Some(ExtendedType::Scalar(scalar)) if !scalar.is_built_in() => match literal {
            Some(Value::Variable(_)) | None => {
                <D as Dispatcher<C>>::validate_custom_scalar_variable(
                    dispatcher,
                    type_name.as_str(),
                    value,
                )
            }
            Some(literal) => <D as Dispatcher<C>>::validate_custom_scalar_literal(
                dispatcher,
                type_name.as_str(),
                literal,
            ),
        },
        Some(ExtendedType::InputObject(input)) => {
            if let Some(object) = value.as_object() {
                for (name, definition) in &input.fields {
                    if let Some(value) = object.get(name.as_str()) {
                        let field_literal = match literal {
                            Some(Value::Variable(_)) => literal,
                            Some(Value::Object(fields)) => fields
                                .iter()
                                .find(|(field_name, _)| field_name == name)
                                .map(|(_, value)| value.as_ref()),
                            _ => None,
                        };
                        validate_custom_scalar_value::<C, D>(
                            schema,
                            &definition.ty,
                            value,
                            field_literal,
                            dispatcher,
                        )?;
                    }
                }
            }
            Ok(())
        }
        _ => Ok(()),
    }
}

#[allow(clippy::too_many_arguments)]
async fn execute_root_field<'a, C, D>(
    schema: &'a Valid<Schema>,
    prepared: &'a PreparedRequest,
    object_type: &'a ObjectType,
    response_key: Name,
    fields: Vec<&'a Field>,
    introspection_data: &'a JsonMap,
    dispatcher: &'a D,
    context: &'a C,
) -> (Name, Option<CompletionResult>, Vec<GraphQLError>)
where
    C: Sync,
    D: Dispatcher<C> + Sync,
{
    let field = fields[0];
    if matches!(field.name.as_str(), "__schema" | "__type") {
        return (
            response_key.clone(),
            introspection_data
                .get(response_key.as_str())
                .cloned()
                .map(Ok),
            Vec::new(),
        );
    }

    let path = vec![ResponseDataPathSegment::Field(response_key.clone())];
    let mut errors = Vec::new();
    let definition = match schema.type_field(object_type.name.as_str(), field.name.as_str()) {
        Ok(definition) => definition,
        Err(_) => {
            errors.push(*new_execution_error(
                prepared,
                &path,
                format!("Could not resolve field definition '{}'.", field.name),
                field.name.location(),
            ));

            return (response_key, Some(Err(PropagateNull)), errors);
        }
    };

    let (value, has_error) = match field.name.as_str() {
        "__typename" => (
            ResolvedValue::Json(JsonValue::from(
                prepared.operation.selection_set.ty.as_str(),
            )),
            false,
        ),
        _ => match coerce_argument_values(schema, prepared, &path, field, definition) {
            Ok(arguments) => match dispatcher
                .resolve_with_literals(
                    context,
                    FieldCoordinate {
                        parent_type: object_type.name.as_str(),
                        field: field.name.as_str(),
                    },
                    &arguments.values,
                    &arguments.literals,
                )
                .await
            {
                Ok(value) => (value, false),
                Err(error) => {
                    errors.push(*resolver_error_to_graphql_error(
                        prepared, &fields, &path, error,
                    ));
                    (ResolvedValue::Json(JsonValue::Null), true)
                }
            },
            Err(error) => {
                errors.push(*error);
                (ResolvedValue::Json(JsonValue::Null), true)
            }
        },
    };

    if has_error
        && matches!(&value, ResolvedValue::Json(value) if value.is_null())
        && definition.ty.is_non_null()
    {
        return (response_key, Some(Err(PropagateNull)), errors);
    }

    let value = complete_value(
        schema,
        prepared,
        &fields,
        &definition.ty,
        value,
        context,
        &path,
        &mut errors,
    )
    .await;

    (response_key, Some(value), errors)
}

fn resolver_error_to_graphql_error(
    prepared: &PreparedRequest,
    fields: &[&Field],
    path: &[ResponseDataPathSegment],
    resolver_error: ResolverError,
) -> Box<GraphQLError> {
    let ResolverError {
        message,
        extensions,
    } = resolver_error;

    let mut error = new_field_execution_error(prepared, fields, path, message);

    error.extensions = extensions.unwrap_or_default();

    error
}

fn collect_fields<'a>(
    schema: &Valid<Schema>,
    prepared: &'a PreparedRequest,
) -> IndexMap<Name, Vec<&'a Field>> {
    let mut fields = IndexMap::default();
    let mut visited_fragments = HashSet::default();
    let object_type = schema
        .get_object(prepared.operation.selection_set.ty.as_str())
        .expect("validated operation root must be an object");

    collect_selections(
        schema,
        object_type,
        prepared,
        &prepared.operation.selection_set.selections,
        &mut visited_fragments,
        &mut fields,
    );

    fields
}

fn collect_selections<'a>(
    schema: &Valid<Schema>,
    object_type: &ObjectType,
    prepared: &'a PreparedRequest,
    selections: &'a [Selection],
    visited_fragments: &mut HashSet<&'a Name>,
    fields: &mut IndexMap<Name, Vec<&'a Field>>,
) {
    selections
        .iter()
        .filter(|selection| should_include(selection, &prepared.variables))
        .for_each(|selection| match selection {
            Selection::Field(field) => {
                fields
                    .entry(field.response_key().clone())
                    .or_default()
                    .push(field);
            }
            Selection::FragmentSpread(spread) => {
                if visited_fragments.insert(&spread.fragment_name)
                    && let Some(fragment) = prepared.document.fragments.get(&spread.fragment_name)
                    && does_fragment_type_apply(schema, object_type, fragment.type_condition())
                {
                    collect_selections(
                        schema,
                        object_type,
                        prepared,
                        &fragment.selection_set.selections,
                        visited_fragments,
                        fields,
                    );
                }
            }
            Selection::InlineFragment(fragment) => {
                if fragment.type_condition.as_ref().is_none_or(|condition| {
                    does_fragment_type_apply(schema, object_type, condition)
                }) {
                    collect_selections(
                        schema,
                        object_type,
                        prepared,
                        &fragment.selection_set.selections,
                        visited_fragments,
                        fields,
                    );
                }
            }
        });
}

fn does_fragment_type_apply(
    schema: &Valid<Schema>,
    object_type: &ObjectType,
    condition: &Name,
) -> bool {
    match schema.types.get(condition) {
        Some(ExtendedType::Object(_)) => condition == &object_type.name,
        Some(ExtendedType::Interface(_)) => object_type.implements_interfaces.contains(condition),
        Some(ExtendedType::Union(union)) => union.members.contains(&object_type.name),
        _ => false,
    }
}

#[derive(Debug, Default)]
struct CoercedArguments {
    values: JsonMap,
    literals: ScalarLiteralMap,
}

fn coerce_argument_values(
    schema: &Valid<Schema>,
    prepared: &PreparedRequest,
    path: &[ResponseDataPathSegment],
    field: &Field,
    field_definition: &FieldDefinition,
) -> Result<CoercedArguments, Box<GraphQLError>> {
    field_definition.arguments.iter().try_fold(
        CoercedArguments::default(),
        |mut coerced, definition| {
            let argument_value = field
                .specified_argument_by_name(definition.name.as_str())
                .or(definition.default_value.as_ref());
            let Some(argument_value) = argument_value else {
                if definition.ty.is_non_null() {
                    return Err(new_execution_error(
                        prepared,
                        path,
                        format!("missing value for required argument '{}'", definition.name),
                        definition.location(),
                    ));
                }

                return Ok(coerced);
            };

            let (value, literal) = if let Some(variable_name) = argument_value.as_variable() {
                match prepared.variables.get(variable_name.as_str()) {
                    Some(variable_value)
                        if variable_value.is_null() && definition.ty.is_non_null() =>
                    {
                        return Err(new_execution_error(
                            prepared,
                            path,
                            format!("null value for non-null argument '{}'", definition.name),
                            argument_value.location(),
                        ));
                    }
                    Some(variable_value) => {
                        let definition = prepared
                            .operation
                            .variables
                            .iter()
                            .find(|definition| definition.name.as_str() == variable_name.as_str())
                            .expect("validated variable reference must have a definition");
                        let literal =
                            variable_literal(schema, prepared, definition, variable_value, path)?;
                        (variable_value.clone(), literal.as_ref().clone())
                    }
                    None => match definition.default_value.as_ref() {
                        Some(default_value) => {
                            let (value, literal) = coerce_input_value(
                                schema,
                                prepared,
                                path,
                                &definition.ty,
                                default_value,
                            )?;
                            (value, literal.as_ref().clone())
                        }
                        None if definition.ty.is_non_null() => {
                            return Err(new_execution_error(
                                prepared,
                                path,
                                format!(
                                    "missing value for required argument '{}'",
                                    definition.name
                                ),
                                argument_value.location(),
                            ));
                        }
                        None => return Ok(coerced),
                    },
                }
            } else {
                let (value, literal) =
                    coerce_input_value(schema, prepared, path, &definition.ty, argument_value)?;
                (value, literal.as_ref().clone())
            };

            coerced.values.insert(definition.name.as_str(), value);
            coerced
                .literals
                .insert(definition.name.as_str().to_owned(), literal);
            Ok(coerced)
        },
    )
}

fn coerce_input_value(
    schema: &Valid<Schema>,
    prepared: &PreparedRequest,
    path: &[ResponseDataPathSegment],
    ty: &Type,
    value: &Node<Value>,
) -> Result<(JsonValue, Node<Value>), Box<GraphQLError>> {
    if let Some(variable_name) = value.as_variable() {
        return match prepared.variables.get(variable_name.as_str()) {
            Some(variable_value) if variable_value.is_null() && ty.is_non_null() => {
                Err(new_execution_error(
                    prepared,
                    path,
                    format!("null variable '${variable_name}' for non-null type '{ty}'"),
                    value.location(),
                ))
            }
            Some(variable_value) => Ok((variable_value.clone(), value.clone())),
            None if ty.is_non_null() => Err(new_execution_error(
                prepared,
                path,
                format!("missing variable '${variable_name}' for non-null type '{ty}'"),
                value.location(),
            )),
            None => Ok((JsonValue::Null, value.clone())),
        };
    }

    if value.is_null() {
        return if ty.is_non_null() {
            Err(new_execution_error(
                prepared,
                path,
                format!("null value for non-null type '{ty}'"),
                value.location(),
            ))
        } else {
            Ok((JsonValue::Null, value.clone()))
        };
    }

    let type_name = match ty {
        Type::List(inner) | Type::NonNullList(inner) => {
            let coerced = value
                .as_list()
                .unwrap_or(std::slice::from_ref(value))
                .iter()
                .map(|value| coerce_input_value(schema, prepared, path, inner, value))
                .collect::<Result<Vec<_>, _>>()?;
            let (values, literals): (Vec<_>, Vec<_>) = coerced.into_iter().unzip();

            return Ok((values.into(), Node::new(Value::List(literals))));
        }
        Type::Named(name) | Type::NonNullNamed(name) => name,
    };

    if type_name.as_str() == "ID"
        && let Value::Int(integer) = value.as_ref()
    {
        return Ok((JsonValue::from(integer.as_str()), value.clone()));
    }

    if let Some(ExtendedType::InputObject(input)) = schema.types.get(type_name) {
        let object = value.as_object().ok_or_else(|| {
            new_execution_error(
                prepared,
                path,
                format!("could not coerce value to input object '{type_name}'"),
                value.location(),
            )
        })?;

        let (values, literals) = input.fields.iter().try_fold(
            (JsonMap::new(), Vec::new()),
            |(mut coerced, mut literals), (name, definition)| {
                let specified = object
                    .iter()
                    .find(|(field_name, _)| field_name == name)
                    .map(|(_, value)| value);
                let value = match specified {
                    Some(value)
                        if value.as_variable().is_some_and(|variable_name| {
                            !prepared.variables.contains_key(variable_name.as_str())
                        }) =>
                    {
                        definition.default_value.as_ref()
                    }
                    Some(value) => Some(value),
                    None => definition.default_value.as_ref(),
                };

                match value {
                    Some(value) => {
                        let (value, literal) =
                            coerce_input_value(schema, prepared, path, &definition.ty, value)?;
                        coerced.insert(name.as_str(), value);
                        literals.push((name.clone(), literal));
                    }
                    None if definition.ty.is_non_null() => {
                        return Err(new_execution_error(
                            prepared,
                            path,
                            format!("missing value for required input field '{type_name}.{name}'"),
                            definition.location(),
                        ));
                    }
                    None => {}
                }

                Ok((coerced, literals))
            },
        )?;

        return Ok((values.into(), Node::new(Value::Object(literals))));
    }

    if schema.types.get(type_name).is_some_and(
        |definition| matches!(definition, ExtendedType::Scalar(scalar) if !scalar.is_built_in()),
    ) {
        let coerced = match value.as_ref() {
            Value::String(value) => value.clone(),
            Value::Enum(value) => value.as_str().to_owned(),
            value => value.to_string(),
        };
        return Ok((JsonValue::from(coerced), value.clone()));
    }

    input_literal_to_json(value)
        .map_err(|error| new_execution_error(prepared, path, error.message, error.location))
        .map(|coerced| (coerced, value.clone()))
}

fn variable_literal(
    schema: &Valid<Schema>,
    prepared: &PreparedRequest,
    definition: &reed_compiler::ast::VariableDefinition,
    value: &JsonValue,
    path: &[ResponseDataPathSegment],
) -> Result<Node<Value>, Box<GraphQLError>> {
    match prepared.provided_variables.get(definition.name.as_str()) {
        Some(provided) => variable_value_literal(
            schema,
            prepared,
            path,
            &definition.ty,
            provided,
            value,
            &definition.name,
        ),
        None => match definition.default_value.as_ref() {
            Some(default) => {
                coerce_input_value(schema, prepared, path, &definition.ty, default).map(|(_, v)| v)
            }
            None => Ok(Node::new(Value::Variable(definition.name.clone()))),
        },
    }
}

#[allow(clippy::too_many_arguments)]
fn variable_value_literal(
    schema: &Valid<Schema>,
    prepared: &PreparedRequest,
    path: &[ResponseDataPathSegment],
    ty: &Type,
    provided: &JsonValue,
    coerced: &JsonValue,
    variable_name: &Name,
) -> Result<Node<Value>, Box<GraphQLError>> {
    let marker = || Node::new(Value::Variable(variable_name.clone()));
    if provided.is_null() || coerced.is_null() {
        return Ok(marker());
    }

    let type_name = match ty {
        Type::List(inner) | Type::NonNullList(inner) => {
            let Some(items) = coerced.as_array() else {
                return Ok(marker());
            };
            let literals = items
                .iter()
                .enumerate()
                .map(|(index, item)| {
                    let provided = provided
                        .as_array()
                        .and_then(|items| items.get(index))
                        .unwrap_or(provided);
                    variable_value_literal(
                        schema,
                        prepared,
                        path,
                        inner,
                        provided,
                        item,
                        variable_name,
                    )
                })
                .collect::<Result<Vec<_>, _>>()?;
            return Ok(Node::new(Value::List(literals)));
        }
        Type::Named(name) | Type::NonNullNamed(name) => name,
    };

    let Some(ExtendedType::InputObject(input)) = schema.types.get(type_name) else {
        return Ok(marker());
    };
    let Some(object) = coerced.as_object() else {
        return Ok(marker());
    };
    let provided = provided.as_object();
    let fields = input
        .fields
        .iter()
        .filter_map(|(name, field)| {
            let coerced = object.get(name.as_str())?;
            Some(
                match provided.and_then(|object| object.get(name.as_str())) {
                    Some(provided) => variable_value_literal(
                        schema,
                        prepared,
                        path,
                        &field.ty,
                        provided,
                        coerced,
                        variable_name,
                    ),
                    None => field
                        .default_value
                        .as_ref()
                        .map(|default| {
                            coerce_input_value(schema, prepared, path, &field.ty, default)
                                .map(|(_, literal)| literal)
                        })
                        .unwrap_or_else(|| Ok(marker())),
                }
                .map(|literal| (name.clone(), literal)),
            )
        })
        .collect::<Result<Vec<_>, _>>()?;
    Ok(Node::new(Value::Object(fields)))
}

fn new_execution_error(
    prepared: &PreparedRequest,
    path: &[ResponseDataPathSegment],
    message: impl Into<String>,
    location: Option<SourceSpan>,
) -> Box<GraphQLError> {
    let mut error = GraphQLError::new(message, location, &prepared.document.sources);

    error.path = path.to_vec();

    Box::new(error)
}

fn new_field_execution_error(
    prepared: &PreparedRequest,
    fields: &[&Field],
    path: &[ResponseDataPathSegment],
    message: impl Into<String>,
) -> Box<GraphQLError> {
    let mut error = new_execution_error(prepared, path, message, fields[0].name.location());
    error
        .locations
        .extend(fields[1..].iter().filter_map(|field| {
            field
                .name
                .location()?
                .line_column(&prepared.document.sources)
        }));
    error
}

struct PropagateNull;

type CompletionResult = Result<JsonValue, PropagateNull>;
type CompletionFuture<'a> =
    ::core::pin::Pin<Box<dyn Future<Output = CompletionResult> + Send + 'a>>;

#[allow(clippy::too_many_arguments)]
fn complete_value<'a, C>(
    schema: &'a Valid<Schema>,
    prepared: &'a PreparedRequest,
    fields: &'a [&'a Field],
    ty: &'a Type,
    value: ResolvedValue<C>,
    context: &'a C,
    path: &'a [ResponseDataPathSegment],
    errors: &'a mut Vec<GraphQLError>,
) -> CompletionFuture<'a>
where
    C: Sync + 'a,
{
    Box::pin(async move {
        let field = fields[0];
        let value = match value {
            ResolvedValue::Error(error) => {
                errors.push(*resolver_error_to_graphql_error(
                    prepared, fields, path, error,
                ));
                return complete_null(ty);
            }
            value => value,
        };

        if matches!(&value, ResolvedValue::Json(value) if value.is_null()) {
            if ty.is_non_null() {
                errors.push(*new_field_execution_error(
                    prepared,
                    fields,
                    path,
                    format!(
                        "Cannot return null for non-nullable field '{}'.",
                        field.name
                    ),
                ));
            }
            return complete_null(ty);
        }

        match ty {
            Type::List(item_type) | Type::NonNullList(item_type) => {
                complete_list_value(
                    schema, prepared, fields, ty, item_type, value, context, path, errors,
                )
                .await
            }
            Type::Named(name) | Type::NonNullNamed(name) => {
                complete_named_value(
                    schema, prepared, fields, ty, name, value, context, path, errors,
                )
                .await
            }
        }
    })
}

fn complete_null(ty: &Type) -> CompletionResult {
    if ty.is_non_null() {
        Err(PropagateNull)
    } else {
        Ok(JsonValue::Null)
    }
}

#[allow(clippy::too_many_arguments)]
async fn complete_list_value<C>(
    schema: &Valid<Schema>,
    prepared: &PreparedRequest,
    fields: &[&Field],
    ty: &Type,
    item_type: &Type,
    value: ResolvedValue<C>,
    context: &C,
    path: &[ResponseDataPathSegment],
    errors: &mut Vec<GraphQLError>,
) -> CompletionResult
where
    C: Sync,
{
    let field = fields[0];
    let values = match value {
        ResolvedValue::List(values) => values,
        ResolvedValue::Json(JsonValue::Array(values)) => {
            values.into_iter().map(ResolvedValue::Json).collect()
        }
        _ => {
            errors.push(*new_field_execution_error(
                prepared,
                fields,
                path,
                format!("Expected field '{}' to return a list.", field.name),
            ));
            return complete_null(ty);
        }
    };

    let results = join_all(values.into_iter().enumerate().map(|(index, value)| {
        let mut item_path = path.to_vec();
        item_path.push(ResponseDataPathSegment::ListIndex(index));

        async move {
            let mut item_errors = Vec::new();
            let item = complete_value(
                schema,
                prepared,
                fields,
                item_type,
                value,
                context,
                &item_path,
                &mut item_errors,
            )
            .await;
            (item, item_errors)
        }
    }))
    .await;

    let mut completed = Vec::with_capacity(results.len());
    let mut propagate_null = false;
    for (item, mut item_errors) in results {
        errors.append(&mut item_errors);
        match item {
            Ok(value) => completed.push(value),
            Err(PropagateNull) => propagate_null = true,
        }
    }

    if propagate_null {
        complete_null(ty)
    } else {
        Ok(completed.into())
    }
}

#[allow(clippy::too_many_arguments)]
async fn complete_named_value<C>(
    schema: &Valid<Schema>,
    prepared: &PreparedRequest,
    fields: &[&Field],
    ty: &Type,
    name: &Name,
    value: ResolvedValue<C>,
    context: &C,
    path: &[ResponseDataPathSegment],
    errors: &mut Vec<GraphQLError>,
) -> CompletionResult
where
    C: Sync,
{
    if matches!(
        schema.types.get(name),
        Some(ExtendedType::Object(_) | ExtendedType::Interface(_) | ExtendedType::Union(_))
    ) {
        complete_object_value(
            schema, prepared, fields, ty, name, value, context, path, errors,
        )
        .await
    } else {
        complete_leaf_value(schema, prepared, fields, ty, name, value, path, errors)
    }
}

#[allow(clippy::too_many_arguments)]
async fn complete_object_value<C>(
    schema: &Valid<Schema>,
    prepared: &PreparedRequest,
    fields: &[&Field],
    ty: &Type,
    name: &Name,
    value: ResolvedValue<C>,
    context: &C,
    path: &[ResponseDataPathSegment],
    errors: &mut Vec<GraphQLError>,
) -> CompletionResult
where
    C: Sync,
{
    let field = fields[0];
    let ResolvedValue::Object(object) = value else {
        errors.push(*new_field_execution_error(
            prepared,
            fields,
            path,
            format!("Expected field '{}' to return an Object value.", field.name),
        ));
        return complete_null(ty);
    };

    let Some(object_type) = schema.get_object(object.type_name()) else {
        errors.push(*new_field_execution_error(
            prepared,
            fields,
            path,
            format!("Unknown runtime Object type '{}'.", object.type_name()),
        ));
        return complete_null(ty);
    };

    if !does_fragment_type_apply(schema, object_type, name) {
        errors.push(*new_field_execution_error(
            prepared,
            fields,
            path,
            format!(
                "Runtime Object type '{}' is not valid for '{name}'.",
                object_type.name
            ),
        ));
        return complete_null(ty);
    }

    let mut subfields = IndexMap::default();
    let mut visited_fragments = HashSet::default();
    for field in fields {
        collect_selections(
            schema,
            object_type,
            prepared,
            &field.selection_set.selections,
            &mut visited_fragments,
            &mut subfields,
        );
    }

    let results = join_all(subfields.into_iter().map(|(response_key, fields)| {
        let mut field_path = path.to_vec();
        field_path.push(ResponseDataPathSegment::Field(response_key.clone()));
        complete_object_field(
            schema,
            prepared,
            response_key,
            fields,
            object.as_ref(),
            object_type,
            context,
            field_path,
        )
    }))
    .await;

    let mut completed = JsonMap::new();
    let mut propagate_null = false;
    for (response_key, field_value, mut field_errors) in results {
        errors.append(&mut field_errors);
        match field_value {
            Ok(value) => {
                completed.insert(response_key.as_str(), value);
            }
            Err(PropagateNull) => propagate_null = true,
        }
    }

    if propagate_null {
        complete_null(ty)
    } else {
        Ok(completed.into())
    }
}

#[allow(clippy::too_many_arguments)]
async fn complete_object_field<'a, C>(
    schema: &'a Valid<Schema>,
    prepared: &'a PreparedRequest,
    response_key: Name,
    fields: Vec<&'a Field>,
    object: &'a dyn ResolvedObject<C>,
    object_type: &'a ObjectType,
    context: &'a C,
    path: Vec<ResponseDataPathSegment>,
) -> (Name, CompletionResult, Vec<GraphQLError>)
where
    C: Sync,
{
    let selected_field = fields[0];
    if selected_field.name.as_str() == "__typename" {
        return (
            response_key,
            Ok(JsonValue::from(object_type.name.as_str())),
            Vec::new(),
        );
    }

    let definition = schema
        .type_field(object_type.name.as_str(), selected_field.name.as_str())
        .expect("validated Object selection must have a field definition");
    let mut errors = Vec::new();
    let (field_value, has_error) = resolve_object_field(
        schema,
        prepared,
        &fields,
        definition,
        object,
        object_type,
        context,
        &path,
        &mut errors,
    )
    .await;

    if has_error
        && matches!(&field_value, ResolvedValue::Json(value) if value.is_null())
        && definition.ty.is_non_null()
    {
        return (response_key, Err(PropagateNull), errors);
    }

    let field_value = complete_value(
        schema,
        prepared,
        &fields,
        &definition.ty,
        field_value,
        context,
        &path,
        &mut errors,
    )
    .await;

    (response_key, field_value, errors)
}

#[allow(clippy::too_many_arguments)]
async fn resolve_object_field<C>(
    schema: &Valid<Schema>,
    prepared: &PreparedRequest,
    fields: &[&Field],
    definition: &FieldDefinition,
    object: &dyn ResolvedObject<C>,
    object_type: &ObjectType,
    context: &C,
    path: &[ResponseDataPathSegment],
    errors: &mut Vec<GraphQLError>,
) -> (ResolvedValue<C>, bool)
where
    C: Sync,
{
    let field = fields[0];
    let arguments = match coerce_argument_values(schema, prepared, path, field, definition) {
        Ok(arguments) => arguments,
        Err(error) => {
            errors.push(*error);
            return (ResolvedValue::Json(JsonValue::Null), true);
        }
    };

    match object
        .resolve_with_literals(
            context,
            FieldCoordinate {
                parent_type: object_type.name.as_str(),
                field: field.name.as_str(),
            },
            &arguments.values,
            &arguments.literals,
        )
        .await
    {
        Ok(value) => (value, false),
        Err(error) => {
            errors.push(*resolver_error_to_graphql_error(
                prepared, fields, path, error,
            ));
            (ResolvedValue::Json(JsonValue::Null), true)
        }
    }
}

#[allow(clippy::too_many_arguments)]
fn complete_leaf_value<C>(
    schema: &Valid<Schema>,
    prepared: &PreparedRequest,
    fields: &[&Field],
    ty: &Type,
    name: &Name,
    value: ResolvedValue<C>,
    path: &[ResponseDataPathSegment],
    errors: &mut Vec<GraphQLError>,
) -> CompletionResult {
    let field = fields[0];
    let ResolvedValue::Json(value) = value else {
        errors.push(*new_field_execution_error(
            prepared,
            fields,
            path,
            format!("Expected field '{}' to return a leaf value.", field.name),
        ));
        return complete_null(ty);
    };

    let message = match schema.types.get(name) {
        Some(ExtendedType::Scalar(_)) if valid_builtin_scalar(name, &value) => return Ok(value),
        Some(ExtendedType::Scalar(_)) if is_builtin_scalar(name) => {
            format!("Expected field '{}' to return a {name}.", field.name)
        }
        Some(ExtendedType::Scalar(_)) => return Ok(value),
        Some(ExtendedType::Enum(enum_type))
            if value
                .as_str()
                .is_some_and(|value| enum_type.values.contains_key(value)) =>
        {
            return Ok(value);
        }
        Some(ExtendedType::Enum(_)) => {
            format!(
                "Expected field '{}' to return a {name} enum value.",
                field.name
            )
        }
        Some(_) => format!("Result completion for type '{name}' is not supported."),
        None => format!("Unknown output type '{name}'."),
    };

    errors.push(*new_field_execution_error(prepared, fields, path, message));
    complete_null(ty)
}

fn valid_builtin_scalar(name: &Name, value: &JsonValue) -> bool {
    match name.as_str() {
        "Int" => value
            .as_i64()
            .is_some_and(|value| i32::try_from(value).is_ok()),
        "Float" => value.as_f64().is_some_and(f64::is_finite),
        "String" | "ID" => value.is_string(),
        "Boolean" => value.as_bool().is_some(),
        _ => false,
    }
}

fn is_builtin_scalar(name: &Name) -> bool {
    matches!(name.as_str(), "Int" | "Float" | "String" | "Boolean" | "ID")
}

fn directive_condition(selection: &Selection, name: &str, variables: &JsonMap) -> Option<bool> {
    let value = selection
        .directives()
        .get(name)?
        .specified_argument_by_name("if")?;

    match value.as_ref() {
        Value::Boolean(value) => Some(*value),
        Value::Variable(name) => variables.get(name.as_str())?.as_bool(),
        _ => None,
    }
}

fn should_include(selection: &Selection, variables: &JsonMap) -> bool {
    !directive_condition(selection, "skip", variables).unwrap_or(false)
        && directive_condition(selection, "include", variables).unwrap_or(true)
}

#[cfg(test)]
mod tests {
    use std::{
        future::{Future, pending, poll_fn},
        sync::{
            Arc, Mutex,
            atomic::{AtomicUsize, Ordering},
        },
        task::Poll,
        time::Duration,
    };

    use crate::{FieldCoordinate, Request, ResolvedValue, ResolverError, request::prepare_request};
    use reed_compiler::{
        Schema,
        response::{
            JsonMap, JsonValue, ResponseDataPathSegment,
            serde_json_bytes::{json, to_value},
        },
    };
    use tokio::sync::Barrier;

    struct TestContext<'a> {
        greeting: &'a str,
    }

    struct TestDispatcher;

    struct FailingDispatcher;

    struct NullDispatcher;

    struct ValueDispatcher(JsonValue);

    struct InvalidAbstractDispatcher(&'static str);

    struct InvalidAbstractObject(&'static str);

    struct CountingDispatcher(AtomicUsize);

    struct CountingFailingDispatcher(AtomicUsize);

    struct RecoveringDispatcher(AtomicUsize);

    struct OrderingDispatcher(Mutex<Vec<String>>);

    struct ConcurrentDispatcher {
        barrier: Arc<Barrier>,
    }

    enum ConcurrentObject {
        Siblings {
            barrier: Arc<Barrier>,
            fail_first: bool,
        },
        Item {
            barrier: Arc<Barrier>,
            index: usize,
        },
    }

    struct PendingDispatcher {
        started: AtomicUsize,
        dropped: AtomicUsize,
    }

    struct DropCounter<'a>(&'a AtomicUsize);

    impl Drop for DropCounter<'_> {
        fn drop(&mut self) {
            self.0.fetch_add(1, Ordering::Relaxed);
        }
    }

    struct MergedObjectDispatcher {
        viewer_calls: AtomicUsize,
        friend_calls: Arc<AtomicUsize>,
    }

    struct MergedObject {
        name: &'static str,
        id: &'static str,
        friend_calls: Arc<AtomicUsize>,
    }

    impl<'context> super::Dispatcher<TestContext<'context>> for TestDispatcher {
        async fn resolve<'a>(
            &'a self,
            context: &'a TestContext<'context>,
            coordinate: FieldCoordinate<'a>,
            arguments: &'a JsonMap,
        ) -> Result<ResolvedValue<TestContext<'context>>, ResolverError> {
            tokio::task::yield_now().await;

            assert_eq!(coordinate.field, "hello");

            let name = arguments.get("name").and_then(JsonValue::as_str).unwrap();

            Ok(ResolvedValue::Json(json!(format!(
                "{}, {name}",
                context.greeting
            ))))
        }
    }

    impl super::Dispatcher<()> for FailingDispatcher {
        async fn resolve<'a>(
            &'a self,
            _context: &'a (),
            _coordinate: FieldCoordinate<'a>,
            _arguments: &'a JsonMap,
        ) -> Result<ResolvedValue, ResolverError> {
            Err(ResolverError::new("Greeting failed.").with_extension("code", "GREETING_FAILED"))
        }
    }

    impl super::Dispatcher<()> for NullDispatcher {
        async fn resolve<'a>(
            &'a self,
            _context: &'a (),
            _coordinate: FieldCoordinate<'a>,
            _arguments: &'a JsonMap,
        ) -> Result<ResolvedValue, ResolverError> {
            Ok(ResolvedValue::Json(JsonValue::Null))
        }
    }

    impl super::Dispatcher<()> for ValueDispatcher {
        async fn resolve<'a>(
            &'a self,
            _context: &'a (),
            _coordinate: FieldCoordinate<'a>,
            _arguments: &'a JsonMap,
        ) -> Result<ResolvedValue, ResolverError> {
            Ok(ResolvedValue::Json(self.0.clone()))
        }
    }

    impl super::Dispatcher<()> for InvalidAbstractDispatcher {
        async fn resolve<'a>(
            &'a self,
            _context: &'a (),
            _coordinate: FieldCoordinate<'a>,
            _arguments: &'a JsonMap,
        ) -> Result<ResolvedValue, ResolverError> {
            Ok(ResolvedValue::Object(Box::new(InvalidAbstractObject(
                self.0,
            ))))
        }
    }

    impl crate::ResolvedObject<()> for InvalidAbstractObject {
        fn type_name(&self) -> &'static str {
            self.0
        }

        fn resolve<'a>(
            &'a self,
            _context: &'a (),
            _coordinate: FieldCoordinate<'a>,
            _arguments: &'a JsonMap,
        ) -> ::core::pin::Pin<
            Box<
                dyn ::core::future::Future<Output = Result<ResolvedValue, ResolverError>>
                    + Send
                    + 'a,
            >,
        > {
            Box::pin(async { panic!("invalid abstract Object fields must not execute") })
        }
    }

    impl super::Dispatcher<()> for CountingDispatcher {
        async fn resolve<'a>(
            &'a self,
            _context: &'a (),
            _coordinate: FieldCoordinate<'a>,
            _arguments: &'a JsonMap,
        ) -> Result<ResolvedValue, ResolverError> {
            self.0.fetch_add(1, Ordering::Relaxed);

            Ok(ResolvedValue::Json(json!("unexpected resolver call")))
        }
    }

    impl super::Dispatcher<()> for CountingFailingDispatcher {
        async fn resolve<'a>(
            &'a self,
            _context: &'a (),
            _coordinate: FieldCoordinate<'a>,
            _arguments: &'a JsonMap,
        ) -> Result<ResolvedValue, ResolverError> {
            self.0.fetch_add(1, Ordering::Relaxed);

            Err(ResolverError::new("Greeting failed.").with_extension("code", "GREETING_FAILED"))
        }
    }

    impl super::Dispatcher<()> for RecoveringDispatcher {
        async fn resolve<'a>(
            &'a self,
            _context: &'a (),
            _coordinate: FieldCoordinate<'a>,
            _arguments: &'a JsonMap,
        ) -> Result<ResolvedValue, ResolverError> {
            if self.0.fetch_add(1, Ordering::Relaxed) == 0 {
                Err(ResolverError::new("Greeting failed."))
            } else {
                Ok(ResolvedValue::Json(json!("Hello")))
            }
        }
    }

    impl super::Dispatcher<()> for OrderingDispatcher {
        async fn resolve<'a>(
            &'a self,
            _context: &'a (),
            coordinate: FieldCoordinate<'a>,
            _arguments: &'a JsonMap,
        ) -> Result<ResolvedValue, ResolverError> {
            self.0.lock().unwrap().push(coordinate.field.to_owned());

            Ok(ResolvedValue::Json(json!(coordinate.field)))
        }
    }

    impl super::Dispatcher<()> for ConcurrentDispatcher {
        async fn resolve<'a>(
            &'a self,
            _context: &'a (),
            coordinate: FieldCoordinate<'a>,
            _arguments: &'a JsonMap,
        ) -> Result<ResolvedValue, ResolverError> {
            match (coordinate.parent_type, coordinate.field) {
                ("Query", "first" | "second") => {
                    self.barrier.wait().await;
                    Ok(ResolvedValue::Json(json!(coordinate.field)))
                }
                ("Query", "nested") => Ok(ResolvedValue::Object(Box::new(
                    ConcurrentObject::Siblings {
                        barrier: Arc::clone(&self.barrier),
                        fail_first: true,
                    },
                ))),
                ("Query", "items") => Ok(ResolvedValue::List(
                    (0..3)
                        .map(|index| {
                            ResolvedValue::Object(Box::new(ConcurrentObject::Item {
                                barrier: Arc::clone(&self.barrier),
                                index,
                            }))
                        })
                        .collect(),
                )),
                ("Mutation", "update") => Ok(ResolvedValue::Object(Box::new(
                    ConcurrentObject::Siblings {
                        barrier: Arc::clone(&self.barrier),
                        fail_first: false,
                    },
                ))),
                _ => Err(ResolverError::new(format!(
                    "Unknown {} field '{}'",
                    coordinate.parent_type, coordinate.field
                ))),
            }
        }
    }

    impl crate::ResolvedObject<()> for ConcurrentObject {
        fn type_name(&self) -> &'static str {
            match self {
                Self::Siblings { .. } => "Nested",
                Self::Item { .. } => "Item",
            }
        }

        fn resolve<'a>(
            &'a self,
            _context: &'a (),
            coordinate: FieldCoordinate<'a>,
            _arguments: &'a JsonMap,
        ) -> ::core::pin::Pin<
            Box<
                dyn ::core::future::Future<Output = Result<ResolvedValue, ResolverError>>
                    + Send
                    + 'a,
            >,
        > {
            Box::pin(async move {
                match self {
                    Self::Siblings {
                        barrier,
                        fail_first,
                    } => {
                        barrier.wait().await;
                        if *fail_first && coordinate.field == "first" {
                            Err(ResolverError::new("Nested field failed.")
                                .with_extension("code", "NESTED_FAILED"))
                        } else {
                            Ok(ResolvedValue::Json(json!(coordinate.field)))
                        }
                    }
                    Self::Item { barrier, index } => {
                        barrier.wait().await;
                        if index % 2 == 0 {
                            Err(ResolverError::new(format!("Item {index} failed."))
                                .with_extension("code", "ITEM_FAILED"))
                        } else {
                            Ok(ResolvedValue::Json(json!(*index)))
                        }
                    }
                }
            })
        }
    }

    impl super::Dispatcher<()> for PendingDispatcher {
        async fn resolve<'a>(
            &'a self,
            _context: &'a (),
            _coordinate: FieldCoordinate<'a>,
            _arguments: &'a JsonMap,
        ) -> Result<ResolvedValue, ResolverError> {
            self.started.fetch_add(1, Ordering::Relaxed);
            let _drop_counter = DropCounter(&self.dropped);
            pending().await
        }
    }

    impl super::Dispatcher<()> for MergedObjectDispatcher {
        async fn resolve<'a>(
            &'a self,
            _context: &'a (),
            coordinate: FieldCoordinate<'a>,
            _arguments: &'a JsonMap,
        ) -> Result<ResolvedValue, ResolverError> {
            assert_eq!(
                coordinate,
                FieldCoordinate {
                    parent_type: "Query",
                    field: "viewer",
                }
            );
            self.viewer_calls.fetch_add(1, Ordering::Relaxed);

            Ok(ResolvedValue::Object(Box::new(MergedObject {
                name: "Viewer",
                id: "viewer-id",
                friend_calls: Arc::clone(&self.friend_calls),
            })))
        }
    }

    impl crate::ResolvedObject<()> for MergedObject {
        fn type_name(&self) -> &'static str {
            "User"
        }

        fn resolve<'a>(
            &'a self,
            _context: &'a (),
            coordinate: FieldCoordinate<'a>,
            _arguments: &'a JsonMap,
        ) -> ::core::pin::Pin<
            Box<
                dyn ::core::future::Future<Output = Result<ResolvedValue, ResolverError>>
                    + Send
                    + 'a,
            >,
        > {
            Box::pin(async move {
                match coordinate.field {
                    "name" => Ok(ResolvedValue::Json(json!(self.name))),
                    "id" => Ok(ResolvedValue::Json(json!(self.id))),
                    "friend" => {
                        self.friend_calls.fetch_add(1, Ordering::Relaxed);
                        Ok(ResolvedValue::Object(Box::new(Self {
                            name: "Friend",
                            id: "friend-id",
                            friend_calls: Arc::clone(&self.friend_calls),
                        })))
                    }
                    field => Err(ResolverError::new(format!("Unknown User field '{field}'"))),
                }
            })
        }
    }

    fn assert_send<T: Send>(_: &T) {}

    fn coerce_arguments(schema_source: &str, request: Request) -> JsonMap {
        let schema = Schema::parse_and_validate(schema_source, "schema.graphql").unwrap();
        let prepared = prepare_request(&schema, &request).unwrap();
        let field = super::collect_fields(&schema, &prepared)
            .into_values()
            .next()
            .unwrap()[0];

        super::coerce_argument_values(&schema, &prepared, &[], field, &field.definition)
            .unwrap()
            .values
    }

    async fn execute_value(field_type: &str, value: JsonValue) -> JsonValue {
        let schema = Schema::parse_and_validate(
            format!("type Query {{ values: {field_type} }}"),
            "schema.graphql",
        )
        .unwrap();
        let request = Request::new("query { values }");
        let dispatcher = ValueDispatcher(value);

        to_value(super::execute(&schema, &request, &dispatcher, &()).await).unwrap()
    }

    async fn execute_list_with_null_item(field_type: &str) -> JsonValue {
        execute_value(field_type, json!(["A", null, "B"])).await
    }

    #[test]
    fn direct_root_field_is_collected_by_response_key() {
        let schema = Schema::parse_and_validate(
            "type Query { hello(name: String!): String! }",
            "schema.graphql",
        )
        .unwrap();

        let request = Request::new(
            r#"
                query {
                    greeting: hello(name: "Sheri")
                }
            "#,
        );

        let prepared = prepare_request(&schema, &request).unwrap();

        let fields = super::collect_fields(&schema, &prepared);
        let collected = fields.get("greeting").unwrap();

        assert_eq!(fields.len(), 1);
        assert_eq!(collected.len(), 1);
        assert_eq!(collected[0].name.as_str(), "hello");
    }

    #[tokio::test(flavor = "current_thread")]
    async fn field_inside_named_fragment_is_executed() {
        let schema = Schema::parse_and_validate(
            "type Query { hello(name: String!): String! }",
            "schema.graphql",
        )
        .unwrap();
        let request = Request::new(
            r#"
                query {
                    ...Greeting
                }

                fragment Greeting on Query {
                    hello(name: "Sheri")
                }
            "#,
        );
        let context = TestContext { greeting: "Hello" };

        let response = super::execute(&schema, &request, &TestDispatcher, &context).await;

        assert_eq!(
            to_value(response).unwrap(),
            json!({
                "data": {
                    "hello": "Hello, Sheri"
                }
            })
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn merged_fields_are_executed_once() {
        let schema =
            Schema::parse_and_validate("type Query { hello: String! }", "schema.graphql").unwrap();
        let request = Request::new(
            r#"
                query {
                    hello
                    ...Greeting
                }

                fragment Greeting on Query {
                    hello
                }
            "#,
        );
        let dispatcher = CountingDispatcher(AtomicUsize::new(0));

        let response = super::execute(&schema, &request, &dispatcher, &()).await;

        assert_eq!(
            to_value(response).unwrap(),
            json!({ "data": { "hello": "unexpected resolver call" } })
        );
        assert_eq!(dispatcher.0.load(Ordering::Relaxed), 1);
    }

    #[tokio::test(flavor = "current_thread")]
    async fn merged_field_error_is_reported_once_with_all_locations() {
        let schema =
            Schema::parse_and_validate("type Query { hello: String }", "schema.graphql").unwrap();
        let request = Request::new(
            "query {\n  greeting: hello\n  ...Again\n}\nfragment Again on Query {\n  greeting: hello\n}",
        );
        let dispatcher = CountingFailingDispatcher(AtomicUsize::new(0));

        let response = to_value(super::execute(&schema, &request, &dispatcher, &()).await).unwrap();

        assert_eq!(response["data"], json!({ "greeting": null }));
        let errors = response["errors"].as_array().unwrap();
        assert_eq!(errors.len(), 1);
        assert_eq!(errors[0]["path"], json!(["greeting"]));
        assert_eq!(errors[0]["extensions"]["code"], "GREETING_FAILED");
        assert_eq!(
            errors[0]["locations"],
            json!([
                { "line": 2, "column": 13 },
                { "line": 6, "column": 13 },
            ])
        );
        assert_eq!(dispatcher.0.load(Ordering::Relaxed), 1);
    }

    #[tokio::test(flavor = "current_thread")]
    async fn merged_field_completion_error_is_reported_once_with_all_locations() {
        let schema =
            Schema::parse_and_validate("type Query { hello: String }", "schema.graphql").unwrap();
        let request = Request::new(
            "query {\n  greeting: hello\n  ...Again\n}\nfragment Again on Query {\n  greeting: hello\n}",
        );
        let dispatcher = ValueDispatcher(json!({ "unexpected": true }));

        let response = to_value(super::execute(&schema, &request, &dispatcher, &()).await).unwrap();

        assert_eq!(response["data"], json!({ "greeting": null }));
        let errors = response["errors"].as_array().unwrap();
        assert_eq!(errors.len(), 1);
        assert_eq!(errors[0]["path"], json!(["greeting"]));
        assert_eq!(
            errors[0]["locations"],
            json!([
                { "line": 2, "column": 13 },
                { "line": 6, "column": 13 },
            ])
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn merged_non_null_completion_error_is_reported_once_with_all_locations() {
        let schema =
            Schema::parse_and_validate("type Query { hello: String! }", "schema.graphql").unwrap();
        let request = Request::new(
            "query {\n  greeting: hello\n  ...Again\n}\nfragment Again on Query {\n  greeting: hello\n}",
        );

        let response =
            to_value(super::execute(&schema, &request, &NullDispatcher, &()).await).unwrap();

        assert_eq!(response["data"], JsonValue::Null);
        let errors = response["errors"].as_array().unwrap();
        assert_eq!(errors.len(), 1);
        assert_eq!(errors[0]["path"], json!(["greeting"]));
        assert_eq!(
            errors[0]["locations"],
            json!([
                { "line": 2, "column": 13 },
                { "line": 6, "column": 13 },
            ])
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn merged_list_completion_error_is_reported_once_with_all_locations() {
        let schema =
            Schema::parse_and_validate("type Query { hello: [String] }", "schema.graphql").unwrap();
        let request = Request::new(
            "query {\n  greeting: hello\n  ...Again\n}\nfragment Again on Query {\n  greeting: hello\n}",
        );
        let dispatcher = ValueDispatcher(json!("not a list"));

        let response = to_value(super::execute(&schema, &request, &dispatcher, &()).await).unwrap();

        assert_eq!(response["data"], json!({ "greeting": null }));
        let errors = response["errors"].as_array().unwrap();
        assert_eq!(errors.len(), 1);
        assert_eq!(errors[0]["path"], json!(["greeting"]));
        assert_eq!(
            errors[0]["locations"],
            json!([
                { "line": 2, "column": 13 },
                { "line": 6, "column": 13 },
            ])
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn merged_object_completion_error_is_reported_once_with_all_locations() {
        let schema = Schema::parse_and_validate(
            "type Query { hello: User } type User { name: String! }",
            "schema.graphql",
        )
        .unwrap();
        let request = Request::new(
            "query {\n  greeting: hello { name }\n  ...Again\n}\nfragment Again on Query {\n  greeting: hello { name }\n}",
        );
        let dispatcher = ValueDispatcher(json!({ "name": "Sheri" }));

        let response = to_value(super::execute(&schema, &request, &dispatcher, &()).await).unwrap();

        assert_eq!(response["data"], json!({ "greeting": null }));
        let errors = response["errors"].as_array().unwrap();
        assert_eq!(errors.len(), 1);
        assert_eq!(errors[0]["path"], json!(["greeting"]));
        assert_eq!(
            errors[0]["locations"],
            json!([
                { "line": 2, "column": 13 },
                { "line": 6, "column": 13 },
            ])
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn merged_invalid_runtime_type_error_is_reported_once_with_all_locations() {
        let schema = Schema::parse_and_validate(
            "union Search = User | Organization \
             type Query { search: Search } \
             type User { name: String! } \
             type Organization { name: String! } \
             type Post { name: String! }",
            "schema.graphql",
        )
        .unwrap();
        let request = Request::new(
            "query {\n  result: search { __typename }\n  ...Again\n}\nfragment Again on Query {\n  result: search { __typename }\n}",
        );

        let response = to_value(
            super::execute(&schema, &request, &InvalidAbstractDispatcher("Post"), &()).await,
        )
        .unwrap();

        assert_eq!(response["data"], json!({ "result": null }));
        let errors = response["errors"].as_array().unwrap();
        assert_eq!(errors.len(), 1);
        assert_eq!(errors[0]["path"], json!(["result"]));
        assert_eq!(
            errors[0]["locations"],
            json!([
                { "line": 2, "column": 11 },
                { "line": 6, "column": 11 },
            ])
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn merged_unknown_runtime_type_error_is_reported_once_with_all_locations() {
        let schema = Schema::parse_and_validate(
            "type Query { viewer: User } type User { name: String! }",
            "schema.graphql",
        )
        .unwrap();
        let request = Request::new(
            "query {\n  result: viewer { __typename }\n  ...Again\n}\nfragment Again on Query {\n  result: viewer { __typename }\n}",
        );

        let response = to_value(
            super::execute(&schema, &request, &InvalidAbstractDispatcher("Ghost"), &()).await,
        )
        .unwrap();

        assert_eq!(response["data"], json!({ "result": null }));
        let errors = response["errors"].as_array().unwrap();
        assert_eq!(errors.len(), 1);
        assert_eq!(errors[0]["path"], json!(["result"]));
        assert_eq!(
            errors[0]["locations"],
            json!([
                { "line": 2, "column": 11 },
                { "line": 6, "column": 11 },
            ])
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn merged_object_fields_complete_every_subselection_once() {
        let schema = Schema::parse_and_validate(
            "type Query { viewer: User! } type User { name: String! id: ID! friend: User! }",
            "schema.graphql",
        )
        .unwrap();
        let request = Request::new(
            r#"
                query {
                    viewer { name friend { name } }
                    viewer { id friend { id } }
                }
            "#,
        );
        let friend_calls = Arc::new(AtomicUsize::new(0));
        let dispatcher = MergedObjectDispatcher {
            viewer_calls: AtomicUsize::new(0),
            friend_calls: Arc::clone(&friend_calls),
        };

        let response = super::execute(&schema, &request, &dispatcher, &()).await;

        assert_eq!(
            to_value(response).unwrap(),
            json!({
                "data": {
                    "viewer": {
                        "name": "Viewer",
                        "id": "viewer-id",
                        "friend": {
                            "name": "Friend",
                            "id": "friend-id",
                        },
                    },
                },
            })
        );
        assert_eq!(dispatcher.viewer_calls.load(Ordering::Relaxed), 1);
        assert_eq!(friend_calls.load(Ordering::Relaxed), 1);
    }

    #[tokio::test(flavor = "current_thread")]
    async fn skipped_field_is_not_executed() {
        let schema =
            Schema::parse_and_validate("type Query { hello: String! }", "schema.graphql").unwrap();
        let request = Request::new("query { hello @skip(if: true) }");
        let dispatcher = CountingDispatcher(AtomicUsize::new(0));

        let response = super::execute(&schema, &request, &dispatcher, &()).await;

        assert_eq!(to_value(response).unwrap(), json!({ "data": {} }));
        assert_eq!(dispatcher.0.load(Ordering::Relaxed), 0);
    }

    #[tokio::test(flavor = "current_thread")]
    async fn fragment_include_directive_uses_coerced_variable() {
        let schema =
            Schema::parse_and_validate("type Query { hello: String! }", "schema.graphql").unwrap();
        let variables = json!({ "include": false }).as_object().unwrap().clone();
        let request = Request::new(
            r#"
                query Greeting($include: Boolean!) {
                    ...GreetingFields @include(if: $include)
                }

                fragment GreetingFields on Query {
                    hello
                }
            "#,
        )
        .with_variables(variables);
        let dispatcher = CountingDispatcher(AtomicUsize::new(0));

        let response = super::execute(&schema, &request, &dispatcher, &()).await;

        assert_eq!(to_value(response).unwrap(), json!({ "data": {} }));
        assert_eq!(dispatcher.0.load(Ordering::Relaxed), 0);
    }

    #[tokio::test(flavor = "current_thread")]
    async fn non_applicable_inline_fragment_is_not_executed() {
        let schema = Schema::parse_and_validate(
            r#"
                interface Node { id: ID! }
                type Query implements Node { id: ID! }
                type Other implements Node { id: ID!, secret: String! }
            "#,
            "schema.graphql",
        )
        .unwrap();
        let request = Request::new(
            r#"
                query {
                    ... on Node {
                        ... on Other {
                            secret
                        }
                    }
                }
            "#,
        );
        let dispatcher = CountingDispatcher(AtomicUsize::new(0));

        let response = super::execute(&schema, &request, &dispatcher, &()).await;

        assert_eq!(to_value(response).unwrap(), json!({ "data": {} }));
        assert_eq!(dispatcher.0.load(Ordering::Relaxed), 0);
    }

    #[test]
    fn literal_argument_is_coerced_for_resolver_dispatch() {
        let schema = Schema::parse_and_validate(
            "type Query { hello(name: String!): String! }",
            "schema.graphql",
        )
        .unwrap();

        let request = Request::new(
            r#"
                query {
                    hello(name: "Sheri")
                }
            "#,
        );

        let prepared = prepare_request(&schema, &request).unwrap();
        let fields = super::collect_fields(&schema, &prepared);
        let field = fields.get("hello").unwrap()[0];

        let arguments =
            super::coerce_argument_values(&schema, &prepared, &[], field, &field.definition)
                .unwrap();

        assert_eq!(arguments.values.get("name"), Some(&json!("Sheri")));
    }

    #[test]
    fn finite_defaults_can_be_reused_across_list_items_and_arguments() {
        let arguments = coerce_arguments(
            "input R { next: R = {next: null} } \
             type Query { hello(first: [R], second: R): String }",
            Request::new("{ hello(first: [{}, {}], second: {}) }"),
        );

        assert_eq!(
            JsonValue::Object(arguments),
            json!({
                "first": [{ "next": { "next": null } }, { "next": { "next": null } }],
                "second": { "next": { "next": null } }
            })
        );
    }

    #[test]
    fn coerced_variable_is_used_as_an_argument() {
        let variables = json!({ "name": "Sheri" }).as_object().unwrap().clone();
        let request = Request::new(
            r#"
                query Greeting($name: String!) {
                    hello(name: $name)
                }
            "#,
        )
        .with_variables(variables);

        let arguments = coerce_arguments("type Query { hello(name: String!): String! }", request);

        assert_eq!(arguments.get("name"), Some(&json!("Sheri")));
    }

    #[test]
    fn omitted_nullable_argument_is_absent_and_explicit_null_is_preserved() {
        let arguments = coerce_arguments(
            "type Query { hello(omitted: String, explicit: String): String! }",
            Request::new("query { hello(explicit: null) }"),
        );

        assert!(!arguments.contains_key("omitted"));
        assert_eq!(arguments.get("explicit"), Some(&json!(null)));
    }

    #[test]
    fn argument_default_is_used_when_the_argument_is_omitted() {
        let arguments = coerce_arguments(
            r#"type Query { hello(name: String! = "Sheri"): String! }"#,
            Request::new("query { hello }"),
        );

        assert_eq!(arguments.get("name"), Some(&json!("Sheri")));
    }

    #[test]
    fn input_object_defaults_and_single_value_list_coercion_are_applied() {
        let arguments = coerce_arguments(
            r#"
                input GreetingInput {
                    name: String!
                    titles: [String!]! = "Detective"
                }

                type Query {
                    hello(input: GreetingInput!): String!
                }
            "#,
            Request::new(r#"query { hello(input: { name: "Sheri" }) }"#),
        );

        assert_eq!(
            arguments.get("input"),
            Some(&json!({
                "name": "Sheri",
                "titles": ["Detective"]
            }))
        );
    }

    #[test]
    fn variable_inside_an_input_object_uses_its_coerced_value() {
        let variables = json!({ "name": "Sheri" }).as_object().unwrap().clone();
        let request = Request::new(
            r#"
                query Greeting($name: String!) {
                    hello(input: { name: $name })
                }
            "#,
        )
        .with_variables(variables);

        let arguments = coerce_arguments(
            r#"
                input GreetingInput { name: String! }
                type Query { hello(input: GreetingInput!): String! }
            "#,
            request,
        );

        assert_eq!(arguments.get("input"), Some(&json!({ "name": "Sheri" })));
    }

    #[test]
    fn argument_coercion_error_contains_alias_aware_path() {
        let schema = Schema::parse_and_validate(
            r#"
                type Query {
                    hello(name: String! = "Sheri"): String!
                }
            "#,
            "schema.graphql",
        )
        .unwrap();

        let variables = json!({ "name": null }).as_object().unwrap().clone();
        let request = Request::new(
            r#"
                query Greeting($name: String) {
                    greeting: hello(name: $name)
                }
            "#,
        )
        .with_variables(variables);

        let prepared = prepare_request(&schema, &request).unwrap();
        let field = super::collect_fields(&schema, &prepared)
            .get("greeting")
            .unwrap()[0];
        let path = vec![ResponseDataPathSegment::Field(field.response_key().clone())];

        let error =
            super::coerce_argument_values(&schema, &prepared, &path, field, &field.definition)
                .unwrap_err();

        assert_eq!(error.path, path);
    }

    #[test]
    fn resolver_error_is_enriched_with_location_and_alias_aware_path() {
        let schema =
            Schema::parse_and_validate("type Query { hello: String }", "schema.graphql").unwrap();
        let request = Request::new(
            r#"
                query {
                    greeting: hello
                }
            "#,
        );

        let prepared = prepare_request(&schema, &request).unwrap();
        let field = super::collect_fields(&schema, &prepared)
            .get("greeting")
            .unwrap()[0];
        let path = vec![ResponseDataPathSegment::Field(field.response_key().clone())];
        let resolver_error =
            ResolverError::new("Greeting failed.").with_extension("code", "GREETING_FAILED");

        let error =
            super::resolver_error_to_graphql_error(&prepared, &[field], &path, resolver_error);

        assert_eq!(error.message, "Greeting failed.");
        assert_eq!(
            error.extensions.get("code"),
            Some(&json!("GREETING_FAILED"))
        );
        assert!(!error.locations.is_empty());
        assert_eq!(error.path, path);
    }

    #[tokio::test(flavor = "current_thread")]
    async fn handwritten_dispatch_receives_coerced_arguments_and_borrowed_context() {
        let schema = Schema::parse_and_validate(
            "type Query { hello(name: String!): String! }",
            "schema.graphql",
        )
        .unwrap();
        let request = Request::new(r#"query { hello(name: "Sheri") }"#);
        let greeting = String::from("Hello");
        let context = TestContext {
            greeting: &greeting,
        };
        let dispatcher = TestDispatcher;

        let future = super::execute(&schema, &request, &dispatcher, &context);

        assert_send(&future);

        let response = future.await;

        assert_eq!(
            to_value(response).unwrap(),
            json!({
                "data": {
                    "hello": "Hello, Sheri"
                }
            })
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn overlapping_requests_keep_their_context_values() {
        let schema = Schema::parse_and_validate(
            "type Query { hello(name: String!): String! }",
            "schema.graphql",
        )
        .unwrap();
        let request = Request::new(r#"query { hello(name: "Sheri") }"#);
        let first_context = TestContext { greeting: "Hello" };
        let second_context = TestContext {
            greeting: "Bonjour",
        };
        let dispatcher = TestDispatcher;

        let (first, second) = tokio::join!(
            super::execute(&schema, &request, &dispatcher, &first_context),
            super::execute(&schema, &request, &dispatcher, &second_context),
        );

        assert_eq!(
            to_value(first).unwrap(),
            json!({ "data": { "hello": "Hello, Sheri" } })
        );
        assert_eq!(
            to_value(second).unwrap(),
            json!({ "data": { "hello": "Bonjour, Sheri" } })
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn execution_succeeds_after_a_domain_error() {
        let schema =
            Schema::parse_and_validate("type Query { hello: String! }", "schema.graphql").unwrap();
        let request = Request::new("query { hello }");
        let dispatcher = RecoveringDispatcher(AtomicUsize::new(0));

        let failed = to_value(super::execute(&schema, &request, &dispatcher, &()).await).unwrap();
        let succeeded =
            to_value(super::execute(&schema, &request, &dispatcher, &()).await).unwrap();

        assert_eq!(failed.get("data"), Some(&JsonValue::Null));
        assert!(
            failed
                .get("errors")
                .and_then(JsonValue::as_array)
                .is_some_and(|errors| !errors.is_empty())
        );
        assert_eq!(succeeded, json!({ "data": { "hello": "Hello" } }));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn query_root_fields_make_progress_concurrently_and_preserve_response_order() {
        let schema = Schema::parse_and_validate(
            "type Query { first: String!, second: String! }",
            "schema.graphql",
        )
        .unwrap();
        let request = Request::new("query { second first }");
        let dispatcher = ConcurrentDispatcher {
            barrier: Arc::new(Barrier::new(2)),
        };

        let response = tokio::time::timeout(
            Duration::from_secs(1),
            super::execute(&schema, &request, &dispatcher, &()),
        )
        .await
        .expect("independent query root fields must make progress concurrently");

        assert_eq!(
            serde_json::to_string(&response).unwrap(),
            r#"{"data":{"second":"second","first":"first"}}"#
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn nested_fields_make_progress_concurrently_and_preserve_error_metadata() {
        let schema = Schema::parse_and_validate(
            "type Query { nested: Nested } type Nested { first: String!, second: String! }",
            "schema.graphql",
        )
        .unwrap();
        let request = Request::new("query { nested { failed: first second } }");
        let dispatcher = ConcurrentDispatcher {
            barrier: Arc::new(Barrier::new(2)),
        };

        let response = tokio::time::timeout(
            Duration::from_secs(1),
            super::execute(&schema, &request, &dispatcher, &()),
        )
        .await
        .expect("independent nested fields must make progress concurrently");
        let response = to_value(response).unwrap();

        assert_eq!(response["data"]["nested"], JsonValue::Null);
        assert_eq!(response["errors"][0]["message"], "Nested field failed.");
        assert_eq!(response["errors"][0]["path"], json!(["nested", "failed"]));
        assert_eq!(response["errors"][0]["extensions"]["code"], "NESTED_FAILED");
    }

    #[tokio::test(flavor = "current_thread")]
    async fn list_items_make_progress_concurrently_and_preserve_ordered_errors() {
        let schema = Schema::parse_and_validate(
            "type Query { items: [Item!]! } type Item { value: Int }",
            "schema.graphql",
        )
        .unwrap();
        let request = Request::new("query { results: items { value } }");
        let dispatcher = ConcurrentDispatcher {
            barrier: Arc::new(Barrier::new(3)),
        };

        let response = tokio::time::timeout(
            Duration::from_secs(1),
            super::execute(&schema, &request, &dispatcher, &()),
        )
        .await
        .expect("independent list items must make progress concurrently");
        let response = to_value(response).unwrap();

        assert_eq!(
            response["data"]["results"],
            json!([{ "value": null }, { "value": 1 }, { "value": null }])
        );
        assert_eq!(response["errors"].as_array().unwrap().len(), 2);
        assert_eq!(
            response["errors"][0]["path"],
            json!(["results", 0, "value"])
        );
        assert_eq!(response["errors"][0]["extensions"]["code"], "ITEM_FAILED");
        assert_eq!(
            response["errors"][1]["path"],
            json!(["results", 2, "value"])
        );
        assert_eq!(response["errors"][1]["extensions"]["code"], "ITEM_FAILED");
    }

    #[tokio::test(flavor = "current_thread")]
    async fn nested_mutation_fields_use_normal_concurrent_execution() {
        let schema = Schema::parse_and_validate(
            r#"
                type Query { noop: Boolean }
                type Mutation { update: Nested! }
                type Nested { first: String!, second: String! }
            "#,
            "schema.graphql",
        )
        .unwrap();
        let request = Request::new("mutation { update { second first } }");
        let dispatcher = ConcurrentDispatcher {
            barrier: Arc::new(Barrier::new(2)),
        };

        let response = tokio::time::timeout(
            Duration::from_secs(1),
            super::execute(&schema, &request, &dispatcher, &()),
        )
        .await
        .expect("nested mutation fields must use normal concurrent execution");

        assert_eq!(
            serde_json::to_string(&response).unwrap(),
            r#"{"data":{"update":{"second":"second","first":"first"}}}"#
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn dropping_request_future_drops_all_started_field_futures() {
        let schema = Schema::parse_and_validate(
            "type Query { first: String, second: String }",
            "schema.graphql",
        )
        .unwrap();
        let request = Request::new("query { first second }");
        let dispatcher = PendingDispatcher {
            started: AtomicUsize::new(0),
            dropped: AtomicUsize::new(0),
        };
        let mut execution = Box::pin(super::execute(&schema, &request, &dispatcher, &()));

        poll_fn(|context| {
            assert!(execution.as_mut().poll(context).is_pending());
            Poll::Ready(())
        })
        .await;
        drop(execution);

        assert_eq!(dispatcher.started.load(Ordering::Relaxed), 2);
        assert_eq!(dispatcher.dropped.load(Ordering::Relaxed), 2);
    }

    #[tokio::test(flavor = "current_thread")]
    async fn mutation_root_fields_execute_in_document_order() {
        let schema = Schema::parse_and_validate(
            r#"
                type Query { noop: String }
                type Mutation { first: String!, second: String! }
            "#,
            "schema.graphql",
        )
        .unwrap();
        let request = Request::new("mutation { first second }");
        let dispatcher = OrderingDispatcher(Mutex::new(Vec::new()));

        let response = super::execute(&schema, &request, &dispatcher, &()).await;

        assert_eq!(
            to_value(response).unwrap(),
            json!({ "data": { "first": "first", "second": "second" } })
        );
        assert_eq!(
            *dispatcher.0.lock().unwrap(),
            ["first".to_owned(), "second".to_owned()]
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn variable_input_reaches_dispatcher_and_produces_greeting() {
        let schema = Schema::parse_and_validate(
            "type Query { hello(name: String!): String! }",
            "schema.graphql",
        )
        .unwrap();
        let variables = json!({ "name": "Sheri" }).as_object().unwrap().clone();
        let request = Request::new(
            r#"
                query Greeting($name: String!) {
                    hello(name: $name)
                }
            "#,
        )
        .with_variables(variables);
        let greeting = String::from("Hello");
        let context = TestContext {
            greeting: &greeting,
        };

        let response = super::execute(&schema, &request, &TestDispatcher, &context).await;

        assert_eq!(
            to_value(response).unwrap(),
            json!({
                "data": {
                    "hello": "Hello, Sheri"
                }
            })
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn invalid_variable_inputs_do_not_invoke_dispatcher() {
        let schema = Schema::parse_and_validate(
            "type Query { hello(name: String!): String! }",
            "schema.graphql",
        )
        .unwrap();
        let cases = [
            ("missing", JsonMap::new()),
            ("null", json!({ "name": null }).as_object().unwrap().clone()),
            (
                "incompatible",
                json!({ "name": 42 }).as_object().unwrap().clone(),
            ),
        ];
        let dispatcher = CountingDispatcher(AtomicUsize::new(0));

        for (case, variables) in cases {
            let request = Request::new(
                r#"
                    query Greeting($name: String!) {
                        hello(name: $name)
                    }
                "#,
            )
            .with_variables(variables);

            let response = super::execute(&schema, &request, &dispatcher, &()).await;
            let response = to_value(response).unwrap();

            assert!(response.get("data").is_none(), "{case}");
            assert!(
                response
                    .get("errors")
                    .and_then(JsonValue::as_array)
                    .is_some_and(|errors| !errors.is_empty()),
                "{case}"
            );
            assert_eq!(dispatcher.0.load(Ordering::Relaxed), 0, "{case}");
        }
    }

    #[tokio::test(flavor = "current_thread")]
    async fn nested_custom_scalar_variables_are_validated_before_dispatch() {
        struct ScalarVariableDispatcher {
            resolver_calls: AtomicUsize,
            values: Mutex<Vec<JsonValue>>,
        }

        impl super::Dispatcher<()> for ScalarVariableDispatcher {
            fn validate_custom_scalar_variable(
                &self,
                scalar_name: &str,
                value: &JsonValue,
            ) -> Result<(), ResolverError> {
                assert_eq!(scalar_name, "Token");
                self.values.lock().unwrap().push(value.clone());
                if value == "bad" {
                    Err(ResolverError::new("invalid token").with_extension("code", "INVALID_TOKEN"))
                } else {
                    Ok(())
                }
            }

            async fn resolve<'a>(
                &'a self,
                _context: &'a (),
                _coordinate: FieldCoordinate<'a>,
                _arguments: &'a JsonMap,
            ) -> Result<ResolvedValue, ResolverError> {
                self.resolver_calls.fetch_add(1, Ordering::Relaxed);
                Ok(ResolvedValue::Json(json!("accepted")))
            }
        }

        let schema = Schema::parse_and_validate(
            r#"
                scalar Token
                input Payload { values: [Token!]! }
                type Query { inspect(payload: Payload!): String! }
            "#,
            "schema.graphql",
        )
        .unwrap();
        let variables = json!({
            "payload": { "values": ["ok", "bad"] },
        })
        .as_object()
        .unwrap()
        .clone();
        let request = Request::new("query($payload: Payload!) { inspect(payload: $payload) }")
            .with_variables(variables);
        let dispatcher = ScalarVariableDispatcher {
            resolver_calls: AtomicUsize::new(0),
            values: Mutex::new(Vec::new()),
        };

        let response = super::execute(&schema, &request, &dispatcher, &()).await;
        let response = to_value(response).unwrap();

        assert!(response.get("data").is_none());
        assert_eq!(response["errors"][0]["message"], "invalid token");
        assert_eq!(response["errors"][0]["extensions"]["code"], "INVALID_TOKEN");
        assert!(response["errors"][0].get("path").is_none());
        assert_eq!(dispatcher.resolver_calls.load(Ordering::Relaxed), 0);
        assert_eq!(
            *dispatcher.values.lock().unwrap(),
            [json!("ok"), json!("bad")]
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn resolver_error_on_non_null_root_field_propagates_null_to_data() {
        let schema =
            Schema::parse_and_validate("type Query { hello: String! }", "schema.graphql").unwrap();
        let request = Request::new("query { hello }");

        let response = super::execute(&schema, &request, &FailingDispatcher, &()).await;
        let response = to_value(response).unwrap();

        assert_eq!(response.get("data"), Some(&JsonValue::Null));
        assert_eq!(response["errors"][0]["message"], "Greeting failed.");
        assert_eq!(response["errors"][0]["path"], json!(["hello"]));
        assert_eq!(
            response["errors"][0]["extensions"]["code"],
            "GREETING_FAILED"
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn non_null_field_returning_null_adds_field_error() {
        let schema =
            Schema::parse_and_validate("type Query { hello: String! }", "schema.graphql").unwrap();
        let request = Request::new("query { hello }");

        let response = super::execute(&schema, &request, &NullDispatcher, &()).await;
        let response = to_value(response).unwrap();

        assert_eq!(response.get("data"), Some(&JsonValue::Null));

        let errors = response
            .get("errors")
            .and_then(JsonValue::as_array)
            .expect("a non-null field returning null must produce an error");

        assert_eq!(errors.len(), 1);
        assert!(
            errors[0]["message"]
                .as_str()
                .is_some_and(|message| !message.is_empty())
        );
        assert_eq!(errors[0]["path"], json!(["hello"]));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn nullable_list_preserves_null_items() {
        let response = execute_list_with_null_item("[String]").await;

        assert_eq!(
            response,
            json!({
                "data": {
                    "values": ["A", null, "B"]
                }
            })
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn non_null_list_preserves_nullable_items() {
        let response = execute_list_with_null_item("[String]!").await;

        assert_eq!(
            response,
            json!({
                "data": {
                    "values": ["A", null, "B"]
                }
            })
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn null_non_null_item_nullifies_nullable_list() {
        let response = execute_list_with_null_item("[String!]").await;

        assert_eq!(response["data"]["values"], JsonValue::Null);
        assert_eq!(response["errors"][0]["path"], json!(["values", 1]));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn null_non_null_item_propagates_through_non_null_list() {
        let response = execute_list_with_null_item("[String!]!").await;

        assert_eq!(response["data"], JsonValue::Null);
        assert_eq!(response["errors"][0]["path"], json!(["values", 1]));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn nullable_list_returning_non_list_becomes_null_with_execution_error() {
        let response = execute_value("[String]", json!("not a list")).await;

        assert_eq!(response["data"]["values"], JsonValue::Null);
        assert!(
            response["errors"][0]["message"]
                .as_str()
                .is_some_and(|message| !message.is_empty())
        );
        assert_eq!(response["errors"][0]["path"], json!(["values"]));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn non_null_list_returning_non_list_propagates_null_to_root() {
        let response = execute_value("[String]!", json!("not a list")).await;

        assert_eq!(response["data"], JsonValue::Null);
        assert!(
            response["errors"][0]["message"]
                .as_str()
                .is_some_and(|message| !message.is_empty())
        );
        assert_eq!(response["errors"][0]["path"], json!(["values"]));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn string_field_returning_incompatible_value_becomes_execution_error() {
        let schema =
            Schema::parse_and_validate("type Query { hello: String }", "schema.graphql").unwrap();
        let request = Request::new("query { hello }");
        let dispatcher = ValueDispatcher(json!({ "unexpected": true }));

        let response = super::execute(&schema, &request, &dispatcher, &()).await;
        let response = to_value(response).unwrap();

        assert_eq!(response["data"]["hello"], JsonValue::Null);
        assert!(
            response["errors"][0]["message"]
                .as_str()
                .is_some_and(|message| !message.is_empty())
        );
        assert_eq!(response["errors"][0]["path"], json!(["hello"]));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn leaf_completion_accepts_int_results_at_boundaries() {
        for value in [i32::MIN, 0, i32::MAX] {
            let response = execute_value("Int!", json!(value)).await;

            assert_eq!(response, json!({ "data": { "values": value } }));
        }
    }

    #[tokio::test(flavor = "current_thread")]
    async fn leaf_completion_accepts_finite_float_results() {
        for value in [0.0, -0.0, 1.5, -2.5, f64::MIN, f64::MAX] {
            let response = execute_value("Float!", json!(value)).await;

            assert_eq!(response, json!({ "data": { "values": value } }));
        }
    }

    #[tokio::test(flavor = "current_thread")]
    async fn leaf_completion_accepts_boolean_results() {
        for value in [false, true] {
            let response = execute_value("Boolean!", json!(value)).await;

            assert_eq!(response, json!({ "data": { "values": value } }));
        }
    }

    #[tokio::test(flavor = "current_thread")]
    async fn leaf_completion_preserves_id_string_results() {
        for value in ["001", "", "Sheri"] {
            let response = execute_value("ID!", json!(value)).await;

            assert_eq!(response, json!({ "data": { "values": value } }));
        }
    }

    #[tokio::test(flavor = "current_thread")]
    async fn leaf_completion_accepts_enum_members_and_rejects_unknown_results() {
        for field_type in ["Status", "Status!"] {
            let schema = Schema::parse_and_validate(
                format!("enum Status {{ OPEN CLOSED }} type Query {{ values: {field_type} }}"),
                "schema.graphql",
            )
            .unwrap();
            let request = Request::new("{ values }");

            for value in [json!("OPEN"), json!("CLOSED"), json!("UNKNOWN"), json!(42)] {
                let valid = value == json!("OPEN") || value == json!("CLOSED");
                let response = to_value(
                    super::execute(&schema, &request, &ValueDispatcher(value.clone()), &()).await,
                )
                .unwrap();

                if valid {
                    assert_eq!(response, json!({ "data": { "values": value } }));
                } else {
                    let expected_data = if field_type == "Status!" {
                        JsonValue::Null
                    } else {
                        json!({ "values": null })
                    };
                    assert_eq!(response["data"], expected_data);
                    assert_eq!(response["errors"].as_array().unwrap().len(), 1);
                    assert_eq!(response["errors"][0]["path"], json!(["values"]));
                }
            }
        }
    }

    #[tokio::test(flavor = "current_thread")]
    async fn leaf_completion_rejects_invalid_scalar_results() {
        for (name, value) in [
            ("Int", json!(i64::from(i32::MIN) - 1)),
            ("Int", json!(i64::from(i32::MAX) + 1)),
            ("Int", json!({ "invalid": true })),
            ("Float", json!({ "invalid": true })),
            ("Boolean", json!({ "invalid": true })),
            ("ID", json!({ "invalid": true })),
        ] {
            for non_null in [false, true] {
                let field_type = format!("{name}{}", if non_null { "!" } else { "" });
                let response = execute_value(&field_type, value.clone()).await;
                let expected_data = if non_null {
                    JsonValue::Null
                } else {
                    json!({ "values": null })
                };

                assert_eq!(response["data"], expected_data, "{field_type}: {value}");
                assert_eq!(response["errors"].as_array().unwrap().len(), 1);
                assert_eq!(response["errors"][0]["path"], json!(["values"]));
                assert!(
                    response["errors"][0]["message"]
                        .as_str()
                        .is_some_and(|message| !message.is_empty())
                );
            }
        }
    }

    #[tokio::test(flavor = "current_thread")]
    async fn leaf_completion_invalid_int_item_follows_list_nullability() {
        for (field_type, expected_data) in [
            ("[Int]", json!({ "values": [1, null, 2] })),
            ("[Int]!", json!({ "values": [1, null, 2] })),
            ("[Int!]", json!({ "values": null })),
            ("[Int!]!", JsonValue::Null),
        ] {
            let response = execute_value(field_type, json!([1, i64::from(i32::MAX) + 1, 2])).await;

            assert_eq!(response["data"], expected_data, "{field_type}");
            assert_eq!(response["errors"].as_array().unwrap().len(), 1);
            assert_eq!(response["errors"][0]["path"], json!(["values", 1]));
        }
    }

    #[tokio::test(flavor = "current_thread")]
    async fn embedded_conversion_errors_follow_list_nullability_and_preserve_extensions() {
        struct ConversionErrorDispatcher;

        impl super::Dispatcher<()> for ConversionErrorDispatcher {
            async fn resolve<'a>(
                &'a self,
                _: &'a (),
                _: FieldCoordinate<'a>,
                _: &'a JsonMap,
            ) -> Result<ResolvedValue, ResolverError> {
                Ok(ResolvedValue::List(vec![
                    ResolvedValue::Json(json!(1.5)),
                    ResolvedValue::Error(
                        ResolverError::new("Invalid Float result.")
                            .with_extension("code", "INVALID_FLOAT"),
                    ),
                    ResolvedValue::Json(json!(2.5)),
                ]))
            }
        }

        for (field_type, expected_data) in [
            ("[Float]", json!({ "numbers": [1.5, null, 2.5] })),
            ("[Float]!", json!({ "numbers": [1.5, null, 2.5] })),
            ("[Float!]", json!({ "numbers": null })),
            ("[Float!]!", JsonValue::Null),
        ] {
            let schema = Schema::parse_and_validate(
                format!("type Query {{ values: {field_type} }}"),
                "schema.graphql",
            )
            .unwrap();
            let request = Request::new("{ numbers: values }");
            let response =
                to_value(super::execute(&schema, &request, &ConversionErrorDispatcher, &()).await)
                    .unwrap();

            assert_eq!(response["data"], expected_data, "{field_type}");
            assert_eq!(response["errors"].as_array().unwrap().len(), 1);
            assert_eq!(response["errors"][0]["path"], json!(["numbers", 1]));
            assert_eq!(response["errors"][0]["message"], "Invalid Float result.");
            assert_eq!(response["errors"][0]["extensions"]["code"], "INVALID_FLOAT");
            assert_eq!(
                response["errors"][0]["locations"],
                json!([{ "line": 1, "column": 12 }])
            );
        }
    }

    #[tokio::test(flavor = "current_thread")]
    async fn unsupported_object_result_becomes_execution_error() {
        let schema = Schema::parse_and_validate(
            "type Query { viewer: User } type User { name: String! }",
            "schema.graphql",
        )
        .unwrap();
        let request = Request::new("query { viewer { name } }");
        let dispatcher = ValueDispatcher(json!({ "name": "Sheri" }));

        let response = super::execute(&schema, &request, &dispatcher, &()).await;
        let response = to_value(response).unwrap();

        assert_eq!(response["data"]["viewer"], JsonValue::Null);
        assert!(
            response["errors"][0]["message"]
                .as_str()
                .is_some_and(|message| !message.is_empty())
        );
        assert_eq!(response["errors"][0]["path"], json!(["viewer"]));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn invalid_abstract_runtime_type_becomes_one_execution_error() {
        let schema = Schema::parse_and_validate(
            "union Search = User | Organization \
             type Query { search: Search } \
             type User { name: String! } \
             type Organization { name: String! } \
             type Post { name: String! }",
            "schema.graphql",
        )
        .unwrap();
        let request = Request::new("{ result: search { __typename } }");

        let response = to_value(
            super::execute(&schema, &request, &InvalidAbstractDispatcher("Post"), &()).await,
        )
        .unwrap();

        assert_eq!(response["data"], json!({ "result": null }));
        assert_eq!(response["errors"].as_array().unwrap().len(), 1);
        assert_eq!(
            response["errors"][0]["message"],
            "Runtime Object type 'Post' is not valid for 'Search'."
        );
        assert_eq!(response["errors"][0]["path"], json!(["result"]));
        assert_eq!(
            response["errors"][0]["locations"],
            json!([{ "line": 1, "column": 11 }])
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn interface_selection_uses_object_field_nullability() {
        let schema = Schema::parse_and_validate(
            "interface Root { hello: String } \
             type Query implements Root { hello: String! }",
            "schema.graphql",
        )
        .unwrap();
        let request = Request::new("{ ... on Root { greeting: hello } }");

        let response =
            to_value(super::execute(&schema, &request, &NullDispatcher, &()).await).unwrap();

        assert_eq!(response["data"], JsonValue::Null);
        assert_eq!(response["errors"].as_array().unwrap().len(), 1);
        assert_eq!(response["errors"][0]["path"], json!(["greeting"]));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn interface_selection_uses_object_argument_default() {
        let schema = Schema::parse_and_validate(
            r#"
                interface Root { hello(name: String! = "Interface"): String! }
                type Query implements Root { hello(name: String! = "Sheri"): String! }
            "#,
            "schema.graphql",
        )
        .unwrap();
        let request = Request::new("{ ... on Root { hello } }");
        let context = TestContext { greeting: "Hello" };

        let response =
            to_value(super::execute(&schema, &request, &TestDispatcher, &context).await).unwrap();

        assert_eq!(response, json!({ "data": { "hello": "Hello, Sheri" } }));
    }

    #[test]
    fn variable_list_default_is_coerced_to_a_list() {
        let arguments = coerce_arguments(
            "type Query { hello(names: [String!]): String }",
            Request::new(r#"query($names: [String!] = "Sheri") { hello(names: $names) }"#),
        );

        assert_eq!(arguments.get("names"), Some(&json!(["Sheri"])));
    }

    #[test]
    fn variable_object_default_applies_input_field_defaults() {
        let arguments = coerce_arguments(
            r#"
                input GreetingInput { name: String! = "Sheri" }
                type Query { hello(input: GreetingInput): String }
            "#,
            Request::new("query($input: GreetingInput = {}) { hello(input: $input) }"),
        );

        assert_eq!(arguments.get("input"), Some(&json!({ "name": "Sheri" })));
    }

    #[test]
    fn nested_list_default_is_coerced_to_a_list() {
        let arguments = coerce_arguments(
            r#"
                input GreetingInput { names: [String!] = "Sheri" }
                type Query { hello(input: GreetingInput): String }
            "#,
            Request::new("query($input: GreetingInput = {}) { hello(input: $input) }"),
        );

        assert_eq!(arguments.get("input"), Some(&json!({ "names": ["Sheri"] })));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn dispatch_identifies_root_fields_by_schema_coordinate() {
        struct RecordingDispatcher(Mutex<Vec<String>>);

        impl super::Dispatcher<()> for RecordingDispatcher {
            async fn resolve<'a>(
                &'a self,
                _context: &'a (),
                coordinate: FieldCoordinate<'a>,
                _arguments: &'a JsonMap,
            ) -> Result<ResolvedValue, ResolverError> {
                self.0
                    .lock()
                    .unwrap()
                    .push(format!("{}.{}", coordinate.parent_type, coordinate.field));
                Ok(ResolvedValue::Json(json!("Hello")))
            }
        }

        let schema = Schema::parse_and_validate(
            "type Query { hello: String! } type Mutation { hello: String! }",
            "schema.graphql",
        )
        .unwrap();
        let dispatcher = RecordingDispatcher(Mutex::new(Vec::new()));

        for document in ["query { hello }", "mutation { hello }"] {
            let response =
                to_value(super::execute(&schema, &Request::new(document), &dispatcher, &()).await)
                    .unwrap();
            assert_eq!(response, json!({ "data": { "hello": "Hello" } }));
        }

        let calls = dispatcher.0.lock().unwrap();
        assert_eq!(
            calls.as_slice(),
            ["Query.hello".to_owned(), "Mutation.hello".to_owned()]
        );
    }

    #[tokio::test(flavor = "current_thread")]
    async fn typename_returns_root_type_without_invoking_dispatcher() {
        let schema = Schema::parse_and_validate(
            "schema { query: ReadRoot mutation: WriteRoot } \
             type ReadRoot { hello: String } type WriteRoot { hello: String }",
            "schema.graphql",
        )
        .unwrap();
        let dispatcher = CountingDispatcher(AtomicUsize::new(0));

        for (document, expected) in [
            ("query { kind: __typename }", "ReadRoot"),
            ("mutation { kind: __typename }", "WriteRoot"),
        ] {
            let response =
                to_value(super::execute(&schema, &Request::new(document), &dispatcher, &()).await)
                    .unwrap();

            assert_eq!(response, json!({ "data": { "kind": expected } }));
            assert_eq!(dispatcher.0.load(Ordering::Relaxed), 0);
        }
    }

    #[tokio::test(flavor = "current_thread")]
    async fn introspection_exposes_the_generated_schema_without_dispatch() {
        let schema = Schema::parse_and_validate(
            "type Query { hello(name: String!): String! }",
            "schema.graphql",
        )
        .unwrap();
        let dispatcher = CountingDispatcher(AtomicUsize::new(0));
        let request = Request::new(
            "{ __schema { queryType { name } types { kind name fields { name args { name type { kind name ofType { kind name } } } type { kind name ofType { kind name } } } } } }",
        );
        let response = to_value(super::execute(&schema, &request, &dispatcher, &()).await).unwrap();

        assert!(response.get("errors").is_none(), "{response:?}");
        assert_eq!(response["data"]["__schema"]["queryType"]["name"], "Query");
        let types = response["data"]["__schema"]["types"].as_array().unwrap();
        let query = types.iter().find(|ty| ty["name"] == "Query").unwrap();
        assert_eq!(query["kind"], "OBJECT");
        let hello = query["fields"]
            .as_array()
            .unwrap()
            .iter()
            .find(|field| field["name"] == "hello")
            .unwrap();
        assert_eq!(
            hello["type"],
            json!({ "kind": "NON_NULL", "name": null, "ofType": { "kind": "SCALAR", "name": "String" } })
        );
        assert_eq!(hello["args"][0]["name"], "name");
        assert_eq!(hello["args"][0]["type"], hello["type"]);
        assert_eq!(dispatcher.0.load(Ordering::Relaxed), 0);
    }

    #[tokio::test(flavor = "current_thread")]
    async fn introspection_exposes_custom_scalar_specified_by_url() {
        let schema = Schema::parse_and_validate(
            r#"
                scalar DateTime @specifiedBy(url: "https://example.com/date-time")
                type Query { now: DateTime }
            "#,
            "schema.graphql",
        )
        .unwrap();
        let dispatcher = CountingDispatcher(AtomicUsize::new(0));
        let request = Request::new(r#"{ __type(name: "DateTime") { kind name specifiedByURL } }"#);
        let response = to_value(super::execute(&schema, &request, &dispatcher, &()).await).unwrap();

        assert_eq!(
            response,
            json!({
                "data": {
                    "__type": {
                        "kind": "SCALAR",
                        "name": "DateTime",
                        "specifiedByURL": "https://example.com/date-time",
                    }
                }
            }),
        );
        assert_eq!(dispatcher.0.load(Ordering::Relaxed), 0);
    }

    #[tokio::test(flavor = "current_thread")]
    async fn type_lookup_honors_aliases_fragments_and_unknown_names() {
        let schema =
            Schema::parse_and_validate("type Query { hello: String! }", "schema.graphql").unwrap();
        let dispatcher = CountingDispatcher(AtomicUsize::new(0));
        let request = Request::new(
            "query($name: String!) { known: __type(name: $name) { ...TypeName } missing: __type(name: \"Missing\") { name } } fragment TypeName on __Type { name kind }",
        ).with_variables(json!({ "name": "Query" }).as_object().unwrap().clone());
        let response = to_value(super::execute(&schema, &request, &dispatcher, &()).await).unwrap();

        assert_eq!(
            response,
            json!({ "data": { "known": { "name": "Query", "kind": "OBJECT" }, "missing": null } })
        );
        assert_eq!(dispatcher.0.load(Ordering::Relaxed), 0);
    }

    #[tokio::test(flavor = "current_thread")]
    async fn disabled_introspection_rejects_schema_queries_before_dispatch() {
        let schema =
            Schema::parse_and_validate("type Query { hello: String! }", "schema.graphql").unwrap();
        let dispatcher = CountingDispatcher(AtomicUsize::new(0));
        let request = Request::new("{ __schema { queryType { name } } hello }");
        let response = to_value(
            super::execute_with_options(
                &schema,
                &request,
                &dispatcher,
                &(),
                super::ExecutionOptions {
                    introspection: false,
                },
            )
            .await,
        )
        .unwrap();

        assert!(response.get("data").is_none());
        assert_eq!(
            response["errors"][0]["locations"][0],
            json!({ "line": 1, "column": 3 })
        );
        assert_eq!(dispatcher.0.load(Ordering::Relaxed), 0);

        let typename = to_value(
            super::execute_with_options(
                &schema,
                &Request::new("{ __typename }"),
                &dispatcher,
                &(),
                super::ExecutionOptions {
                    introspection: false,
                },
            )
            .await,
        )
        .unwrap();
        assert_eq!(typename, json!({ "data": { "__typename": "Query" } }));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn introspection_and_resolver_fields_share_one_response() {
        let schema =
            Schema::parse_and_validate("type Query { hello: String! }", "schema.graphql").unwrap();
        let dispatcher = CountingDispatcher(AtomicUsize::new(0));
        let request = Request::new("{ hello schema: __schema { queryType { name } } }");
        let response = to_value(super::execute(&schema, &request, &dispatcher, &()).await).unwrap();

        assert_eq!(
            response,
            json!({ "data": { "hello": "unexpected resolver call", "schema": { "queryType": { "name": "Query" } } } })
        );
        assert_eq!(dispatcher.0.load(Ordering::Relaxed), 1);
    }

    #[tokio::test(flavor = "current_thread")]
    async fn schema_query_returns_nested_type_metadata() {
        let schema = Schema::parse_and_validate(
            "type Query { hello(name: String!): String! }",
            "schema.graphql",
        )
        .unwrap();
        let request = Request::new(
            r#"
            query IntrospectionQuery {
              __schema {
                queryType { name }
                mutationType { name }
                types { ...FullType }
                directives { name locations args { name type { ...TypeRef } } }
              }
            }
            fragment FullType on __Type {
              kind name description
              fields(includeDeprecated: true) {
                name args { name type { ...TypeRef } }
                type { ...TypeRef }
              }
              inputFields { name type { ...TypeRef } }
              enumValues(includeDeprecated: true) { name }
              interfaces { name }
              possibleTypes { name }
            }
            fragment TypeRef on __Type {
              kind name ofType { kind name ofType { kind name } }
            }
            "#,
        );
        let response =
            to_value(super::execute(&schema, &request, &NullDispatcher, &()).await).unwrap();

        assert!(response.get("errors").is_none(), "{response:?}");
        assert_eq!(response["data"]["__schema"]["queryType"]["name"], "Query");
        assert_eq!(
            response["data"]["__schema"]["mutationType"],
            JsonValue::Null
        );
        assert!(
            response["data"]["__schema"]["directives"]
                .as_array()
                .is_some_and(|directives| !directives.is_empty())
        );
        let query = response["data"]["__schema"]["types"]
            .as_array()
            .unwrap()
            .iter()
            .find(|ty| ty["name"] == "Query")
            .unwrap();
        let hello = &query["fields"][0];
        assert_eq!(hello["name"], "hello");
        assert_eq!(hello["args"][0]["name"], "name");
        assert_eq!(hello["type"]["kind"], "NON_NULL");
        assert_eq!(hello["type"]["ofType"]["name"], "String");
    }

    // TODO: Temporary unsupported-feature contract: remove this rejection test when
    // subscription execution is implemented and replace it with response-stream tests.
    #[tokio::test(flavor = "current_thread")]
    async fn unsupported_subscription_is_rejected_before_dispatch() {
        let schema = Schema::parse_and_validate(
            "type Query { hello: String } type Subscription { hello: String }",
            "schema.graphql",
        )
        .unwrap();
        let request = Request::new("subscription { hello }");
        let dispatcher = CountingDispatcher(AtomicUsize::new(0));

        let response = to_value(super::execute(&schema, &request, &dispatcher, &()).await).unwrap();

        assert_eq!(dispatcher.0.load(Ordering::Relaxed), 0);
        assert!(response.get("data").is_none());
        assert!(
            response["errors"]
                .as_array()
                .is_some_and(|errors| !errors.is_empty())
        );
    }
}