@@ -5,11 +5,16 @@ import {
55 type GqlModuleOptions ,
66} from '@nestjs/graphql' ;
77import type { RouteOptions as FastifyRoute } from 'fastify' ;
8+ import type { ExecutionArgs } from 'graphql' ;
9+ import { makeHandler as makeGqlWSHandler } from 'graphql-ws/use/@fastify/websocket' ;
810import {
911 createYoga ,
12+ type envelop ,
1013 type YogaServerInstance ,
1114 type YogaServerOptions ,
1215} from 'graphql-yoga' ;
16+ import { AsyncResource } from 'node:async_hooks' ;
17+ import type { WebSocket } from 'ws' ;
1318import { type GqlContextType } from '~/common' ;
1419import { HttpAdapter , type IRequest } from '../http' ;
1520import { type IResponse } from '../http/types' ;
@@ -55,7 +60,14 @@ export class Driver extends AbstractDriver<DriverConfig> {
5560 } ) ;
5661
5762 fastify . route ( {
58- method : [ 'GET' , 'POST' , 'OPTIONS' ] ,
63+ method : 'GET' ,
64+ url : this . yoga . graphqlEndpoint ,
65+ handler : this . httpHandler ,
66+ // Allow this same path to handle websocket upgrade requests.
67+ wsHandler : this . makeWsHandler ( options ) ,
68+ } ) ;
69+ fastify . route ( {
70+ method : [ 'POST' , 'OPTIONS' ] ,
5971 url : this . yoga . graphqlEndpoint ,
6072 handler : this . httpHandler ,
6173 } ) ;
@@ -77,6 +89,97 @@ export class Driver extends AbstractDriver<DriverConfig> {
7789 . send ( res . body ) ;
7890 } ;
7991
92+ /**
93+ * This code ties fastify, yoga, and graphql-ws together.
94+ * Execution layers in order:
95+ * 1. fastify route (http path matching)
96+ * 2. fastify's websocket plugin (http upgrade request & websocket open/close)
97+ * This allows our fastify hooks to be executed.
98+ * And provides a consistent Fastify `Request` type,
99+ * instead of a raw `IncomingMessage`.
100+ * 3. `graphql-ws`'s fastify handler (adapts #2 to graphql-ws)
101+ * 4. `graphql-ws` (handles specific gql protocol over websockets)
102+ * 5. `graphql-yoga` is unwrapped to `envelop`.
103+ * Yoga just wraps `envelop` and handles more of the http layer.
104+ * We really just reference `envelop` hooks with our "Yoga Plugins".
105+ * So this allows our "yoga" plugins to be executed.
106+ */
107+ private makeWsHandler ( options : DriverConfig ) {
108+ const asyncContextBySocket = new WeakMap < WebSocket , AsyncResource > ( ) ;
109+ interface WsExecutionArgs extends ExecutionArgs {
110+ socket : WebSocket ;
111+ envelop : ReturnType < ReturnType < typeof envelop > > ;
112+ }
113+
114+ // The graphql-ws handler which accepts the fastify websocket/request and
115+ // orchestrates the subscription setup & execution.
116+ // This forwards to yoga/envelop.
117+ // This was adapted from yoga's graphql-ws example.
118+ // https://github.com/dotansimha/graphql-yoga/tree/main/examples/graphql-ws
119+ const fastifyWsHandler = makeGqlWSHandler <
120+ Record < string , unknown > ,
121+ { socket : WebSocket ; request : IRequest }
122+ > ( {
123+ schema : options . schema ! ,
124+ // Custom execute/subscribe functions that really just defer to a
125+ // unique envelop (yoga) instance per request.
126+ execute : ( wsArgs ) => {
127+ const { envelop, socket, ...args } = wsArgs as WsExecutionArgs ;
128+ return asyncContextBySocket . get ( socket ) ! . runInAsyncScope ( ( ) => {
129+ return envelop . execute ( args ) ;
130+ } ) ;
131+ } ,
132+ subscribe : ( wsArgs ) => {
133+ const { envelop, socket, ...args } = wsArgs as WsExecutionArgs ;
134+ // Because this is called via socket.onmessage, we don't have
135+ // the same async context we started with.
136+ // Grab and resume it.
137+ return asyncContextBySocket . get ( socket ) ! . runInAsyncScope ( ( ) => {
138+ return envelop . subscribe ( args ) ;
139+ } ) ;
140+ } ,
141+ // Create a unique envelop/yoga instance for each subscription.
142+ // This allows "yoga" plugins that are really just envelop hooks
143+ // to be executed.
144+ onSubscribe : async ( ctx , id , payload ) => {
145+ const {
146+ extra : { request, socket } ,
147+ } = ctx ;
148+ const envelop = this . yoga . getEnveloped ( {
149+ req : request ,
150+ socket,
151+ params : payload , // Same(ish?) shape as YogaInitialContext.params
152+ } ) ;
153+
154+ const args : WsExecutionArgs = {
155+ schema : envelop . schema ,
156+ operationName : payload . operationName ,
157+ document : envelop . parse ( payload . query ) ,
158+ variableValues : payload . variables ,
159+ contextValue : await envelop . contextFactory ( ) ,
160+ // These are needed in our execute()/subscribe() declared above.
161+ // Public examples put these functions in the context, but I don't
162+ // like exposing that implementation detail to the rest of the app.
163+ envelop,
164+ socket,
165+ } ;
166+
167+ const errors = envelop . validate ( args . schema , args . document ) ;
168+ if ( errors . length ) {
169+ return errors ;
170+ }
171+ return args ;
172+ } ,
173+ } ) ;
174+
175+ const wsHandler : FastifyRoute [ 'wsHandler' ] = function ( socket , req ) {
176+ // Save a reference to the current async context, so we can resume it.
177+ asyncContextBySocket . set ( socket , new AsyncResource ( 'graphql-ws' ) ) ;
178+ return fastifyWsHandler . call ( this , socket , req ) ;
179+ } ;
180+ return wsHandler ;
181+ }
182+
80183 async stop ( ) {
81184 // noop
82185 }
0 commit comments