feat: @defer support

index.js

@@ -1,5 +1,6 @@
 'use strict'
 
+const Negotiator = require('negotiator')
 const fp = require('fastify-plugin')
 let LRU = require('tiny-lru')
 const routes = require('./lib/routes')
@@ -18,10 +19,10 @@ const {
   extendSchema,
   validate,
   validateSchema,
-  specifiedRules,
-  execute
+  specifiedRules
 } = require('graphql')
 const { buildExecutionContext } = require('graphql/execution/execute')
+const { Readable } = require('stream')
 const queryDepth = require('./lib/queryDepth')
 const buildFederationSchema = require('./lib/federation')
 const { initGateway } = require('./lib/gateway')
@@ -47,6 +48,7 @@ const {
   preExecutionHandler,
   onResolutionHandler
 } = require('./lib/handlers')
+const { executeGraphql, MEDIA_TYPES } = require('./lib/util')
 
 // Required for module bundlers
 // istanbul ignore next
@@ -546,7 +548,7 @@ const plugin = fp(async function (app, opts) {
       return maybeFormatErrors(execution, context)
     }
 
-    const execution = await execute({
+    const execution = await executeGraphql({
       schema: modifiedSchema || fastifyGraphQl.schema,
       document: modifiedDocument || document,
       rootValue: root,
@@ -555,9 +557,95 @@ const plugin = fp(async function (app, opts) {
       operationName
     })
 
+    /* istanbul ignore next */
+    if (execution.initialResult) {
+      const acceptHeader = reply.request.raw.headers.accept
+
+      if (
+        !(
+          acceptHeader &&
+          new Negotiator({
+            headers: { accept: acceptHeader }
+          }).mediaType([
+            // mediaType() will return the first one that matches, so if the client
+            // doesn't include the deferSpec parameter it will match this one here,
+            // which isn't good enough.
+            MEDIA_TYPES.MULTIPART_MIXED_NO_DEFER_SPEC,
+            MEDIA_TYPES.MULTIPART_MIXED_EXPERIMENTAL
+          ]) === MEDIA_TYPES.MULTIPART_MIXED_EXPERIMENTAL
+        )
+      ) {
+        // The client ran an operation that would yield multiple parts, but didn't
+        // specify `accept: multipart/mixed`. We return an error.
+        throw new Error(
+          'Mercurius server received an operation that uses incremental delivery ' +
+          '(@defer or @stream), but the client does not accept multipart/mixed ' +
+          'HTTP responses. To enable incremental delivery support, add the HTTP ' +
+          "header 'Accept: multipart/mixed; deferSpec=20220824'.",
+          // Use 406 Not Accepted TODO
+          { extensions: { http: { status: 406 } } }
+        )
+      }
+
+      reply.header('content-type', 'multipart/mixed; boundary="-"; deferSpec=20220824')
+
+      return Readable.from(
+        writeMultipartBody(
+          execution.initialResult,
+          execution.subsequentResults
+        )
+      )
+    }
+
     return maybeFormatErrors(execution, context)
   }
 
+  /* istanbul ignore next */
+  async function * writeMultipartBody (initialResult, subsequentResults) {
+    yield `\r\n---\r\ncontent-type: application/json; charset=utf-8\r\n\r\n${JSON.stringify(
+      orderInitialIncrementalExecutionResultFields(initialResult)
+    )}\r\n---${initialResult.hasNext ? '' : '--'}\r\n`
+
+    for await (const result of subsequentResults) {
+      yield `content-type: application/json; charset=utf-8\r\n\r\n${JSON.stringify(
+        orderSubsequentIncrementalExecutionResultFields(result)
+      )}\r\n---${result.hasNext ? '' : '--'}\r\n`
+    }
+  }
+
+  /* istanbul ignore next */
+  function orderInitialIncrementalExecutionResultFields (result) {
+    return {
+      hasNext: result.hasNext,
+      errors: result.errors,
+      data: result.data,
+      incremental: orderIncrementalResultFields(result.incremental),
+      extensions: result.extensions
+    }
+  }
+
+  /* istanbul ignore next */
+  function orderSubsequentIncrementalExecutionResultFields (result) {
+    return {
+      hasNext: result.hasNext,
+      incremental: orderIncrementalResultFields(result.incremental),
+      extensions: result.extensions
+    }
+  }
+
+  /* istanbul ignore next */
+  function orderIncrementalResultFields (incremental) {
+    return incremental?.map((i) => ({
+      hasNext: i.hasNext,
+      errors: i.errors,
+      path: i.path,
+      label: i.label,
+      data: i.data,
+      items: i.items,
+      extensions: i.extensions
+    }))
+  }
+
   async function maybeFormatErrors (execution, context) {
     execution = addErrorsToExecutionResult(execution, context.errors)
 

lib/errors.js

@@ -35,12 +35,14 @@ function toGraphQLError (err) {
 
   const gqlError = new GraphQLError(
     err.message,
-    err.nodes,
-    err.source,
-    err.positions,
-    err.path,
-    err,
-    err.extensions
+    {
+      nodes: err.nodes,
+      source: err.source,
+      positions: err.positions,
+      path: err.path,
+      originalError: err,
+      extensions: err.extensions
+    }
   )
 
   gqlError.locations = err.locations

lib/util.js

@@ -1,5 +1,8 @@
 'use strict'
 
+const { execute } = require('graphql')
+const { experimentalExecuteIncrementally } = require('graphql/execution')
+
 function hasDirective (directiveName, node) {
   if (!node.directives || node.directives.length < 1) {
     return false
@@ -23,7 +26,26 @@ function hasExtensionDirective (node) {
   }
 }
 
+const canUseIncrementalExecution = !!require('graphql/execution').experimentalExecuteIncrementally
+
+// istanbul ignore next
+function executeGraphql (args) {
+  if (canUseIncrementalExecution) {
+    return experimentalExecuteIncrementally(args)
+  }
+
+  return execute(args)
+}
+
+const MEDIA_TYPES = {
+  MULTIPART_MIXED_NO_DEFER_SPEC: 'multipart/mixed',
+  MULTIPART_MIXED_EXPERIMENTAL: 'multipart/mixed; deferSpec=20220824'
+}
+
 module.exports = {
   hasDirective,
-  hasExtensionDirective
+  hasExtensionDirective,
+  canUseIncrementalExecution,
+  executeGraphql,
+  MEDIA_TYPES
 }

package.json

@@ -1,6 +1,6 @@
 {
   "name": "mercurius",
-  "version": "11.1.0",
+  "version": "11.2.0",
   "description": "Fastify GraphQL adapter with gateway and subscription support",
   "main": "index.js",
   "types": "index.d.ts",
@@ -64,6 +64,7 @@
     "graphql": "^16.0.0",
     "graphql-jit": "^0.7.3",
     "mqemitter": "^5.0.0",
+    "negotiator": "^0.6.3",
     "p-map": "^4.0.0",
     "readable-stream": "^4.0.0",
     "safe-stable-stringify": "^2.3.0",

test/directives.js

@@ -14,6 +14,7 @@ const {
   getResolversFromSchema
 } = require('@graphql-tools/utils')
 const buildFederationSchema = require('../lib/federation')
+const { canUseIncrementalExecution } = require('../lib/util')
 
 class ValidationError extends Error {
   constructor (message, extensions) {
@@ -962,3 +963,147 @@ type User {
     }
   })
 })
+
+if (canUseIncrementalExecution) {
+  const schema = `
+    directive @defer(
+      if: Boolean! = true
+      label: String
+    ) on FRAGMENT_SPREAD | INLINE_FRAGMENT
+
+    type Query {
+      allProducts: [Product!]!
+    }
+
+    type Product {
+      delivery: DeliveryEstimates!
+      sku: String!
+      id: ID!
+    }
+
+    type DeliveryEstimates {
+      estimatedDelivery: String!
+      fastestDelivery: String!
+    }`
+
+  const allProducts = new Array(1).fill(0).map((_, index) => ({
+    id: `${index}`,
+    sku: 'sku'
+  }))
+
+  const resolvers = {
+    Query: {
+      allProducts: () => {
+        return allProducts
+      }
+    },
+    Product: {
+      delivery: () => {
+        return {
+          estimatedDelivery: '25.01.2000',
+          fastestDelivery: '25.01.2000'
+        }
+      }
+    }
+  }
+
+  const query = `
+    query deferVariation {
+      allProducts {
+        delivery {
+          ...MyFragment @defer
+        }
+        sku
+        id
+      }
+    }
+
+    fragment MyFragment on DeliveryEstimates {
+      estimatedDelivery
+      fastestDelivery
+    }
+  `
+
+  test('errors with @defer when used with wrong "accept" header', async t => {
+    const wrongAcceptValues = [
+      '',
+      'application/json',
+      'multipart/mixed',
+      'multipart/mixed; deferSpec=12345'
+    ]
+
+    for (const accept of wrongAcceptValues) {
+      const app = Fastify()
+      await app.register(mercurius, { schema, resolvers, graphiql: true })
+
+      const res = await app.inject({
+        method: 'POST',
+        url: '/graphql',
+        headers: {
+          'content-type': 'application/json',
+          accept
+        },
+        body: JSON.stringify({ query })
+      })
+
+      t.match(res, {
+        statusCode: 200,
+        body: JSON.stringify({
+          data: null,
+          errors: [{
+            message: "Mercurius server received an operation that uses incremental delivery (@defer or @stream), but the client does not accept multipart/mixed HTTP responses. To enable incremental delivery support, add the HTTP header 'Accept: multipart/mixed; deferSpec=20220824'."
+          }]
+        })
+      })
+
+      await app.close()
+    }
+
+    t.end()
+  })
+
+  test('works with @defer when used with correct "accept" header', async t => {
+    const correctAcceptValues = [
+      'multipart/mixed; deferSpec=20220824',
+      'multipart/mixed; deferSpec=20220824, application/json',
+      'application/json, multipart/mixed; deferSpec=20220824'
+    ]
+
+    for (const accept of correctAcceptValues) {
+      const app = Fastify()
+      await app.register(mercurius, { schema, resolvers, graphiql: true })
+
+      const res = await app.inject({
+        method: 'POST',
+        url: '/graphql',
+        headers: {
+          'content-type': 'application/json',
+          accept
+        },
+        body: JSON.stringify({ query })
+      })
+
+      t.match(res, {
+        statusCode: 200,
+        headers: {
+          'content-type': 'multipart/mixed; boundary="-"; deferSpec=20220824'
+        },
+        body: `\r
+---\r
+content-type: application/json; charset=utf-8\r
+\r
+{"hasNext":true,"data":{"allProducts":[{"delivery":{},"sku":"sku","id":"0"}]}}\r
+---\r
+content-type: application/json; charset=utf-8\r
+\r
+{"hasNext":false,"incremental":[{"path":["allProducts",0,"delivery"],"data":{"estimatedDelivery":"25.01.2000","fastestDelivery":"25.01.2000"}}]}\r
+-----\r
+`
+      })
+
+      await app.close()
+    }
+
+    t.end()
+  })
+}