hive-router 0.2.0

use std::sync::Arc;
use std::time::{Duration, Instant};

use crate::executor::executors::dedupe::unique_leader_fingerprint;
use crate::executor::executors::inflight::InFlightRole;
use crate::executor::executors::map::InflightRequestsMap;
use crate::executor::executors::multipart_subscribe;
use crate::executor::executors::sse;
use crate::executor::executors::subscription_buffer;
use crate::executor::hooks::on_subgraph_http_request::{
    OnSubgraphHttpRequestHookPayload, OnSubgraphHttpResponseHookPayload,
};
use crate::executor::json_writer::write_named_operation;
use crate::executor::plugin_context::PluginRequestState;
use crate::executor::plugin_trait::{EndControlFlow, StartControlFlow};
use crate::executor::plugins::hooks;
use crate::executor::response::subgraph_response::SubgraphResponse;
use crate::query_planner::planner::plan_nodes::CustomScalarPaths;
use crate::telemetry::logging::targets;
use crate::telemetry::metrics::catalog::values::GraphQLResponseStatus;
use crate::telemetry::metrics::http_client_metrics::HttpClientRequestStateCapture;
use crate::telemetry::metrics::subscription_metrics::SubscriptionTransport;
use crate::telemetry::TelemetryContext;
use futures::stream::BoxStream;

use async_trait::async_trait;

use bytes::{BufMut, Bytes};
use http::HeaderMap;
use http::HeaderValue;
use http::StatusCode;
use http_body_util::BodyExt;
use http_body_util::Full;
use hyper::Version;
use hyper_rustls::HttpsConnector;
use hyper_util::client::legacy::{connect::HttpConnector, Client};
use tokio::sync::Semaphore;
use tracing::{debug, trace};

use crate::executor::executors::common::SubgraphExecutionRequest;
use crate::executor::executors::error::SubgraphExecutorError;
use crate::executor::utils::consts::CLOSE_BRACE;
use crate::executor::utils::consts::COLON;
use crate::executor::utils::consts::COMMA;
use crate::executor::utils::consts::QUOTE;
use crate::executor::{executors::common::SubgraphExecutor, json_writer::write_and_escape_string};
use crate::telemetry::traces::spans::http_request::HttpClientRequestSpan;
use crate::telemetry::traces::spans::http_request::HttpInflightRequestSpan;
use tracing::Instrument;

pub struct HTTPSubgraphExecutor {
    pub subgraph_name: String,
    pub endpoint: http::Uri,
    pub http_client: Arc<HttpClient>,
    pub header_map: HeaderMap,
    pub semaphore: Arc<Semaphore>,
    pub dedupe_enabled: bool,
    pub in_flight_requests: InflightRequestsMap,
    pub telemetry_context: Arc<TelemetryContext>,
    pub subgraph_buffer_capacity: usize,
}

const FIRST_VARIABLE_STR: &[u8] = b",\"variables\":{";
const FIRST_QUOTE_STR: &[u8] = b"{\"query\":";
const OPERATION_NAME_STR: &[u8] = b",\"operationName\":";

pub type HttpClient = Client<HttpsConnector<HttpConnector>, Full<Bytes>>;

struct FetchedSubgraphResponse<'a> {
    response: SubgraphHttpResponse,
    http_request_capture: HttpClientRequestStateCapture<'a>,
    transport_duration: Duration,
}

struct HttpRequestTelemetryCapture<'a> {
    capture: HttpClientRequestStateCapture<'a>,
    response_body_size: u64,
    transport_duration: Duration,
}

