fix: run on_upstream_http_request hook in federation source

libs/engine/src/source/federation_source.rs

@@ -3,10 +3,12 @@ use crate::gateway::ConductorGatewayRouteData;
 use base64::{engine, Engine};
 use conductor_common::execute::RequestExecutionContext;
 use conductor_common::graphql::GraphQLResponse;
+use conductor_common::http::ConductorHttpRequest;
 use conductor_config::{FederationSourceConfig, SupergraphSourceConfig};
 use federation_query_planner::execute_federation;
 use federation_query_planner::supergraph::{parse_supergraph, Supergraph};
 use minitrace_reqwest::{traced_reqwest, TracedHttpClient};
+use reqwest::Method;
 use std::collections::HashMap;
 use std::{future::Future, pin::Pin};
 
@@ -190,7 +192,7 @@ impl SourceRuntime for FederationSourceRuntime {
 
   fn execute<'a>(
     &'a self,
-    _route_data: &'a ConductorGatewayRouteData,
+    route_data: &'a ConductorGatewayRouteData,
     request_context: &'a mut RequestExecutionContext,
   ) -> Pin<Box<(dyn Future<Output = Result<GraphQLResponse, SourceError>> + 'a)>> {
     Box::pin(wasm_polyfills::call_async(async move {
@@ -199,13 +201,26 @@ impl SourceRuntime for FederationSourceRuntime {
         .take()
         .expect("GraphQL request isn't available at the time of execution");
 
-      //   let source_req = &mut downstream_request.request;
-
-      // TODO: this needs to be called by conductor execution when fetching subgarphs
-      //   route_data
-      //     .plugin_manager
-      //     .on_upstream_graphql_request(source_req)
-      //     .await;
+      let source_req = &downstream_request.request;
+      let endpoint = &route_data.endpoint;
+
+      // TODO: improve this by implementing https://github.com/the-guild-org/conductor-t2/issues/205
+      let mut conductor_http_request = ConductorHttpRequest {
+        body: source_req.into(),
+        uri: endpoint.to_string(),
+        query_string: "".to_string(),
+        method: Method::POST,
+        headers: Default::default(),
+      };
+
+      route_data
+        .plugin_manager
+        .on_upstream_http_request(request_context, &mut conductor_http_request)
+        .await;
+
+      if request_context.is_short_circuit() {
+        return Err(SourceError::ShortCircuit);
+      }
 
       let operation = downstream_request.parsed_operation;