feat: add support for @defer directive

src/core/app_context.rs

@@ -55,6 +55,7 @@ impl AppContext {
                                     http_filter,
                                     is_list,
                                     dedupe,
+                                    is_dependent,
                                     ..
                                 } => {
                                     let is_list = *is_list;
@@ -73,6 +74,7 @@ impl AppContext {
                                         http_filter: http_filter.clone(),
                                         is_list,
                                         dedupe,
+                                        is_dependent: *is_dependent,
                                     }));
 
                                     http_data_loaders.push(data_loader);

src/core/blueprint/operators/http.rs

@@ -61,8 +61,14 @@
                 .on_request
                 .clone()
                 .or(config_module.upstream.on_request.clone())
                 .map(|on_request| HttpFilter { on_request });
 
+            // check if the resolver is independent or not.
+            let is_dependent = http.query.iter().any(|q| q.value.contains("{{.value") || q.value.contains("{{value")) || http
+                    .body
+                    .as_ref()
+                    .map_or(false, |b| b.contains("{{.value") || b.contains("{{value"));
+
             let io = if !http.batch_key.is_empty() && http.method == Method::GET {
                 // Find a query parameter that contains a reference to the {{.value}} key
                 let key = http.query.iter().find_map(|q| {
@@ -77,6 +83,7 @@
                     http_filter,
                     is_list,
                     dedupe,
+                    is_dependent,
                 })
             } else {
                 IR::IO(IO::Http {
@@ -86,6 +93,7 @@
                     http_filter,
                     is_list,
                     dedupe,
+                    is_dependent,
                 })
             };
             (io, &http.select)

src/core/ir/eval.rs

@@ -146,6 +146,7 @@ impl IR {
 
                     Ok(ConstValue::object(obj))
                 }
+                IR::Deferred { ir, .. } => ir.eval(ctx).await,
             }
         })
     }

src/core/ir/model.rs

@@ -13,6 +13,15 @@ use crate::core::graphql::{self};
 use crate::core::http::HttpFilter;
 use crate::core::{grpc, http};
 
+#[derive(Clone, Debug)]
+#[allow(dead_code)]
+pub struct IrId(usize);
+impl IrId {
+    pub fn new(id: usize) -> Self {
+        Self(id)
+    }
+}
+
 #[derive(Clone, Debug, Display)]
 pub enum IR {
     Dynamic(DynamicValue<Value>),
@@ -30,6 +39,11 @@ pub enum IR {
     Entity(HashMap<String, IR>),
     /// Apollo Federation _service resolver
     Service(String),
+    Deferred {
+        id: IrId,
+        ir: Box<IR>,
+        path: Vec<String>,
+    },
 }
 
 #[derive(Clone, Debug)]
@@ -48,6 +62,7 @@ pub enum IO {
         http_filter: Option<HttpFilter>,
         is_list: bool,
         dedupe: bool,
+        is_dependent: bool,
     },
     GraphQL {
         req_template: graphql::RequestTemplate,
@@ -68,6 +83,16 @@ pub enum IO {
 }
 
 impl IO {
+    // TODO: fix is_dependent for GraphQL and Grpc.
+    pub fn is_dependent(&self) -> bool {
+        match self {
+            IO::Http { is_dependent, .. } => *is_dependent,
+            IO::GraphQL { .. } => true,
+            IO::Grpc { .. } => true,
+            IO::Js { .. } => false,
+        }
+    }
+
     pub fn dedupe(&self) -> bool {
         match self {
             IO::Http { dedupe, .. } => *dedupe,
@@ -174,6 +199,9 @@ impl IR {
                             .collect(),
                     ),
                     IR::Service(sdl) => IR::Service(sdl),
+                    IR::Deferred { ir, path, id } => {
+                        IR::Deferred { ir: Box::new(ir.modify(modifier)), path, id }
+                    }
                 }
             }
         }

src/core/jit/builder.rs

@@ -278,7 +278,34 @@
                         .map(|cond| cond.node.on.node.as_str())
                         .unwrap_or(type_condition);
 
-                    fields.extend(self.iter(&fragment.selection_set.node, type_of, fragments));
+                    // what if we've directive defined in the fragment?
+                    let mut directives = Vec::with_capacity(fragment.directives.len());
+                    for directive in fragment.directives.iter() {
+                        let directive = &directive.node;
+                        if directive.name.node == "skip" || directive.name.node == "include" {
+                            continue;
+                        }
+                        let arguments = directive
+                            .arguments
+                            .iter()
+                            .map(|(k, v)| (k.node.to_string(), v.node.clone()))
+                            .collect::<Vec<_>>();
+
+                        directives
+                            .push(JitDirective { name: directive.name.to_string(), arguments });
+                    }
+
+                    if directives.iter().find(|d| d.name == "defer").is_some() {
+                        let mut child_fields =
+                            self.iter(&fragment.selection_set.node, type_of, fragments);
+
+                        child_fields.iter_mut().for_each(|f| {
+                            f.directives.extend(directives.clone());
+                        });
+                        fields.extend(child_fields);
+                    } else {
+                        fields.extend(self.iter(&fragment.selection_set.node, type_of, fragments));
+                    }
                 }
             }
         }

