fraiseql-server 2.17.0

//! GraphQL HTTP handlers and execution logic.

use std::{sync::atomic::Ordering, time::Instant};

use axum::{
    Json,
    extract::{Query, State},
    http::HeaderMap,
};
use fraiseql_core::{
    apq::{ApqMetrics, ApqStorage},
    security::SecurityContext,
};
use fraiseql_error::FraiseQLError;
use tracing::{debug, error, warn};

use super::{
    app_state::AppState,
    request::{GraphQLGetParams, GraphQLRequest, GraphQLResponse},
};
use crate::{
    error::{ErrorResponse, GraphQLError},
    extractors::{OptionalSecurityContext, PeerIp},
    tracing_utils,
};

/// GraphQL HTTP handler for POST requests.
///
/// Handles POST requests to the GraphQL endpoint:
/// 1. Extract W3C trace context from traceparent header (if present)
/// 2. Validate GraphQL request (depth, complexity)
/// 3. Parse GraphQL request body
/// 4. Execute query via Executor with optional `SecurityContext`
/// 5. Return GraphQL response with proper error formatting
///
/// Tracks execution timing and operation name for monitoring.
/// Provides GraphQL spec-compliant error responses.
/// Supports W3C Trace Context for distributed tracing.
/// Supports OIDC authentication for RLS policy evaluation.
///
/// # Errors
///
/// Returns appropriate HTTP status codes based on error type.
#[tracing::instrument(skip_all, fields(operation_name))]
#[doc(hidden)] // Internal-pub: axum route handler wired via Server::route; downstream uses Server::serve(), not this fn directly.
pub async fn graphql_handler(
    State(state): State<AppState>,
    headers: HeaderMap,
    PeerIp(peer_ip): PeerIp,
    OptionalSecurityContext(security_context): OptionalSecurityContext,
    // #958: an SSE `@stream` delivery outlives its request, so it re-checks the
    // principal per batch — which needs the decoded `jti`/`iat` the auth middleware
    // put in the extensions. Absent for anonymous and service-account callers, which
    // is exactly the signal that no revocation re-check applies.
    token_claims: Option<axum::Extension<crate::middleware::oidc_auth::SessionTokenClaims>>,
    Json(request): Json<GraphQLRequest>,
) -> Result<axum::response::Response, ErrorResponse> {
    // #1562: every stage below reads this request's one parse of its document, and a
    // document the executor parsed before is not parsed again.
    let documents = state.executor().documents().clone();
    documents
        .scope(async move {
            // Extract trace context from W3C headers
            let trace_context = tracing_utils::extract_trace_context(&headers);
            if trace_context.is_some() {
                debug!("Extracted W3C trace context from incoming request");
            }

            if security_context.is_some() {
                debug!("Authenticated request with security context");
            }

            // GraphQL-over-SSE (#387): opt-in, negotiated by Accept. This branch runs
            // INSIDE the authenticated route, so it inherits the full middleware stack.
            if let Some(wire) = state
                .graphql_incremental_enabled
                .then(|| incremental::negotiate(&headers))
                .flatten()
            {
                return Ok(with_caching(
                    Box::pin(sse::handle_sse(
                        state,
                        wire,
                        headers,
                        peer_ip,
                        security_context,
                        token_claims.map(|axum::Extension(claims)| claims),
                        request,
                        Method::Post,
                    ))
                    .await,
                ));
            }

            Ok(answer(
                execute_graphql_request(
                    state,
                    request,
                    Method::Post,
                    trace_context,
                    security_context,
                    &headers,
                    &peer_ip,
                )
                .await,
            ))
        })
        .await
}

