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,41 @@ server.register(mercurius, {
 server.listen(3002)
 ```
 
+#### Batched Queries to services
+
+To fully leverage the DataLoader pattern we can tell the Gateway which of its services support [batched queries](batched-queries.md).  
+In this case the service will receive a request body with an array of queries to execute.  
+Enabling batched queries for a service that doesn't support it will generate errors.
+
+
+```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'  
+        allowBatchedQueries: true             
+      },
+      {
+        name: 'company',
+        url: 'http://localhost:3001/graphql', 
+        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,9 +17,8 @@ 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')
 
 function isDefaultType (type) {
@@ -354,72 +352,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/get-query-result.js

@@ -0,0 +1,177 @@
+'use strict'
+
+const { preGatewayExecutionHandler } = require('../handlers')
+
+/**
+ * @typedef {Object.<string, { variables: Object, document: import('graphql').DocumentNode }>} GroupedQueries
+ */
+
+/**
+ * Group GraphQL queries by their string and map them to their variables and document.
+ * @param {Array} queries
+ * @returns {GroupedQueries}
+ */
+function groupQueriesByDefinition (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 }
+}
+
+/**
+ * Fetches queries result from the service with batching (1 request for all the queries).
+ * @param {Object} params
+ * @param {Object} params.service The service that will receive one request with the batched queries
+ * @returns {Array} result
+ */
+async function fetchBactchedResult ({ 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 })
+}
+
+/**
+ *
+ * @param {Object} params
+ * @param {Array<Number[]>} params.resultIndexes Array used to map results with queries
+ * @param {Array<Object>} params.data Array of data returned from GraphQL end point
+ * @returns {Array} result
+ */
+function 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
+}
+
+/**
+ * Fetches queries result from the service without batching (1 request for each query)
+ * @param {Object} params
+ * @param {GroupedQueries} params.mergeQueriesResult
+ * @param {Object} params.service The service that will receive requests for the queries
+ * @returns {Array} result
+ */
+async function fetchResult ({ 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 })
+}
+
+/**
+ * Fetches queries results from their shared service and returns array of data.
+ * It batches queries into one request if allowBatchedQueries is true for the service.
+ * @param {Object} params
+ * @param {Array} params.queries The list of queries to be executed
+ * @param {Object} params.service The service to send requests to
+ * @returns {Array} The array of results
+ */
+async function getQueryResult ({ context, queries, serviceDefinition, service }) {
+  const mergeQueriesResult = groupQueriesByDefinition(queries)
+  const params = {
+    mergeQueriesResult,
+    service,
+    serviceDefinition,
+    queries,
+    context
+  }
+
+  if (serviceDefinition.allowBatchedQueries) {
+    return fetchBactchedResult({ ...params })
+  }
+
+  return fetchResult({ ...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 })