$request.headers and $response.headers for selection mapping in connectors

apollo-federation/src/sources/connect/validation/expression.rs

@@ -53,6 +53,7 @@ impl<'schema> Context<'schema> {
             ),
             (Namespace::Config, Shape::unknown()),
             (Namespace::Context, Shape::unknown()),
+            (Namespace::Request, Shape::unknown()),
         ]
         .into_iter()
         .collect();
@@ -299,6 +300,46 @@ mod tests {
         assert!(validate(&expression(selection), &context).is_err());
     }
 
+    #[rstest]
+    #[case("$args.int")]
+    #[case("$config.abc")]
+    #[case("$context.def")]
+    #[case("$request.headers.'apollo-client-name'")]
+    fn valid_variables(#[case] selection: &str) {
+        let schema = Schema::parse(SCHEMA, "schema").unwrap();
+        let connect = name!("connect");
+        let source = name!("source");
+        let schema_info = SchemaInfo::new(&schema, "", &connect, &source);
+        let object = schema.get_object("Query").unwrap();
+        let field = object.fields.get("aField").unwrap();
+        let directive = field.directives.get("connect").unwrap();
+        let coordinate = ConnectDirectiveCoordinate {
+            field_coordinate: FieldCoordinate { field, object },
+            directive,
+        };
+        let context = Context::for_connect_request(&schema_info, coordinate);
+        validate(&expression(selection), &context).unwrap();
+    }
+
+    #[rstest]
+    #[case("$status")]
+    #[case("$response.headers.etag")]
+    fn invalid_variables(#[case] selection: &str) {
+        let schema = Schema::parse(SCHEMA, "schema").unwrap();
+        let connect = name!("connect");
+        let source = name!("source");
+        let schema_info = SchemaInfo::new(&schema, "", &connect, &source);
+        let object = schema.get_object("Query").unwrap();
+        let field = object.fields.get("aField").unwrap();
+        let directive = field.directives.get("connect").unwrap();
+        let coordinate = ConnectDirectiveCoordinate {
+            field_coordinate: FieldCoordinate { field, object },
+            directive,
+        };
+        let context = Context::for_connect_request(&schema_info, coordinate);
+        assert!(validate(&expression(selection), &context).is_err());
+    }
+
     #[rstest]
     #[case("$args.int")]
     #[case("$args.string")]

apollo-federation/src/sources/connect/validation/snapshots/validation_tests@headers__invalid_namespace_in_header_variables.graphql.snap

@@ -1,6 +1,6 @@
 ---
 source: apollo-federation/src/sources/connect/validation/mod.rs
