graphql-ws

To implement authentication, use the onConnect and onSubscribe hooks in the makeServer configuration. You can access the connection context (including the HTTP request) via the ctx.extra field. If authentication fails, you can throw an error or return false from onConnect to close the connection with a 4403: Forbidden code. For more granular control, you can use ctx.extra.socket.close() within onSubscribe.

import { CloseCode, makeServer } from 'graphql-ws';
import { WebSocketServer } from 'ws';
import { validate } from './my-auth';
import { schema } from './my-graphql-schema';

interface Extra {
  readonly request: http.IncomingMessage;
}

const gqlServer = makeServer<Extra>({
  schema,
  onConnect: async (ctx) => {
    await handleAuth(ctx.extra.request);
  },
  onSubscribe: async (ctx) => {
    await handleAuth(ctx.extra.request);
  },
});

// ... implementation of wsServer.on('connection') using gqlServer.opened

When using Apollo Server Express, you need to manage the lifecycle of both the HTTP server and the WebSocket server. Use useServer to create the WebSocket server and capture the returned dispose function (often named serverCleanup). In the Apollo Server plugins configuration, use ApolloServerPluginDrainHttpServer for the HTTP server and a custom drainServer hook to call serverCleanup.dispose() for the WebSocket server.

import { createServer } from 'http';
import { ApolloServerPluginDrainHttpServer } from 'apollo-server-core';
import { ApolloServer } from 'apollo-server-express';
import express from 'express';
import { useServer } from 'graphql-ws/use/ws';
import { WebSocketServer } from 'ws';
import { schema } from './my-graphql-schema';

const app = express();
const httpServer = createServer(app);
const wsServer = new WebSocketServer({
  server: httpServer,
  path: '/graphql',
});

const serverCleanup = useServer({ schema }, wsServer);

const apolloServer = new ApolloServer({
  schema,
  plugins: [
    ApolloServerPluginDrainHttpServer({ httpServer }),
    {
      async serverWillStart() {
        return {
          async drainServer() {
            await serverCleanup.dispose();
          },
        };
      },
    },
  ],
});

await apolloServer.start();
apolloServer.applyMiddleware({ app });
httpServer.listen(4000);

To handle authentication tokens that expire, use the connectionParams option in createClient as an async function. If a token is expired, you can trigger a refresh inside this function. Additionally, use the on: { connected, closed } lifecycle hooks. In connected, set a timeout to manually close the socket with CloseCode.Forbidden when the token is expected to expire. In closed, if the event code is CloseCode.Forbidden, set a flag to trigger a token refresh on the next reconnection attempt.

// 📺 client
import { CloseCode, createClient } from 'graphql-ws';

let shouldRefreshToken = false,
  tokenExpiryTimeout = null;

const client = createClient({
  url: 'ws://server-validates.auth:4000/graphql',
  connectionParams: async () => {
    if (shouldRefreshToken) {
      await refreshCurrentToken();
      shouldRefreshToken = false;
    }
    return { token: getCurrentToken() };
  },
  on: {
    connected: (socket) => {
      clearTimeout(tokenExpiryTimeout);
      tokenExpiryTimeout = setTimeout(() => {
        if (socket.readyState === WebSocket.OPEN)
          socket.close(CloseCode.Forbidden, 'Forbidden');
      }, getCurrentTokenExpiresIn());
    },
    closed: (event) => {
      if (event.code === CloseCode.Forbidden) shouldRefreshToken = true;
    },
  },
});

To support persisted queries, the server's onSubscribe hook should check the payload.extensions.persistedQuery field. If the ID exists in your queries store, return the pre-parsed ExecutionArgs. If not, you can either throw an error to reject the request or allow the client to send the full query as a fallback.

// 🛸 server
import { ExecutionArgs, parse } from 'graphql';
import { useServer } from 'graphql-ws/use/ws';
import { WebSocketServer } from 'ws';
import { schema } from './my-graphql-schema';

type QueryID = string;
const queriesStore: Record<QueryID, ExecutionArgs> = {
  iWantTheGreetings: {
    schema,
    document: parse('subscription Greetings { greetings }'),
  },
};

const wsServer = new WebSocketServer({ port: 4000, path: '/graphql' });

useServer(
  {
    onSubscribe: (_ctx, _id, payload) => {
      const persistedQuery = queriesStore[payload.extensions?.persistedQuery];
      if (persistedQuery) {
        return {
          ...persistedQuery,
          variableValues: payload.variables,
        };
      }
      throw new Error('404: Query Not Found');
    },
  },
  wsServer,
);

// 📺 client
import { createClient } from 'graphql-ws';

const client = createClient({
  url: 'ws://persisted.graphql:4000/queries',
});

(async () => {
  await new Promise((resolve, reject) => {
    client.subscribe(
      {
        query: '',
        extensions: { persistedQuery: 'iWantTheGreetings' },
      },
      {
        next: () => {},
        error: reject,
        complete: resolve,
      },
    );
  });
})();

If your WebSocket environment does not support standard WS-level pings/pongs, you can implement them at the graphql-ws protocol level. Use stringifyMessage with MessageType.Ping to send a ping. You can then use the onPong callback in the server.opened configuration to clear any termination timeouts when a pong is received.

