Mesh subscriptions returning an error

[platoyuval]

I have a simple mesh with only one underlying service. It is an Apollo GraphQL server with a schema that has one query and one subscription (based on a public example), with CORS disabled (set to true).

When I start the mesh it runs fine. Using to the underlying service Apollo Sandbox or the Mesh graphiql UI works for both query and subscription.

I then created a small React web app that connects to the underlying service (port 4100) and subscribes as expected.
However, when pointing to the mesh (port 4200) it fails and shows an error message:

Internal error occurred during message handling. Please check your implementation. TypeError: Cannot read properties of undefined (reading 'headers')
    at onSubscribe (/Users/XX/mesh-examples/node_modules/@graphql-mesh/cli/bin.js:671:82)
    at onMessage (/Users/XX/mesh-examples/node_modules/graphql-ws/lib/server.js:146:120)
    at WebSocket.<anonymous> (/Users/XX/mesh-examples/node_modules/graphql-ws/lib/use/ws.js:83:27)
    at WebSocket.emit (node:events:513:28)
    at WebSocket.emit (node:domain:489:12)
    at Receiver.receiverOnMessage (/Users/XX/demo-graphql-mesh/mesh-examples/node_modules/ws/lib/websocket.js:1137:20)
    at Receiver.emit (node:events:513:28)
    at Receiver.emit (node:domain:489:12)
    at Receiver.dataMessage (/Users/XX/demo-graphql-mesh/mesh-examples/node_modules/ws/lib/receiver.js:528:14)
    at Receiver.getData (/Users/XX/demo-graphql-mesh/mesh-examples/node_modules/ws/lib/receiver.js:446:17)

I then built a little NodeJS websockets client, to perhaps see if this has something to do with react.
Directly with the underlying service (port 4100) it works as expected but the same issue occurred with the mesh (port 4200).

Any ideas?

The graphql-mesh code (cli/bin.js::onSusbcribe) that fails seems to not check if the connectionParams argument is assigned. But why is it unassigned?

      const { dispose: stopGraphQLWSServer } = ws$1.useServer({
            onSubscribe: async ({ connectionParams, extra: { request } }, msg) => {
                var _a;
                // spread connectionParams.headers to upgrade request headers.
                // we completely ignore the root connectionParams because
                // [@graphql-tools/url-loader adds the headers inside the "headers" field](https://github.com/ardatan/graphql-tools/blob/9a13357c4be98038c645f6efb26f0584828177cf/packages/loaders/url/src/index.ts#L597)
                for (const [key, value] of Object.entries((_a = connectionParams.headers) !== null && _a !== void 0 ? _a : {})) {
                    // dont overwrite existing upgrade headers due to security reasons
                    if (!(key.toLowerCase() in request.headers)) {
                        request.headers[key.toLowerCase()] = value;
                    }
                }

Underlying service code:

import { createServer } from 'http'

import { ApolloServer } from '@apollo/server'
import { expressMiddleware } from '@apollo/server/express4'
import { ApolloServerPluginDrainHttpServer } from '@apollo/server/plugin/drainHttpServer'
import { makeExecutableSchema } from '@graphql-tools/schema'
import bodyParser from 'body-parser'
import cors from 'cors'
import express from 'express'
import { PubSub } from 'graphql-subscriptions'
import { useServer } from 'graphql-ws/lib/use/ws'
import { WebSocketServer } from 'ws'

const PORT = 4100
const pubsub = new PubSub()

const PubSubEvent_NumberIncremented = 'NUMBER_INCREMENTED'

// Schema definition
const typeDefs = `#graphql
  type Query {
    currentNumber: Int
  }

  type Subscription {
    numberIncremented: Int
  }
`

// Resolver map
const resolvers = {
  Query: {
    currentNumber() {
      return currentNumber
    }
  },
  Subscription: {
    numberIncremented: {
      subscribe: () => pubsub.asyncIterator([PubSubEvent_NumberIncremented])
    }
  }
}

// Create schema, which will be used separately by ApolloServer and
// the WebSocket server.
const schema = makeExecutableSchema({ typeDefs, resolvers })

// Create an Express app and HTTP server; we will attach the WebSocket
// server and the ApolloServer to this HTTP server.
const app = express()
const httpServer = createServer(app)

// Set up WebSocket server.
const wsServer = new WebSocketServer({
  server: httpServer,
  path: '/graphql'
})
const serverCleanup = useServer({ schema }, wsServer)

// Set up ApolloServer.
const server = new ApolloServer({
  schema,
  plugins: [
    // Proper shutdown for the HTTP server.
    ApolloServerPluginDrainHttpServer({ httpServer }),

    // Proper shutdown for the WebSocket server.
    {
      async serverWillStart() {
        return {
          async drainServer() {
            await serverCleanup.dispose()
          }
        }
      }
    }
  ]
})

await server.start()
app.use(
  '/graphql',
  // for POC only, allow all origins
  cors<cors.CorsRequest>({ origin: true }),
  bodyParser.json(),
  expressMiddleware(server)
)

// Now that our HTTP server is fully set up, actually listen.
httpServer.listen(PORT, () => {
  console.log(`🚀 Query endpoint ready at http://localhost:${PORT}/graphql`)
  console.log(`🚀 Subscription endpoint ready at ws://localhost:${PORT}/graphql`)
})

// In the background, increment a number every second and notify subscribers when it changes.
let currentNumber = 0
function incrementNumber() {
  currentNumber++
  pubsub.publish(PubSubEvent_NumberIncremented, { numberIncremented: currentNumber })
  setTimeout(incrementNumber, 1000)
}

// Start incrementing
incrementNumber()

Mesh definition:

sources:
  - name: Subscriptions
    handler:
      graphql:
        endpoint: http://localhost:4100/graphql

serve:
  port: 4200

NodeJS WS Client:

import { createClient } from 'graphql-ws'
import WebSocket from 'ws'

const client = createClient({
  url: 'ws://localhost:4200/graphql',
  webSocketImpl: WebSocket
})

;(async () => {
  const onNext = (value) => {
    /* handle incoming values */
    console.log(`+ onNext - ${JSON.stringify(value)}`)
  }

  let unsubscribe = () => {
    /* complete the subscription */
    console.log('+ unsubscribe')
  }

  await new Promise((resolve, reject) => {
    unsubscribe = client.subscribe(
      {
        query: 'subscription { numberIncremented }'
      },
      {
        next: onNext,
        error: reject,
        complete: resolve
      }
    )
  }).catch((reason) => {
    console.log(`Failed, reason: ${JSON.stringify(reason)}`)
  })
})()

[platoyuval]

It seems similar to this past issue: https://github.com/Urigo/graphql-mesh/pull/4270/commits

[platoyuval]

I actually found a workaround by passing connectionParams

const client = createClient({
  url: 'ws://localhost:4200/graphql',
  webSocketImpl: WebSocket,
  connectionParams: {
    headers: {
      authorization: 'blah'
    }
  }
})

It now works!

But I think the graphql mesh code needs to be fixed to make the connectionParams optional.