Repository: mercurius-js/mercurius (js)

Issue:
GraphiQL fetcher uses HTTP for subscriptions
I have been experimenting with subscriptions and the way mercurius currently uses GraphiQL is slightly wrong. The [fetcher](https://github.com/mercurius-js/mercurius/blob/e0f672c4a3dc2054e9557d863183aa58e265f34b/static/main.js#L21-L30) always uses the fetch API which should not be used for subscriptions which use WebSockets.

Apparently, there is a package (`@graphiql/create-fetcher`) which comes with a function that allows for an easy way to create a fetcher, there is some good documentation on this here: https://github.com/graphql/graphiql/blob/main/packages/graphiql-toolkit/docs/create-fetcher.md

Do you agree with me that the current way of doing it is wrong? All subscriptions just return `null` when requested via GraphiQL or even throw an error if the response is non-nullable.

I can maybe look into making a PR about this next week if we agree on using `@graphiql/create-fetcher`.

Pull request description:
Fix #568
Fix for https://github.com/mercurius-js/mercurius/issues/568
This is a draft.

I have tried using `GraphiQL.createFetcher` in order to create a fetcher that works properly for subscriptions.

However, the WebSocket is getting closed with code **1002** every time I try to open a subscription. This has already been reported in https://github.com/graphql/graphiql/issues/1800 but the issue is still the same after changing the GraphiQL version.

Any ideas?

Patch:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 6975dbb..295749d 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -21,6 +21,6 @@ jobs:
     needs: test
     runs-on: ubuntu-latest
     steps:
-      - uses: fastify/github-action-merge-dependabot@v2.4.0
+      - uses: fastify/github-action-merge-dependabot@v2.5.0
         with:
           github-token: ${{secrets.GITHUB_TOKEN}}
diff --git a/docs/typescript.md b/docs/typescript.md
index 111ae0e..96713a2 100644
--- a/docs/typescript.md
+++ b/docs/typescript.md
@@ -61,7 +61,7 @@ app.register(mercurius, {
   context: buildContext
 })
 
-mercuriusCodegen(mercurius, {
+mercuriusCodegen(app, {
   // Commonly relative to your root package.json
   targetPath: './src/graphql/generated.ts'
 }).catch(console.error)
diff --git a/lib/federation.js b/lib/federation.js
index 3b56ce2..4c8736f 100644
--- a/lib/federation.js
+++ b/lib/federation.js
@@ -92,6 +92,10 @@ function getStubTypes (schemaDefinitions, isGateway) {
   const directiveDefinitions = []
 
   for (const definition of schemaDefinitions) {
+    if (definition.kind === 'SchemaDefinition') {
+      continue
+    }
+
     const typeName = definition.name.value
     const isTypeExtensionByDirective = hasExtensionDirective(definition)
     /* istanbul ignore else we are not interested in nodes that does not match if statements */
diff --git a/lib/subscription-connection.js b/lib/subscription-connection.js
index 330dd9a..3413624 100644
--- a/lib/subscription-connection.js
+++ b/lib/subscription-connection.js
@@ -49,9 +49,9 @@ module.exports = class SubscriptionConnection {
   }
 
   async handleConnection () {
-    for await (const [message] of on(this.socket, 'message')) {
+    for await (const [message, isBinary] of on(this.socket, 'message')) {
       try {
-        await this.handleMessage(message)
+        await this.handleMessage(message, isBinary)
       } catch (err) {
         this.fastify.log.error(err)
         this.handleConnectionClose()
@@ -59,10 +59,10 @@ module.exports = class SubscriptionConnection {
     }
   }
 
-  async handleMessage (message) {
+  async handleMessage (message, isBinary) {
     let data
     try {
-      data = sJSON.parse(message)
+      data = sJSON.parse(isBinary ? message : message.toString())
     } catch (e) {
       this.sendMessage(GQL_ERROR, null, 'Message must be a JSON string')
       return
@@ -197,7 +197,7 @@ module.exports = class SubscriptionConnection {
         pubsub: sc,
         lruGatewayResolvers: this.lruGatewayResolvers,
         reply: {
-          [kLoaders]: subscriptionLoaders && subscriptionLoaders.create(),
+          [kLoaders]: subscriptionLoaders && subscriptionLoaders.create(context),
           [kEntityResolvers]: this.entityResolvers,
           request: { headers: {} }
         }
diff --git a/package.json b/package.json
index afa6ae3..af3da03 100644
--- a/package.json
+++ b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "mercurius",
-  "version": "8.2.1",
+  "version": "8.4.1",
   "description": "Fastify GraphQL adapter with gateway and subscription support",
   "main": "index.js",
   "types": "index.d.ts",
@@ -71,7 +71,7 @@
     "single-user-cache": "^0.5.0",
     "tiny-lru": "^7.0.6",
     "undici": "^4.1.0",
-    "ws": "^7.5.2"
+    "ws": "^8.2.2"
   },
   "tsd": {
     "directory": "test/types"
diff --git a/static/main.js b/static/main.js
index 0e265d1..2de7bdb 100644
--- a/static/main.js
+++ b/static/main.js
@@ -1,4 +1,4 @@
-/* global fetch:false React:false ReactDOM:false GraphiQL:false */
+/* global React:false ReactDOM:false GraphiQL:false SubscriptionsTransportWs: false */
 
 const importer = {
   url: (url) => {
@@ -17,20 +17,17 @@ const importer = {
 }
 
 function render () {
-  async function fetcher (params, opts) {
-    const res = await fetch(window.GRAPHQL_ENDPOINT, {
-      method: 'post',
-      headers: {
-        ...opts.headers,
-        Accept: 'application/json',
-        'Content-Type': 'application/json'
-      },
-      body: JSON.stringify(params),
-      credentials: 'include'
-    })
+  const host = window.location.host
 
-    return res.json()
-  }
+  const websocketProtocol = window.location.protocol === 'https:' ? 'wss:' : 'ws'
+
+  const url = `${window.location.protocol}//${host}${window.GRAPHQL_ENDPOINT}`
+  const subscriptionUrl = `${websocketProtocol}://${host}${window.GRAPHQL_ENDPOINT}`
+
+  const fetcher = GraphiQL.createFetcher({
+    url,
+    legacyClient: new SubscriptionsTransportWs.SubscriptionClient(subscriptionUrl)
+  })
 
   ReactDOM.render(
     React.createElement(GraphiQL, {
@@ -48,7 +45,7 @@ if ('serviceWorker' in navigator) {
     .register('./graphiql/sw.js')
     .then(function () {
       const link = document.createElement('link')
-      link.href = 'https://unpkg.com/graphiql@1.4.0/graphiql.css'
+      link.href = 'https://unpkg.com/graphiql@1.4.2/graphiql.css'
       link.type = 'text/css'
       link.rel = 'stylesheet'
       link.media = 'screen,print'
@@ -57,7 +54,8 @@ if ('serviceWorker' in navigator) {
       return importer.urls([
         'https://unpkg.com/react@16.8.0/umd/react.production.min.js',
         'https://unpkg.com/react-dom@16.8.0/umd/react-dom.production.min.js',
-        'https://unpkg.com/graphiql@1.4.0/graphiql.min.js'
+        'https://unpkg.com/graphiql@1.4.2/graphiql.min.js',
+        'https://unpkg.com/subscriptions-transport-ws@0.9.19/browser/client.js'
       ])
     }).then(render)
 } else {
diff --git a/static/sw.js b/static/sw.js
index 6caec8f..db99ee9 100644
--- a/static/sw.js
+++ b/static/sw.js
@@ -5,10 +5,11 @@ self.addEventListener('install', function (e) {
     caches.open('graphiql-v1.4.0').then(function (cache) {
       return cache.addAll([
         './main.js',
-        'https://unpkg.com/graphiql@1.4.0/graphiql.css',
+        'https://unpkg.com/graphiql@1.4.2/graphiql.css',
         'https://unpkg.com/react@16.8.0/umd/react.production.min.js',
         'https://unpkg.com/react-dom@16.8.0/umd/react-dom.production.min.js',
-        'https://unpkg.com/graphiql@1.4.0/graphiql.min.js'
+        'https://unpkg.com/graphiql@1.4.2/graphiql.min.js',
+        'https://unpkg.com/subscriptions-transport-ws@0.9.19/browser/client.js'
       ])
     })
   )


Test patch:
diff --git a/test/federation.js b/test/federation.js
index 8045e59..ff3b1a2 100644
--- a/test/federation.js
+++ b/test/federation.js
@@ -1025,3 +1025,65 @@ test('federation schema is built correctly with type extension', async (t) => {
     }
   })
 })
+
+test('basic federation support with \'schema\' in the schema', async (t) => {
+  const app = Fastify()
+  const schema = `
+    extend type Query {
+      me: User
+    }
+
+    type User @key(fields: "id") {
+      id: ID!
+      name: String
+      username: String
+    }
+
+    schema {
+      query: Query
+    }
+  `
+
+  const resolvers = {
+    Query: {
+      me: () => {
+        return {
+          id: '1',
+          name: 'John',
+          username: '@john'
+        }
+      }
+    },
+    User: {
+      __resolveReference: (object) => {
+        return {
+          id: object.id,
+          name: 'John',
+          username: '@john'
+        }
+      }
+    }
+  }
+
+  app.register(GQL, {
+    schema,
+    resolvers,
+    federationMetadata: true
+  })
+
+  await app.ready()
+
+  const query = '{ _service { sdl } }'
+  const res = await app.inject({
+    method: 'GET',
+    url: `/graphql?query=${query}`
+  })
+
+  t.same(JSON.parse(res.body), {
+    data: {
+      _service: {
+        sdl: schema
+      }
+    }
+  })
+})
diff --git a/test/subscription-client.js b/test/subscription-client.js
index 4d85dae..2510383 100644
--- a/test/subscription-client.js
+++ b/test/subscription-client.js
@@ -11,8 +11,8 @@ test('subscription client calls the publish method with the correct payload', (t
   const port = server.address().port
 
   server.on('connection', function connection (ws) {
-    ws.on('message', function incoming (message) {
-      const data = JSON.parse(message)
+    ws.on('message', function incoming (message, isBinary) {
+      const data = JSON.parse(isBinary ? message : message.toString())
       if (data.type === 'connection_init') {
         ws.send(JSON.stringify({ id: undefined, type: 'connection_ack' }))
       } else if (data.type === 'start') {
@@ -46,8 +46,8 @@ test('subscription client calls the publish method with null after GQL_COMPLETE
   const port = server.address().port
 
   server.on('connection', function connection (ws) {
-    ws.on('message', function incoming (message) {
-      const data = JSON.parse(message)
+    ws.on('message', function incoming (message, isBinary) {
+      const data = JSON.parse(isBinary ? message : message.toString())
       if (data.type === 'connection_init') {
         ws.send(JSON.stringify({ id: undefined, type: 'connection_ack' }))
       } else if (data.type === 'start') {
@@ -99,13 +99,16 @@ test('subscription client tries to reconnect when server closes', (t) => {
   function connectionCallback () {
     if (shouldCloseServer) {
       server.close()
+      for (const ws of server.clients) {
+        ws.terminate()
+      }
       shouldCloseServer = false
       server = new WS.Server({ port }, () => {
         createSubscription()
       })
       server.on('connection', function connection (ws) {
-        ws.on('message', (message) => {
-          const data = JSON.parse(message)
+        ws.on('message', function incoming (message, isBinary) {
+          const data = JSON.parse(isBinary ? message : message.toString())
           if (data.type === 'connection_init') {
             ws.send(JSON.stringify({ id: undefined, type: 'connection_ack' }))
           } else if (data.type === 'start') {
@@ -150,8 +153,8 @@ test('subscription client multiple subscriptions is handled by one operation', t
   const port = server.address().port
 
   server.on('connection', function connection (ws) {
-    ws.on('message', function incoming (message) {
-      const data = JSON.parse(message)
+    ws.on('message', function incoming (message, isBinary) {
+      const data = JSON.parse(isBinary ? message : message.toString())
       if (data.type === 'connection_init') {
         ws.send(JSON.stringify({ id: undefined, type: 'connection_ack' }))
       } else if (data.type === 'start') {
@@ -182,8 +185,8 @@ test('subscription client multiple subscriptions unsubscribe removes only one su
   const port = server.address().port
 
   server.on('connection', function connection (ws) {
-    ws.on('message', function incoming (message) {
-      const data = JSON.parse(message)
+    ws.on('message', function incoming (message, isBinary) {
+      const data = JSON.parse(isBinary ? message : message.toString())
       if (data.type === 'stop') {
         ws.send(JSON.stringify({ id: '1', type: 'complete' }))
       }
@@ -220,8 +223,8 @@ test('subscription client closes the connection after GQL_CONNECTION_ERROR type
   const port = server.address().port
 
   server.on('connection', function connection (ws) {
-    ws.on('message', function incoming (message) {
-      const data = JSON.parse(message)
+    ws.on('message', function incoming (message, isBinary) {
+      const data = JSON.parse(isBinary ? message : message.toString())
       if (data.type === 'connection_init') {
         ws.send(JSON.stringify({ id: '1', type: 'connection_error' }))
       }
@@ -248,8 +251,8 @@ test('subscription client connectionInitPayload is correctly passed', (t) => {
   const port = server.address().port
 
   server.on('connection', function connection (ws) {
-    ws.on('message', function incoming (message) {
-      const data = JSON.parse(message)
+    ws.on('message', function incoming (message, isBinary) {
+      const data = JSON.parse(isBinary ? message : message.toString())
       if (data.type === 'connection_init') {
         t.same(data.payload, connectionInitPayload)
         client.close()
@@ -294,8 +297,8 @@ test('subscription client sending empty object payload on connection init', (t)
   const port = server.address().port
 
   server.on('connection', function connection (ws) {
-    ws.on('message', function incoming (message) {
-      const data = JSON.parse(message)
+    ws.on('message', function incoming (message, isBinary) {
+      const data = JSON.parse(isBinary ? message : message.toString())
       if (data.type === 'connection_init') {
         t.same(data.payload, {})
         ws.send(JSON.stringify({ id: '1', type: 'complete' }))
@@ -329,8 +332,8 @@ test('subscription client not throwing error on GQL_CONNECTION_KEEP_ALIVE type p
   })
 
   server.on('connection', function connection (ws) {
-    ws.on('message', function incoming (message) {
-      const data = JSON.parse(message)
+    ws.on('message', function incoming (message, isBinary) {
+      const data = JSON.parse(isBinary ? message : message.toString())
       if (data.type === 'start') {
         ws.send(JSON.stringify({ id: '1', type: 'complete' }))
       }
@@ -369,8 +372,8 @@ test('subscription client should throw on createSubscription if connection is no
   const port = server.address().port
 
   server.on('connection', function connection (ws) {
-    ws.on('message', function incoming (message) {
-      const data = JSON.parse(message)
+    ws.on('message', function incoming (message, isBinary) {
+      const data = JSON.parse(isBinary ? message : message.toString())
       if (data.type === 'connection_init') {
         ws.send(JSON.stringify({ id: undefined, type: 'connection_error' }))
       }
@@ -400,8 +403,8 @@ test('subscription client should pass the error payload to failedConnectionCallb
   const errorPayload = { message: 'error' }
 
   server.on('connection', function connection (ws) {
-    ws.on('message', function incoming (message) {
-      const data = JSON.parse(message)
+    ws.on('message', function incoming (message, isBinary) {
+      const data = JSON.parse(isBinary ? message : message.toString())
       if (data.type === 'connection_init') {
         ws.send(JSON.stringify({ id: undefined, type: 'connection_error', payload: errorPayload }))
       }
diff --git a/test/subscription.js b/test/subscription.js
index 4da9703..0841f2b 100644
--- a/test/subscription.js
+++ b/test/subscription.js
@@ -1876,3 +1876,176 @@ test('subscription server works with fastify-websocket', t => {
     }))
   })
 })
+
+test('subscription passes context to its loaders', t => {
+  const app = Fastify()
+  t.teardown(async () => {
+    await app.close()
+  })
+
+  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: String
+    }
+
+    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 loaders = {
+    Notification: {
+      message: async (queries, context) => {
+        t.ok(context, 'context is not missing')
+        const { username } = context
+        return queries.map(({ obj }) => `${obj.message} ${username}`)
+      }
+    }
+  }
+
+  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, { pubsub, topic }) => pubsub.subscribe(topic)
+      }
+    }
+  }
+
+  app.register(GQL, {
+    schema,
+    resolvers,
+    loaders,
+    subscription: {
+      emitter,
+      context: () => ({
+        topic: 'NOTIFICATION_ADDED',
+        username: 'foobar'
+      })
+    }
+  })
+
+  app.listen(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: 'Hello World foobar'
+              }
+            }
+          }
+        }))
+
+        client.end()
+        t.end()
+      } else if (data.id === 2 && data.type === 'complete') {
+        sendTestMutation()
+      }
+    })
+  })
+})