/// GraphQL HTTP handler for GET requests.
///
/// Handles GET requests to the GraphQL endpoint per the GraphQL over HTTP spec.
/// Query parameters:
/// - `query`: Required, the GraphQL query string (URL-encoded)
/// - `variables`: Optional, JSON-encoded variables object (URL-encoded)
/// - `operationName`: Optional, name of the operation to execute
///
/// Supports W3C Trace Context via traceparent header for distributed tracing.
///
/// Example:
/// ```text
/// GET /graphql?query={users{id,name}}&variables={"limit":10}
/// ```
///
/// # Errors
///
/// Returns `413 Payload Too Large` (via `ErrorResponse`) when the query string
/// exceeds `AppState::max_get_query_bytes` (default 100 `KiB`, configurable via
/// `ServerConfig::max_get_query_bytes`). Returns other HTTP status codes for
/// additional error conditions.
///
/// # Note
///
/// Per GraphQL over HTTP spec, GET requests should only be used for queries,
/// not mutations (which should use POST). This handler does not enforce that
/// restriction but logs a warning for mutation-like queries.
#[tracing::instrument(skip_all, fields(operation_name))]
#[doc(hidden)] // Internal-pub: axum route handler wired via Server::route; downstream uses Server::serve(), not this fn directly.
pub async fn graphql_get_handler(
    State(state): State<AppState>,
    headers: HeaderMap,
    PeerIp(peer_ip): PeerIp,
    OptionalSecurityContext(security_context): OptionalSecurityContext,
    // #958: see `graphql_handler` — the SSE branch needs the token's revocation claims.
    token_claims: Option<axum::Extension<crate::middleware::oidc_auth::SessionTokenClaims>>,
    Query(params): Query<GraphQLGetParams>,
) -> Result<axum::response::Response, ErrorResponse> {
    // #1562: every stage below reads this request's one parse of its document, and a
    // document the executor parsed before is not parsed again.
    let documents = state.executor().documents().clone();
    documents.scope(async move {
        // Reject oversized GET queries early to prevent DoS via query parsing.
        let max_get_bytes = state.max_get_query_bytes;
        if params.query.as_ref().is_some_and(|q| q.len() > max_get_bytes) {
            return Err(ErrorResponse::from_error(GraphQLError::payload_too_large(format!(
                "GET query string exceeds maximum allowed length ({max_get_bytes} bytes)"
            ))));
        }

        // Parse variables from JSON string.
        // Apply the same size cap as the query string — the URL-length limit imposed
        // by reverse proxies/OS is real but not enforced by axum itself, so we guard
        // explicitly to prevent parser DoS from a very large variables value.
        let variables = if let Some(vars_str) = params.variables {
            if vars_str.len() > max_get_bytes {
                return Err(ErrorResponse::from_error(GraphQLError::payload_too_large(format!(
                    "GET variables string exceeds maximum allowed length ({max_get_bytes} bytes)"
                ))));
            }
            match serde_json::from_str::<serde_json::Value>(&vars_str) {
                Ok(v) => Some(v),
                Err(e) => {
                    // Log the fault, never the payload. `variables` is client-supplied
                    // and may hold PII or bearer tokens; emitting up to
                    // `max_get_query_bytes` of it at `warn!` put that into every log
                    // sink the deployment ships to (#730). The serde error already
                    // carries the line/column, which is what a diagnosis needs.
                    warn!(
                        error = %e,
                        variables_bytes = vars_str.len(),
                        "Failed to parse variables JSON in GET request"
                    );
                    return Err(ErrorResponse::from_error(GraphQLError::request(format!(
                        "Invalid variables JSON: {e}"
                    ))));
                },
            }
        } else {
            None
        };

        // #1564: an APQ hash travels as JSON-encoded `extensions`, which is what lets a CDN
        // key the request by its URL. Same cap and same handling as `variables`.
        let extensions = if let Some(ext_str) = params.extensions {
            if ext_str.len() > max_get_bytes {
                return Err(ErrorResponse::from_error(GraphQLError::payload_too_large(format!(
                    "GET extensions string exceeds maximum allowed length ({max_get_bytes} bytes)"
                ))));
            }
            match serde_json::from_str::<serde_json::Value>(&ext_str) {
                Ok(v) => Some(v),
                Err(e) => {
                    warn!(
                        error = %e,
                        extensions_bytes = ext_str.len(),
                        "Failed to parse extensions JSON in GET request"
                    );
                    return Err(ErrorResponse::from_error(GraphQLError::request(format!(
                        "Invalid extensions JSON: {e}"
                    ))));
                },
            }
        } else {
            None
        };

        // A mutation over GET is refused with 405 once the document is resolved
        // (`stages::resolve_query_body`): with an APQ hash the document is not in the URL,
        // so it cannot be judged before.

        let trace_context = tracing_utils::extract_trace_context(&headers);
        if trace_context.is_some() {
            debug!("Extracted W3C trace context from incoming request");
        }

        let request = GraphQLRequest {
            query: params.query,
            variables,
            operation_name: params.operation_name,
            extensions,
            document_id: None,
        };

        if security_context.is_some() {
            debug!("Authenticated GET request with security context");
        }

        // GraphQL-over-SSE (#387): the GET arm is what an `EventSource` client can
        // reach (it cannot POST). A mutation is refused once its document is resolved.
        if let Some(wire) = state
            .graphql_incremental_enabled
            .then(|| incremental::negotiate(&headers))
            .flatten()
        {
            return Ok(with_caching(
                Box::pin(sse::handle_sse(
                    state,
                    wire,
                    headers,
                    peer_ip,
                    security_context,
                    token_claims.map(|axum::Extension(claims)| claims),
                    request,
                    Method::Get,
                ))
                .await,
            ));
        }

        Ok(answer(
            execute_graphql_request(
                state,
                request,
                Method::Get,
                trace_context,
                security_context,
                &headers,
                &peer_ip,
            )
            .await,
        ))
    })
    .await
}

