import * as express from 'express' ; import { ExecutionResult } from 'graphql' ; import { Observable } from 'subscriptions-transport-ws' ; import * as request from 'superagent' ; import { v4 as uuid } from 'uuid' ; import * as assert from 'assert' ; import * as _ from 'lodash' ; import { ComponentHelper , ComponentTable , ComponentTag , ComponentTagPath } from './componentHelper' ; import { SubscriptionCoordinator } from './subscriptionCoordinator' ; import { TxConfig , TxCoordinator , Vote } from './txCoordinator' ; import { ComponentPath } from './componentPath' ; import { DvConfig , GatewayConfig , getPort } from './gateway.model' ; import { InputValuesMap , TxInputsValidator } from './txInputsValidator' ; const JSON_INDENTATION = 2 ; const SUCCESS = 200 ; const INTERNAL_SERVER_ERROR = 500 ; export interface Dict { readonly [ key ] : string ; } export interface RequestOptions { readonly params ? : Dict ; readonly headers ? : Dict ; } export enum Method { GET = 'GET' , POST = 'POST' } export interface Params { from : string ; runId : string ; path ? : string ; options ? : string ; } export interface ChildRequest { method : Method ; body : string | Object ; query : Params ; } export interface GatewayRequest { readonly fullComponentName : string ; readonly from : ComponentPath ; readonly reqId : string ; readonly runId ? : string | undefined ; readonly path ? : string | undefined ; readonly options ? : RequestOptions ; } export interface GatewayToConceptRequest extends GatewayRequest { readonly url : string ; readonly method : string ; readonly body : string ; } export interface ConceptResponse < T > { readonly status : number ; readonly text : T ; readonly index ? : number ; } export type port = number ; function stringify ( json ) { return JSON . stringify ( json , undefined , JSON_INDENTATION ) ; } /**
 * Class for batching responses.
 */ export class ResponseBatch { private responses : ConceptResponse < string > [ ] = [ ] ; constructor ( private res , private batchSize ) { } /**
     * Add a response to the batch.
     * A single response is sent once we have reached the batchSize
     */ add ( status , text , index ) { this . responses . push ( { status , text , index } ) ; if ( this . responses . length === this . batchSize ) { this . send ( ) ; } } /**
     * Fail response
     */ fail ( status , text ) { this . responses = [ ] ; for ( let i = 0 ; i < this . batchSize ; i ++ ) { this . add ( status , text , i ) ; } } /**
     * Send the batched response.
     */ private send ( ) { let status = this . responses [ 0 ] . status ; let body = this . responses [ 0 ] . text ; if ( this . batchSize > 1 ) { status = this . responses . filter ( ( { status : s } ) => s !== SUCCESS ) . length === 0 ? SUCCESS : INTERNAL_SERVER_ERROR ; body = this . responses . sort ( ( a , b ) => a . index - b . index ) . map ( ( response ) => ( { status : response . status , body : response . text } ) ) ; } this . res . status ( status ) . send ( body ) ; } } export class RequestInvalidError { constructor ( public readonly message ) { } } export abstract class RequestProcessor { protected readonly txCoordinator : TxCoordinator < GatewayToConceptRequest , ConceptResponse < string > , ResponseBatch > ; protected readonly dstTable : { [ concept ] : port ; } = { } ; protected readonly subscriptionCoordinator : SubscriptionCoordinator ; private static ConceptOf ( node ) { if ( node === undefined ) { return undefined ; } return _ . isEmpty ( node . dvOf ) ? node . tag . split ( '-' ) [ 0 ] : node . dvOf ! ; } /**
     *  Forwards to the concept the given request.
     */ private static async ForwardRequest < T > ( gatewayRequest ) { let url = gatewayRequest . url ; if ( gatewayRequest . path ) { url += gatewayRequest . path ; } url += `/${ gatewayRequest . fullComponentName }` ; let conceptReq = request ( gatewayRequest . method , url ) ; if ( gatewayRequest . options ) { if ( gatewayRequest . options . params ) { conceptReq = conceptReq . query ( gatewayRequest . options . params ) ; } if ( gatewayRequest . options . headers ) { conceptReq = conceptReq . set ( gatewayRequest . options . headers ) ; } } conceptReq . send ( gatewayRequest . body ) ; let response ; try { response = await conceptReq ; } catch ( err ) { response = err . response ; } if ( ! response ) { const errMsg = `Got an undefined response for concept ` + `request ${ stringify ( conceptReq ) }` ; console . error ( errMsg ) ; throw new Error ( errMsg ) ; } return { status : response . status , text : RequestProcessor . JsonParse ( response . text ) } ; } private static NewReqFor ( msg , gcr ) { return _ . assign ( { } , gcr , { path : `/dv-${ gcr . fullComponentName }/${ gcr . reqId }/${ msg }` + gcr . path } ) ; } private static BuildGatewayRequest ( req ) { return { from : ComponentPath . fromString ( req . query . from ) , fullComponentName : req . query . fullComponentName , reqId : uuid ( ) , runId : req . query . runId , path : req . query . path , options : req . query . options ? RequestProcessor . JsonParse ( req . query . options ) : undefined } ; } private static JsonParse ( value ) { try { return JSON . parse ( value ) ; } catch ( e ) { throw new Error ( `Couldn't parse JSON ${ value }. Reason: ${ e }` ) ; } } protected constructor ( config ) { const txConfig = this . getTxConfig ( config ) ; this . txCoordinator = new TxCoordinator < GatewayToConceptRequest , ConceptResponse < string > , ResponseBatch > ( txConfig ) ; this . subscriptionCoordinator = new SubscriptionCoordinator ( ) ; } start ( ) { return this . txCoordinator . start ( ) ; } async processRequest ( req , res ) { return req . query . isTx ? this . processTxRequest ( req , res ) : this . processNonTxRequest ( req , res ) ; } processSubscription ( request ) { if ( ! request [ 'from' ] ) { throw new RequestInvalidError ( 'No from specified' ) ; } const port = this . getToPort ( ComponentPath . fromString ( request [ 'from' ] ) ) ; const url = this . getUrl ( port , request [ 'path' ] , true ) ; return this . subscriptionCoordinator . forwardRequest ( url , request ) ; } unsubscribeAll ( ) { this . subscriptionCoordinator . unsubscribeAll ( ) ; } protected async processTxRequest ( req , res ) { const context = req . body . context ; const childRequests = req . body . requests ; if ( _ . isNil ( childRequests ) ) { throw new Error ( `Got undefined child requests for tx req ${ req }` ) ; } const resBatch = new ResponseBatch ( res , childRequests . length ) ; let gatewayToConceptRequests ; let prunedCohortComponents ; try { gatewayToConceptRequests = childRequests . map ( ( childRequest ) => this . validateRequest ( childRequest ) ) ; const componentPath = gatewayToConceptRequests [ 0 ] . from ; const dvTxNodeIndex = componentPath . indexOfClosestTxNode ( ) ! ; const cohortComponents = this . getCohortComponents ( componentPath , dvTxNodeIndex ) ; prunedCohortComponents = this . pruneCohortComponents ( cohortComponents , childRequests . map ( ( childRequest ) => ComponentPath . fromString ( childRequest . query . from ) . last ( ) ) ) ; const inputValuesMap = { } ; for ( const childRequest of childRequests ) { const componentFqtag = ComponentPath . fromString ( childRequest . query . from ) . last ( ) ; const inputs = childRequest . method === 'GET' ? _ . get ( JSON . parse ( _ . get ( childRequest , 'query.options' ) ) , 'params.inputs.input' ) : _ . get ( childRequest , 'body.inputs.input' ) ; console . log ( `Got inputs ${ stringify ( inputs ) }` ) ; _ . forEach ( inputs , ( value , inputName ) => { _ . set ( inputValuesMap , [ componentFqtag , inputName ] , value ) ; } ) ; } console . log ( `Checking with context ${ stringify ( context ) }` + `Input values map ${ stringify ( inputValuesMap ) }` ) ; TxInputsValidator . Validate ( inputValuesMap , prunedCohortComponents , context ) ; } catch ( e ) { if ( e instanceof RequestInvalidError ) { resBatch . fail ( INTERNAL_SERVER_ERROR , e . message ) ; } else { console . log ( 'Something bad happened' + e . message ) ; throw e ; } return ; } return Promise . all ( gatewayToConceptRequests . map ( ( gatewayToConceptRequest , index ) => this . txCoordinator . processMessage ( gatewayToConceptRequest . runId , gatewayToConceptRequest . from . serialize ( ) , this . getCohorts ( gatewayToConceptRequest . from . serialize ( ) , prunedCohortComponents ) , gatewayToConceptRequest , resBatch , index ) ) ) . then ( ( ) => { } ) ; } protected async processNonTxRequest ( req , res ) { const resBatch = new ResponseBatch ( res , 1 ) ; try { const gatewayToConceptRequest = this . validateRequest ( req ) ; /**
             * We need to check that no other req is expected (another req could
             * be expected if this req is supposed to be part of a tx with more
             * than one cohort)
             */ const componentPath = gatewayToConceptRequest . from ; if ( componentPath . isDvTx ( ) ) { const dvTxNodeIndex = componentPath . indexOfClosestTxNode ( ) ; console . log ( 'Validating request of no-op tx' ) ; const cohortComponents = this . getCohortComponents ( componentPath , dvTxNodeIndex ) ; this . pruneCohortComponents ( cohortComponents , [ componentPath . last ( ) ] ) ; } const conceptRes = await RequestProcessor . ForwardRequest < string > ( gatewayToConceptRequest ) ; resBatch . add ( conceptRes . status , conceptRes . text , 1 ) ; } catch ( e ) { resBatch . add ( INTERNAL_SERVER_ERROR , e . message , 1 ) ; } } protected getComponentFromPath ( componentPath ) { const fqtag = _ . last ( componentPath . nodes ( ) ) ; return { fqtag , tag : fqtag } ; } protected abstract getCohortComponents ( _componentPath , _dvTxNodeIndex ) ; protected abstract pruneCohortComponents ( cohortComponents , receivedRequestFqTags ) ; protected validateRequest ( req ) { if ( ! req . query . from ) { throw new RequestInvalidError ( 'No from specified' ) ; } const gatewayRequest = RequestProcessor . BuildGatewayRequest ( req ) ; const { runId , from : componentPath } = gatewayRequest ; const toPort = this . getToPort ( componentPath ) ; console . log ( `Req from ${ stringify ( gatewayRequest ) }` ) ; console . log ( `Valid request: port: ${ toPort }, ` + `component path: ${ componentPath }, runId: ${ runId }` + ( componentPath . isDvTx ( ) ? `, dvTxId: ${ runId }` : ' not part of a tx' ) ) ; return { ... gatewayRequest , ... { url : `${ this . getUrl ( toPort ) }` , method : req . method , body : req . body } } ; } protected getCohorts ( componentPathId , cohortComponents ) { const componentPath = ComponentPath . fromString ( componentPathId ) ; assert . ok ( componentPath . isDvTx ( ) , `Getting cohorts of a component path that is not part of a ` + `dv-tx: ${ componentPath }` ) ; const dvTxNodeIndex = componentPath . indexOfClosestTxNode ( ) ! ; return _ . map ( cohortComponents , ( component ) => new ComponentPath ( [ ... _ . take ( componentPath . nodes ( ) , dvTxNodeIndex + 1 ) , component . fqtag ] ) . serialize ( ) ) ; } private getToPort ( componentPath ) { const componentTag = this . getComponentFromPath ( componentPath ) ; const to = RequestProcessor . ConceptOf ( componentTag ) ; const toPort = _ . get ( this . dstTable , to ) ; if ( ! componentTag ) { throw new RequestInvalidError ( `Invalid component path: ${ componentPath }` ) ; } if ( toPort === undefined ) { throw new RequestInvalidError ( `Invalid to: ${ to }, my dstTable is ${ stringify ( this . dstTable ) }` ) ; } return toPort ; } private getUrl ( port , path = '' , isSubscription ? ) { const protocol = isSubscription ? 'ws' : 'http' ; return `${ protocol }://localhost:${ port }${ path }` ; } private getTxConfig ( config ) { return { dbHost : config . dbHost , dbPort : config . dbPort , dbUri : config . dbUri , dbName : config . dbName , reinitDbOnStartup : config . reinitDbOnStartup , sendCommitToCohort : ( gcr ) : Promise < void > => { if ( gcr . method === 'GET' ) { return Promise . resolve ( ) ; } return RequestProcessor . ForwardRequest ( RequestProcessor . NewReqFor ( 'commit' , gcr ) ) . then ( ( _unusedResp ) => undefined ) ; } , sendAbortToCohort : ( gcr ) : Promise < void > => { if ( gcr . method === 'GET' ) { return Promise . resolve ( ) ; } return RequestProcessor . ForwardRequest ( RequestProcessor . NewReqFor ( 'abort' , gcr ) ) . then ( ( _unusedResp ) => undefined ) ; } , sendVoteToCohort : ( gcr ) : Promise < Vote < ConceptResponse < string > > > => { return RequestProcessor . ForwardRequest < Vote < string > > ( RequestProcessor . NewReqFor ( 'vote' , gcr ) ) . then ( ( resp ) : Vote < ConceptResponse < string > > => { const vote = { result : resp . text . result , payload : { status : resp . status , text : resp . text . payload } } ; return vote ; } ) ; } , sendAbortToClient : ( causedAbort , _gcr ? , payload ? , txRes ? , index ? ) => { assert . ok ( txRes !== undefined ) ; if ( causedAbort ) { assert . ok ( payload !== undefined ) ; txRes ! . add ( payload ! . status , payload ! . text , index ) ; } else { txRes ! . add ( INTERNAL_SERVER_ERROR , 'the tx that this component is part of aborted' , index ) ; } } , sendToClient : ( payload , txRes ? , index ? ) => { txRes ! . add ( payload . status , payload . text , index ) ; } , onError : ( e , _gcr , txRes ? , index ? ) => { console . error ( e ) ; txRes ! . add ( INTERNAL_SERVER_ERROR , e . message , index ) ; } } ; } } export class AppRequestProcessor extends RequestProcessor { private readonly componentHelper : ComponentHelper ; private readonly dvConfig : DvConfig | undefined ; constructor ( config , dvConfig ? , appComponentTable ? ) { super ( config ) ; this . dvConfig = dvConfig ; _ . forEach ( dvConfig . usedConcepts , ( conceptConfig , name ) => { this . dstTable [ name ] = getPort ( conceptConfig . config . wsPort ) ; } ) ; if ( dvConfig . config ) { this . dstTable [ dvConfig . name ] = getPort ( dvConfig . config . wsPort ) ; } console . log ( `Using dst table ${ stringify ( this . dstTable ) }` ) ; const usedConcepts = _ . map ( dvConfig . usedConcepts , ( usedConceptConfig , alias ) => _ . get ( usedConceptConfig , 'name' , alias ) ) ; const routeComponentSelectors = _ . uniq ( _ . map ( dvConfig . routes , ( { component } ) => `${ dvConfig . name }-${ component }` ) ) ; this . componentHelper = new ComponentHelper ( usedConcepts , appComponentTable , routeComponentSelectors ) ; } protected getComponentFromPath ( componentPath ) { try { return this . componentHelper . getMatchingComponents ( componentPath ) [ 0 ] ; } catch ( e ) { e . message = `Error in component path ${ componentPath }: ${ e . message }` ; throw e ; } } protected getCohortComponents ( componentPath , dvTxNodeIndex ) { let paths ; try { paths = this . componentHelper . getMatchingPaths ( componentPath ) ; } catch ( e ) { e . message = `Error in component path ${ componentPath }: ${ e . message }` ; throw e ; } const debugPaths = _ . map ( paths , ( p ) => ComponentHelper . PickComponentTagPath ( p , [ 'fqtag' ] ) ) ; assert . ok ( paths . length === 1 , `Expected 1 path but got ${ paths . length } for ` + `${ componentPath . serialize ( ) }: ${ stringify ( debugPaths ) }` ) ; const componentTagPath = paths [ 0 ] ; assert . ok ( componentTagPath . length === componentPath . length ( ) , 'Expected the length of the path to match the component path ' + ` length but got ${ stringify ( componentTagPath ) }` ) ; const dvTxNode = componentTagPath [ dvTxNodeIndex ] ; assert . ok ( dvTxNode !== null && dvTxNode !== undefined , 'Expected the tx node to exist on the path' ) ; return _ . reject ( dvTxNode . content , ( component ) => component . tag . split ( '-' ) [ 0 ] === 'dv' || _ . get ( component . inputs , '[save]' ) === 'false' || ! this . componentHelper . shouldHaveExecRequest ( component . tag ) ) ; } param cohortComponents param receivedRequestFqTags return throws protected pruneCohortComponents ( cohortComponents , receivedRequestFqTags ) { const fqTagsSet = new Set ( receivedRequestFqTags ) ; if ( fqTagsSet . size < receivedRequestFqTags . length ) { throw new RequestInvalidError ( 'Got multiple requests from at least one ' + 'fqtag' ) ; } const prunedCohortComponents = [ ] ; for ( const component of cohortComponents ) { const componentRequestExists = fqTagsSet . has ( component . fqtag ) ; if ( ! componentRequestExists && ! this . componentHelper . isRequestOptional ( component . tag ) && ! this . isAppComponent ( component ) ) { throw new RequestInvalidError ( `Did not get any requests from ` + `${ component . fqtag } and ${ component . tag }'s request is not optional` ) ; } if ( componentRequestExists ) { prunedCohortComponents . push ( component ) ; } } if ( prunedCohortComponents . length < receivedRequestFqTags . length ) { throw new RequestInvalidError ( 'Received requests include components that are not part of the cohort' + `. Received: ${ stringify ( receivedRequestFqTags ) }` + `. Expected cohort: ${ stringify ( _ . map ( prunedCohortComponents , 'fqtag' ) ) }` ) ; } return prunedCohortComponents ; } private isAppComponent ( componentTag ) { return this . dvConfig . name === componentTag . tag . split ( '-' ) [ 0 ] ; } } export class DesignerRequestProcessor extends RequestProcessor { private cohortComponents : ComponentTag [ ] ; constructor ( config ) { super ( config ) ; } /**
     * Add a concept (for appless mode for the designer)
     */ addConcept ( name , wsPort , alias ? ) { this . dstTable [ alias || name ] = wsPort ; } /**
     * Remove a concept (for appless mode for the designer)
     */ removeConcept ( name , alias ? ) { delete this . dstTable [ alias || name ] ; } protected async processTxRequest ( req , res ) { const childRequests = req . body . requests ; this . cohortComponents = childRequests . map ( ( chReq ) => { const componentPath = ComponentPath . fromString ( chReq . query . from ) ; const fqtag = componentPath . nodes ( ) [ componentPath . indexOfClosestTxNode ( ) + 1 ] ; return { fqtag , tag : fqtag , inputs : { } } ; } ) ; const resBatch = new ResponseBatch ( res , childRequests . length ) ; let gatewayToConceptRequests ; try { gatewayToConceptRequests = childRequests . map ( ( childRequest ) => this . validateRequest ( childRequest ) ) ; } catch ( e ) { resBatch . fail ( INTERNAL_SERVER_ERROR , e . message ) ; } return Promise . all ( gatewayToConceptRequests . map ( ( gatewayToConceptRequest , index ) => this . txCoordinator . processMessage ( gatewayToConceptRequest . runId , gatewayToConceptRequest . from . serialize ( ) , this . getCohorts ( gatewayToConceptRequest . from . serialize ( ) , this . cohortComponents ) , gatewayToConceptRequest , resBatch , index ) ) ) . then ( ( ) => { } ) ; } protected getCohortComponents ( _componentPath , _dvTxNodeIndex ) { return this . cohortComponents ; } protected pruneCohortComponents ( _cohortComponents , _receivedRequestFqTags ) { return this . cohortComponents ; } }