pub fn build_request_body(
    execution_request: &SubgraphExecutionRequest<'_>,
) -> Result<Vec<u8>, SubgraphExecutorError> {
    let mut body = Vec::with_capacity(4096);
    body.put(FIRST_QUOTE_STR);

    if let Some(operation_name) = &execution_request.operation_name {
        write_named_operation(
            &mut body,
            operation_name.as_bytes(),
            execution_request.document_name_write_pos,
            execution_request.query,
        );
    } else {
        write_and_escape_string(&mut body, execution_request.query);
    }

    let mut first_variable = true;
    if let Some(variables) = &execution_request.variables {
        for (variable_name, variable_value) in variables {
            if first_variable {
                body.put(FIRST_VARIABLE_STR);
                first_variable = false;
            } else {
                body.put(COMMA);
            }
            body.put(QUOTE);
            body.put(variable_name.as_bytes());
            body.put(QUOTE);
            body.put(COLON);
            let value_str = sonic_rs::to_string(variable_value).map_err(|err| {
                SubgraphExecutorError::VariablesSerializationFailure(variable_name.to_string(), err)
            })?;
            body.put(value_str.as_bytes());
        }
    }
    if let Some(raw_variable_values) = &execution_request.raw_variable_values {
        for (variable_name, variable_value) in raw_variable_values {
            if first_variable {
                body.put(FIRST_VARIABLE_STR);
                first_variable = false;
            } else {
                body.put(COMMA);
            }
            body.put(QUOTE);
            body.put(variable_name.as_bytes());
            body.put(QUOTE);
            body.put(COLON);
            body.extend_from_slice(variable_value);
        }
    }
    // "first_variable" should be still true if there are no variables
    if !first_variable {
        body.put(CLOSE_BRACE);
    }

    if let Some(operation_name) = &execution_request.operation_name {
        body.put(OPERATION_NAME_STR);
        body.put(QUOTE);
        body.put(operation_name.as_bytes());
        body.put(QUOTE);
    }

    if let Some(extensions) = &execution_request.extensions {
        if !extensions.is_empty() {
            let as_value = sonic_rs::to_value(extensions).unwrap();

            body.put(COMMA);
            body.put("\"extensions\":".as_bytes());
            body.extend_from_slice(as_value.to_string().as_bytes());
        }
    }

    body.put(CLOSE_BRACE);

    Ok(body)
}

impl HTTPSubgraphExecutor {
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        subgraph_name: String,
        endpoint: http::Uri,
        http_client: Arc<HttpClient>,
        semaphore: Arc<Semaphore>,
        dedupe_enabled: bool,
        in_flight_requests: InflightRequestsMap,
        telemetry_context: Arc<TelemetryContext>,
        subgraph_buffer_capacity: usize,
    ) -> Self {
        let mut header_map = HeaderMap::new();
        header_map.insert(
            http::header::CONTENT_TYPE,
            HeaderValue::from_static("application/json; charset=utf-8"),
        );
        header_map.insert(
            http::header::CONNECTION,
            HeaderValue::from_static("keep-alive"),
        );

        Self {
            subgraph_name,
            endpoint,
            http_client,
            header_map,
            semaphore,
            dedupe_enabled,
            in_flight_requests,
            telemetry_context,
            subgraph_buffer_capacity,
        }
    }
}

pub struct SendRequestOpts<'a> {
    pub http_client: &'a HttpClient,
    pub endpoint: &'a http::Uri,
    pub subgraph_name: &'a str,
    pub method: http::Method,
    pub body: Vec<u8>,
    pub headers: HeaderMap,
    pub timeout: Option<Duration>,
    pub telemetry_context: &'a Arc<TelemetryContext>,
}

async fn send_request<'a>(
    opts: SendRequestOpts<'a>,
) -> Result<FetchedSubgraphResponse<'a>, SubgraphExecutorError> {
    let SendRequestOpts {
        http_client,
        endpoint,
        subgraph_name,
        method,
        body,
        headers,
        timeout,
        telemetry_context,
    } = opts;
    let request_body_size = body.len() as u64;

    let mut req = hyper::Request::builder()
        .method(method)
        .uri(endpoint)
        .version(Version::HTTP_11)
        .body(Full::new(Bytes::from(body)))?;

    *req.headers_mut() = headers;

    debug!(target: targets::HTTP_CLIENT, endpoint = ?endpoint, subgraph = subgraph_name, "making http request");

    let http_request_span = HttpClientRequestSpan::from_request(&req);
    let mut http_request_capture = telemetry_context.metrics.http_client.capture_request(
        &req,
        request_body_size,
        Some(subgraph_name),
    );
    let transport_started_at = Instant::now();

    let response: Result<SubgraphHttpResponse, SubgraphExecutorError> = async {
        // TODO: let's decide at some point if the tracing headers
        //       should be part of the fingerprint or not.
        telemetry_context.inject_context_into_http_headers(req.headers_mut());

        let res_fut = http_client.request(req);

        let res = if let Some(timeout_duration) = timeout {
            tokio::time::timeout(timeout_duration, res_fut).await?
        } else {
            res_fut.await
        }?;

        http_request_span.record_response(&res);
        http_request_capture.set_status_code(res.status().as_u16());

        debug!(
            target: targets::HTTP_CLIENT,
            endpoint = ?endpoint,
            subgraph = subgraph_name,
            status = res.status().as_u16(),
            "subgraph http request completed",
        );

        let (parts, body) = res.into_parts();

        let body = match body.collect().await {
            Ok(body) => body.to_bytes(),
            Err(err) => {
                return Err(SubgraphExecutorError::ResponseBodyReadFailure(
                    endpoint.to_string(),
                    err.to_string(),
                    parts.headers.into(),
                ));
            }
        };

        if body.is_empty() {
            return Err(SubgraphExecutorError::EmptyResponseBody(
                subgraph_name.to_string(),
                parts.headers.into(),
            ));
        }

        Ok(SubgraphHttpResponse {
            status: parts.status,
            body,
            headers: parts.headers.into(),
        })
    }
    .instrument(http_request_span.clone())
    .await;

    let transport_duration = transport_started_at.elapsed();

    match response {
        Ok(response) => Ok(FetchedSubgraphResponse {
            response,
            http_request_capture,
            transport_duration,
        }),
        Err(err) => {
            http_request_span.record_error(err.error_code());
            http_request_capture.finish_error(err.error_code(), transport_duration);
            Err(err)
        }
    }
}