/// The HTTP `QUERY` method (RFC 10008), as a byte string.
///
/// **This is the single seam to swap for native support.** Neither `http` 1.4.x nor
/// axum 0.8.x exposes `Method::QUERY` / `MethodFilter::QUERY` yet
/// (hyperium/http#798, tokio-rs/axum#3799). When they ship, replace this constant
/// and the `MethodRouter::fallback` wiring in `server/routing/graphql.rs` with the
/// typed filter — those two places are the whole workaround.
pub(crate) const HTTP_QUERY_METHOD: &str = "QUERY";

/// `QUERY /graphql` — RFC 10008 (#508).
///
/// Mounted only when `enable_http_query = true`, as a `MethodRouter` fallback, so
/// `GET` and `POST` dispatch is untouched. The body is parsed exactly like a `POST`,
/// then the operation is **restricted to queries**.
///
/// # Why the restriction is mandatory
///
/// `QUERY` is defined as safe and idempotent, which entitles any intermediary —
/// proxy, CDN, retry layer — to replay it. A mutation carried over it could
/// therefore be silently executed more than once. Refusing non-query operations is
/// what makes accepting the method safe at all.
///
/// # Errors
///
/// - `405` when the request method is not `QUERY` (any other unmatched method).
/// - `405` when the document is a `mutation` or `subscription`.
/// - `400` when the body is not a valid `GraphQLRequest`.
pub async fn graphql_query_method_handler(
    State(state): State<AppState>,
    method: axum::http::Method,
    headers: HeaderMap,
    PeerIp(peer_ip): PeerIp,
    OptionalSecurityContext(security_context): OptionalSecurityContext,
    body: axum::body::Bytes,
) -> Result<axum::response::Response, ErrorResponse> {
    // #1562: every stage below reads this request's one parse of its document, and a
    // document the executor parsed before is not parsed again.
    let documents = state.executor().documents().clone();
    documents
        .scope(async move {
            // This handler is a method-router fallback, so it also catches PUT/DELETE/…
            // Anything that is not QUERY gets the same 405 the router would have produced.
            if method.as_str() != HTTP_QUERY_METHOD {
                return Err(ErrorResponse::from_error(GraphQLError::method_not_allowed(
                    "Method not allowed on the GraphQL endpoint",
                )));
            }

            let request: GraphQLRequest = serde_json::from_slice(&body).map_err(|e| {
                ErrorResponse::from_error(GraphQLError::request(format!("Invalid QUERY body: {e}")))
            })?;

            // Queries-only, judged on the resolved document (`stages::resolve_query_body`), with
            // the parser the executor uses, so the gate cannot disagree with what would
            // actually execute. A document that does not parse surfaces the executor's parse
            // error — it executes nothing, so no state can change.
            let trace_context = tracing_utils::extract_trace_context(&headers);

            Ok(answer(
                execute_graphql_request(
                    state,
                    request,
                    Method::Query,
                    trace_context,
                    security_context,
                    &headers,
                    &peer_ip,
                )
                .await,
            ))
        })
        .await
}

/// The method a GraphQL request arrived on, which decides what it may run, whether it
/// may register a persisted query, and whether its response may be stored (#1564).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum Method {
    /// `POST`: any operation.
    Post,
    /// `GET`: no mutation, no persisted-query registration; a response may be cached.
    Get,
    /// `QUERY` (RFC 10008): queries only, no persisted-query registration.
    Query,
}

/// An executed request's response, with the caching headers it may carry.
pub(super) struct Answer {
    /// The GraphQL response.
    pub(super) response: GraphQLResponse,
    /// `Cache-Control` and, on a reusable response, `Vary`.
    pub(super) headers:  HeaderMap,
}