-expression: "format!(\"{:#?}\", errors)"
+expression: "format!(\"{:#?}\", result.errors)"
 input_file: apollo-federation/src/sources/connect/validation/test_data/headers/invalid_namespace_in_header_variables.graphql
 ---
 [
@@ -34,14 +34,14 @@ input_file: apollo-federation/src/sources/connect/validation/test_data/headers/i
     },
     Message {
         code: InvalidHeader,
-        message: "In `@connect(http.headers:)` on `Query.scalar`: $status is not valid here, must be one of $args, $config, $context",
+        message: "In `@connect(http.headers:)` on `Query.scalar`: $status is not valid here, must be one of $args, $config, $context, $request",
         locations: [
             25:62..25:69,
         ],
     },
     Message {
         code: InvalidHeader,
-        message: "In `@connect(http.headers:)` on `Query.scalar`: $this is not valid here, must be one of $args, $config, $context",
+        message: "In `@connect(http.headers:)` on `Query.scalar`: $this is not valid here, must be one of $args, $config, $context, $request",
         locations: [
             26:47..26:52,
         ],

apollo-federation/src/sources/connect/validation/snapshots/validation_tests@invalid_namespace_in_body_selection.graphql.snap

@@ -1,12 +1,12 @@
 ---
 source: apollo-federation/src/sources/connect/validation/mod.rs
-expression: "format!(\"{:#?}\", errors)"
+expression: "format!(\"{:#?}\", result.errors)"
 input_file: apollo-federation/src/sources/connect/validation/test_data/invalid_namespace_in_body_selection.graphql
 ---
 [
     Message {
         code: InvalidJsonSelection,
-        message: "In `@connect(http: {body:})` on `Mutation.createUser`: variable `$status` is not valid at this location, must be one of $args, $config, $context, $this",
+        message: "In `@connect(http: {body:})` on `Mutation.createUser`: variable `$status` is not valid at this location, must be one of $args, $config, $context, $request, $this",
         locations: [
             21:17..21:24,
         ],

apollo-federation/src/sources/connect/validation/snapshots/validation_tests@uri_templates__invalid_namespace_in_url_template_variables.graphql.snap

@@ -1,6 +1,6 @@
 ---
 source: apollo-federation/src/sources/connect/validation/mod.rs
-expression: "format!(\"{:#?}\", errors)"
+expression: "format!(\"{:#?}\", result.errors)"
 input_file: apollo-federation/src/sources/connect/validation/test_data/uri_templates/invalid_namespace_in_url_template_variables.graphql
 ---
 [
@@ -13,7 +13,7 @@ input_file: apollo-federation/src/sources/connect/validation/test_data/uri_templ
     },
     Message {
         code: InvalidUrl,
-        message: "In `GET` in `@connect(http:)` on `Query.invalid`: $status is not valid here, must be one of $args, $config, $context",
+        message: "In `GET` in `@connect(http:)` on `Query.invalid`: $status is not valid here, must be one of $args, $config, $context, $request",
         locations: [
             18:31..18:42,
         ],

apollo-federation/src/sources/connect/validation/snapshots/validation_tests@uri_templates__this_on_root_types.graphql.snap

@@ -1,19 +1,19 @@
 ---
 source: apollo-federation/src/sources/connect/validation/mod.rs
-expression: "format!(\"{:#?}\", errors)"
+expression: "format!(\"{:#?}\", result.errors)"
 input_file: apollo-federation/src/sources/connect/validation/test_data/uri_templates/this_on_root_types.graphql
 ---
 [
     Message {
         code: InvalidUrl,
-        message: "In `GET` in `@connect(http:)` on `Query.requiresThis`: $this is not valid here, must be one of $args, $config, $context",
+        message: "In `GET` in `@connect(http:)` on `Query.requiresThis`: $this is not valid here, must be one of $args, $config, $context, $request",
         locations: [
             11:39..11:51,
         ],
     },
     Message {
         code: InvalidUrl,
-        message: "In `GET` in `@connect(http:)` on `Mutation.requiresThis`: $this is not valid here, must be one of $args, $config, $context",
+        message: "In `GET` in `@connect(http:)` on `Mutation.requiresThis`: $this is not valid here, must be one of $args, $config, $context, $request",
         locations: [
             20:37..20:49,
         ],

apollo-federation/src/sources/connect/variable.rs

@@ -49,6 +49,8 @@ impl<'schema> VariableContext<'schema> {
                     Namespace::Context,
                     Namespace::Status,
                     Namespace::This,
+                    Namespace::Request,
+                    Namespace::Response,
                 ]
             }
             Phase::Request => {
@@ -57,6 +59,7 @@ impl<'schema> VariableContext<'schema> {
                     Namespace::Context,
                     Namespace::This,
                     Namespace::Args,
+                    Namespace::Request,
                 ]
             }
         }
@@ -104,6 +107,8 @@ pub enum Namespace {
     Context,
     Status,
     This,
+    Request,
+    Response,
 }
 
 impl Namespace {
@@ -114,6 +119,8 @@ impl Namespace {
             Self::Context => "$context",
             Self::Status => "$status",
             Self::This => "$this",
+            Self::Request => "$request",
+            Self::Response => "$response",
         }
     }
 }
@@ -128,6 +135,8 @@ impl FromStr for Namespace {
             "$context" => Ok(Self::Context),
             "$status" => Ok(Self::Status),
             "$this" => Ok(Self::This),
+            "$request" => Ok(Self::Request),
+            "$response" => Ok(Self::Response),
             _ => Err(()),
         }
     }

apollo-router/src/plugins/connectors/handle_responses.rs

@@ -76,6 +76,7 @@ impl RawResponse {
         connector: Arc<Connector>,
         context: &Context,
         debug_context: &Option<Arc<Mutex<ConnectorContext>>>,
+        supergraph_request: Arc<http::Request<crate::graphql::Request>>,
     ) -> connector::request_service::Response {
         let mapped_response = match self {
             RawResponse::Error { error, key } => MappedResponse::Error { error, key },
@@ -90,6 +91,8 @@ impl RawResponse {
                     connector.config.as_ref(),
                     context,
                     Some(parts.status.as_u16()),
+                    supergraph_request,
+                    Some(&parts),
                 );
 
                 let (res, apply_to_errors) = key.selection().apply_with_vars(&data, &inputs);
@@ -297,6 +300,7 @@ pub(crate) async fn process_response<T: HttpBody>(
     context: &Context,
     debug_request: Option<ConnectorDebugHttpRequest>,
     debug_context: &Option<Arc<Mutex<ConnectorContext>>>,
+    supergraph_request: Arc<http::Request<crate::graphql::Request>>,
 ) -> connector::request_service::Response {
     match result {
         // This occurs when we short-circuit the request when over the limit
@@ -346,7 +350,13 @@ pub(crate) async fn process_response<T: HttpBody>(
             };
             if is_success {
                 Span::current().record(OTEL_STATUS_CODE, OTEL_STATUS_CODE_OK);
-                raw.map_response(result, connector, context, debug_context)
+                raw.map_response(
+                    result,
+                    connector,
+                    context,
+                    debug_context,
+                    supergraph_request,
+                )
             } else {
                 Span::current().record(OTEL_STATUS_CODE, OTEL_STATUS_CODE_ERROR);
                 raw.map_error(result, connector, context, debug_context)
@@ -549,6 +559,7 @@ mod tests {
     use insta::assert_debug_snapshot;
     use url::Url;
 
+    use crate::graphql;
     use crate::plugins::connectors::handle_responses::process_response;
     use crate::plugins::connectors::make_requests::ResponseKey;
     use crate::services::router;
@@ -600,6 +611,12 @@ mod tests {
             selection: Arc::new(JSONSelection::parse("$.data").unwrap()),
         };
 
+        let supergraph_request = Arc::new(
+            http::Request::builder()
+                .body(graphql::Request::builder().build())
+                .unwrap(),
+        );
+
         let res = super::aggregate_responses(vec![
             process_response(
                 Ok(response1),
@@ -608,6 +625,7 @@ mod tests {
                 &Context::default(),
                 None,
                 &None,
+                supergraph_request.clone(),
             )
             .await
             .mapped_response,
@@ -618,6 +636,7 @@ mod tests {
                 &Context::default(),
                 None,
                 &None,
+                supergraph_request,
             )
             .await
             .mapped_response,
@@ -700,6 +719,12 @@ mod tests {
             selection: Arc::new(JSONSelection::parse("$.data").unwrap()),
         };
 
+        let supergraph_request = Arc::new(
+            http::Request::builder()
+                .body(graphql::Request::builder().build())
+                .unwrap(),
+        );
+
         let res = super::aggregate_responses(vec![
             process_response(
                 Ok(response1),
@@ -708,6 +733,7 @@ mod tests {
                 &Context::default(),
                 None,
                 &None,
+                supergraph_request.clone(),
             )
             .await
             .mapped_response,
@@ -718,6 +744,7 @@ mod tests {
                 &Context::default(),
                 None,
                 &None,
+                supergraph_request,
             )
             .await
             .mapped_response,
@@ -810,6 +837,12 @@ mod tests {
             selection: Arc::new(JSONSelection::parse("$.data").unwrap()),
         };
 
+        let supergraph_request = Arc::new(
+            http::Request::builder()
+                .body(graphql::Request::builder().build())
+                .unwrap(),
+        );
+
         let res = super::aggregate_responses(vec![
             process_response(
                 Ok(response1),
@@ -818,6 +851,7 @@ mod tests {
                 &Context::default(),
                 None,
                 &None,
+                supergraph_request.clone(),
             )
             .await
             .mapped_response,
@@ -828,6 +862,7 @@ mod tests {
                 &Context::default(),
                 None,
                 &None,
+                supergraph_request,
             )
             .await
             .mapped_response,
@@ -942,6 +977,12 @@ mod tests {
             selection: Arc::new(JSONSelection::parse("$.data").unwrap()),
         };
 
+        let supergraph_request = Arc::new(
+            http::Request::builder()
+                .body(graphql::Request::builder().build())
+                .unwrap(),
+        );
+
         let res = super::aggregate_responses(vec![
             process_response(
                 Ok(response_plaintext),
@@ -950,6 +991,7 @@ mod tests {
                 &Context::default(),
                 None,
                 &None,
+                supergraph_request.clone(),
             )
             .await
             .mapped_response,
@@ -960,6 +1002,7 @@ mod tests {
                 &Context::default(),
                 None,
                 &None,
+                supergraph_request.clone(),
             )
             .await
             .mapped_response,
@@ -970,6 +1013,7 @@ mod tests {
                 &Context::default(),
                 None,
                 &None,
+                supergraph_request.clone(),
             )
             .await
             .mapped_response,
@@ -980,6 +1024,7 @@ mod tests {
                 &Context::default(),
                 None,
                 &None,
+                supergraph_request,
             )
             .await
             .mapped_response,
@@ -1168,6 +1213,12 @@ mod tests {
             selection: Arc::new(JSONSelection::parse("$status").unwrap()),
         };
 
+        let supergraph_request = Arc::new(
+            http::Request::builder()
+                .body(graphql::Request::builder().build())
+                .unwrap(),
+        );
+
         let res = super::aggregate_responses(vec![
             process_response(
                 Ok(response1),
@@ -1176,6 +1227,7 @@ mod tests {
                 &Context::default(),
                 None,
                 &None,
+                supergraph_request,
             )
             .await
             .mapped_response,