Repository: mercurius-js/mercurius
Pull request #967 (2023-02-21 17:34:16)

Use error formatter for subscription GQL errors
RE: https://github.com/mercurius-js/mercurius/issues/966

Currently, if there are graphql errors returned from a subscription in the standard "errors" property on a response - these do not get the opportunity to be passed through the potentially provided `errorFormatter`. This means any formatting and any side effects you may have in the error formatter are not run for these cases.

This PR applies the provided error formatter or merely the default error formatter as a fallback. 

NOTE:

I'm not 100% convinced on this change as is, my main thing is that in the case of mutation/query, we get to override the status code for the response in addition to the main payload. This doesn't apply to subscriptions and therefore that is thrown away in that case. It will work, but semantically _**may**_ be slightly weird. Open to feedback and suggestions of course.

mcollina: This seems correct.

Patch:
diff --git a/lib/routes.js b/lib/routes.js
index 8e5a76d..a6ab508 100644
--- a/lib/routes.js
+++ b/lib/routes.js
@@ -325,7 +325,8 @@ module.exports = async function (app, opts) {
       entityResolversFactory,
       subscriptionContextFn,
       keepAlive,
-      fullWsTransport
+      fullWsTransport,
+      errorFormatter
     })
   } else {
     app.route(getOptions)
diff --git a/lib/subscription-connection.js b/lib/subscription-connection.js
index 1740fa0..65d5e1d 100644
--- a/lib/subscription-connection.js
+++ b/lib/subscription-connection.js
@@ -19,7 +19,8 @@ module.exports = class SubscriptionConnection {
     onDisconnect,
     resolveContext,
     keepAlive,
-    fullWsTransport
+    fullWsTransport,
+    errorFormatter
   }) {
     this.fastify = fastify
     this.socket = socket
@@ -34,6 +35,7 @@ module.exports = class SubscriptionConnection {
     this.resolveContext = resolveContext
     this.keepAlive = keepAlive
     this.fullWsTransport = fullWsTransport
+    this.errorFormatter = errorFormatter
     this.headers = {}
 
     this.protocolMessageTypes = getProtocolByName(socket.protocol)
@@ -308,7 +310,10 @@ module.exports = class SubscriptionConnection {
           return this.handleConnectionClose()
         }
       }
-      this.sendMessage(this.protocolMessageTypes.GQL_DATA, id, value)
+
+      const hasErrors = Array.isArray(value.errors) && value.errors.length > 0
+      const response = hasErrors ? this.errorFormatter(value, this.context).response : value
+      this.sendMessage(this.protocolMessageTypes.GQL_DATA, id, response)
     }
 
     this.sendMessage(this.protocolMessageTypes.GQL_COMPLETE, id, null)
diff --git a/lib/subscription.js b/lib/subscription.js
index 9738541..c35cb59 100644
--- a/lib/subscription.js
+++ b/lib/subscription.js
@@ -6,7 +6,7 @@ const { kHooks } = require('./symbols')
 const SubscriptionConnection = require('./subscription-connection')
 const { getProtocolByName } = require('./subscription-protocol')
 
