use crate::pipeline::error::PipelineError;
use crate::pipeline::normalize::GraphQLNormalizationPayload;
use crate::schema_state::SelectedSupergraph;
use crate::shared_state::RouterSharedState;
use hive_router_internal::telemetry::traces::spans::graphql::{
    GraphQLExecuteSpan, GraphQLOperationSpan,
};
use hive_router_plan_executor::execution::client_request_details::ClientRequestDetails;
use hive_router_plan_executor::execution::demand_control::DemandControlExecutionContext;
use hive_router_plan_executor::execution::jwt_forward::JwtAuthForwardingPlan;
use hive_router_plan_executor::execution::operation_name::OperationNameFactory;
use hive_router_plan_executor::execution::plan::{
    execute_query_plan, CoerceVariablesPayload, ExecutionResultExtensions, PlanExecutionOutput,
    QueryPlanExecutionOpts, QueryPlanExecutionResult,
};
use hive_router_plan_executor::headers::response::ResponseHeaderSink;

use hive_router_plan_executor::introspection::resolve::IntrospectionContext;
use hive_router_plan_executor::plugin_context::PluginRequestState;
use hive_router_plan_executor::response::graphql_error::GraphQLError;
use hive_router_query_planner::planner::plan_nodes::QueryPlan;
use http::HeaderName;
use sonic_rs::json;
use std::sync::Arc;
use tracing::Instrument;

pub static EXPOSE_QUERY_PLAN_HEADER: HeaderName = HeaderName::from_static("hive-expose-query-plan");

#[derive(Clone, Debug, PartialEq, Eq)]
pub enum ExposeQueryPlanMode {
    Yes,
    No,
    DryRun,
}

pub struct PlannedRequest<'req> {
    pub normalized_payload: Arc<GraphQLNormalizationPayload>,
    pub query_plan_payload: &'req QueryPlan,
    pub variable_payload: Arc<CoerceVariablesPayload>,
    pub client_request_details: Arc<ClientRequestDetails<'req>>,
    pub initial_errors: Vec<GraphQLError>,
    pub demand_control_execution_context: Option<DemandControlExecutionContext>,
    pub plugin_req_state: Option<PluginRequestState<'req>>,
}

#[inline]
pub async fn execute_plan<'exec>(
    supergraph: &SelectedSupergraph,
    app_state: &RouterSharedState,
    planned_request: PlannedRequest<'exec>,
    span: GraphQLOperationSpan,
    response_header_sink: ResponseHeaderSink,
) -> Result<QueryPlanExecutionResult, PipelineError> {
    let execute_span = GraphQLExecuteSpan::new();

    let introspection_context = IntrospectionContext {
        query: planned_request
            .normalized_payload
            .operation_for_introspection
            .clone(),
        schema: Arc::clone(&supergraph.snapshot.planner.consumer_schema.document),
        metadata: Arc::clone(&supergraph.snapshot.metadata),
        variables: planned_request.variable_payload.clone(),
    };
    async {
        let mut extensions = ExecutionResultExtensions::default();

        let mut expose_query_plan = ExposeQueryPlanMode::No;
        if app_state.router_config.query_planner.allow_expose {
            if let Some(expose_qp_header) = planned_request
                .client_request_details
                .headers
                .get(&EXPOSE_QUERY_PLAN_HEADER)
            {
                let str_value = expose_qp_header.to_str().unwrap_or_default().trim();
                match str_value {
                    "true" => expose_query_plan = ExposeQueryPlanMode::Yes,
                    "dry-run" => expose_query_plan = ExposeQueryPlanMode::DryRun,
                    _ => {}
                }
            }
        }

        if matches!(
            expose_query_plan,
            ExposeQueryPlanMode::Yes | ExposeQueryPlanMode::DryRun
        ) {
            extensions.query_plan = Some(planned_request.query_plan_payload);
        }

        if matches!(expose_query_plan, ExposeQueryPlanMode::DryRun) {
            let body = sonic_rs::to_vec(&json!({
                "extensions": extensions,
            }))
            .map_err(PipelineError::QueryPlanSerializationFailed)?;

            return Ok(QueryPlanExecutionResult::Single(PlanExecutionOutput {
                body,
                ..Default::default()
            }));
        }

        let jwt_auth_forwarding: Option<JwtAuthForwardingPlan> = if app_state
            .router_config
            .jwt
            .is_jwt_extensions_forwarding_enabled()
        {
            planned_request
                .client_request_details
                .jwt
                .build_forwarding_plan(
                    &app_state
                        .router_config
                        .jwt
                        .forward_claims_to_upstream_extensions
                        .field_name,
                )?
        } else {
            None
        };

        let operation_name = planned_request.client_request_details.operation.name;
        let result = execute_query_plan(QueryPlanExecutionOpts {
            query_plan: planned_request.query_plan_payload,
            operation_for_plan: planned_request
                .normalized_payload
                .operation_for_plan
                .clone(),
            projection_plan: planned_request.normalized_payload.projection_plan.clone(),
            headers_plan: app_state.headers_plan.clone(),
            extensions_plan: app_state.extensions_plan.clone(),
            variable_values: planned_request.variable_payload.clone(),
            extensions,
            client_request: planned_request.client_request_details,
            introspection_context: introspection_context.into(),
            operation_type_name: planned_request.normalized_payload.root_type_name.clone(),
            operation_kind: planned_request.normalized_payload.operation_kind.clone(),
            jwt_auth_forwarding: jwt_auth_forwarding.map(|j| j.into()),
            graphql_error_recorder: app_state.telemetry_context.metrics.graphql.error_recorder(),
            demand_control_context: planned_request
                .demand_control_execution_context
                .map(|d| d.into()),
            executors: Arc::clone(&supergraph.runtime.subgraph_executor_map),
            initial_errors: planned_request.initial_errors,
            span,
            plugin_req_state: planned_request.plugin_req_state,
            operation_name_factory: OperationNameFactory::new(
                supergraph.runtime.operation_name_forward_config.clone(),
                operation_name,
            ),
            response_header_sink,
            error_masking_runtime: app_state.error_masking.clone(),
        })
        .await?;

        Ok(result)
    }
    .instrument(execute_span.span)
    .await
}
