feat: optimize gateway by batching queries

docs/federation.md

@@ -1,13 +1,17 @@
 # mercurius
 
-- [Federation metadata support](#federation-metadata-support)
-- [Federation with \_\_resolveReference caching](#federation-with-__resolvereference-caching)
-- [Use GraphQL server as a Gateway for federated schemas](#use-graphql-server-as-a-gateway-for-federated-schemas)
-  - [Periodically refresh federated schemas in Gateway mode](#periodically-refresh-federated-schemas-in-gateway-mode)
-  - [Programmatically refresh federated schemas in Gateway mode](#programmatically-refresh-federated-schemas-in-gateway-mode)
-  - [Using Gateway mode with a schema registry](#using-gateway-mode-with-a-schema-registry)
-  - [Flag service as mandatory in Gateway mode](#flag-service-as-mandatory-in-gateway-mode)
-  - [Using a custom errorHandler for handling downstream service errors in Gateway mode](#using-a-custom-errorhandler-for-handling-downstream-service-errors-in-gateway-mode)
+- [mercurius](#mercurius)
+  - [Federation](#federation)
+    - [Federation metadata support](#federation-metadata-support)
+    - [Federation with \_\_resolveReference caching](#federation-with-__resolvereference-caching)
+    - [Use GraphQL server as a Gateway for federated schemas](#use-graphql-server-as-a-gateway-for-federated-schemas)
+      - [Periodically refresh federated schemas in Gateway mode](#periodically-refresh-federated-schemas-in-gateway-mode)
+      - [Programmatically refresh federated schemas in Gateway mode](#programmatically-refresh-federated-schemas-in-gateway-mode)
+      - [Using Gateway mode with a schema registry](#using-gateway-mode-with-a-schema-registry)
+      - [Flag service as mandatory in Gateway mode](#flag-service-as-mandatory-in-gateway-mode)
+      - [Batched Queries to services](#batched-queries-to-services)
+      - [Using a custom errorHandler for handling downstream service errors in Gateway mode](#using-a-custom-errorhandler-for-handling-downstream-service-errors-in-gateway-mode)
+      - [Securely parse service responses in Gateway mode](#securely-parse-service-responses-in-gateway-mode)
 
 ## Federation
 
@@ -351,6 +355,40 @@ server.register(mercurius, {
 server.listen(3002)
 ```
 
+#### Batched Queries to services
+
+To fully leverage the DataLoader pattern a Gateway assumes that can send a request with batched queries to its services.  
+This configuration can be disabled if the service doesn't support batched queries instead.
+
+
+```js
+const Fastify = require('fastify')
+const mercurius = require('mercurius')
+
+const server = Fastify()
+
+server.register(mercurius, {
+  graphiql: true,
+  gateway: {
+    services: [
+      {
+        name: 'user',
+        url: 'http://localhost:3000/graphql'  // queries will be batched into one request
+                                              // same as setting allowBatchedQueries: true
+      },
+      {
+        name: 'company',
+        url: 'http://localhost:3001/graphql', // one request for each of the queries
+        allowBatchedQueries: false
+      }
+    ]
+  },
+  pollingInterval: 2000
+})
+
+server.listen(3002)
+```
+
 #### Using a custom errorHandler for handling downstream service errors in Gateway mode
 
 Service which uses Gateway mode can process different types of issues that can be obtained from remote services (for example, Network Error, Downstream Error, etc.). A developer can provide a function (`gateway.errorHandler`) that can process these errors.

lib/gateway.js

@@ -4,8 +4,7 @@ const {
   getNamedType,
   isObjectType,
   isScalarType,
-  Kind,
-  parse
+  Kind
 } = require('graphql')
 const { Factory } = require('single-user-cache')
 const buildFederatedSchema = require('./federation')
@@ -18,10 +17,10 @@ const {
   kEntityResolvers
 } = require('./gateway/make-resolver')
 const { MER_ERR_GQL_GATEWAY_REFRESH, MER_ERR_GQL_GATEWAY_INIT } = require('./errors')
-const { preGatewayExecutionHandler } = require('./handlers')
 const findValueTypes = require('./gateway/find-value-types')
-
+const getQueryResult = require('./gateway/get-query-result')
 const allSettled = require('promise.allsettled')
+const defaultServiceDefinition = require('./gateway/default-service-definition')
 
 function isDefaultType (type) {
   return [
@@ -289,7 +288,9 @@ async function buildGateway (gatewayOpts, app) {
     reply[kEntityResolvers] = factory.create()
   })
 
-  for (const [service, serviceDefinition] of Object.entries(serviceMap)) {
+  for (const [service, serviceDefinitionFromGateway] of Object.entries(serviceMap)) {
+    const serviceDefinition = { ...defaultServiceDefinition, ...serviceDefinitionFromGateway }
+
     for (const type of serviceDefinition.types) {
       allTypes.push(serviceDefinition.schema.getTypeMap()[type])
       typeToServiceMap[type] = service
@@ -354,72 +355,12 @@ async function buildGateway (gatewayOpts, app) {
      *
      */
     factory.add(`${service}Entity`, async (queries) => {
-      const q = [...new Set(queries.map(q => q.query))]
-
-      const resultIndexes = []
-      let queryIndex = 0
-      const mergedQueries = queries.reduce((acc, curr) => {
-        if (!acc[curr.query]) {
-          acc[curr.query] = curr.variables
-          resultIndexes[q.indexOf(curr.query)] = []
-        } else {
-          acc[curr.query].representations = [
-            ...acc[curr.query].representations,
-            ...curr.variables.representations
-          ]
-        }
-
-        for (let i = 0; i < curr.variables.representations.length; i++) {
-          resultIndexes[q.indexOf(curr.query)].push(queryIndex)
-        }
-
-        queryIndex++
-
-        return acc
-      }, {})
-
-      const result = []
-
-      // Gateway query here
-      await Promise.all(Object.entries(mergedQueries).map(async ([query, variables], queryIndex, entries) => {
-        // Trigger preGatewayExecution hook for entities
-        let modifiedQuery
-        if (queries[queryIndex].context.preGatewayExecution !== null) {
-          ({ modifiedQuery } = await preGatewayExecutionHandler({
-            schema: serviceDefinition.schema,
-            document: parse(query),
-            context: queries[queryIndex].context,
-            service: { name: service }
-          }))
-        }
-
-        const response = await serviceDefinition.sendRequest({
-          originalRequestHeaders: queries[queryIndex].originalRequestHeaders,
-          body: JSON.stringify({
-            query: modifiedQuery || query,
-            variables
-          }),
-          context: queries[queryIndex].context
-        })
-
-        let entityIndex = 0
-        for (const entity of response.json.data._entities) {
-          if (!result[resultIndexes[queryIndex][entityIndex]]) {
-            result[resultIndexes[queryIndex][entityIndex]] = {
-              ...response,
-              json: {
-                data: {
-                  _entities: [entity]
-                }
-              }
-            }
-          } else {
-            result[resultIndexes[queryIndex][entityIndex]].json.data._entities.push(entity)
-          }
-
-          entityIndex++
-        }
-      }))
+      // context is the same for each query, but unfortunately it's not acessible from onRequest
+      // where we do factory.create(). What is a cleaner option?
+      const context = queries[0].context
+      const result = await getQueryResult({
+        context, queries, serviceDefinition, service
+      })
 
       return result
     }, query => query.id)

lib/gateway/default-service-definition.js

@@ -0,0 +1,5 @@
+const defaultServiceDefinition = {
+  allowBatchedQueries: true
+}
+
+module.exports = defaultServiceDefinition

lib/gateway/get-query-result.js

@@ -0,0 +1,140 @@
+'use strict'
+
+const { preGatewayExecutionHandler } = require('../handlers')
+
+const mergeQueries = (queries) => {
+  const q = [...new Set(queries.map(q => q.query))]
+  const resultIndexes = []
+  const mergedQueries = queries.reduce((acc, curr, queryIndex) => {
+    if (!acc[curr.query]) {
+      acc[curr.query] = {
+        document: curr.document,
+        variables: curr.variables
+      }
+      resultIndexes[q.indexOf(curr.query)] = []
+    } else {
+      acc[curr.query].variables.representations = [
+        ...acc[curr.query].variables.representations,
+        ...curr.variables.representations
+      ]
+    }
+
+    for (let i = 0; i < curr.variables.representations.length; i++) {
+      resultIndexes[q.indexOf(curr.query)].push(queryIndex)
+    }
+
+    return acc
+  }, {})
+
+  return { mergedQueries, resultIndexes }
+}
+
+const getBactchedResult = async ({ mergeQueriesResult, context, serviceDefinition, service }) => {
+  const { mergedQueries, resultIndexes } = mergeQueriesResult
+  const batchedQueries = []
+
+  for (const [query, { document, variables }] of Object.entries(mergedQueries)) {
+    let modifiedQuery
+
+    if (context.preGatewayExecution !== null) {
+      ({ modifiedQuery } = await preGatewayExecutionHandler({
+        schema: serviceDefinition.schema,
+        document,
+        context,
+        service: { name: service }
+      }))
+    }
+
+    batchedQueries.push({
+      operationName: document.definitions.find(d => d.kind === 'OperationDefinition').name.value,
+      query: modifiedQuery || query,
+      variables
+    })
+  }
+
+  const response = await serviceDefinition.sendRequest({
+    originalRequestHeaders: context.reply.request.headers,
+    body: JSON.stringify(batchedQueries),
+    context
+  })
+
+  return buildResult({ resultIndexes, data: response.json })
+}
+
+const buildResult = ({ resultIndexes, data }) => {
+  const result = []
+
+  for (const [queryIndex, queryResponse] of data.entries()) {
+    let entityIndex = 0
+
+    for (const entity of queryResponse.data._entities) {
+      if (!result[resultIndexes[queryIndex][entityIndex]]) {
+        result[resultIndexes[queryIndex][entityIndex]] = {
+          ...queryResponse,
+          json: {
+            data: {
+              _entities: [entity]
+            }
+          }
+        }
+      } else {
+        result[resultIndexes[queryIndex][entityIndex]].json.data._entities.push(entity)
+      }
+
+      entityIndex++
+    }
+  }
+
+  return result
+}
+
+const getResult = async ({ mergeQueriesResult, serviceDefinition, context, service }) => {
+  const { mergedQueries, resultIndexes } = mergeQueriesResult
+  const queriesEntries = Object.entries(mergedQueries)
+  const data = await Promise.all(
+    queriesEntries.map(async ([query, { document, variables }]) => {
+      let modifiedQuery
+
+      if (context.preGatewayExecution !== null) {
+        ({ modifiedQuery } = await preGatewayExecutionHandler({
+          schema: serviceDefinition.schema,
+          document,
+          context,
+          service: { name: service }
+        }))
+      }
+
+      const response = await serviceDefinition.sendRequest({
+        originalRequestHeaders: context.reply.request.headers,
+        body: JSON.stringify({
+          query: modifiedQuery || query,
+          variables
+        }),
+        context
+      })
+
+      return response.json
+    })
+  )
+
+  return buildResult({ data, resultIndexes })
+}
+
+const getQueryResult = async ({ context, queries, serviceDefinition, service }) => {
+  const mergeQueriesResult = mergeQueries(queries)
+  const params = {
+    mergeQueriesResult,
+    service,
+    serviceDefinition,
+    queries,
+    context
+  }
+
+  if (serviceDefinition.allowBatchedQueries) {
+    return getBactchedResult({ ...params })
+  }
+
+  return getResult({ ...params })
+}
+
+module.exports = getQueryResult

lib/gateway/make-resolver.js

@@ -508,10 +508,12 @@ function makeResolver ({ service, createOperation, transformData, isQuery, isRef
 
     const entityResolvers = reply.entityResolversFactory ? reply.entityResolversFactory.create() : reply[kEntityResolvers]
 
+    // This method is declared in gateway.js inside of onRequest
+    // hence it's unique per request.
     const response = await entityResolvers[`${service.name}Entity`]({
+      document: operation,
       query,
       variables,
-      originalRequestHeaders: reply.request.headers,
       context,
       id: queryId
     })

lib/gateway/service-map.js

@@ -221,6 +221,7 @@ async function buildServiceMap (services, errorHandler) {
     }
 
     serviceMap[service.name].name = service.name
+    serviceMap[service.name].allowBatchedQueries = service.allowBatchedQueries
   }
 
   await pmap(services, mapper, { concurrency: 8 })