feat(router): support hive as source for supergraph

Cargo.toml

@@ -54,3 +54,4 @@ tokio-util = { version = "0.7.16" }
 rand = "0.9.2"
 jsonwebtoken = "9.3.1"
 ntex = { version = "2", features = ["tokio"] }
+reqwest = "0.12.23"

bin/router/Cargo.toml

@@ -25,6 +25,7 @@ futures = { workspace = true }
 graphql-parser = { workspace = true }
 graphql-tools = { workspace = true }
 serde = { workspace = true }
+reqwest = { workspace = true }
 sonic-rs = { workspace = true }
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true }
@@ -37,13 +38,14 @@ lazy_static = { workspace = true }
 async-trait = { workspace = true }
 xxhash-rust = { workspace = true }
 rand = { workspace = true }
-regex-automata = "0.4.10"
+ntex = { workspace = true }
 jsonwebtoken = { workspace = true }
 
 mimalloc = { version = "0.1.47", features = ["override"] }
 moka = { version = "0.12.10", features = ["future"] }
 ulid = "1.2.1"
-ntex = { workspace = true }
 tokio-util = "0.7.16"
-reqwest = "0.12.23"
 cookie = "0.18.1"
+arc-swap = "1.7.1"
+notify = "8.2.0"
+regex-automata = "0.4.10"

bin/router/src/lib.rs

@@ -4,6 +4,8 @@ mod jwt;
 mod logger;
 mod pipeline;
 mod shared_state;
+mod supergraph;
+mod supergraph_mgr;
 
 use std::sync::Arc;
 
@@ -14,20 +16,20 @@ use crate::{
     logger::configure_logging,
     pipeline::graphql_request_handler,
     shared_state::RouterSharedState,
+    supergraph_mgr::SupergraphManager,
 };
 
 use hive_router_config::{load_config, HiveRouterConfig};
 use ntex::{
     util::Bytes,
     web::{self, HttpRequest},
 };
