/**
 *
 * handler
 *
 */ import type { IncomingMessage , ServerResponse } from 'http' ; import { ExecutionArgs , getOperationAST , GraphQLSchema , OperationTypeNode , parse , validate as graphqlValidate , execute as graphqlExecute , subscribe as graphqlSubscribe , } from 'graphql' ; import { isObject } from './utils' ; import { RequestParams , StreamEvent , StreamData , StreamDataForID , ExecutionResult , ExecutionPatchResult , TOKEN_HEADER_KEY , TOKEN_QUERY_KEY , } from './common' ; category export type ExecutionContext = object | symbol | number | string | boolean | undefined | null ; category export type OperationResult = Promise < AsyncGenerator < ExecutionResult | ExecutionPatchResult > | AsyncIterable < ExecutionResult | ExecutionPatchResult > | ExecutionResult > | AsyncGenerator < ExecutionResult | ExecutionPatchResult > | AsyncIterable < ExecutionResult | ExecutionPatchResult > | ExecutionResult ; category export interface HandlerOptions < Request extends IncomingMessage = IncomingMessage , Response extends ServerResponse = ServerResponse > { /**
     * The GraphQL schema on which the operations will
     * be executed and validated against.
     *
     * If a function is provided, it will be called on every
     * subscription request allowing you to manipulate schema
     * dynamically.
     *
     * If the schema is left undefined, you're trusted to
     * provide one in the returned `ExecutionArgs` from the
     * `onSubscribe` callback.
     */ schema ? : GraphQLSchema | ( ( req : Request , args : Omit < ExecutionArgs , 'schema' > ) => Promise < GraphQLSchema > | GraphQLSchema ) ; /**
     * A value which is provided to every resolver and holds
     * important contextual information like the currently
     * logged in user, or access to a database.
     *
     * Note that the context function is invoked on each operation only once.
     * Meaning, for subscriptions, only at the point of initialising the subscription;
     * not on every subscription event emission. Read more about the context lifecycle
     * in subscriptions here: https://github.com/graphql/graphql-js/issues/894.
     */ context ? : ExecutionContext | ( ( req : Request , args : ExecutionArgs ) => Promise < ExecutionContext > | ExecutionContext ) ; /**
     * A custom GraphQL validate function allowing you to apply your
     * own validation rules.
     */ validate ? : typeof graphqlValidate ; /**
     * Is the `execute` function from GraphQL which is
     * used to execute the query and mutation operations.
     */ execute ? : ( args : ExecutionArgs ) => OperationResult ; /**
     * Is the `subscribe` function from GraphQL which is
     * used to execute the subscription operation.
     */ subscribe ? : ( args : ExecutionArgs ) => OperationResult ; default authenticate ? : ( req : Request , res : Response ) => Promise < string | undefined | void > | string | undefined | void ; /**
     * Called when a new event stream is connecting BEFORE it is accepted.
     * By accepted, its meant the server responded with a 200 (OK), alongside
     * flushing the necessary event stream headers.
     *
     * If you want to respond to the client with a custom status or body,
     * you should do so using the provided `res` argument which will stop
     * further execution.
     */ onConnecting ? : ( req : Request , res : Response ) => Promise < void > | void ; /**
     * Called when a new event stream has been succesfully connected and
     * accepted, and after all pending messages have been flushed.
     */ onConnected ? : ( req : Request ) => Promise < void > | void ; /**
     * The subscribe callback executed right after processing the request
     * before proceeding with the GraphQL operation execution.
     *
     * If you return `ExecutionArgs` from the callback, it will be used instead of
     * trying to build one internally. In this case, you are responsible for providing
     * a ready set of arguments which will be directly plugged in the operation execution.
     *
     * Omitting the fields `contextValue` from the returned `ExecutionArgs` will use the
     * provided `context` option, if available.
     *
     * If you want to respond to the client with a custom status or body,
     * you should do so using the provided `res` argument which will stop
     * further execution.
     *
     * Useful for preparing the execution arguments following a custom logic. A typical
     * use-case is persisted queries. You can identify the query from the request parameters
     * and supply the appropriate GraphQL operation execution arguments.
     */ onSubscribe ? : ( req : Request , res : Response , params : RequestParams ) => Promise < ExecutionArgs | void > | ExecutionArgs | void ; /**
     * Executed after the operation call resolves. For streaming
     * operations, triggering this callback does not necessarely
     * mean that there is already a result available - it means
     * that the subscription process for the stream has resolved
     * and that the client is now subscribed.
     *
     * The `OperationResult` argument is the result of operation
     * execution. It can be an iterator or already a value.
     *
     * Use this callback to listen for GraphQL operations and
     * execution result manipulation.
     *
     * If you want to respond to the client with a custom status or body,
     * you should do so using the provided `res` argument which will stop
     * further execution.
     *
     * First argument, the request, is always the GraphQL operation
     * request.
     */ onOperation ? : ( req : Request , res : Response , args : ExecutionArgs , result : OperationResult ) => Promise < OperationResult | void > | OperationResult | void ; /**
     * Executed after an operation has emitted a result right before
     * that result has been sent to the client.
     *
     * Results from both single value and streaming operations will
     * invoke this callback.
     *
     * Use this callback if you want to format the execution result
     * before it reaches the client.
     *
     * First argument, the request, is always the GraphQL operation
     * request.
     */ onNext ? : ( req : Request , args : ExecutionArgs , result : ExecutionResult | ExecutionPatchResult ) => Promise < ExecutionResult | ExecutionPatchResult | void > | ExecutionResult | ExecutionPatchResult | void ; /**
     * The complete callback is executed after the operation
     * has completed and the client has been notified.
     *
     * Since the library makes sure to complete streaming
     * operations even after an abrupt closure, this callback
     * will always be called.
     *
     * First argument, the request, is always the GraphQL operation
     * request.
     */ onComplete ? : ( req : Request , args : ExecutionArgs ) => Promise < void > | void ; /**
     * Called when an event stream has disconnected right before the
     * accepting the stream.
     */ onDisconnect ? : ( req : Request ) => Promise < void > | void ; } category export type Handler < Request extends IncomingMessage = IncomingMessage , Response extends ServerResponse = ServerResponse > = ( req : Request , res : Response , body ? : unknown ) => Promise < void > ; interface Stream < Request extends IncomingMessage = IncomingMessage , Response extends ServerResponse = ServerResponse > { /**
     * Does the stream have an open connection to some client.
     */ readonly open : boolean ; /**
     * If the operation behind an ID is an `AsyncIterator` - the operation
     * is streaming; on the contrary, if the operation is `null` - it is simply
     * a reservation, meaning - the operation resolves to a single result or is still
     * pending/being prepared.
     */ ops : Record < string , AsyncGenerator < unknown > | AsyncIterable < unknown > | null > ; /**
     * Use this connection for streaming.
     */ use ( req , res ) : Promise < void > ; /**
     * Stream from provided execution result to used connection.
     */ from ( operationReq , args , result , opId ? ) : Promise < void > ; } category export function createHandler < Request extends IncomingMessage = IncomingMessage , Response extends ServerResponse = ServerResponse > ( options ) { const { schema , context , validate = graphqlValidate , execute = graphqlExecute , subscribe = graphqlSubscribe , authenticate = function extractOrCreateStreamToken ( req ) { const headerToken = req . headers [ TOKEN_HEADER_KEY ] || req . headers [ 'x-graphql-stream-token' ] ; if ( headerToken ) return Array . isArray ( headerToken ) ? headerToken . join ( '' ) : headerToken ; const urlToken = new URL ( req . url ?? '' , 'http://localhost/' ) . searchParams . get ( TOKEN_QUERY_KEY ) ; if ( urlToken ) return urlToken ; return 'xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx' . replace ( /[xy]/g , ( c ) => { const r = ( Math . random ( ) * 16 ) | 0 , v = c == 'x' ? r : ( r & 0x3 ) | 0x8 ; return v . toString ( 16 ) ; } ) ; } , onConnecting , onConnected , onSubscribe , onOperation , onNext , onComplete , onDisconnect , } = options ; const streams = { } ; function createStream ( token ) { let request = null , response = null , pinger , disposed = false ; const pendingMsgs = [ ] ; const ops = { } ; function write ( msg ) { return new Promise < boolean > ( ( resolve , reject ) => { if ( disposed || ! response || ! response . writable ) return resolve ( false ) ; response . write ( msg , ( err ) => { if ( err ) return reject ( err ) ; resolve ( true ) ; } ) ; } ) ; } async function emit < E extends StreamEvent > ( event , data ) { let msg = `event: ${ event }` ; if ( data ) msg += `\ndata: ${ JSON . stringify ( data ) }` ; msg += '\n\n' ; const wrote = await write ( msg ) ; if ( ! wrote ) pendingMsgs . push ( msg ) ; } async function dispose ( ) { if ( disposed ) return ; disposed = true ; if ( typeof token === 'string' ) delete streams [ token ] ; for ( const op of Object . values ( ops ) ) { if ( isAsyncGenerator ( op ) ) await op . return ( undefined ) ; } while ( pendingMsgs . length ) { const msg = pendingMsgs . shift ( ) ! ; await write ( msg ) ; } response ! . end ( ) ; response = null ; clearInterval ( pinger ) ; onDisconnect ?. ( request ! ) ; request = null ; } return { get open ( ) { return disposed || Boolean ( response ) ; } , ops , async use ( req , res ) { request = req ; response = res ; req . socket . setTimeout ( 0 ) ; req . socket . setNoDelay ( true ) ; req . socket . setKeepAlive ( true ) ; res . once ( 'close' , dispose ) ; res . statusCode = 200 ; res . setHeader ( 'Content-Type' , 'text/event-stream; charset=utf-8' ) ; res . setHeader ( 'Cache-Control' , 'no-cache' ) ; res . setHeader ( 'X-Accel-Buffering' , 'no' ) ; if ( req . httpVersionMajor < 2 ) res . setHeader ( 'Connection' , 'keep-alive' ) ; res . flushHeaders ( ) ; await write ( ':\n\n' ) ; pinger = setInterval ( ( ) => write ( ':\n\n' ) , 12000 ) ; while ( pendingMsgs . length ) { const msg = pendingMsgs . shift ( ) ! ; const wrote = await write ( msg ) ; if ( ! wrote ) throw new Error ( 'Unable to flush messages' ) ; } await onConnected ?. ( req ) ; } , async from ( operationReq , args , result , opId ) { if ( isAsyncIterable ( result ) ) { /** multiple emitted results */ for await ( let part of result ) { const maybeResult = await onNext ?. ( operationReq , args , part ) ; if ( maybeResult ) part = maybeResult ; await emit ( 'next' , opId ? { id : opId , payload : part , } : part ) ; } } else { /** single emitted result */ const maybeResult = await onNext ?. ( operationReq , args , result ) ; if ( maybeResult ) result = maybeResult ; await emit ( 'next' , opId ? { id : opId , payload : result , } : result ) ; } await emit ( 'complete' , opId ? { id : opId } : null ) ; if ( ! opId ) await dispose ( ) ; else delete ops [ opId ] ; await onComplete ?. ( operationReq , args ) ; } , } ; } async function prepare ( req , res , params ) { let args , operation ; const maybeExecArgs = await onSubscribe ?. ( req , res , params ) ; if ( maybeExecArgs ) args = maybeExecArgs ; else { if ( ! schema ) throw new Error ( 'The GraphQL schema is not provided' ) ; const { operationName , variables } = params ; let { query } = params ; if ( typeof query === 'string' ) { try { query = parse ( query ) ; } catch { return res . writeHead ( 400 , 'GraphQL query syntax error' ) . end ( ) ; } } const argsWithoutSchema = { operationName , document : query , variableValues : variables , } ; args = { ... argsWithoutSchema , schema : typeof schema === 'function' ? await schema ( req , argsWithoutSchema ) : schema , } ; } try { const ast = getOperationAST ( args . document , args . operationName ) ; if ( ! ast ) throw null ; operation = ast . operation ; } catch { return res . writeHead ( 400 , 'Unable to detect operation AST' ) . end ( ) ; } if ( operation === 'mutation' && req . method === 'GET' ) return res . writeHead ( 405 , 'Cannot perform mutations over GET' , { Allow : 'POST' , } ) . end ( ) ; if ( ! ( 'contextValue' in args ) ) args . contextValue = typeof context === 'function' ? await context ( req , args ) : context ; const validationErrs = validate ( args . schema , args . document ) ; if ( validationErrs . length ) { if ( req . headers . accept === 'text/event-stream' ) { return [ args , function perform ( ) { return { errors : validationErrs } ; } , ] ; } res . writeHead ( 400 , { 'Content-Type' : req . headers . accept === 'application/json' ? 'application/json; charset=utf-8' : 'application/graphql+json; charset=utf-8' , } ) . write ( JSON . stringify ( { errors : validationErrs } ) ) ; return res . end ( ) ; } return [ args , async function perform ( ) { let result = operation === 'subscription' ? subscribe ( args ) : execute ( args ) ; const maybeResult = await onOperation ?. ( req , res , args , result ) ; if ( maybeResult ) result = maybeResult ; return result ; } , ] ; } return async function handler ( req , res , body ) { const token = await authenticate ( req , res ) ; if ( res . writableEnded ) return ; if ( typeof token !== 'string' ) throw new Error ( 'Token was not supplied' ) ; const accept = req . headers . accept ?? '*/*' ; const stream = streams [ token ] ; if ( accept === 'text/event-stream' ) { if ( ! stream ) { let params ; try { params = await parseReq ( req , body ) ; } catch ( err ) { return res . writeHead ( 400 , err . message ) . end ( ) ; } const distinctStream = createStream ( null ) ; distinctStream . ops [ '' ] = null ; const prepared = await prepare ( req , res , params ) ; if ( res . writableEnded ) return ; if ( ! prepared ) throw new Error ( "Operation preparation didn't respond, yet it was not prepared" ) ; const [ args , perform ] = prepared ; const result = await perform ( ) ; if ( res . writableEnded ) { if ( isAsyncGenerator ( result ) ) result . return ( undefined ) ; return ; } if ( isAsyncIterable ( result ) ) distinctStream . ops [ '' ] = result ; await onConnecting ?. ( req , res ) ; if ( res . writableEnded ) return ; await distinctStream . use ( req , res ) ; await distinctStream . from ( req , args , result ) ; return ; } if ( stream . open ) return res . writeHead ( 409 , 'Stream already open' ) . end ( ) ; await onConnecting ?. ( req , res ) ; if ( res . writableEnded ) return ; await stream . use ( req , res ) ; return ; } if ( req . method === 'PUT' ) { if ( ! [ '*/*' , 'text/plain' ] . includes ( accept ) ) return res . writeHead ( 406 ) . end ( ) ; if ( stream ) return res . writeHead ( 409 , 'Stream already registered' ) . end ( ) ; streams [ token ] = createStream ( token ) ; res . writeHead ( 201 , { 'Content-Type' : 'text/plain; charset=utf-8' } ) . write ( token ) ; return res . end ( ) ; } else if ( req . method === 'DELETE' ) { if ( ! stream ) return res . writeHead ( 404 , 'Stream not found' ) . end ( ) ; const opId = new URL ( req . url ?? '' , 'http://localhost/' ) . searchParams . get ( 'operationId' ) ; if ( ! opId ) return res . writeHead ( 400 , 'Operation ID is missing' ) . end ( ) ; const op = stream . ops [ opId ] ; if ( isAsyncGenerator ( op ) ) op . return ( undefined ) ; delete stream . ops [ opId ] ; return res . writeHead ( 200 ) . end ( ) ; } else if ( req . method !== 'GET' && req . method !== 'POST' ) return res . writeHead ( 405 , undefined , { Allow : 'GET, POST, PUT, DELETE' } ) . end ( ) ; else if ( ! stream ) return res . writeHead ( 404 , 'Stream not found' ) . end ( ) ; if ( ! [ '*/*' , 'application/graphql+json' , 'application/json' ] . includes ( accept ) ) return res . writeHead ( 406 ) . end ( ) ; let params ; try { params = await parseReq ( req , body ) ; } catch ( err ) { return res . writeHead ( 400 , err . message ) . end ( ) ; } const opId = String ( params . extensions ?. operationId ?? '' ) ; if ( ! opId ) return res . writeHead ( 400 , 'Operation ID is missing' ) . end ( ) ; if ( opId in stream . ops ) return res . writeHead ( 409 , 'Operation with ID already exists' ) . end ( ) ; stream . ops [ opId ] = null ; const prepared = await prepare ( req , res , params ) ; if ( res . writableEnded ) return ; if ( ! prepared ) throw new Error ( "Operation preparation didn't respond, yet it was not prepared" ) ; const [ args , perform ] = prepared ; if ( ! ( opId in stream . ops ) ) return res . writeHead ( 204 ) . end ( ) ; const result = await perform ( ) ; if ( res . writableEnded ) { if ( isAsyncGenerator ( result ) ) result . return ( undefined ) ; delete stream . ops [ opId ] ; return ; } if ( ! ( opId in stream . ops ) ) { if ( isAsyncGenerator ( result ) ) result . return ( undefined ) ; return res . writeHead ( 204 ) . end ( ) ; } if ( isAsyncIterable ( result ) ) stream . ops [ opId ] = result ; res . writeHead ( 202 ) . end ( ) ; await stream . from ( req , args , result , opId ) ; } ; } async function parseReq < Request extends IncomingMessage = IncomingMessage > ( req , body ) { const params = { } ; if ( req . method === 'GET' ) { await new Promise < void > ( ( resolve , reject ) => { try { const url = new URL ( req . url ?? '' , 'http://localhost/' ) ; params . operationName = url . searchParams . get ( 'operationName' ) ?? undefined ; params . query = url . searchParams . get ( 'query' ) ?? undefined ; const variables = url . searchParams . get ( 'variables' ) ; if ( variables ) params . variables = JSON . parse ( variables ) ; const extensions = url . searchParams . get ( 'extensions' ) ; if ( extensions ) params . extensions = JSON . parse ( extensions ) ; resolve ( ) ; } catch { reject ( new Error ( 'Unparsable URL' ) ) ; } } ) ; } else if ( req . method === 'POST' ) { await new Promise < void > ( ( resolve , reject ) => { const end = ( body ) => { try { const data = typeof body === 'string' ? JSON . parse ( body ) : body ; params . operationName = data . operationName ; params . query = data . query ; params . variables = data . variables ; params . extensions = data . extensions ; resolve ( ) ; } catch { reject ( new Error ( 'Unparsable body' ) ) ; } } ; if ( typeof body === 'string' || isObject ( body ) ) end ( body ) ; else { let body = '' ; req . on ( 'data' , ( chunk ) => ( body += chunk ) ) ; req . on ( 'end' , ( ) => end ( body ) ) ; } } ) ; } else throw new Error ( `Unsupported method ${ req . method }` ) ; if ( ! params . query ) throw new Error ( 'Missing query' ) ; if ( params . variables && typeof params . variables !== 'object' ) throw new Error ( 'Invalid variables' ) ; if ( params . extensions && typeof params . extensions !== 'object' ) throw new Error ( 'Invalid extensions' ) ; return params as RequestParams ; } function isAsyncIterable < T = unknown > ( val ) { return typeof Object ( val ) [ Symbol . asyncIterator ] === 'function' ; } export function isAsyncGenerator < T = unknown > ( val ) { return ( isObject ( val ) && typeof Object ( val ) [ Symbol . asyncIterator ] === 'function' && typeof val . return === 'function' ) ; }