import {
  CloseCode,
  makeServer,
  MessageType,
  stringifyMessage,
} from 'graphql-ws';
import { WebSocketServer } from 'ws';

const server = makeServer({ schema });
const wsServer = new WebSocketServer({ port: 4000, path: '/graphql' });

wsServer.on('connection', (socket, request) => {
  let pinger, pongWait;
  function ping() {
    if (socket.readyState === socket.OPEN) {
      socket.send(stringifyMessage({ type: MessageType.Ping }));
      pongWait = setTimeout(() => {
        clearInterval(pinger);
        socket.close();
      }, 6_000);
    }
  }

  pinger = setInterval(() => ping(), 12_000);

  const closed = server.opened(
    {
      protocol: socket.protocol,
      send: (data) => socket.send(data),
      close: (code, reason) => socket.close(code, reason),
      onMessage: (cb) => socket.on('message', async (event) => {
          try { await cb(event.toString()); } 
          catch (err) { socket.close(CloseCode.InternalServerError, err.message); }
        }),
      onPong: () => clearTimeout(pongWait),
    },
    { socket, request },
  );

  socket.once('close', (code, reason) => {
    clearTimeout(pongWait);
    clearInterval(pinger);
    closed(code, reason);
  });
});

Before executing operations, a client must establish a logical connection within the socket.

1. Client sends ConnectionInit: The client sends a connection_init message. It can optionally include a payload (e.g., for authentication tokens).
2. Server responds with ConnectionAck: If successful, the server sends a connection_ack message. The client is now ready to send Subscribe messages.

Error Cases:

• If ConnectionInit is not received within the server's connectionInitWaitTimeout, the server closes the socket with 4408: Connection initialisation timeout.
• If multiple ConnectionInit messages are received simultaneously, the server closes the socket with 4429: Too many initialisation requests.
• If the server rejects the connection (e.g., failed auth), it should close the socket with 4403: Forbidden.

// Client -> Server
interface ConnectionInitMessage {
  type: 'connection_init';
  payload?: Record<string, unknown> | null;
}

// Server -> Client
interface ConnectionAckMessage {
  type: 'connection_ack';
  payload?: Record<string, unknown> | null;
}

To implement a pattern where the client is notified when a subscription is successfully acknowledged by the server, you can augment the client.subscribe method. The server uses onSubscribe to create an 'awaiter' in the context and onOperation to trigger it. The client sends a unique ackId in the extensions of the subscription payload. The server then sends a MessageType.Ping containing this ackId back to the client, which the client listens for to trigger a callback.

// 📺 client augmented version
type ClientWithSubscribeAck = Omit<Client, 'subscribe'> & {
  subscribe<Data = Record<string, unknown>, Extensions = unknown>(
    payload: SubscribePayload,
    sink: Sink<ExecutionResult<Data, Extensions>>,
    onAck: () => void,
  ): () => void;
};

function createClientWithSubscribeAck(
  options: ClientOptions,
): ClientWithSubscribeAck {
  const client = createClient(options);
  const ackListeners: Record<string, () => void> = {};

  client.on('ping', (_received, payload) => {
    const ackId = payload?.ackId;
    if (typeof ackId === 'string') {
      ackListeners[ackId]?.();
      delete ackListeners[ackId];
    }
  });

  return {
    ...client,
    subscribe: (payload, sink, onAck) => {
      const ackId = Math.random().toString();
      ackListeners[ackId] = onAck;
      return client.subscribe(
        {
          ...payload,
          extensions: { ...payload.extensions, ackId },
        },
        sink,
      );
    },
  };
}

To execute a GraphQL operation (query, mutation, or subscription), follow this lifecycle:

1. Subscribe (Client -> Server): The client sends a subscribe message with a unique id and a payload containing the query, optional operationName, variables, and extensions.
2. Execution:
   • If the id is already in use, the server closes the socket with 4409: Subscriber for <unique-operation-id> already exists.
   • If the operation is invalid or unauthorized, the server sends an Error message or closes the socket with 4401: Unauthorized.
3. Results (Next): For streaming operations, the server sends one or more next messages containing the ExecutionResult.
4. Termination:
   • Error (Server -> Client): If an error occurs, the server sends an error message and the operation terminates.
   • Complete (Bidirectional):
     • Server -> Client: Indicates the stream has finished. If an Error was sent, no Complete is sent.
     • Client -> Server: Indicates the client is stopping the subscription.

Note: Because the connection is full-duplex, both parties should be prepared to ignore messages for IDs they consider already completed.

// Client -> Server
interface SubscribeMessage {
  id: '<unique-operation-id>';
  type: 'subscribe';
  payload: {
    operationName?: string | null;
    query: string;
    variables?: Record<string, unknown> | null;
    extensions?: Record<string, unknown> | null;
  };
}

// Server -> Client (Result)
import { ExecutionResult } from 'graphql';
interface NextMessage {
  id: '<unique-operation-id>';
  type: 'next';
  payload: ExecutionResult;
}

// Server -> Client (Error)
import { GraphQLError } from 'graphql';
interface ErrorMessage {
  id: '<unique-operation-id>';
  type: 'error';
  payload: GraphQLError[];
}

// Bidirectional (Completion)
interface CompleteMessage {
  id: '<unique-operation-id>';
  type: 'complete';
}