import { BinaryDataManager , IBinaryDataConfig , IProcessMessage , UserSettings , WorkflowExecute , } from 'n8n-core' ; import { ExecutionError , IDataObject , IExecuteResponsePromiseData , IExecuteWorkflowInfo , ILogger , INodeExecutionData , INodeType , INodeTypeData , IRun , ITaskData , IWorkflowExecuteAdditionalData , IWorkflowExecuteHooks , LoggerProxy , Workflow , WorkflowExecuteMode , WorkflowHooks , WorkflowOperationError , } from 'n8n-workflow' ; import { CredentialsOverwrites , CredentialTypes , Db , ExternalHooks , GenericHelpers , IWorkflowExecuteProcess , IWorkflowExecutionDataProcessWithExecution , NodeTypes , WebhookHelpers , WorkflowExecuteAdditionalData , WorkflowHelpers , } from '.' ; import { getLogger } from './Logger' ; import * as config from '../config' ; import { InternalHooksManager } from './InternalHooksManager' ; export class WorkflowRunnerProcess { data : IWorkflowExecutionDataProcessWithExecution | undefined ; logger : ILogger ; startedAt = new Date ( ) ; workflow : Workflow | undefined ; workflowExecute : WorkflowExecute | undefined ; executionIdCallback : ( executionId : string ) => void | undefined ; childExecutions : { [ key ] : IWorkflowExecuteProcess ; } = { } ; static async stopProcess ( ) { setTimeout ( ( ) => { process . exit ( 0 ) ; } , 30000 ) ; } async runWorkflow ( inputData ) { process . on ( 'SIGTERM' , WorkflowRunnerProcess . stopProcess ) ; process . on ( 'SIGINT' , WorkflowRunnerProcess . stopProcess ) ; const logger = ( this . logger = getLogger ( ) ) ; LoggerProxy . init ( logger ) ; this . data = inputData ; logger . verbose ( 'Initializing n8n sub-process' , { pid : process . pid , workflowId : this . data . workflowData . id , } ) ; let className ; let tempNode ; let filePath ; this . startedAt = new Date ( ) ; const nodeTypesData = { } ; for ( const nodeTypeName of Object . keys ( this . data . nodeTypeData ) ) { className = this . data . nodeTypeData [ nodeTypeName ] . className ; filePath = this . data . nodeTypeData [ nodeTypeName ] . sourcePath ; const tempModule = require ( filePath ) ; try { const nodeObject = new tempModule [ className ] ( ) ; if ( nodeObject . getNodeType !== undefined ) { tempNode = nodeObject . getNodeType ( ) ; } else { tempNode = nodeObject ; } tempNode = new tempModule [ className ] ( ) as INodeType ; } catch ( error ) { throw new Error ( `Error loading node "${ nodeTypeName }" from: "${ filePath }"` ) ; } nodeTypesData [ nodeTypeName ] = { type : tempNode , sourcePath : filePath , } ; } const nodeTypes = NodeTypes ( ) ; await nodeTypes . init ( nodeTypesData ) ; const credentialTypes = CredentialTypes ( ) ; await credentialTypes . init ( inputData . credentialsTypeData ) ; const credentialsOverwrites = CredentialsOverwrites ( ) ; await credentialsOverwrites . init ( inputData . credentialsOverwrite ) ; const externalHooks = ExternalHooks ( ) ; await externalHooks . init ( ) ; const instanceId = ( await UserSettings . prepareUserSettings ( ) ) . instanceId ?? '' ; const { cli } = await GenericHelpers . getVersions ( ) ; InternalHooksManager . init ( instanceId , cli , nodeTypes ) ; const binaryDataConfig = config . get ( 'binaryDataManager' ) as IBinaryDataConfig ; await BinaryDataManager . init ( binaryDataConfig ) ; let shouldInitializaDb = false ; inputData . workflowData . nodes . map ( ( node ) => { if ( Object . keys ( node . credentials === undefined ? { } : node . credentials ) . length > 0 ) { shouldInitializaDb = true ; } } ) ; if ( shouldInitializaDb ) { await Db . init ( ) ; } else if ( inputData . workflowData . settings !== undefined && inputData . workflowData . settings . saveExecutionProgress === true ) { await Db . init ( ) ; } else if ( inputData . workflowData . settings !== undefined && inputData . workflowData . settings . saveExecutionProgress !== false && ( config . get ( 'executions.saveExecutionProgress' ) as boolean ) ) { await Db . init ( ) ; } else if ( inputData . workflowData . settings === undefined && ( config . get ( 'executions.saveExecutionProgress' ) as boolean ) ) { await Db . init ( ) ; } let workflowTimeout = config . get ( 'executions.timeout' ) as number ; if ( this . data . workflowData . settings && this . data . workflowData . settings . executionTimeout ) { workflowTimeout = this . data . workflowData . settings . executionTimeout as number ; } if ( workflowTimeout > 0 ) { workflowTimeout = Math . min ( workflowTimeout , config . get ( 'executions.maxTimeout' ) as number ) ; } this . workflow = new Workflow ( { id : this . data . workflowData . id as string | undefined , name : this . data . workflowData . name , nodes : this . data . workflowData . nodes , connections : this . data . workflowData . connections , active : this . data . workflowData . active , nodeTypes , staticData : this . data . workflowData . staticData , settings : this . data . workflowData . settings , } ) ; const additionalData = await WorkflowExecuteAdditionalData . getBase ( undefined , workflowTimeout <= 0 ? undefined : Date . now ( ) + workflowTimeout * 1000 ) ; additionalData . hooks = this . getProcessForwardHooks ( ) ; additionalData . hooks . hookFunctions . sendResponse = [ async ( response ) : Promise < void > => { await sendToParentProcess ( 'sendResponse' , { response : WebhookHelpers . encodeWebhookResponse ( response ) , } ) ; } , ] ; additionalData . executionId = inputData . executionId ; additionalData . sendMessageToUI = async ( source , message ) => { if ( workflowRunner . data ! . executionMode !== 'manual' ) { return ; } try { await sendToParentProcess ( 'sendMessageToUI' , { source , message } ) ; } catch ( error ) { this . logger . error ( `There was a problem sending UI data to parent process: "${ error . message }"` ) ; } } ; const executeWorkflowFunction = additionalData . executeWorkflow ; additionalData . executeWorkflow = async ( workflowInfo , additionalData , inputData ? ) : Promise < Array < INodeExecutionData [ ] | null > | IRun > => { const workflowData = await WorkflowExecuteAdditionalData . getWorkflowData ( workflowInfo ) ; const runData = await WorkflowExecuteAdditionalData . getRunData ( workflowData , inputData ) ; await sendToParentProcess ( 'startExecution' , { runData } ) ; const executionId = await new Promise ( ( resolve ) => { this . executionIdCallback = ( executionId ) => { resolve ( executionId ) ; } ; } ) ; let result ; try { const executeWorkflowFunctionOutput = ( await executeWorkflowFunction ( workflowInfo , additionalData , inputData , executionId , workflowData , runData ) ) as { workflowExecute : WorkflowExecute ; workflow : Workflow ; } as IWorkflowExecuteProcess ; const { workflowExecute } = executeWorkflowFunctionOutput ; this . childExecutions [ executionId ] = executeWorkflowFunctionOutput ; const { workflow } = executeWorkflowFunctionOutput ; result = await workflowExecute . processRunExecutionData ( workflow ) ; await externalHooks . run ( 'workflow.postExecute' , [ result , workflowData , executionId ] ) ; void InternalHooksManager . getInstance ( ) . onWorkflowPostExecute ( executionId , workflowData , result ) ; await sendToParentProcess ( 'finishExecution' , { executionId , result } ) ; delete this . childExecutions [ executionId ] ; } catch ( e ) { await sendToParentProcess ( 'finishExecution' , { executionId } ) ; delete this . childExecutions [ executionId ] ; throw e ; } await sendToParentProcess ( 'finishExecution' , { executionId , result } ) ; const returnData = WorkflowHelpers . getDataLastExecutedNodeData ( result ) ; return returnData ! . data ! . main ; } ; if ( this . data . executionData !== undefined ) { this . workflowExecute = new WorkflowExecute ( additionalData , this . data . executionMode , this . data . executionData ) ; return this . workflowExecute . processRunExecutionData ( this . workflow ) ; } if ( this . data . runData === undefined || this . data . startNodes === undefined || this . data . startNodes . length === 0 || this . data . destinationNode === undefined ) { this . workflowExecute = new WorkflowExecute ( additionalData , this . data . executionMode ) ; return this . workflowExecute . run ( this . workflow , undefined , this . data . destinationNode ) ; } this . workflowExecute = new WorkflowExecute ( additionalData , this . data . executionMode ) ; return this . workflowExecute . runPartialWorkflow ( this . workflow , this . data . runData , this . data . startNodes , this . data . destinationNode ) ; } param { string } hook param { any [ ] } parameters memberof async sendHookToParentProcess ( hook , parameters ) { try { await sendToParentProcess ( 'processHook' , { hook , parameters , } ) ; } catch ( error ) { this . logger . error ( `There was a problem sending hook: "${ hook }"` , { parameters , error } ) ; } } returns getProcessForwardHooks ( ) { const hookFunctions = { nodeExecuteBefore : [ async ( nodeName ) : Promise < void > => { await this . sendHookToParentProcess ( 'nodeExecuteBefore' , [ nodeName ] ) ; } , ] , nodeExecuteAfter : [ async ( nodeName , data ) : Promise < void > => { await this . sendHookToParentProcess ( 'nodeExecuteAfter' , [ nodeName , data ] ) ; } , ] , workflowExecuteBefore : [ async ( ) : Promise < void > => { await this . sendHookToParentProcess ( 'workflowExecuteBefore' , [ ] ) ; } , ] , workflowExecuteAfter : [ async ( fullRunData , newStaticData ? ) : Promise < void > => { await this . sendHookToParentProcess ( 'workflowExecuteAfter' , [ fullRunData , newStaticData ] ) ; } , ] , } ; const preExecuteFunctions = WorkflowExecuteAdditionalData . hookFunctionsPreExecute ( ) ; for ( const key of Object . keys ( preExecuteFunctions ) ) { if ( hookFunctions [ key ] === undefined ) { hookFunctions [ key ] = [ ] ; } hookFunctions [ key ] ! . push . apply ( hookFunctions [ key ] , preExecuteFunctions [ key ] ) ; } return new WorkflowHooks ( hookFunctions , this . data ! . executionMode , this . data ! . executionId , this . data ! . workflowData , { sessionId : this . data ! . sessionId , retryOf : this . data ! . retryOf as string } ) ; } } param { string } type param { * } data returns { Promise < void > } async function sendToParentProcess ( type , data ) { return new Promise ( ( resolve , reject ) => { process . send ! ( { type , data , } , ( error ) => { if ( error ) { return reject ( error ) ; } resolve ( ) ; } ) ; } ) ; } const workflowRunner = new WorkflowRunnerProcess ( ) ; process . on ( 'message' , async ( message ) => { try { if ( message . type === 'startWorkflow' ) { await sendToParentProcess ( 'start' , { } ) ; const runData = await workflowRunner . runWorkflow ( message . data ) ; await sendToParentProcess ( 'end' , { runData , } ) ; process . exit ( ) ; } else if ( message . type === 'stopExecution' || message . type === 'timeout' ) { let runData ; if ( workflowRunner . workflowExecute !== undefined ) { const executionIds = Object . keys ( workflowRunner . childExecutions ) ; for ( const executionId of executionIds ) { const childWorkflowExecute = workflowRunner . childExecutions [ executionId ] ; runData = childWorkflowExecute . workflowExecute . getFullRunData ( workflowRunner . childExecutions [ executionId ] . startedAt ) ; const timeOutError = message . type === 'timeout' ? new WorkflowOperationError ( 'Workflow execution timed out!' ) : new WorkflowOperationError ( 'Workflow-Execution has been canceled!' ) ; await childWorkflowExecute . workflowExecute . processSuccessExecution ( workflowRunner . childExecutions [ executionId ] . startedAt , childWorkflowExecute . workflow , timeOutError ) ; } runData = workflowRunner . workflowExecute . getFullRunData ( workflowRunner . startedAt ) ; const timeOutError = message . type === 'timeout' ? new WorkflowOperationError ( 'Workflow execution timed out!' ) : new WorkflowOperationError ( 'Workflow-Execution has been canceled!' ) ; await workflowRunner . workflowExecute . processSuccessExecution ( workflowRunner . startedAt , workflowRunner . workflow ! , timeOutError ) ; } else { runData = { data : { resultData : { runData : { } , } , } , finished : false , mode : workflowRunner . data ? workflowRunner . data . executionMode : ( 'own' as WorkflowExecuteMode ) , startedAt : workflowRunner . startedAt , stoppedAt : new Date ( ) , } ; workflowRunner . sendHookToParentProcess ( 'workflowExecuteAfter' , [ runData ] ) ; } await sendToParentProcess ( message . type === 'timeout' ? message . type : 'end' , { runData , } ) ; process . exit ( ) ; } else if ( message . type === 'executionId' ) { workflowRunner . executionIdCallback ( message . data . executionId ) ; } } catch ( error ) { const executionError = { ... error , name : error . name || 'Error' , message : error . message , stack : error . stack , } as ExecutionError ; await sendToParentProcess ( 'processError' , { executionError , } ) ; process . exit ( ) ; } } ) ;