/// The HTTP response for an executed request: its body with its caching headers, or the
/// error with `no-store`.
fn answer(result: Result<Answer, ErrorResponse>) -> axum::response::Response {
    match result {
        Ok(Answer { response, headers }) => {
            let mut http = axum::response::IntoResponse::into_response(response);
            http.headers_mut().extend(headers);
            http
        },
        Err(error) => with_caching(Err(error)),
    }
}

/// A response no cache may store: an error, or an SSE stream.
fn with_caching(
    result: Result<axum::response::Response, ErrorResponse>,
) -> axum::response::Response {
    let mut http = match result {
        Ok(http) => http,
        Err(error) => axum::response::IntoResponse::into_response(error),
    };
    http_cache::apply(http.headers_mut(), http_cache::Caching::NoStore, None);
    http
}

/// Returns the operation type when `query` is **not** a plain query — i.e.
/// `Some("mutation")` or `Some("subscription")`. `None` for queries and for
/// documents that fail to parse (which execute nothing).
pub(crate) fn non_query_operation(query: &str) -> Option<&'static str> {
    let parsed = fraiseql_core::graphql::parse_query(query).ok()?;
    match parsed.operation_type.as_str() {
        "mutation" => Some("mutation"),
        "subscription" => Some("subscription"),
        _ => None,
    }
}

/// Extract the mutation name from a GraphQL query string, if the operation is a mutation.
///
/// Returns `Some(root_field_name)` when the query parses successfully and the operation
/// type is `"mutation"`. Returns `None` for queries, subscriptions, or parse errors.
///
/// Used to look up before-mutation hooks: a single `HashMap::get` on the trigger
/// registry — O(1) and allocation-free when no hooks are registered.
pub(crate) fn detect_mutation_name(query: &str) -> Option<String> {
    let parsed = fraiseql_core::graphql::parse_query(query).ok()?;
    if parsed.operation_type == "mutation" {
        Some(parsed.root_field)
    } else {
        None
    }
}

/// Extract client IP address from headers.
///
/// # Security
///
/// Does NOT trust X-Forwarded-For or X-Real-IP headers, as these are trivially
/// spoofable by attackers to bypass rate limiting. Returns "unknown" as a safe
/// fallback — callers requiring real IPs should use `ConnectInfo<SocketAddr>`
/// or `ProxyConfig::extract_client_ip()` with validated proxy chains.
#[cfg(feature = "auth")]
#[allow(dead_code)] // Reason: used only in tests that verify spoofable headers are ignored
pub(crate) fn extract_ip_from_headers(_headers: &HeaderMap) -> String {
    // SECURITY: Spoofable headers removed. Use ConnectInfo<SocketAddr> or
    // ProxyConfig::extract_client_ip() for validated IP extraction.
    "unknown".to_string()
}

/// Extract the APQ SHA-256 hash from the `extensions.persistedQuery` field, if present.
pub(crate) fn extract_apq_hash(extensions: Option<&serde_json::Value>) -> Option<&str> {
    extensions?.get("persistedQuery")?.get("sha256Hash")?.as_str()
}

/// Extract a trusted document ID from the request.
///
/// Supports three formats:
/// 1. `documentId` (GraphQL over HTTP spec)
/// 2. `extensions.persistedQuery.sha256Hash` (Apollo APQ format)
/// 3. `extensions.doc_id` (Relay format)
fn extract_document_id(request: &GraphQLRequest) -> Option<String> {
    // 1. Top-level documentId field (GraphQL over HTTP spec)
    if let Some(ref doc_id) = request.document_id {
        return Some(doc_id.clone());
    }
    // 2. Extensions-based formats
    if let Some(ext) = request.extensions.as_ref() {
        // Relay format: extensions.doc_id
        if let Some(doc_id) = ext.get("doc_id").and_then(|v| v.as_str()) {
            return Some(doc_id.to_string());
        }
        // Apollo APQ format: extensions.persistedQuery.sha256Hash (also used for APQ)
        if let Some(hash) = ext
            .get("persistedQuery")
            .and_then(|pq| pq.get("sha256Hash"))
            .and_then(|h| h.as_str())
        {
            return Some(hash.to_string());
        }
    }
    None
}