src/core/jit/exec_const.rs

@@ -95,9 +95,12 @@
         let exe = Executor::new(&plan, exec);
         let store = exe.store().await;
         let synth = Synth::new(&plan, store, vars);
 
         let resp: Response<serde_json_borrow::Value> = exe.execute(&synth).await;
 
+
+        // add the pending to response.
+
         if is_introspection_query {
             let async_req = async_graphql::Request::from(request).only_introspection();
             let async_resp = app_ctx.execute(async_req).await;

src/core/jit/model.rs

@@ -291,19 +291,24 @@
     pub min_cache_ttl: Option<NonZeroU64>,
     pub selection: Vec<Field<Input>>,
     pub before: Option<IR>,
+    pub deferred_fields: Vec<Field<Input>>,
 }
 
 impl<Input> OperationPlan<Input> {
     pub fn try_map<Output, Error>(
         self,
         map: impl Fn(Input) -> Result<Output, Error>,
     ) -> Result<OperationPlan<Output>, Error> {
-        let mut selection = vec![];
-
+        let mut selection = Vec::with_capacity(self.selection.len());
         for n in self.selection {
             selection.push(n.try_map(&map)?);
         }
 
+        let mut deferred_selection = Vec::with_capacity(self.deferred_fields.len());
+        for n in self.deferred_fields {
+            deferred_selection.push(n.try_map(&map)?);
+        }
+
         Ok(OperationPlan {
             selection,
             root_name: self.root_name,
@@ -315,6 +320,7 @@
             is_protected: self.is_protected,
             min_cache_ttl: self.min_cache_ttl,
             before: self.before,
+            deferred_fields: deferred_selection,
         })
     }
 }
@@ -342,6 +348,7 @@
             is_protected: false,
             min_cache_ttl: None,
             before: Default::default(),
+            deferred_fields: Default::default(),
         }
     }
 
@@ -600,5 +607,11 @@
         let actual = plan(r#"{ users { id comments {body} } }"#);
 
         assert!(actual.is_dedupe);
     }
+
+    #[test]
+    fn test_defer(){
+        let proposed_plan = plan(r#"{ users { id ... @defer(label: "comment-defer") { comments { body } } } }"#);
+        insta::assert_debug_snapshot!(proposed_plan);
+    }
 }

src/core/jit/request.rs

@@ -50,6 +50,8 @@ impl Request<ConstValue> {
             .pipe(transform::AuthPlanner::new())
             .pipe(transform::CheckDedupe::new())
             .pipe(transform::CheckCache::new())
+            .pipe(transform::WrapDefer::new())
+            .pipe(transform::DeferPlanner::new())
             .transform(plan)
             .to_result()
             // both transformers are infallible right now

src/core/jit/response.rs

@@ -19,7 +19,10 @@
     pub extensions: Vec<(String, Value)>,
 
     #[serde(skip)]
     pub cache_control: CacheControl,
+
+    #[serde(skip_serializing_if = "Vec::is_empty")]
+    pub pending: Vec<Value>
 }
 
 impl<V: Default> Default for Response<V> {
@@ -29,6 +32,7 @@
             errors: Default::default(),
             extensions: Default::default(),
             cache_control: Default::default(),
+            pending: Default::default(),
         }
     }
 }