-function createConnectionHandler ({ subscriber, fastify, onConnect, onDisconnect, entityResolversFactory, subscriptionContextFn, keepAlive, fullWsTransport }) {
+function createConnectionHandler ({ subscriber, fastify, onConnect, onDisconnect, entityResolversFactory, subscriptionContextFn, keepAlive, fullWsTransport, errorFormatter }) {
   return async (connection, request) => {
     const { socket } = connection
 
@@ -47,7 +47,8 @@ function createConnectionHandler ({ subscriber, fastify, onConnect, onDisconnect
       context,
       resolveContext,
       keepAlive,
-      fullWsTransport
+      fullWsTransport,
+      errorFormatter
     })
 
     /* istanbul ignore next */
@@ -61,7 +62,7 @@ function createConnectionHandler ({ subscriber, fastify, onConnect, onDisconnect
 }
 
 module.exports = async function (fastify, opts) {
-  const { getOptions, subscriber, verifyClient, onConnect, onDisconnect, entityResolversFactory, subscriptionContextFn, keepAlive, fullWsTransport } = opts
+  const { getOptions, subscriber, verifyClient, onConnect, onDisconnect, entityResolversFactory, subscriptionContextFn, keepAlive, fullWsTransport, errorFormatter } = opts
 
   // If `fastify.websocketServer` exists, it means `@fastify/websocket` already registered.
   // Without this check, @fastify/websocket will be registered multiple times and raises FST_ERR_DEC_ALREADY_PRESENT.
@@ -84,7 +85,8 @@ module.exports = async function (fastify, opts) {
       entityResolversFactory,
       subscriptionContextFn,
       keepAlive,
-      fullWsTransport
+      fullWsTransport,
+      errorFormatter
     })
   })
 }


Test patch:
diff --git a/test/errors.js b/test/errors.js
index 107810a..5a92f7c 100644
--- a/test/errors.js
+++ b/test/errors.js
@@ -2,10 +2,13 @@
 
 const { test } = require('tap')
 const Fastify = require('fastify')
+const mq = require('mqemitter')
+const WebSocket = require('ws')
 const GQL = require('..')
 const { ErrorWithProps } = GQL
 const { kRequestContext } = require('../lib/symbols')
 const split = require('split2')
+const { GraphQLError } = require('graphql-jit/dist/error')
 
 test('ErrorWithProps - support status code in the constructor', async (t) => {
   const error = new ErrorWithProps('error', { }, 500)
@@ -359,6 +362,200 @@ test('errors - custom error formatter that uses default error formatter', async
   t.equal(body.errors[0].message, 'Bad Resolver')
 })
 
+test('subscription server sends correct error if there\'s a graphql error', t => {
+  const app = Fastify()
+  t.teardown(() => app.close())
+
+  const sendTestQuery = () => {
+    app.inject({
+      method: 'POST',
+      url: '/graphql',
+      body: {
+        query: `
+          query {
+            notifications {
+              id
+              message
+            }
+          }
+        `
+      }
+    }, () => {
+      sendTestMutation()
+    })
+  }
+
+  const sendTestMutation = () => {
+    app.inject({
+      method: 'POST',
+      url: '/graphql',
+      body: {
+        query: `
+          mutation {
+            addNotification(message: "Hello World") {
+              id
+            }
+          }
+        `
+      }
+    }, () => {})
+  }
+
+  const emitter = mq()
+  const schema = `
+    type Notification {
+      id: ID!
+      message: Int
+    }
+
+    type Query {
+      notifications: [Notification]
+    }
+
+    type Mutation {
+      addNotification(message: String): Notification
+    }
+
+    type Subscription {
+      notificationAdded: Notification
+    }
+  `
+
+  let idCount = 1
+  const notifications = [{
+    id: idCount,
+    message: 'Notification message'
+  }]
+
+  const resolvers = {
+    Query: {
+      notifications: () => notifications
+    },
+    Mutation: {
+      addNotification: async (_, { message }) => {
+        const id = idCount++
+        const notification = {
+          id,
+          message
+        }
+        notifications.push(notification)
+        await emitter.emit({
+          topic: 'NOTIFICATION_ADDED',
+          payload: {
+            notificationAdded: notification
+          }
+        })
+
+        return notification
+      }
+    },
+    Subscription: {
+      notificationAdded: {
+        subscribe: (root, args, ctx) => {
+          return ctx.pubsub.subscribe('NOTIFICATION_ADDED')
+        }
+      }
+    }
+  }
+
+  app.register(GQL, {
+    schema,
+    resolvers,
+    errorFormatter: (execution, ctx) => {
+      const formatted = GQL.defaultErrorFormatter(execution, ctx)
+      const errors = execution.errors.map(() => {
+        return new GraphQLError('Whoops! Something went wrong.')
+      })
+
+      return {
+        statusCode: formatted.statusCode,
+        response: {
+          data: formatted.response.data,
+          errors
+        }
+      }
+    },
+    subscription: {
+      emitter
+    }
+  })
+
+  app.listen({ port: 0 }, err => {
+    t.error(err)
+
+    const ws = new WebSocket('ws://localhost:' + (app.server.address()).port + '/graphql', 'graphql-ws')
+    const client = WebSocket.createWebSocketStream(ws, { encoding: 'utf8', objectMode: true })
+    t.teardown(client.destroy.bind(client))
+    client.setEncoding('utf8')
+
+    client.write(JSON.stringify({
+      type: 'connection_init'
+    }))
+
+    client.write(JSON.stringify({
+      id: 1,
+      type: 'start',
+      payload: {
+        query: `
+          subscription {
+            notificationAdded {
+              id
+              message
+            }
+          }
+        `
+      }
+    }))
+
+    client.write(JSON.stringify({
+      id: 2,
+      type: 'start',
+      payload: {
+        query: `
+          subscription {
+            notificationAdded {
+              id
+              message
+            }
+          }
+        `
+      }
+    }))
+
+    client.write(JSON.stringify({
+      id: 2,
+      type: 'stop'
+    }))
+
+    client.on('data', chunk => {
+      const data = JSON.parse(chunk)
+
+      if (data.id === 1 && data.type === 'data') {
+        t.equal(chunk, JSON.stringify({
+          type: 'data',
+          id: 1,
+          payload: {
+            data: {
+              notificationAdded: {
+                id: '1',
+                message: null
+              }
+            },
+            errors: [{
+              message: 'Whoops! Something went wrong.'
+            }]
+          }
+        }))
+
+        client.end()
+        t.end()
+      } else if (data.id === 2 && data.type === 'complete') {
+        sendTestQuery()
+      }
+    })
+  })
+})
+
 test('POST query with a resolver which which throws and a custom error formatter', async (t) => {
   const app = Fastify()
 
diff --git a/test/subscription.js b/test/subscription.js
index e76e330..99dd7cc 100644
--- a/test/subscription.js
+++ b/test/subscription.js
@@ -1059,6 +1059,188 @@ test('subscription server sends correct error if execution throws', t => {
   })
 })
 
+test('subscription server sends correct error if there\'s a graphql error', t => {
+  const app = Fastify()
+  t.teardown(() => app.close())
+
+  const sendTestQuery = () => {
+    app.inject({
+      method: 'POST',
+      url: '/graphql',
+      body: {
+        query: `
+          query {
+            notifications {
+              id
+              message
+            }
+          }
+        `
+      }
+    }, () => {
+      sendTestMutation()
+    })
+  }
+
+  const sendTestMutation = () => {
+    app.inject({
+      method: 'POST',
+      url: '/graphql',
+      body: {
+        query: `
+          mutation {
+            addNotification(message: "Hello World") {
+              id
+            }
+          }
+        `
+      }
+    }, () => {})
+  }
+
+  const emitter = mq()
+  const schema = `
+    type Notification {
+      id: ID!
+      message: Int
+    }
+
+    type Query {
+      notifications: [Notification]
+    }
+
+    type Mutation {
+      addNotification(message: String): Notification
+    }
+
+    type Subscription {
+      notificationAdded: Notification
+    }
+  `
+
+  let idCount = 1
+  const notifications = [{
+    id: idCount,
+    message: 'Notification message'
+  }]
+
+  const resolvers = {
+    Query: {
+      notifications: () => notifications
+    },
+    Mutation: {
+      addNotification: async (_, { message }) => {
+        const id = idCount++
+        const notification = {
+          id,
+          message
+        }
+        notifications.push(notification)
+        await emitter.emit({
+          topic: 'NOTIFICATION_ADDED',
+          payload: {
+            notificationAdded: notification
+          }
+        })
+
+        return notification
+      }
+    },
+    Subscription: {
+      notificationAdded: {
+        subscribe: (root, args, ctx) => {
+          return ctx.pubsub.subscribe('NOTIFICATION_ADDED')
+        }
+      }
+    }
+  }
+
+  app.register(GQL, {
+    schema,
+    resolvers,
+    subscription: {
+      emitter
+    }
+  })
+
+  app.listen({ port: 0 }, err => {
+    t.error(err)
+
+    const ws = new WebSocket('ws://localhost:' + (app.server.address()).port + '/graphql', 'graphql-ws')
+    const client = WebSocket.createWebSocketStream(ws, { encoding: 'utf8', objectMode: true })
+    t.teardown(client.destroy.bind(client))
+    client.setEncoding('utf8')
+
+    client.write(JSON.stringify({
+      type: 'connection_init'
+    }))
+
+    client.write(JSON.stringify({
+      id: 1,
+      type: 'start',
+      payload: {
+        query: `
+          subscription {
+            notificationAdded {
+              id
+              message
+            }
+          }
+        `
+      }
+    }))
+
+    client.write(JSON.stringify({
+      id: 2,
+      type: 'start',
+      payload: {
+        query: `
+          subscription {
+            notificationAdded {
+              id
+              message
+            }
+          }
+        `
+      }
+    }))
+
+    client.write(JSON.stringify({
+      id: 2,
+      type: 'stop'
+    }))
+
+    client.on('data', chunk => {
+      const data = JSON.parse(chunk)
+
+      if (data.id === 1 && data.type === 'data') {
+        t.equal(chunk, JSON.stringify({
+          type: 'data',
+          id: 1,
+          payload: {
+            data: {
+              notificationAdded: {
+                id: '1',
+                message: null
+              }
+            },
+            errors: [{
+              message: 'Int cannot represent non-integer value: "Hello World"',
+              locations: [{ line: 5, column: 15 }],
+              path: ['notificationAdded', 'message']
+            }]
+          }
+        }))
+
+        client.end()
+        t.end()
+      } else if (data.id === 2 && data.type === 'complete') {
+        sendTestQuery()
+      }
+    })
+  })
+})
+
 test('subscription server exposes pubsub', t => {
   const app = Fastify()
   t.teardown(() => app.close())