import { Server , IncomingMessage , ServerResponse , OutgoingHttpHeaders } from 'http' ; import { HttpRequestHandler , mixed , Middleware } from '../../interfaces' ; import { subscribe as graphqlSubscribe , ExecutionResult , ExecutionArgs , specifiedRules , validate , GraphQLError , parse , DocumentNode , execute , getOperationAST , } from 'graphql' ; import * as WebSocket from 'ws' ; import { SubscriptionServer , ConnectionContext , ExecutionParams } from 'subscriptions-transport-ws' ; import { GRAPHQL_TRANSPORT_WS_PROTOCOL } from 'graphql-ws' ; import { useServer } from 'graphql-ws/lib/use/ws' ; import parseUrl = require ( 'parseurl' ) ; import { pluginHookFromOptions } from '../pluginHook' ; import { isEmpty } from './createPostGraphileHttpRequestHandler' ; import { makeLiveSubscribe } from './liveSubscribe' ; interface Deferred < T > extends Promise < T > { resolve : ( input ? : T | PromiseLike < T > | undefined ) => void ; reject : ( error : Error ) => void ; } function lowerCaseKeys ( obj ) { return Object . keys ( obj ) . reduce ( ( memo , key ) => { memo [ key . toLowerCase ( ) ] = obj [ key ] ; return memo ; } , { } ) ; } function deferred < T = void > ( ) { let resolve ; let reject ; const promise = new Promise < T > ( ( _resolve , _reject ) : void => { resolve = _resolve ; reject = _reject ; } ) ; return Object . assign ( promise , { resolve , reject , } ) ; } export async function enhanceHttpServerWithWebSockets < Request extends IncomingMessage = IncomingMessage , Response extends ServerResponse = ServerResponse > ( websocketServer , postgraphileMiddleware , subscriptionServerOptions ? ) { if ( websocketServer [ '__postgraphileSubscriptionsEnabled' ] ) { return ; } websocketServer [ '__postgraphileSubscriptionsEnabled' ] = true ; const { options , getGraphQLSchema , withPostGraphileContextFromReqRes , handleErrors , } = postgraphileMiddleware ; const pluginHook = pluginHookFromOptions ( options ) ; const liveSubscribe = makeLiveSubscribe ( { pluginHook , options } ) ; const graphqlRoute = ( subscriptionServerOptions && subscriptionServerOptions . graphqlRoute ) || ( options . externalUrlBase || '' ) + ( options . graphqlRoute || '/graphql' ) ; const { subscriptions , live , websockets = subscriptions || live ? [ 'v0' , 'v1' ] : [ ] } = options ; if ( ! websockets ?. length ) { throw new Error ( `Invalid value for \`websockets\` option: '${ JSON . stringify ( websockets ) }'` ) ; } const schema = await getGraphQLSchema ( ) ; const keepalivePromisesByContextKey = { } ; const contextKey = ( ws , opId ) : string => ws [ 'postgraphileId' ] + '|' + opId ; const releaseContextForSocketAndOpId = ( ws , opId ) : void => { const promise = keepalivePromisesByContextKey [ contextKey ( ws , opId ) ] ; if ( promise ) { promise . resolve ( ) ; keepalivePromisesByContextKey [ contextKey ( ws , opId ) ] = null ; } } ; const addContextForSocketAndOpId = ( context , ws , opId ) : Deferred < void > => { releaseContextForSocketAndOpId ( ws , opId ) ; const promise = deferred ( ) ; promise [ 'context' ] = context ; keepalivePromisesByContextKey [ contextKey ( ws , opId ) ] = promise ; return promise ; } ; const applyMiddleware = async ( middlewares = [ ] , req , res ) : Promise < void > => { for ( const middleware of middlewares ) { await new Promise < void > ( ( resolve , reject ) : void => { middleware ( req , res , err => ( err ? reject ( err ) : resolve ( ) ) ) ; } ) ; } } ; const reqResFromSocket = async ( socket ) : Promise < { req : Request ; res : Response ; } > => { const req = socket [ '__postgraphileReq' ] ; if ( ! req ) { throw new Error ( 'req could not be extracted' ) ; } let dummyRes = socket [ '__postgraphileRes' ] ; if ( req . res ) { throw new Error ( "Please get in touch with Benjie; we weren't expecting req.res to be present but we want to reserve it for future usage." ) ; } if ( ! dummyRes ) { dummyRes = new ServerResponse ( req ) as Response ; dummyRes . writeHead = ( statusCode , _statusMessage ? , headers ? ) : void => { if ( statusCode && statusCode > 200 ) { console . error ( `Something used 'writeHead' to write a '${ statusCode }' error for websockets - check the middleware you're passing!` ) ; socket . close ( ) ; } else if ( headers ) { console . error ( "Passing headers to 'writeHead' is not supported with websockets currently - check the middleware you're passing" ) ; socket . close ( ) ; } } ; await applyMiddleware ( options . websocketMiddlewares , req , dummyRes ) ; socket [ '__postgraphileRes' ] = dummyRes ; } return { req , res : dummyRes } ; } ; const getContext = ( socket , opId , isSubscription ) : Promise < mixed > => { const singleStatement = isSubscription ; return new Promise ( ( resolve , reject ) : void => { reqResFromSocket ( socket ) . then ( ( { req , res } ) => withPostGraphileContextFromReqRes ( req , res , { singleStatement } , context => { const promise = addContextForSocketAndOpId ( context , socket , opId ) ; resolve ( promise [ 'context' ] ) ; return promise ; } ) ) . then ( null , reject ) ; } ) ; } ; const staticValidationRules = pluginHook ( 'postgraphile:validationRules:static' , specifiedRules , { options , } ) ; let socketId = 0 ; let v0Wss = null ; if ( websockets . includes ( 'v0' ) ) { v0Wss = new WebSocket . Server ( { noServer : true } ) ; SubscriptionServer . create ( { schema , validationRules : staticValidationRules , execute : options . websocketOperations === 'all' ? execute : ( ) => { throw new Error ( 'Only subscriptions are allowed over websocket transport' ) ; } , subscribe : options . live ? liveSubscribe : graphqlSubscribe , onConnect ( connectionParams , _socket , connectionContext ) { const { socket , request } = connectionContext ; socket [ 'postgraphileId' ] = ++ socketId ; if ( ! request ) { throw new Error ( 'No request!' ) ; } const normalizedConnectionParams = lowerCaseKeys ( connectionParams ) ; request [ 'connectionParams' ] = connectionParams ; request [ 'normalizedConnectionParams' ] = normalizedConnectionParams ; socket [ '__postgraphileReq' ] = request ; if ( ! request . headers . authorization && normalizedConnectionParams [ 'authorization' ] ) { request . headers . authorization = String ( normalizedConnectionParams [ 'authorization' ] ) ; } socket [ 'postgraphileHeaders' ] = { ... normalizedConnectionParams , ... request . headers , } ; } , async onOperation ( message , params , socket ) { const opId = message . id ; params . schema = await getGraphQLSchema ( ) ; const { req , res } = await reqResFromSocket ( socket ) ; const meta = { } ; const formatResponse = < TExecutionResult extends ExecutionResult = ExecutionResult > ( response ) : TExecutionResult => { if ( response . errors ) { response . errors = handleErrors ( response . errors , req , res ) ; } if ( ! isEmpty ( meta ) ) { response [ 'meta' ] = meta ; } return response ; } ; params . formatResponse = formatResponse ; const hookedParams = pluginHook ? pluginHook ( 'postgraphile:ws:onOperation' , params , { message , params , socket , options , } ) : params ; const finalParams = { ... hookedParams , query : typeof hookedParams . query !== 'string' ? hookedParams . query : parse ( hookedParams . query ) , } ; const operation = getOperationAST ( finalParams . query , finalParams . operationName ) ; const isSubscription = ! ! operation && operation . operation === 'subscription' ; const context = await getContext ( socket , opId , isSubscription ) ; Object . assign ( params . context , context ) ; const moreValidationRules = pluginHook ( 'postgraphile:validationRules' , [ ] , { options , req , res , variables : params . variables , operationName : params . operationName , meta , } ) ; if ( moreValidationRules . length ) { const validationErrors = validate ( params . schema , finalParams . query , moreValidationRules ) ; if ( validationErrors . length ) { const error = new Error ( 'Query validation failed: \n' + validationErrors . map ( e => e . message ) . join ( '\n' ) ) ; error [ 'errors' ] = validationErrors ; return Promise . reject ( error ) ; } } return finalParams ; } , onOperationComplete ( socket , opId ) { releaseContextForSocketAndOpId ( socket , opId ) ; } , keepAlive : 15000 , ... subscriptionServerOptions , } , v0Wss ) ; } let v1Wss = null ; if ( websockets . includes ( 'v1' ) ) { v1Wss = new WebSocket . Server ( { noServer : true } ) ; useServer ( { schema , execute : options . websocketOperations === 'all' ? execute : ( ) => { throw new Error ( 'Only subscriptions are allowed over WebSocket transport' ) ; } , subscribe : options . live ? liveSubscribe : graphqlSubscribe , onConnect ( ctx ) { const { socket , request } = ctx . extra ; socket [ 'postgraphileId' ] = ++ socketId ; socket [ '__postgraphileReq' ] = request ; const normalizedConnectionParams = lowerCaseKeys ( ctx . connectionParams || { } ) ; request [ 'connectionParams' ] = ctx . connectionParams || { } ; request [ 'normalizedConnectionParams' ] = normalizedConnectionParams ; if ( ! request . headers . authorization && normalizedConnectionParams [ 'authorization' ] ) { request . headers . authorization = String ( normalizedConnectionParams [ 'authorization' ] ) ; } socket [ 'postgraphileHeaders' ] = { ... normalizedConnectionParams , ... request . headers , } ; } , async onSubscribe ( ctx , msg ) { const schema = await getGraphQLSchema ( ) ; const { payload } = msg ; const args = { schema , contextValue : { } , operationName : payload . operationName , document : payload . query ? parse ( payload . query ) : null , variableValues : payload . variables , } ; const hookedArgs = ( pluginHook ? pluginHook ( 'postgraphile:ws:onSubscribe' , args , { context : ctx , message : msg , options , } ) : args ) as ExecutionArgs ; const operation = args . document ? getOperationAST ( args . document , hookedArgs . operationName ) : null ; const isSubscription = ! ! operation && operation . operation === 'subscription' ; const context = await getContext ( ctx . extra . socket , msg . id , isSubscription ) ; Object . assign ( hookedArgs . contextValue , context ) ; const validationErrors = validate ( hookedArgs . schema , hookedArgs . document , staticValidationRules ) ; if ( validationErrors . length ) { return validationErrors ; } const { req , res } = await reqResFromSocket ( ctx . extra . socket ) ; const moreValidationRules = pluginHook ( 'postgraphile:validationRules' , [ ] , { options , req , res , variables : hookedArgs . variableValues , operationName : hookedArgs . operationName , } ) ; if ( moreValidationRules . length ) { const moreValidationErrors = validate ( hookedArgs . schema , hookedArgs . document , moreValidationRules ) ; if ( moreValidationErrors . length ) { return moreValidationErrors ; } } return hookedArgs ; } , async onError ( ctx , msg , errors ) { releaseContextForSocketAndOpId ( ctx . extra . socket , msg . id ) ; const { req , res } = await reqResFromSocket ( ctx . extra . socket ) ; return handleErrors ( errors , req , res ) ; } , async onNext ( ctx , _msg , _args , result ) { if ( result . errors ) { const { req , res } = await reqResFromSocket ( ctx . extra . socket ) ; result . errors = handleErrors ( result . errors , req , res ) ; return result ; } } , onComplete ( ctx , msg ) { releaseContextForSocketAndOpId ( ctx . extra . socket , msg . id ) ; } , } , v1Wss , subscriptionServerOptions ?. keepAlive ) ; } websocketServer . on ( 'upgrade' , ( req , socket , head ) => { const { pathname = '' } = parseUrl ( req ) || { } ; const isGraphqlRoute = pathname === graphqlRoute ; if ( isGraphqlRoute ) { const protocol = req . headers [ 'sec-websocket-protocol' ] ; const protocols = Array . isArray ( protocol ) ? protocol : protocol ?. split ( ',' ) . map ( p => p . trim ( ) ) ; const wss = v0Wss && protocols ?. includes ( 'graphql-ws' ) && ! protocols . includes ( GRAPHQL_TRANSPORT_WS_PROTOCOL ) ? v0Wss : v1Wss ; if ( wss ) { wss . handleUpgrade ( req , socket , head , ws => { wss . emit ( 'connection' , ws , req ) ; } ) ; } } } ) ; }