/// Resolve an APQ request: look up or register a persisted query.
///
/// Returns the resolved query body, or an error if the query is not found and no body was
/// provided (the client should resend with the full body). A hash sent with its body is
/// verified, and stored only when `register` (a `POST`): on a method that may not write it
/// is executed as sent.
///
/// # Errors
///
/// Returns [`ErrorResponse`] if the hash doesn't match the body, or if the
/// hash is unknown and no query body was provided (client must retry with full body).
pub(crate) async fn resolve_apq(
    apq_store: &dyn ApqStorage,
    apq_metrics: &ApqMetrics,
    hash: &str,
    query_body: Option<&str>,
    register: bool,
) -> Result<String, ErrorResponse> {
    if let Some(body) = query_body {
        // Hash + body present: verify, and register where the method may write.
        if !fraiseql_core::apq::verify_hash(body, hash) {
            apq_metrics.record_error();
            return Err(ErrorResponse::from_error(GraphQLError::persisted_query_mismatch()));
        }
        if !register {
            return Ok(body.to_owned());
        }
        // Store the query (best-effort; log on failure).
        if let Err(e) = apq_store.set(hash.to_owned(), body.to_owned()).await {
            warn!(error = %e, "Failed to store APQ query — proceeding without caching");
            apq_metrics.record_error();
        } else {
            apq_metrics.record_store();
        }
        Ok(body.to_owned())
    } else {
        // Hash only: look up.
        match apq_store.get(hash).await {
            Ok(Some(stored)) => {
                apq_metrics.record_hit();
                Ok(stored)
            },
            Ok(None) => {
                apq_metrics.record_miss();
                Err(ErrorResponse::from_error(GraphQLError::persisted_query_not_found()))
            },
            Err(e) => {
                warn!(error = %e, "APQ store lookup failed — treating as miss");
                apq_metrics.record_error();
                Err(ErrorResponse::from_error(GraphQLError::persisted_query_not_found()))
            },
        }
    }
}

