import { ActiveWorkflows , NodeExecuteFunctions } from 'n8n-core' ; import { IDeferredPromise , IExecuteData , IExecuteResponsePromiseData , IGetExecutePollFunctions , IGetExecuteTriggerFunctions , INode , INodeExecutionData , IRunExecutionData , IWorkflowExecuteAdditionalData as IWorkflowExecuteAdditionalDataWorkflow , NodeHelpers , WebhookHttpMethod , Workflow , WorkflowActivateMode , WorkflowExecuteMode , LoggerProxy as Logger , } from 'n8n-workflow' ; import * as express from 'express' ; import { Db , IActivationError , IResponseCallbackData , IWebhookDb , IWorkflowDb , IWorkflowExecutionDataProcess , NodeTypes , ResponseHelper , WebhookHelpers , WorkflowExecuteAdditionalData , WorkflowHelpers , WorkflowRunner , ExternalHooks , } from '.' ; import config = require ( '../config' ) ; const WEBHOOK_PROD_UNREGISTERED_HINT = `The workflow must be active for a production URL to run successfully. You can activate the workflow using the toggle in the top-right of the editor. Note that unlike test URL calls, production URL calls aren't shown on the canvas (only in the executions list)` ; export class ActiveWorkflowRunner { private activeWorkflows : ActiveWorkflows | null = null ; private activationErrors : { [ key ] : IActivationError ; } = { } ; async init ( ) { const workflowsData = ( await Db . collections . Workflow ! . find ( { active : true , } ) ) as IWorkflowDb [ ] ; if ( ! config . get ( 'endpoints.skipWebhoooksDeregistrationOnShutdown' ) ) { await Db . collections . Webhook ?. clear ( ) ; } this . activeWorkflows = new ActiveWorkflows ( ) ; if ( workflowsData . length !== 0 ) { console . info ( ' ================================' ) ; console . info ( '   Start Active Workflows:' ) ; console . info ( ' ================================' ) ; for ( const workflowData of workflowsData ) { console . log ( `   - ${ workflowData . name }` ) ; Logger . debug ( `Initializing active workflow "${ workflowData . name }" (startup)` , { workflowName : workflowData . name , workflowId : workflowData . id , } ) ; try { await this . add ( workflowData . id . toString ( ) , 'init' , workflowData ) ; Logger . verbose ( `Successfully started workflow "${ workflowData . name }"` , { workflowName : workflowData . name , workflowId : workflowData . id , } ) ; console . log ( `     => Started` ) ; } catch ( error ) { console . log ( `     => ERROR: Workflow could not be activated:` ) ; console . log ( `               ${ error . message }` ) ; Logger . error ( `Unable to initialize workflow "${ workflowData . name }" (startup)` , { workflowName : workflowData . name , workflowId : workflowData . id , } ) ; } } Logger . verbose ( 'Finished initializing active workflows (startup)' ) ; } const externalHooks = ExternalHooks ( ) ; await externalHooks . run ( 'activeWorkflows.initialized' , [ ] ) ; } async initWebhooks ( ) { this . activeWorkflows = new ActiveWorkflows ( ) ; } returns { Promise < void > } memberof async removeAll ( ) { const activeWorkflowId = [ ] ; Logger . verbose ( 'Call to remove all active workflows received (removeAll)' ) ; if ( this . activeWorkflows !== null ) { activeWorkflowId . push . apply ( activeWorkflowId , this . activeWorkflows . allActiveWorkflows ( ) ) ; } const activeWorkflows = await this . getActiveWorkflows ( ) ; activeWorkflowId . push . apply ( activeWorkflowId , activeWorkflows . map ( ( workflow ) => workflow . id ) ) ; const removePromises = [ ] ; for ( const workflowId of activeWorkflowId ) { removePromises . push ( this . remove ( workflowId ) ) ; } await Promise . all ( removePromises ) ; } param { WebhookHttpMethod } httpMethod param { string } path param { express . Request } req param { express . Response } res returns { Promise < object > } memberof async executeWebhook ( httpMethod , path , req , res ) { Logger . debug ( `Received webhoook "${ httpMethod }" for path "${ path }"` ) ; if ( this . activeWorkflows === null ) { throw new ResponseHelper . ResponseError ( 'The "activeWorkflows" instance did not get initialized yet.' , 404 , 404 ) ; } req . params = { } ; if ( path . endsWith ( '/' ) ) { path = path . slice ( 0 , - 1 ) ; } let webhook = ( await Db . collections . Webhook ?. findOne ( { webhookPath : path , method : httpMethod , } ) ) as IWebhookDb ; let webhookId ; if ( webhook === undefined ) { const pathElements = path . split ( '/' ) ; webhookId = pathElements . shift ( ) ; const dynamicWebhooks = await Db . collections . Webhook ?. find ( { webhookId , method : httpMethod , pathLength : pathElements . length , } ) ; if ( dynamicWebhooks === undefined || dynamicWebhooks . length === 0 ) { throw new ResponseHelper . ResponseError ( `The requested webhook "${ httpMethod } ${ path }" is not registered.` , 404 , 404 , WEBHOOK_PROD_UNREGISTERED_HINT ) ; } let maxMatches = 0 ; const pathElementsSet = new Set ( pathElements ) ; dynamicWebhooks . forEach ( ( dynamicWebhook ) => { const staticElements = dynamicWebhook . webhookPath . split ( '/' ) . filter ( ( ele ) => ! ele . startsWith ( ':' ) ) ; const allStaticExist = staticElements . every ( ( staticEle ) => pathElementsSet . has ( staticEle ) ) ; if ( allStaticExist && staticElements . length > maxMatches ) { maxMatches = staticElements . length ; webhook = dynamicWebhook ; } else if ( staticElements . length === 0 && ! webhook ) { webhook = dynamicWebhook ; } } ) ; if ( webhook === undefined ) { throw new ResponseHelper . ResponseError ( `The requested webhook "${ httpMethod } ${ path }" is not registered.` , 404 , 404 , WEBHOOK_PROD_UNREGISTERED_HINT ) ; } path = webhook . webhookPath ; webhook . webhookPath . split ( '/' ) . forEach ( ( ele , index ) => { if ( ele . startsWith ( ':' ) ) { req . params [ ele . slice ( 1 ) ] = pathElements [ index ] ; } } ) ; } const workflowData = await Db . collections . Workflow ! . findOne ( webhook . workflowId ) ; if ( workflowData === undefined ) { throw new ResponseHelper . ResponseError ( `Could not find workflow with id "${ webhook . workflowId }"` , 404 , 404 ) ; } const nodeTypes = NodeTypes ( ) ; const workflow = new Workflow ( { id : webhook . workflowId . toString ( ) , name : workflowData . name , nodes : workflowData . nodes , connections : workflowData . connections , active : workflowData . active , nodeTypes , staticData : workflowData . staticData , settings : workflowData . settings , } ) ; const additionalData = await WorkflowExecuteAdditionalData . getBase ( ) ; const webhookData = NodeHelpers . getNodeWebhooks ( workflow , workflow . getNode ( webhook . node ) as INode , additionalData ) . filter ( ( webhook ) => { return webhook . httpMethod === httpMethod && webhook . path === path ; } ) [ 0 ] ; const workflowStartNode = workflow . getNode ( webhookData . node ) ; if ( workflowStartNode === null ) { throw new ResponseHelper . ResponseError ( 'Could not find node to process webhook.' , 404 , 404 ) ; } return new Promise ( ( resolve , reject ) => { const executionMode = 'webhook' ; WebhookHelpers . executeWebhook ( workflow , webhookData , workflowData , workflowStartNode , executionMode , undefined , undefined , undefined , req , res , ( error , data ) => { if ( error !== null ) { return reject ( error ) ; } resolve ( data ) ; } ) ; } ) ; } param { string } path returns { Promise < string [ ] > } memberof async getWebhookMethods ( path ) { const webhooks = ( await Db . collections . Webhook ?. find ( { webhookPath : path } ) ) as IWebhookDb [ ] ; const webhookMethods = webhooks . map ( ( webhook ) => webhook . method ) ; return webhookMethods ; } returns { string [ ] } memberof async getActiveWorkflows ( ) { const activeWorkflows = ( await Db . collections . Workflow ?. find ( { where : { active : true } , select : [ 'id' ] , } ) ) as IWorkflowDb [ ] ; return activeWorkflows . filter ( ( workflow ) => this . activationErrors [ workflow . id . toString ( ) ] === undefined ) ; } param { string } id returns { boolean } memberof async isActive ( id ) { const workflow = ( await Db . collections . Workflow ?. findOne ( { id : Number ( id ) } ) ) as IWorkflowDb ; return workflow ?. active ; } param { string } id returns { ( IActivationError | undefined ) } memberof getActivationError ( id ) { if ( this . activationErrors [ id ] === undefined ) { return undefined ; } return this . activationErrors [ id ] ; } param { Workflow } workflow param { IWorkflowExecuteAdditionalDataWorkflow } additionalData param { WorkflowExecuteMode } mode returns { Promise < void > } memberof async addWorkflowWebhooks ( workflow , additionalData , mode , activation ) { const webhooks = WebhookHelpers . getWorkflowWebhooks ( workflow , additionalData , undefined , true ) ; let path = '' as string | undefined ; for ( const webhookData of webhooks ) { const node = workflow . getNode ( webhookData . node ) as INode ; node . name = webhookData . node ; path = webhookData . path ; const webhook = { workflowId : webhookData . workflowId , webhookPath : path , node : node . name , method : webhookData . httpMethod , } as IWebhookDb ; if ( webhook . webhookPath . startsWith ( '/' ) ) { webhook . webhookPath = webhook . webhookPath . slice ( 1 ) ; } if ( webhook . webhookPath . endsWith ( '/' ) ) { webhook . webhookPath = webhook . webhookPath . slice ( 0 , - 1 ) ; } if ( ( path . startsWith ( ':' ) || path . includes ( '/:' ) ) && node . webhookId ) { webhook . webhookId = node . webhookId ; webhook . pathLength = webhook . webhookPath . split ( '/' ) . length ; } try { await Db . collections . Webhook ?. insert ( webhook ) ; const webhookExists = await workflow . runWebhookMethod ( 'checkExists' , webhookData , NodeExecuteFunctions , mode , activation , false ) ; if ( webhookExists !== true ) { await workflow . runWebhookMethod ( 'create' , webhookData , NodeExecuteFunctions , mode , activation , false ) ; } } catch ( error ) { if ( activation === 'init' && config . get ( 'endpoints.skipWebhoooksDeregistrationOnShutdown' ) && error . name === 'QueryFailedError' ) { continue ; } try { await this . removeWorkflowWebhooks ( workflow . id as string ) ; } catch ( error ) { console . error ( `Could not remove webhooks of workflow "${ workflow . id }" because of error: "${ error . message }"` ) ; } let errorMessage = '' ; if ( error . name === 'QueryFailedError' ) { errorMessage = `The webhook path [${ webhook . webhookPath }] and method [${ webhook . method }] already exist.` ; } else if ( error . detail ) { errorMessage = error . detail ; } else { errorMessage = error . message ; } throw error ; } } await WorkflowHelpers . saveStaticData ( workflow ) ; } param { string } workflowId returns memberof async removeWorkflowWebhooks ( workflowId ) { const workflowData = await Db . collections . Workflow ! . findOne ( workflowId ) ; if ( workflowData === undefined ) { throw new Error ( `Could not find workflow with id "${ workflowId }"` ) ; } const nodeTypes = NodeTypes ( ) ; const workflow = new Workflow ( { id : workflowId , name : workflowData . name , nodes : workflowData . nodes , connections : workflowData . connections , active : workflowData . active , nodeTypes , staticData : workflowData . staticData , settings : workflowData . settings , } ) ; const mode = 'internal' ; const additionalData = await WorkflowExecuteAdditionalData . getBase ( ) ; const webhooks = WebhookHelpers . getWorkflowWebhooks ( workflow , additionalData , undefined , true ) ; for ( const webhookData of webhooks ) { await workflow . runWebhookMethod ( 'delete' , webhookData , NodeExecuteFunctions , mode , 'update' , false ) ; } await WorkflowHelpers . saveStaticData ( workflow ) ; const webhook = { workflowId : workflowData . id , } as IWebhookDb ; await Db . collections . Webhook ?. delete ( webhook ) ; } param { IWorkflowDb } workflowData param { INode } node param { INodeExecutionData [ ] [ ] } data param { IWorkflowExecuteAdditionalDataWorkflow } additionalData param { WorkflowExecuteMode } mode returns memberof async runWorkflow ( workflowData , node , data , additionalData , mode , responsePromise ? ) { const nodeExecutionStack = [ { node , data : { main : data , } , } , ] ; const executionData = { startData : { } , resultData : { runData : { } , } , executionData : { contextData : { } , nodeExecutionStack , waitingExecution : { } , } , } ; const runData = { executionMode : mode , executionData , workflowData , } ; const workflowRunner = new WorkflowRunner ( ) ; return workflowRunner . run ( runData , true , undefined , undefined , responsePromise ) ; } param { IWorkflowDb } workflowData param { IWorkflowExecuteAdditionalDataWorkflow } additionalData param { WorkflowExecuteMode } mode returns { IGetExecutePollFunctions } memberof getExecutePollFunctions ( workflowData , additionalData , mode , activation ) { return ( workflow , node ) => { const returnFunctions = NodeExecuteFunctions . getExecutePollFunctions ( workflow , node , additionalData , mode , activation ) ; returnFunctions . __emit = ( data ) : void => { Logger . debug ( `Received event to trigger execution for workflow "${ workflow . name }"` ) ; this . runWorkflow ( workflowData , node , data , additionalData , mode ) ; } ; return returnFunctions ; } ; } param { IWorkflowDb } workflowData param { IWorkflowExecuteAdditionalDataWorkflow } additionalData param { WorkflowExecuteMode } mode returns { IGetExecuteTriggerFunctions } memberof getExecuteTriggerFunctions ( workflowData , additionalData , mode , activation ) { return ( workflow , node ) => { const returnFunctions = NodeExecuteFunctions . getExecuteTriggerFunctions ( workflow , node , additionalData , mode , activation ) ; returnFunctions . emit = ( data , responsePromise ? ) : void => { Logger . debug ( `Received trigger for workflow "${ workflow . name }"` ) ; WorkflowHelpers . saveStaticData ( workflow ) ; this . runWorkflow ( workflowData , node , data , additionalData , mode , responsePromise ) . catch ( ( error ) => console . error ( error ) ) ; } ; return returnFunctions ; } ; } param { string } workflowId param { IWorkflowDb } workflowData returns { Promise < void > } memberof async add ( workflowId , activation , workflowData ? ) { if ( this . activeWorkflows === null ) { throw new Error ( `The "activeWorkflows" instance did not get initialized yet.` ) ; } let workflowInstance ; try { if ( workflowData === undefined ) { workflowData = ( await Db . collections . Workflow ! . findOne ( workflowId ) ) as IWorkflowDb ; } if ( ! workflowData ) { throw new Error ( `Could not find workflow with id "${ workflowId }".` ) ; } const nodeTypes = NodeTypes ( ) ; workflowInstance = new Workflow ( { id : workflowId , name : workflowData . name , nodes : workflowData . nodes , connections : workflowData . connections , active : workflowData . active , nodeTypes , staticData : workflowData . staticData , settings : workflowData . settings , } ) ; const canBeActivated = workflowInstance . checkIfWorkflowCanBeActivated ( [ 'n8n-nodes-base.start' , ] ) ; if ( ! canBeActivated ) { Logger . error ( `Unable to activate workflow "${ workflowData . name }"` ) ; throw new Error ( `The workflow can not be activated because it does not contain any nodes which could start the workflow. Only workflows which have trigger or webhook nodes can be activated.` ) ; } const mode = 'trigger' ; const additionalData = await WorkflowExecuteAdditionalData . getBase ( ) ; const getTriggerFunctions = this . getExecuteTriggerFunctions ( workflowData , additionalData , mode , activation ) ; const getPollFunctions = this . getExecutePollFunctions ( workflowData , additionalData , mode , activation ) ; await this . addWorkflowWebhooks ( workflowInstance , additionalData , mode , activation ) ; if ( workflowInstance . getTriggerNodes ( ) . length !== 0 || workflowInstance . getPollNodes ( ) . length !== 0 ) { await this . activeWorkflows . add ( workflowId , workflowInstance , additionalData , mode , activation , getTriggerFunctions , getPollFunctions ) ; Logger . verbose ( `Successfully activated workflow "${ workflowData . name }"` , { workflowId , workflowName : workflowData . name , } ) ; } if ( this . activationErrors [ workflowId ] !== undefined ) { delete this . activationErrors [ workflowId ] ; } } catch ( error ) { this . activationErrors [ workflowId ] = { time : new Date ( ) . getTime ( ) , error : { message : error . message , } , } ; throw error ; } await WorkflowHelpers . saveStaticData ( workflowInstance ! ) ; } param { string } workflowId returns { Promise < void > } memberof async remove ( workflowId ) { if ( this . activeWorkflows !== null ) { try { await this . removeWorkflowWebhooks ( workflowId ) ; } catch ( error ) { console . error ( `Could not remove webhooks of workflow "${ workflowId }" because of error: "${ error . message }"` ) ; } if ( this . activationErrors [ workflowId ] !== undefined ) { delete this . activationErrors [ workflowId ] ; } if ( this . activeWorkflows . isActive ( workflowId ) ) { await this . activeWorkflows . remove ( workflowId ) ; Logger . verbose ( `Successfully deactivated workflow "${ workflowId }"` , { workflowId } ) ; } return ; } throw new Error ( `The "activeWorkflows" instance did not get initialized yet.` ) ; } } let workflowRunnerInstance ; export function getInstance ( ) { if ( workflowRunnerInstance === undefined ) { workflowRunnerInstance = new ActiveWorkflowRunner ( ) ; } return workflowRunnerInstance ; }