pub enum DeduplicationHint {
    Deduped {
        fingerprint: u64,
        leader_id: u64,
        is_leader: bool,
    },
    NotDeduped,
}

#[async_trait]
impl SubgraphExecutor for HTTPSubgraphExecutor {
    fn executor_name(&self) -> &str {
        "http"
    }

    fn endpoint(&self) -> &http::Uri {
        &self.endpoint
    }

    async fn execute<'a>(
        &self,
        mut execution_request: SubgraphExecutionRequest<'a>,
        timeout: Option<Duration>,
        plugin_req_state: Option<&'a PluginRequestState<'a>>,
    ) -> Result<SubgraphResponse<'static>, SubgraphExecutorError> {
        let mut body = build_request_body(&execution_request)?;

        self.header_map.iter().for_each(|(key, value)| {
            execution_request.headers.insert(key, value.clone());
        });

        let mut method = http::Method::POST;
        let mut deduplicate_request = !self.dedupe_enabled || !execution_request.dedupe;

        let mut on_end_callbacks = vec![];
        let mut response = None;

        if let Some(plugin_req_state) = plugin_req_state.as_ref() {
            let mut start_payload = OnSubgraphHttpRequestHookPayload {
                subgraph_name: &self.subgraph_name,
                endpoint: &self.endpoint,
                method,
                body,
                execution_request,
                deduplicate_request,
                context: &plugin_req_state.context,
                request_context: plugin_req_state
                    .request_context
                    .for_plugin::<hooks::OnSubgraphHttp>(),
            };
            for plugin in plugin_req_state.plugins.as_ref() {
                let result = plugin.on_subgraph_http_request(start_payload).await;
                start_payload = result.payload;
                match result.control_flow {
                    StartControlFlow::Proceed => { /* continue to next plugin */ }
                    StartControlFlow::EndWithResponse(early_response) => {
                        response = Some(early_response);
                        // Break so other plugins are not called
                        break;
                    }
                    StartControlFlow::OnEnd(callback) => {
                        on_end_callbacks.push(callback);
                    }
                }
            }
            // Give the ownership back to variables
            method = start_payload.method;
            execution_request = start_payload.execution_request;
            body = start_payload.body;
            deduplicate_request = start_payload.deduplicate_request;
        }

        let mut deduplication_hint = DeduplicationHint::NotDeduped;
        let mut http_request_capture = None;

        let mut response = match response {
            Some(resp) => resp,
            None => {
                let send_request_opts = SendRequestOpts {
                    http_client: &self.http_client,
                    endpoint: &self.endpoint,
                    subgraph_name: &self.subgraph_name,
                    method,
                    body,
                    headers: execution_request.headers,
                    timeout,
                    telemetry_context: &self.telemetry_context,
                };

                if deduplicate_request {
                    // This unwrap is safe because the semaphore is never closed during the application's lifecycle.
                    // `acquire()` only fails if the semaphore is closed, so this will always return `Ok`.
                    let _permit = self.semaphore.acquire().await.unwrap();
                    let fetched_response = send_request(send_request_opts).await?;
                    http_request_capture = Some(HttpRequestTelemetryCapture {
                        capture: fetched_response.http_request_capture,
                        response_body_size: fetched_response.response.body.len() as u64,
                        transport_duration: fetched_response.transport_duration,
                    });
                    fetched_response.response
                } else {
                    let fingerprint = send_request_opts.fingerprint();

                    let inflight_span = HttpInflightRequestSpan::new(
                        &send_request_opts.method,
                        send_request_opts.endpoint,
                        &send_request_opts.headers,
                        &send_request_opts.body,
                    );

                    let result: Result<_, SubgraphExecutorError> = async {
                        let claim = self.in_flight_requests.claim(fingerprint);
                        let mut leader_http_request_capture = None;
                        let (shared_response, role) = claim
                            .get_or_try_init(|| async {
                                let res = {
                                    // This unwrap is safe because the semaphore is never closed during the application's lifecycle.
                                    // `acquire()` only fails if the semaphore is closed, so this will always return `Ok`.
                                    let _permit = self.semaphore.acquire().await.unwrap();
                                    send_request(send_request_opts).await
                                };

                                res.map(|fetched_response| {
                                    leader_http_request_capture =
                                        Some(HttpRequestTelemetryCapture {
                                            capture: fetched_response.http_request_capture,
                                            response_body_size: fetched_response.response.body.len()
                                                as u64,
                                            transport_duration: fetched_response.transport_duration,
                                        });
                                    (fetched_response.response, unique_leader_fingerprint())
                                })
                            })
                            .await?;

                        let (shared_response, leader_id) = shared_response.as_ref();
                        let shared_response = shared_response.clone();
                        let leader_id = *leader_id;

                        if role == InFlightRole::Leader {
                            inflight_span.record_as_leader(&leader_id);
                        } else {
                            inflight_span.record_as_joiner(&leader_id);
                        }

                        inflight_span
                            .record_response(&shared_response.body, &shared_response.status);

                        deduplication_hint = DeduplicationHint::Deduped {
                            fingerprint,
                            leader_id,
                            is_leader: role == InFlightRole::Leader,
                        };

                        Ok((shared_response, leader_http_request_capture))
                    }
                    .instrument(inflight_span.clone())
                    .await;

                    let (shared_response, leader_http_request_capture) = result?;
                    if let Some(capture) = leader_http_request_capture {
                        http_request_capture = Some(capture);
                    }

                    shared_response
                }
            }
        };

        if !on_end_callbacks.is_empty() {
            let plugin_state_ref = plugin_req_state
                .as_ref()
                .expect("plugin state not available, but on_end_callbacks are present");
            let mut end_payload = OnSubgraphHttpResponseHookPayload {
                context: &plugin_state_ref.context,
                request_context: plugin_state_ref
                    .request_context
                    .for_plugin::<hooks::OnSubgraphHttp>(),
                response,
                deduplication_hint,
            };
            for callback in on_end_callbacks {
                let result = callback(end_payload);
                end_payload = result.payload;
                match result.control_flow {
                    EndControlFlow::Proceed => { /* continue to next plugin */ }
                    EndControlFlow::EndWithResponse(early_response) => {
                        end_payload.response = early_response;
                        // Break so other plugins are not called
                        break;
                    }
                }
            }
            // Give the ownership back to variables
            response = end_payload.response;
        }

        let (is_content_type_valid, content_type) =
            validate_response_content_type(&response.headers);

        if !is_content_type_valid {
            return Err(SubgraphExecutorError::InvalidContentType(
                self.subgraph_name.clone(),
                content_type.unwrap_or("").to_string(),
                response.headers,
            ));
        }

        let response_result =
            response.deserialize_http_response(execution_request.custom_scalar_paths);

        if let Some(mut http_request_capture) = http_request_capture {
            finish_capture_from_subgraph_result(
                &mut http_request_capture.capture,
                http_request_capture.response_body_size,
                http_request_capture.transport_duration,
                &response_result,
            );
        }

        response_result
    }

    async fn subscribe<'a>(
        &self,
        execution_request: SubgraphExecutionRequest<'a>,
        connection_timeout: Option<Duration>,
    ) -> Result<
        BoxStream<'static, Result<SubgraphResponse<'static>, SubgraphExecutorError>>,
        SubgraphExecutorError,
    > {
        let custom_scalar_paths = execution_request.custom_scalar_paths.cloned();
        let buffer_capacity = self.subgraph_buffer_capacity;
        let body = build_request_body(&execution_request)?;

        let mut req = hyper::Request::builder()
            .method(http::Method::POST)
            .uri(&self.endpoint)
            .version(Version::HTTP_11)
            .body(Full::new(Bytes::from(body)))?;

        let mut headers = execution_request.headers;
        self.header_map.iter().for_each(|(key, value)| {
            headers.insert(key, value.clone());
        });

        // Prefer multipart over SSE for subscriptions
        // https://www.apollographql.com/docs/graphos/routing/operations/subscriptions/multipart-protocol
        headers.insert(
            http::header::ACCEPT,
            HeaderValue::from_static(
                r#"multipart/mixed;subscriptionSpec="1.0", text/event-stream"#,
            ),
        );
        *req.headers_mut() = headers;

        debug!(
            target: targets::HTTP_CLIENT,
            subgraph = self.subgraph_name,
            endpoint = ?self.endpoint,
            "establishing subscription connection"
        );

        let res_fut = self.http_client.request(req);

        let res = if let Some(timeout_duration) = connection_timeout {
            tokio::time::timeout(timeout_duration, res_fut).await?
        } else {
            res_fut.await
        }?;

        debug!(
            target: targets::HTTP_CLIENT,
            subgraph = self.subgraph_name,
            endpoint = ?self.endpoint,
            status = res.status().as_u16(),
            "subscription connection established"
        );

        if !res.status().is_success() {
            return Err(SubgraphExecutorError::StreamStatusCodeNotOk(res.status()));
        }

        let (parts, body_stream) = res.into_parts();

        let content_type = parts
            .headers
            .get(http::header::CONTENT_TYPE)
            .and_then(|v| v.to_str().ok())
            .unwrap_or("");

        let is_multipart = content_type.starts_with("multipart/mixed");
        let is_sse = content_type.starts_with("text/event-stream");

        if !is_multipart && !is_sse {
            return Err(SubgraphExecutorError::UnsupportedContentTypeError(
                content_type.to_string(),
            ));
        }

        if is_multipart {
            debug!(
                target: targets::HTTP_CLIENT,
                subgraph = self.subgraph_name,
                "using multipart HTTP for subscription",
            );

            let boundary = multipart_subscribe::parse_boundary_from_header(content_type)
                .map_err(|e| SubgraphExecutorError::MultipartBoundaryParseFailure(e.to_string()))?;
            let stream = multipart_subscribe::parse_to_stream(
                boundary,
                body_stream,
                custom_scalar_paths.clone(),
            );

            let op_guard = self
                .telemetry_context
                .metrics
                .subscriptions
                .active_subgraph_operation(&self.subgraph_name);
            let conn_guard = self
                .telemetry_context
                .metrics
                .subscriptions
                .active_subgraph_connection(
                    &self.subgraph_name,
                    SubscriptionTransport::HttpMultipart,
                );

            let mapped = Box::pin(async_stream::stream! {
                let _op_guard = op_guard;
                let _conn_guard = conn_guard;

                trace!(target: targets::HTTP_CLIENT, "multipart subscription stream started");

                for await result in stream {
                    match result {
                        Ok(response) => {
                            trace!(target: targets::HTTP_CLIENT, response = ?response, "multipart subscription event received");

                            yield Ok(response);
                        }
                        Err(e) => {
                            yield Err(SubgraphExecutorError::MultipartStreamError(e.to_string()));
                            return;
                        }
                    }
                }
            });

            // buffer decouples the emitting subgraph from slow downstream consumers, dropping
            // messages under backpressure instead of throttling the subgraph
            Ok(subscription_buffer::buffered(
                mapped,
                buffer_capacity,
                self.telemetry_context.clone(),
                SubscriptionTransport::HttpMultipart,
                self.subgraph_name.clone(),
                self.endpoint.to_string(),
            ))
        } else {
            debug!(
                target: targets::HTTP_CLIENT,
                subgraph = self.subgraph_name,
                endpoint = ?self.endpoint,
                "using SSE for subscription connection to subgraph",
            );

            let stream = sse::parse_to_stream(body_stream, custom_scalar_paths.clone());

            let op_guard = self
                .telemetry_context
                .metrics
                .subscriptions
                .active_subgraph_operation(&self.subgraph_name);
            let conn_guard = self
                .telemetry_context
                .metrics
                .subscriptions
                .active_subgraph_connection(&self.subgraph_name, SubscriptionTransport::HttpSse);

            let mapped = Box::pin(async_stream::stream! {
                let _op_guard = op_guard;
                let _conn_guard = conn_guard;

                trace!(target: targets::HTTP_CLIENT, "SSE subscription stream started");
                for await result in stream {
                    match result {
                        Ok(response) => {
                            trace!(target: targets::HTTP_CLIENT, response = ?response, "SSE subscription event received");
                            yield Ok(response);
                        }
                        Err(e) => {
                            yield Err(SubgraphExecutorError::SseStreamError(e.to_string()));
                            return;
                        }
                    }
                }
            });

            // buffer decouples the emitting subgraph from slow downstream consumers, dropping
            // messages under backpressure instead of throttling the subgraph
            Ok(subscription_buffer::buffered(
                mapped,
                buffer_capacity,
                self.telemetry_context.clone(),
                SubscriptionTransport::HttpSse,
                self.subgraph_name.clone(),
                self.endpoint.to_string(),
            ))
        }
    }
}