/// Shared GraphQL execution logic for both GET and POST handlers.
#[tracing::instrument(skip_all, fields(operation_name = request.operation_name.as_deref().unwrap_or("anonymous")))]
async fn execute_graphql_request(
    state: AppState,
    mut request: GraphQLRequest,
    method: Method,
    #[cfg(feature = "federation")] _trace_context: Option<
        fraiseql_core::federation::FederationTraceContext,
    >,
    #[cfg(not(feature = "federation"))] _trace_context: Option<()>,
    security_context: Option<SecurityContext>,
    headers: &HeaderMap,
    peer_ip: &str,
) -> Result<Answer, ErrorResponse> {
    // ── Who is asking ────────────────────────────────────────────────────────
    //
    // Every `await` on a stage is `Box::pin`ned. An awaited `async fn` embeds its
    // whole state machine in the caller's future, so the nine stages compose into
    // a type deep enough that rustc 1.97 refuses to compute the handler's layout
    // ("queries overflow the depth limit"). Boxing also keeps the request future
    // small enough for `clippy::large_futures`, which this crate has tripped
    // repeatedly on `Server`-shaped futures. The cost is one allocation per stage
    // per request, against a database round-trip.
    let mut security_context =
        Box::pin(stages::authenticate(&state, headers, security_context)).await?;
    security_context = stages::stamp_trace_context(headers, security_context);
    #[cfg(feature = "auth")]
    Box::pin(stages::enrich_identity(&state, &mut security_context)).await?;

    // ── What they are asking ─────────────────────────────────────────────────
    let query = Box::pin(stages::resolve_query_body(&state, &mut request, method)).await?;

    let start_time = Instant::now();
    let metrics = &state.metrics;
    metrics.queries_total.fetch_add(1, Ordering::Relaxed);

    // Reason (F041): per-request execution log moved to `debug!`. At >100 RPS
    // this event drowns the operator's `info!`-level signal-to-noise ratio.
    // `info!` is reserved for startup/shutdown/schema-reload events.
    debug!(
        query_length = query.len(),
        has_variables = request.variables.is_some(),
        operation_name = ?request.operation_name,
        "Executing GraphQL query"
    );

    // ── Whether they may ─────────────────────────────────────────────────────
    stages::enforce_introspection_policy(&state, &query, security_context.as_ref())?;
    stages::validate_request(&state, &query, &request, peer_ip)?;

    #[cfg(feature = "federation")]
    let cb_entity_types =
        stages::check_federation_circuit_breakers(&state, &query, request.variables.as_ref())?;

    // Resolve the tenant key — the token's tenant for an authenticated caller, the
    // client hints otherwise — through the same seam the MCP transport uses (#858). A
    // refusal keeps its own class: a header naming a tenant the token is not bound to
    // is FORBIDDEN, a malformed header a validation error.
    let tenant_key =
        super::tenant_dispatch::resolve_tenant_key(&state, security_context.as_ref(), headers)
            .map_err(|e| ErrorResponse::from_error(GraphQLError::from_fraiseql_error(&e)))?;

    // ── Idempotency (#747) ───────────────────────────────────────────────────
    // A mutation carrying an `Idempotency-Key` header executes at most once per
    // key: a repeat with the same body replays the stored response, a repeat
    // with a different body is a 409 conflict. This is the receiving half of
    // the saga at-least-once dispatch contract — a peer coordinator re-sends a
    // step's mutation under the same key after an ambiguous failure (timeout,
    // connection reset after send) or a crash-recovery replay, and this check
    // is what turns those re-sends into one logical effect. Scoped by tenant and
    // by principal, so a key never replays across tenants nor to a caller other
    // than the one whose gates produced the stored response; mutations only
    // arrive via POST (the GET handler rejects them).
    let idempotency_key = if detect_mutation_name(&query).is_some() {
        headers.get("idempotency-key").and_then(|v| v.to_str().ok()).map(|client_key| {
            let scope = crate::routes::idempotency::IdempotencyScope {
                tenant:    tenant_key.clone(),
                principal: security_context.as_ref().map(|c| c.user_id.to_string()),
                method:    "POST".to_string(),
                path:      "/graphql".to_string(),
            };
            let body_hash = crate::routes::idempotency::hash_body(&serde_json::json!({
                "query": query,
                "variables": request.variables,
                "operationName": request.operation_name,
            }));
            (scope.key(client_key), body_hash)
        })
    } else {
        None
    };
    if let Some((ref key, body_hash)) = idempotency_key {
        match state.idempotency_store.check(key, body_hash).await {
            crate::routes::idempotency::IdempotencyCheck::Replay(stored) => {
                debug!("Replaying stored response for repeated Idempotency-Key mutation");
                let mut headers = HeaderMap::new();
                http_cache::apply(&mut headers, http_cache::Caching::NoStore, None);
                return Ok(Answer {
                    response: GraphQLResponse {
                        body: stored.body.unwrap_or(serde_json::Value::Null),
                    },
                    headers,
                });
            },
            crate::routes::idempotency::IdempotencyCheck::Conflict => {
                return Err(ErrorResponse::from_error(GraphQLError::idempotency_conflict()));
            },
            crate::routes::idempotency::IdempotencyCheck::New => {},
        }
    }

    // The `before:mutation` chain used to run here, once per request, keyed on the
    // first root field and handed this `variables` map. That shape was bypassable
    // three ways (#1327), so enforcement moved into the engine: the chain now runs
    // from `execute_mutation_impl` — per executed root, in document order, with the
    // arguments the write binds from — on every transport. It is installed on the
    // executor's `RuntimeConfig` at serve time by `prepare_functions_runtime`.
    let variables = request.variables;

    // ── Execution ────────────────────────────────────────────────────────────
    // Dispatch, the suspended-tenant gate and the per-tenant quotas all live in
    // the shared seam so the MCP transport enforces the identical policy (#858).
    // `dispatch` holds the concurrency permit for the rest of this scope, which
    // is why it is not extracted into a stage.
    let dispatch = super::tenant_dispatch::dispatch_to_tenant(&state, tenant_key.as_deref())
        .map_err(|e| ErrorResponse::from_error(tenant_dispatch_error(&e)))?;
    let executor = &dispatch.executor;

    // M-quotas (cost): one estimate serves both budget enforcement and
    // observability (#379). Rejection surfaces at the same chokepoint as the
    // other per-tenant quotas, in the shared seam for the same reason (#858).
    let estimated_cost =
        super::tenant_dispatch::estimate_request_cost(&query, variables.as_ref(), executor);
    super::tenant_dispatch::charge_cost_budget(
        &state,
        tenant_key.as_deref(),
        security_context.as_ref(),
        estimated_cost,
    )
    .map_err(|e| ErrorResponse::from_error(tenant_dispatch_error(&e)))?;

    // #1564: whether the response was produced for a principal, read before
    // `security_context` is consumed.
    let principal = security_context.is_some();
    // Preserve subject for audit logging before security_context is consumed.
    #[cfg(feature = "auth")]
    let audit_subject = security_context.as_ref().map(|ctx| ctx.user_id.to_string());
    // Error propagation is deferred so the circuit-breaker outcome is recorded first.
    // GraphQL § 6.1 — the request's `operationName` selects which operation runs.
    // Before this was threaded, a document carrying two operations always ran the
    // first one, whatever the client named.
    let operation_name = request.operation_name.as_deref();
    // #1512: the request's locale, from the explicit `extensions` argument, the headers and
    // the resolved identity, under this tenant's `[locale]`; #1520: the headers its session
    // variables name.
    let scope = crate::request_scope::RequestScope::resolve(
        executor.schema(),
        Some(headers),
        Some(&crate::request_scope::json_argument(request.extensions.as_ref())),
        security_context.as_ref(),
    )
    .map_err(|e| ErrorResponse::from_error(GraphQLError::from_fraiseql_error(&e)))?;
    // Boxed for the reason the stages above are: the executor future is deep.
    let exec_result = Box::pin(crate::request_scope::scoped(scope, async {
        if let Some(sec_ctx) = security_context {
            executor
                .execute_operation_with_security(
                    &query,
                    variables.as_ref(),
                    &sec_ctx,
                    operation_name,
                )
                .await
        } else {
            executor.execute_operation(&query, variables.as_ref(), operation_name).await
        }
    }))
    .await;

    // Record circuit breaker outcome for federation entity queries
    #[cfg(feature = "federation")]
    if !cb_entity_types.is_empty() {
        if let Some(ref cb_manager) = state.circuit_breaker {
            if exec_result.is_ok() {
                for entity_type in &cb_entity_types {
                    cb_manager.record_success(entity_type);
                }
            } else {
                for entity_type in &cb_entity_types {
                    cb_manager.record_failure(entity_type);
                }
            }
        }
    }

    // Propagate execution errors with metrics
    let op_name = request.operation_name.as_deref().unwrap_or("");
    let result = exec_result.map_err(|e| {
        let elapsed = start_time.elapsed();
        #[allow(clippy::cast_possible_truncation)]
        // Reason: microsecond counter cannot exceed u64 in any practical uptime
        let elapsed_us = elapsed.as_micros() as u64;
        error!(
            error = %e,
            elapsed_ms = elapsed.as_millis(),
            operation_name = ?request.operation_name,
            "Query execution failed"
        );
        metrics.queries_error.fetch_add(1, Ordering::Relaxed);
        metrics.execution_errors_total.fetch_add(1, Ordering::Relaxed);
        // Record duration even for failed queries
        metrics.queries_duration_us.fetch_add(elapsed_us, Ordering::Relaxed);
        metrics.operation_metrics.record(op_name, elapsed_us, true);

        // S46: emit AuthorizationDenied audit event for compliance (SOC 2).
        // Must be emitted before error sanitization so we log the real reason.
        #[cfg(feature = "auth")]
        if matches!(e, fraiseql_core::FraiseQLError::Authorization { .. }) {
            use fraiseql_auth::audit::logger::{
                AuditEntry, AuditEventType, SecretType, get_audit_logger,
            };
            let resource =
                if let fraiseql_core::FraiseQLError::Authorization { ref resource, .. } = e {
                    resource.clone().unwrap_or_else(|| op_name.to_string())
                } else {
                    op_name.to_string()
                };
            get_audit_logger().log_entry(AuditEntry {
                event_type:    AuditEventType::AuthorizationDenied,
                secret_type:   SecretType::JwtToken,
                subject:       audit_subject.clone(),
                operation:     op_name.to_string(),
                success:       false,
                error_message: Some(resource),
                context:       Some(format!("peer_ip={peer_ip}")),
                chain_hash:    None,
            });
        }

        let err = state.error_sanitizer.sanitize(GraphQLError::from_fraiseql_error(&e));
        ErrorResponse::from_error(err)
    })?;

    let elapsed = start_time.elapsed();
    #[allow(clippy::cast_possible_truncation)]
    // Reason: microsecond counter cannot exceed u64 in any practical uptime
    let elapsed_us = elapsed.as_micros() as u64;

    // Record successful query metrics
    metrics.queries_success.fetch_add(1, Ordering::Relaxed);
    metrics.queries_duration_us.fetch_add(elapsed_us, Ordering::Relaxed);
    metrics.db_queries_total.fetch_add(1, Ordering::Relaxed);
    metrics.db_queries_duration_us.fetch_add(elapsed_us, Ordering::Relaxed);
    metrics.operation_metrics.record(op_name, elapsed_us, false);

    // #379 acceptance: the audit trail records the estimated cost alongside
    // tenant and operation for every executed request, and the running sum is
    // exported at /metrics so budgets can be sized from observed traffic.
    // Per-request, but on a dedicated target (the `mutation_audit` pattern) so
    // operators subscribe to it explicitly instead of it flooding the default
    // `info!` stream.
    if let Some(cost) = estimated_cost {
        metrics.queries_cost_total.fetch_add(cost, Ordering::Relaxed);
        tracing::info!(
            target: "fraiseql::cost_audit",
            cost,
            tenant = tenant_key.as_deref().unwrap_or(""),
            operation = %op_name,
            "operation cost"
        );
    }

    // Record federation-specific metrics for federation queries
    #[cfg(feature = "federation")]
    if fraiseql_core::federation::is_federation_query(&query) {
        metrics.record_entity_resolution(elapsed_us, true);
    }

    debug!(
        elapsed_ms = elapsed.as_millis(),
        operation_name = ?request.operation_name,
        "Query executed successfully"
    );

    // ── Post-processing ──────────────────────────────────────────────────────
    #[allow(unused_mut)]
    // Reason: mut is required by decrypt_response_fields(&mut ...) under the secrets feature
    let mut response_json = result;

    #[cfg(feature = "secrets")]
    Box::pin(stages::decrypt_response_fields(&state, &mut response_json)).await?;

    // Idempotency (#747): persist the successful response so a re-send of the
    // same mutation under the same key replays it instead of executing again.
    // Only success is stored — a failed mutation stays retryable under its key.
    if let Some((key, body_hash)) = idempotency_key {
        state
            .idempotency_store
            .store(
                key,
                body_hash,
                crate::routes::idempotency::StoredResponse {
                    status:  200,
                    headers: Vec::new(),
                    body:    Some(response_json.clone()),
                },
            )
            .await;
    }

    // #1564: only a GET response may be stored, and only as `http_cache` decides.
    let caching = if method == Method::Get {
        http_cache::decide(
            &http_cache::Answered {
                body: &response_json,
                ttl: executor.response_cache_ttl(&query, operation_name),
                principal,
                tenant: tenant_key.is_some(),
            },
            headers,
            executor.schema(),
        )
    } else {
        http_cache::Caching::NoStore
    };
    let mut cache_headers = HeaderMap::new();
    http_cache::apply(&mut cache_headers, caching, Some(executor.schema()));

    Ok(Answer {
        response: GraphQLResponse {
            body: response_json,
        },
        headers:  cache_headers,
    })
}