-
-use hive_router_query_planner::utils::parsing::parse_schema;
 use tracing::info;
 
 async fn graphql_endpoint_handler(
     mut request: HttpRequest,
     body_bytes: Bytes,
+    supergraph_manager: web::types::State<Arc<SupergraphManager>>,
     app_state: web::types::State<Arc<RouterSharedState>>,
 ) -> impl web::Responder {
     // If an early CORS response is needed, return it immediately.
@@ -39,7 +41,9 @@ async fn graphql_endpoint_handler(
         return Some(early_response);
     }
 
-    let mut res = graphql_request_handler(&mut request, body_bytes, app_state.get_ref()).await;
+    let supergraph = supergraph_manager.current();
+    let mut res =
+        graphql_request_handler(&mut request, body_bytes, &supergraph, app_state.get_ref()).await;
 
     // Apply CORS headers to the final response if CORS is configured.
     if let Some(cors) = app_state.cors.as_ref() {
@@ -55,11 +59,13 @@ pub async fn router_entrypoint() -> Result<(), Box<dyn std::error::Error>> {
     configure_logging(&router_config.log);
     let addr = router_config.http.address();
     let mut bg_tasks_manager = BackgroundTasksManager::new();
-    let shared_state = configure_app_from_config(router_config, &mut bg_tasks_manager).await?;
+    let (shared_state, supergraph_manager) =
+        configure_app_from_config(router_config, &mut bg_tasks_manager).await?;
 
     let maybe_error = web::HttpServer::new(move || {
         web::App::new()
             .state(shared_state.clone())
+            .state(supergraph_manager.clone())
             .configure(configure_ntex_app)
             .default_service(web::to(landing_page_handler))
     })
@@ -77,18 +83,16 @@ pub async fn router_entrypoint() -> Result<(), Box<dyn std::error::Error>> {
 pub async fn configure_app_from_config(
     router_config: HiveRouterConfig,
     bg_tasks_manager: &mut BackgroundTasksManager,
-) -> Result<Arc<RouterSharedState>, Box<dyn std::error::Error>> {
-    let supergraph_sdl = router_config.supergraph.load().await?;
-    let parsed_schema = parse_schema(&supergraph_sdl);
-
+) -> Result<(Arc<RouterSharedState>, Arc<SupergraphManager>), Box<dyn std::error::Error>> {
     let jwt_runtime = match &router_config.jwt {
         Some(jwt_config) => Some(JwtAuthRuntime::init(bg_tasks_manager, jwt_config).await?),
         None => None,
     };
 
-    let shared_state = RouterSharedState::new(parsed_schema, router_config, jwt_runtime)?;
+    let supergraph_manager = Arc::new(SupergraphManager::new_from_config(&router_config).await?);
+    let shared_state = Arc::new(RouterSharedState::new(router_config, jwt_runtime)?);
 
-    Ok(shared_state)
+    Ok((shared_state, supergraph_manager))
 }
 
 pub fn configure_ntex_app(cfg: &mut web::ServiceConfig) {

bin/router/src/pipeline/coerce_variables.rs

@@ -11,7 +11,7 @@ use tracing::{error, trace, warn};
 use crate::pipeline::error::{PipelineError, PipelineErrorFromAcceptHeader, PipelineErrorVariant};
 use crate::pipeline::execution_request::ExecutionRequest;
 use crate::pipeline::normalize::GraphQLNormalizationPayload;
-use crate::shared_state::RouterSharedState;
+use crate::supergraph_mgr::SupergraphData;
 
 #[derive(Clone, Debug)]
 pub struct CoerceVariablesPayload {
@@ -21,7 +21,7 @@ pub struct CoerceVariablesPayload {
 #[inline]
 pub fn coerce_request_variables(
     req: &HttpRequest,
-    app_state: &Arc<RouterSharedState>,
+    supergraph: &Arc<SupergraphData>,
     execution_params: ExecutionRequest,
     normalized_operation: &Arc<GraphQLNormalizationPayload>,
 ) -> Result<CoerceVariablesPayload, PipelineError> {
@@ -38,7 +38,7 @@ pub fn coerce_request_variables(
     match collect_variables(
         &normalized_operation.operation_for_plan,
         execution_params.variables,
-        &app_state.schema_metadata,
+        &supergraph.metadata,
     ) {
         Ok(values) => {
             trace!(

bin/router/src/pipeline/execution.rs

@@ -7,6 +7,7 @@ use crate::pipeline::coerce_variables::CoerceVariablesPayload;
 use crate::pipeline::error::{PipelineError, PipelineErrorFromAcceptHeader, PipelineErrorVariant};
 use crate::pipeline::normalize::GraphQLNormalizationPayload;
 use crate::shared_state::RouterSharedState;
+use crate::supergraph_mgr::SupergraphData;
 use hive_router_plan_executor::execute_query_plan;
 use hive_router_plan_executor::execution::jwt_forward::JwtAuthForwardingPlan;
 use hive_router_plan_executor::execution::plan::{
@@ -31,6 +32,7 @@ enum ExposeQueryPlanMode {
 pub async fn execute_plan<'a>(
     req: &mut HttpRequest,
     query: Cow<'a, str>,
+    supergraph: &Arc<SupergraphData>,
     app_state: &Arc<RouterSharedState>,
     normalized_payload: &Arc<GraphQLNormalizationPayload>,
     query_plan_payload: &Arc<QueryPlan>,
@@ -63,8 +65,8 @@ pub async fn execute_plan<'a>(
 
     let introspection_context = IntrospectionContext {
         query: normalized_payload.operation_for_introspection.as_ref(),
-        schema: &app_state.planner.consumer_schema.document,
-        metadata: &app_state.schema_metadata,
+        schema: &supergraph.planner.consumer_schema.document,
+        metadata: &supergraph.metadata,
     };
 
     let jwt_context = {
@@ -101,8 +103,8 @@ pub async fn execute_plan<'a>(
         },
         introspection_context: &introspection_context,
         operation_type_name: normalized_payload.root_type_name,
-        executors: &app_state.subgraph_executor_map,
         jwt_auth_forwarding: &jwt_forward_plan,
+        executors: &supergraph.subgraph_executor_map,
     })
     .await
     .map_err(|err| {

bin/router/src/pipeline/mod.rs

@@ -26,6 +26,7 @@ use crate::{
         validation::validate_operation_with_cache,
     },
     shared_state::RouterSharedState,
+    supergraph_mgr::SupergraphData,
 };
 
 pub mod coerce_variables;
@@ -47,6 +48,7 @@ static GRAPHIQL_HTML: &str = include_str!("../../static/graphiql.html");
 pub async fn graphql_request_handler(
     req: &mut HttpRequest,
     body_bytes: Bytes,
+    supergraph: &Arc<SupergraphData>,
     state: &Arc<RouterSharedState>,
 ) -> web::HttpResponse {
     if req.method() == Method::GET && req.accepts_content_type(*TEXT_HTML_CONTENT_TYPE) {
@@ -55,14 +57,7 @@ pub async fn graphql_request_handler(
             .body(GRAPHIQL_HTML);
     }
 
-    if let Some(jwt) = &state.jwt_auth_runtime {
-        match jwt.validate_request(req) {
-            Ok(_) => (),
-            Err(err) => return err.make_response(),
-        }
-    }
-
-    match execute_pipeline(req, body_bytes, state).await {
+    match execute_pipeline(req, body_bytes, supergraph, state).await {
         Ok(response) => {
             let response_bytes = Bytes::from(response.body);
             let response_headers = response.headers;
@@ -93,26 +88,29 @@ pub async fn graphql_request_handler(
 pub async fn execute_pipeline(
     req: &mut HttpRequest,
     body_bytes: Bytes,
+    supergraph: &Arc<SupergraphData>,
     state: &Arc<RouterSharedState>,
 ) -> Result<PlanExecutionOutput, PipelineError> {
     perform_csrf_prevention(req, &state.router_config.csrf)?;
 
     let execution_request = get_execution_request(req, body_bytes).await?;
     let parser_payload = parse_operation_with_cache(req, state, &execution_request).await?;
-    validate_operation_with_cache(req, state, &parser_payload).await?;
+    validate_operation_with_cache(req, supergraph, state, &parser_payload).await?;
 
     let progressive_override_ctx = request_override_context()?;
     let normalize_payload =
-        normalize_request_with_cache(req, state, &execution_request, &parser_payload).await?;
+        normalize_request_with_cache(req, supergraph, state, &execution_request, &parser_payload)
+            .await?;
     let query = Cow::Owned(execution_request.query.clone());
     let variable_payload =
-        coerce_request_variables(req, state, execution_request, &normalize_payload)?;
+        coerce_request_variables(req, supergraph, execution_request, &normalize_payload)?;
 
     let query_plan_cancellation_token =
         CancellationToken::with_timeout(state.router_config.query_planner.timeout);
 
     let query_plan_payload = plan_operation_with_cache(
         req,
+        supergraph,
         state,
         &normalize_payload,
         &progressive_override_ctx,
@@ -123,6 +121,7 @@ pub async fn execute_pipeline(
     let execution_result = execute_plan(
         req,
         query,
+        supergraph,
         state,
         &normalize_payload,
         &query_plan_payload,

bin/router/src/pipeline/normalize.rs

@@ -12,6 +12,7 @@ use crate::pipeline::error::{PipelineError, PipelineErrorFromAcceptHeader, Pipel
 use crate::pipeline::execution_request::ExecutionRequest;
 use crate::pipeline::parser::GraphQLParserPayload;
 use crate::shared_state::RouterSharedState;
+use crate::supergraph_mgr::SupergraphData;
 use tracing::{error, trace};
 
 #[derive(Debug)]
@@ -26,6 +27,7 @@ pub struct GraphQLNormalizationPayload {
 #[inline]
 pub async fn normalize_request_with_cache(
     req: &HttpRequest,
+    supergraph: &Arc<SupergraphData>,
     app_state: &Arc<RouterSharedState>,
     execution_params: &ExecutionRequest,
     parser_payload: &GraphQLParserPayload,
@@ -51,7 +53,7 @@ pub async fn normalize_request_with_cache(
             Ok(payload)
         }
         None => match normalize_operation(
-            &app_state.planner.supergraph,
+            &supergraph.planner.supergraph,
             &parser_payload.parsed_operation,
             execution_params.operation_name.as_deref(),
         ) {
@@ -64,7 +66,7 @@ pub async fn normalize_request_with_cache(
 
                 let operation = doc.operation;
                 let (root_type_name, projection_plan) =
-                    FieldProjectionPlan::from_operation(&operation, &app_state.schema_metadata);
+                    FieldProjectionPlan::from_operation(&operation, &supergraph.metadata);
                 let partitioned_operation = partition_operation(operation);
 
                 let payload = GraphQLNormalizationPayload {

bin/router/src/pipeline/query_plan.rs

@@ -5,6 +5,7 @@ use crate::pipeline::error::{PipelineError, PipelineErrorFromAcceptHeader, Pipel
 use crate::pipeline::normalize::GraphQLNormalizationPayload;
 use crate::pipeline::progressive_override::{RequestOverrideContext, StableOverrideContext};
 use crate::shared_state::RouterSharedState;
+use crate::supergraph_mgr::SupergraphData;
 use hive_router_query_planner::planner::plan_nodes::QueryPlan;
 use hive_router_query_planner::utils::cancellation::CancellationToken;
 use ntex::web::HttpRequest;
@@ -13,13 +14,14 @@ use xxhash_rust::xxh3::Xxh3;
 #[inline]
 pub async fn plan_operation_with_cache(
     req: &HttpRequest,
+    supergraph: &Arc<SupergraphData>,
     app_state: &Arc<RouterSharedState>,
     normalized_operation: &Arc<GraphQLNormalizationPayload>,
     request_override_context: &RequestOverrideContext,
     cancellation_token: &CancellationToken,
 ) -> Result<Arc<QueryPlan>, PipelineError> {
     let stable_override_context =
-        StableOverrideContext::new(&app_state.planner.supergraph, request_override_context);
+        StableOverrideContext::new(&supergraph.planner.supergraph, request_override_context);
 
     let filtered_operation_for_plan = &normalized_operation.operation_for_plan;
     let plan_cache_key =
@@ -37,7 +39,7 @@ pub async fn plan_operation_with_cache(
                 }));
             }
 
-            app_state
+            supergraph
                 .planner
                 .plan_from_normalized_operation(
                     filtered_operation_for_plan,

bin/router/src/pipeline/validation.rs

@@ -3,17 +3,19 @@ use std::sync::Arc;
 use crate::pipeline::error::{PipelineError, PipelineErrorFromAcceptHeader, PipelineErrorVariant};
 use crate::pipeline::parser::GraphQLParserPayload;
 use crate::shared_state::RouterSharedState;
+use crate::supergraph_mgr::SupergraphData;
 use graphql_tools::validation::validate::validate;
 use ntex::web::HttpRequest;
 use tracing::{error, trace};
 
 #[inline]
 pub async fn validate_operation_with_cache(
     req: &HttpRequest,
+    supergraph: &Arc<SupergraphData>,
     app_state: &Arc<RouterSharedState>,
     parser_payload: &GraphQLParserPayload,
 ) -> Result<(), PipelineError> {
-    let consumer_schema_ast = &app_state.planner.consumer_schema.document;
+    let consumer_schema_ast = &supergraph.planner.consumer_schema.document;
 
     let validation_result = match app_state
         .validate_cache