fn finish_capture_from_subgraph_result(
    capture: &mut HttpClientRequestStateCapture<'_>,
    response_body_size: u64,
    transport_duration: Duration,
    response_result: &Result<SubgraphResponse<'_>, SubgraphExecutorError>,
) {
    let error_code = response_result.as_ref().err().map(|err| err.error_code());
    let graphql_response_status = if response_result.as_ref().is_ok_and(|response| {
        response
            .errors
            .as_ref()
            .is_none_or(|errors| errors.is_empty())
    }) {
        GraphQLResponseStatus::Ok
    } else {
        GraphQLResponseStatus::Error
    };

    capture.finish(
        response_body_size,
        transport_duration,
        graphql_response_status,
        error_code,
    );
}

#[derive(Default, Clone)]
pub struct SubgraphHttpResponse {
    pub status: StatusCode,
    pub headers: Arc<HeaderMap>,
    pub body: Bytes,
}

const VALID_GRAPHQL_CONTENT_TYPES: [&str; 2] =
    ["application/json", "application/graphql-response+json"];

#[inline]
fn is_valid_graphql_content_type(header_value: &str) -> bool {
    // Split by `;` to remove any charset or boundary parameters
    // and then take the first one.
    // We can't rely on simple string comparison because of additianal attributes (like charset),
    // and we can't rely on starts_with because there are valid values that start with `application/json` and can confuse our check.
    let essence = header_value.split(';').next().unwrap_or("").trim();

    VALID_GRAPHQL_CONTENT_TYPES
        .iter()
        .any(|valid| valid.eq_ignore_ascii_case(essence))
}