/// Map a tenant-dispatch error from [`AppState::executor_for_tenant`] to the
/// correct GraphQL error code (#332).
///
/// `executor_for_tenant` returns [`FraiseQLError::Authorization`] for an unknown
/// tenant key (→ 403 Forbidden) and [`FraiseQLError::ServiceUnavailable`] for a
/// suspended tenant (→ 503 with a `Retry-After` header carrying `retry_after`).
/// Previously both collapsed to 403, discarding the variant and the retry hint.
pub(super) fn tenant_dispatch_error(error: &FraiseQLError) -> GraphQLError {
    match error {
        FraiseQLError::ServiceUnavailable { retry_after, .. } => {
            GraphQLError::service_unavailable(error.to_string(), *retry_after)
        },
        // Per-tenant concurrency limit reached (M-quotas) → 429 Too Many Requests.
        FraiseQLError::RateLimited { .. } => GraphQLError::rate_limited(error.to_string()),
        // Cost rejections carry their own codes (#379): a per-request ceiling
        // is permanent for the operation (200 + errors[], no retry invitation),
        // an exhausted rolling window is retryable (429 + Retry-After).
        FraiseQLError::CostExceeded {
            retry_after_secs, ..
        } => match retry_after_secs {
            Some(secs) => GraphQLError::cost_budget_exhausted(error.to_string(), *secs),
            None => GraphQLError::operation_cost_exceeded(error.to_string()),
        },
        // Unknown tenant key (Authorization) and any other dispatch error stay
        // 403 Forbidden, preserving the prior behaviour.
        _ => GraphQLError::new(error.to_string(), crate::error::ErrorCode::Forbidden),
    }
}

mod http_cache;
mod incremental;
mod sse;
mod stages;

#[cfg(test)]
mod tests;