#[inline]
fn validate_response_content_type(headers: &HeaderMap) -> (bool, Option<&str>) {
    let content_type = headers
        .get(http::header::CONTENT_TYPE)
        .and_then(|v| v.to_str().ok());

    let Some(content_type) = content_type else {
        return (false, None);
    };

    if !is_valid_graphql_content_type(content_type) {
        return (false, Some(content_type));
    }

    (true, Some(content_type))
}

impl SubgraphHttpResponse {
    fn deserialize_http_response(
        self,
        custom_scalar_paths: Option<&CustomScalarPaths>,
    ) -> Result<SubgraphResponse<'static>, SubgraphExecutorError> {
        SubgraphResponse::deserialize_from_bytes(self.body, custom_scalar_paths)
            .map(|mut resp: SubgraphResponse| {
                resp.headers = Some(self.headers.clone());
                resp.status = Some(self.status);
                resp
            })
            .map_err(|e| match e {
                SubgraphExecutorError::ResponseDeserializationFailure(err, _) => {
                    SubgraphExecutorError::ResponseDeserializationFailure(err, Some(self.headers))
                }
                other => other,
            })
    }
}

#[cfg(test)]
mod tests {
    use http::HeaderMap;
    use serde_json::json;

    use super::build_request_body;
    use crate::executor::executors::common::SubgraphExecutionRequest;

    #[test]
    fn build_request_body_with_forwarded_operation_name_and_no_variables_is_valid_json() {
        let execution_request = SubgraphExecutionRequest {
            query: "query { me { id } }",
            document_name_write_pos: 5,
            dedupe: false,
            operation_name: Some("GetMe_accounts_0".to_string()),
            variables: None,
            headers: HeaderMap::new(),
            raw_variable_values: None,
            extensions: None,
            custom_scalar_paths: None,
            connection_fingerprint: None,
        };

        let body = build_request_body(&execution_request).expect("request body should serialize");
        let json_body: serde_json::Value =
            serde_json::from_slice(&body).expect("request body should be valid JSON");

        assert_eq!(
            json_body,
            json!({
                "query": "query GetMe_accounts_0 { me { id } }",
                "operationName": "GetMe_accounts_0",
            })
        );
    }
}