2019-06-23 03:35:23 -07:00
import {
2021-02-20 04:51:06 -08:00
ActiveExecutions ,
2020-01-25 23:48:38 -08:00
CredentialsHelper ,
2019-06-23 03:35:23 -07:00
Db ,
2020-05-05 15:59:58 -07:00
ExternalHooks ,
2019-06-23 03:35:23 -07:00
IExecutionDb ,
IExecutionFlattedDb ,
2021-02-08 23:59:32 -08:00
IExecutionResponse ,
2019-06-23 03:35:23 -07:00
IPushDataExecutionFinished ,
IWorkflowBase ,
2021-04-17 07:44:07 -07:00
IWorkflowExecuteProcess ,
2019-08-08 11:38:25 -07:00
IWorkflowExecutionDataProcess ,
2019-12-19 14:07:55 -08:00
NodeTypes ,
2019-06-23 03:35:23 -07:00
Push ,
ResponseHelper ,
WebhookHelpers ,
2020-01-20 08:22:12 -08:00
WorkflowCredentials ,
2019-06-23 03:35:23 -07:00
WorkflowHelpers ,
} from './' ;
import {
UserSettings ,
2019-12-19 14:07:55 -08:00
WorkflowExecute ,
2019-09-19 05:14:37 -07:00
} from 'n8n-core' ;
2019-06-23 03:35:23 -07:00
import {
2019-08-08 11:38:25 -07:00
IDataObject ,
2019-12-19 14:07:55 -08:00
IExecuteData ,
2020-01-02 15:13:53 -08:00
IExecuteWorkflowInfo ,
2019-12-19 14:07:55 -08:00
INode ,
INodeExecutionData ,
2020-10-22 06:46:03 -07:00
INodeParameters ,
2019-06-23 03:35:23 -07:00
IRun ,
2019-12-19 14:07:55 -08:00
IRunExecutionData ,
2019-06-23 03:35:23 -07:00
ITaskData ,
2019-08-08 11:38:25 -07:00
IWorkflowCredentials ,
2019-06-23 03:35:23 -07:00
IWorkflowExecuteAdditionalData ,
2019-08-08 11:38:25 -07:00
IWorkflowExecuteHooks ,
2019-12-19 14:07:55 -08:00
IWorkflowHooksOptionalParameters ,
2021-05-01 20:43:01 -07:00
LoggerProxy as Logger ,
2019-12-19 14:07:55 -08:00
Workflow ,
2019-06-23 03:35:23 -07:00
WorkflowExecuteMode ,
2019-12-19 14:07:55 -08:00
WorkflowHooks ,
2019-06-23 03:35:23 -07:00
} from 'n8n-workflow' ;
2019-07-21 10:47:41 -07:00
import * as config from '../config' ;
2019-06-23 03:35:23 -07:00
2021-05-01 20:43:01 -07:00
import { LessThanOrEqual } from 'typeorm' ;
2020-07-09 06:17:47 -07:00
2020-12-12 09:36:47 -08:00
const ERROR_TRIGGER_TYPE = config . get ( 'nodes.errorTriggerType' ) as string ;
2019-06-23 03:35:23 -07:00
/ * *
2020-12-12 09:36:47 -08:00
* Checks if there was an error and if errorWorkflow or a trigger is defined . If so it collects
2019-06-23 03:35:23 -07:00
* all the data and executes it
*
* @param { IWorkflowBase } workflowData The workflow which got executed
* @param { IRun } fullRunData The run which produced the error
2019-12-19 14:07:55 -08:00
* @param { WorkflowExecuteMode } mode The mode in which the workflow got started in
2019-06-23 03:35:23 -07:00
* @param { string } [ executionId ] The id the execution got saved as
* /
2019-09-11 08:48:14 -07:00
function executeErrorWorkflow ( workflowData : IWorkflowBase , fullRunData : IRun , mode : WorkflowExecuteMode , executionId? : string , retryOf? : string ) : void {
2020-12-12 09:36:47 -08:00
// Check if there was an error and if so if an errorWorkflow or a trigger is set
2020-02-19 09:29:03 -08:00
let pastExecutionUrl : string | undefined = undefined ;
if ( executionId !== undefined ) {
pastExecutionUrl = ` ${ WebhookHelpers . getWebhookBaseUrl ( ) } execution/ ${ executionId } ` ;
}
2020-12-12 09:36:47 -08:00
if ( fullRunData . data . resultData . error !== undefined ) {
2019-06-23 03:35:23 -07:00
const workflowErrorData = {
execution : {
id : executionId ,
2020-02-19 09:29:03 -08:00
url : pastExecutionUrl ,
2019-06-23 03:35:23 -07:00
error : fullRunData.data.resultData.error ,
lastNodeExecuted : fullRunData.data.resultData.lastNodeExecuted ! ,
mode ,
2019-09-11 08:48:14 -07:00
retryOf ,
2019-06-23 03:35:23 -07:00
} ,
workflow : {
id : workflowData.id !== undefined ? workflowData . id . toString ( ) as string : undefined ,
name : workflowData.name ,
2020-10-22 06:46:03 -07:00
} ,
2019-06-23 03:35:23 -07:00
} ;
2020-12-12 09:41:11 -08:00
2019-06-23 03:35:23 -07:00
// Run the error workflow
2020-12-12 09:41:11 -08:00
// To avoid an infinite loop do not run the error workflow again if the error-workflow itself failed and it is its own error-workflow.
2020-12-12 11:59:17 -08:00
if ( workflowData . settings !== undefined && workflowData . settings . errorWorkflow && ! ( mode === 'error' && workflowData . id && workflowData . settings . errorWorkflow . toString ( ) === workflowData . id . toString ( ) ) ) {
2021-05-05 13:18:14 -07:00
Logger . verbose ( ` Start external error workflow ` , { executionId , errorWorkflowId : workflowData.settings.errorWorkflow.toString ( ) , workflowId : workflowData.id } ) ;
2020-12-12 09:41:11 -08:00
// If a specific error workflow is set run only that one
2020-12-12 09:36:47 -08:00
WorkflowHelpers . executeErrorWorkflow ( workflowData . settings . errorWorkflow as string , workflowErrorData ) ;
2020-12-12 09:41:11 -08:00
} else if ( mode !== 'error' && workflowData . id !== undefined && workflowData . nodes . some ( ( node ) = > node . type === ERROR_TRIGGER_TYPE ) ) {
2021-05-05 13:18:14 -07:00
Logger . verbose ( ` Start internal error workflow ` , { executionId , workflowId : workflowData.id } ) ;
2020-12-12 09:41:11 -08:00
// If the workflow contains
WorkflowHelpers . executeErrorWorkflow ( workflowData . id . toString ( ) , workflowErrorData ) ;
2020-12-12 09:36:47 -08:00
}
2019-06-23 03:35:23 -07:00
}
}
2020-07-09 06:17:47 -07:00
/ * *
* Prunes Saved Execution which are older than configured .
* Throttled to be executed just once in configured timeframe .
*
* /
2020-07-12 03:48:32 -07:00
let throttling = false ;
2020-07-09 07:52:38 -07:00
function pruneExecutionData ( ) : void {
if ( ! throttling ) {
2021-05-01 20:43:01 -07:00
Logger . verbose ( 'Pruning execution data from database' ) ;
2020-07-09 07:52:38 -07:00
throttling = true ;
2020-07-12 03:48:32 -07:00
const timeout = config . get ( 'executions.pruneDataTimeout' ) as number ; // in seconds
2020-07-09 07:52:38 -07:00
const maxAge = config . get ( 'executions.pruneDataMaxAge' ) as number ; // in h
const date = new Date ( ) ; // today
date . setHours ( date . getHours ( ) - maxAge ) ;
// throttle just on success to allow for self healing on failure
2020-07-12 03:48:32 -07:00
Db . collections . Execution ! . delete ( { stoppedAt : LessThanOrEqual ( date . toISOString ( ) ) } )
2021-02-20 04:51:06 -08:00
. then ( data = >
setTimeout ( ( ) = > {
throttling = false ;
} , timeout * 1000 )
) . catch ( err = > throttling = false ) ;
2020-07-09 06:17:47 -07:00
}
}
2019-06-23 03:35:23 -07:00
2019-08-08 11:38:25 -07:00
/ * *
2019-12-19 14:07:55 -08:00
* Returns hook functions to push data to Editor - UI
2019-08-08 11:38:25 -07:00
*
* @returns { IWorkflowExecuteHooks }
* /
2019-12-19 14:07:55 -08:00
function hookFunctionsPush ( ) : IWorkflowExecuteHooks {
2019-06-23 03:35:23 -07:00
return {
nodeExecuteBefore : [
2019-12-19 14:07:55 -08:00
async function ( this : WorkflowHooks , nodeName : string ) : Promise < void > {
2019-08-08 11:38:25 -07:00
// Push data to session which started workflow before each
// node which starts rendering
2019-12-19 14:07:55 -08:00
if ( this . sessionId === undefined ) {
2019-06-23 03:35:23 -07:00
return ;
}
2021-05-01 20:43:01 -07:00
Logger . debug ( ` Executing hook on node " ${ nodeName } " (hookFunctionsPush) ` , { executionId : this.executionId , sessionId : this.sessionId , workflowId : this.workflowData.id } ) ;
2019-06-23 03:35:23 -07:00
2019-08-28 06:28:47 -07:00
const pushInstance = Push . getInstance ( ) ;
2019-08-08 11:38:25 -07:00
pushInstance . send ( 'nodeExecuteBefore' , {
2019-12-19 14:07:55 -08:00
executionId : this.executionId ,
2019-06-23 03:35:23 -07:00
nodeName ,
2019-12-19 14:07:55 -08:00
} , this . sessionId ) ;
2019-06-23 03:35:23 -07:00
} ,
] ,
nodeExecuteAfter : [
2019-12-19 14:07:55 -08:00
async function ( this : WorkflowHooks , nodeName : string , data : ITaskData ) : Promise < void > {
2019-08-08 11:38:25 -07:00
// Push data to session which started workflow after each rendered node
2019-12-19 14:07:55 -08:00
if ( this . sessionId === undefined ) {
2019-06-23 03:35:23 -07:00
return ;
}
2021-05-01 20:43:01 -07:00
Logger . debug ( ` Executing hook on node " ${ nodeName } " (hookFunctionsPush) ` , { executionId : this.executionId , sessionId : this.sessionId , workflowId : this.workflowData.id } ) ;
2019-06-23 03:35:23 -07:00
2019-08-28 06:28:47 -07:00
const pushInstance = Push . getInstance ( ) ;
2019-08-08 11:38:25 -07:00
pushInstance . send ( 'nodeExecuteAfter' , {
2019-12-19 14:07:55 -08:00
executionId : this.executionId ,
2019-06-23 03:35:23 -07:00
nodeName ,
data ,
2019-12-19 14:07:55 -08:00
} , this . sessionId ) ;
2019-06-23 03:35:23 -07:00
} ,
] ,
2019-07-24 05:25:30 -07:00
workflowExecuteBefore : [
2019-12-19 14:07:55 -08:00
async function ( this : WorkflowHooks ) : Promise < void > {
2021-05-01 20:43:01 -07:00
Logger . debug ( ` Executing hook (hookFunctionsPush) ` , { executionId : this.executionId , sessionId : this.sessionId , workflowId : this.workflowData.id } ) ;
2021-02-13 11:40:27 -08:00
// Push data to session which started the workflow
if ( this . sessionId === undefined ) {
return ;
2021-02-09 14:32:40 -08:00
}
2021-02-13 11:40:27 -08:00
const pushInstance = Push . getInstance ( ) ;
pushInstance . send ( 'executionStarted' , {
executionId : this.executionId ,
mode : this.mode ,
startedAt : new Date ( ) ,
retryOf : this.retryOf ,
2021-05-01 20:43:01 -07:00
workflowId : this.workflowData.id , sessionId : this.sessionId as string ,
2021-02-13 11:40:27 -08:00
workflowName : this.workflowData.name ,
} , this . sessionId ) ;
2020-10-22 06:46:03 -07:00
} ,
2019-07-24 05:25:30 -07:00
] ,
2019-06-23 03:35:23 -07:00
workflowExecuteAfter : [
2019-12-19 14:07:55 -08:00
async function ( this : WorkflowHooks , fullRunData : IRun , newStaticData : IDataObject ) : Promise < void > {
2021-05-01 20:43:01 -07:00
Logger . debug ( ` Executing hook (hookFunctionsPush) ` , { executionId : this.executionId , sessionId : this.sessionId , workflowId : this.workflowData.id } ) ;
2021-02-13 11:40:27 -08:00
// Push data to session which started the workflow
if ( this . sessionId === undefined ) {
return ;
2021-02-09 14:32:40 -08:00
}
2021-02-13 11:40:27 -08:00
// Clone the object except the runData. That one is not supposed
// to be send. Because that data got send piece by piece after
// each node which finished executing
const pushRunData = {
. . . fullRunData ,
data : {
. . . fullRunData . data ,
resultData : {
. . . fullRunData . data . resultData ,
runData : { } ,
} ,
} ,
} ;
// Push data to editor-ui once workflow finished
2021-05-01 20:43:01 -07:00
Logger . debug ( ` Save execution progress to database for execution ID ${ this . executionId } ` , { executionId : this.executionId , workflowId : this.workflowData.id } ) ;
2021-02-13 11:40:27 -08:00
// TODO: Look at this again
const sendData : IPushDataExecutionFinished = {
executionId : this.executionId ,
data : pushRunData ,
retryOf : this.retryOf ,
} ;
const pushInstance = Push . getInstance ( ) ;
pushInstance . send ( 'executionFinished' , sendData , this . sessionId ) ;
2019-12-19 14:07:55 -08:00
} ,
2020-10-22 06:46:03 -07:00
] ,
2019-12-19 14:07:55 -08:00
} ;
}
2020-11-13 14:31:27 -08:00
export function hookFunctionsPreExecute ( parentProcessMode? : string ) : IWorkflowExecuteHooks {
const externalHooks = ExternalHooks ( ) ;
return {
workflowExecuteBefore : [
async function ( this : WorkflowHooks , workflow : Workflow ) : Promise < void > {
await externalHooks . run ( 'workflow.preExecute' , [ workflow , this . mode ] ) ;
} ,
] ,
2021-02-08 23:59:32 -08:00
nodeExecuteAfter : [
async function ( nodeName : string , data : ITaskData , executionData : IRunExecutionData ) : Promise < void > {
if ( this . workflowData . settings !== undefined ) {
if ( this . workflowData . settings . saveExecutionProgress === false ) {
return ;
} else if ( this . workflowData . settings . saveExecutionProgress !== true && ! config . get ( 'executions.saveExecutionProgress' ) as boolean ) {
return ;
}
} else if ( ! config . get ( 'executions.saveExecutionProgress' ) as boolean ) {
return ;
}
2021-03-10 01:50:07 -08:00
try {
2021-05-01 20:43:01 -07:00
Logger . debug ( ` Save execution progress to database for execution ID ${ this . executionId } ` , { executionId : this.executionId , nodeName } ) ;
2021-03-10 01:50:07 -08:00
const execution = await Db . collections . Execution ! . findOne ( this . executionId ) ;
2021-02-08 23:59:32 -08:00
2021-03-10 01:50:07 -08:00
if ( execution === undefined ) {
// Something went badly wrong if this happens.
// This check is here mostly to make typescript happy.
return undefined ;
}
const fullExecutionData : IExecutionResponse = ResponseHelper . unflattenExecutionData ( execution ) ;
2021-02-08 23:59:32 -08:00
2021-03-10 01:50:07 -08:00
if ( fullExecutionData . finished ) {
// We already received ´ workflowExecuteAfter´ webhook, so this is just an async call
// that was left behind. We skip saving because the other call should have saved everything
// so this one is safe to ignore
return ;
}
2021-02-08 23:59:32 -08:00
2021-03-10 01:50:07 -08:00
if ( fullExecutionData . data === undefined ) {
fullExecutionData . data = {
startData : {
} ,
resultData : {
runData : { } ,
} ,
executionData : {
contextData : { } ,
nodeExecutionStack : [ ] ,
waitingExecution : { } ,
} ,
} ;
}
2021-02-08 23:59:32 -08:00
2021-03-10 01:50:07 -08:00
if ( Array . isArray ( fullExecutionData . data . resultData . runData [ nodeName ] ) ) {
// Append data if array exists
fullExecutionData . data . resultData . runData [ nodeName ] . push ( data ) ;
} else {
// Initialize array and save data
fullExecutionData . data . resultData . runData [ nodeName ] = [ data ] ;
}
fullExecutionData . data . executionData = executionData . executionData ;
2021-02-08 23:59:32 -08:00
2021-03-10 01:50:07 -08:00
// Set last executed node so that it may resume on failure
fullExecutionData . data . resultData . lastNodeExecuted = nodeName ;
2021-02-08 23:59:32 -08:00
2021-03-10 01:50:07 -08:00
const flattenedExecutionData = ResponseHelper . flattenExecutionData ( fullExecutionData ) ;
2021-02-13 11:40:27 -08:00
2021-03-10 01:50:07 -08:00
await Db . collections . Execution ! . update ( this . executionId , flattenedExecutionData as IExecutionFlattedDb ) ;
} catch ( err ) {
// TODO: Improve in the future!
// Errors here might happen because of database access
// For busy machines, we may get "Database is locked" errors.
// We do this to prevent crashes and executions ending in `unknown` state.
2021-05-01 20:43:01 -07:00
Logger . error ( ` Failed saving execution progress to database for execution ID ${ this . executionId } (hookFunctionsPreExecute, nodeExecuteAfter) ` , { . . . err , executionId : this.executionId , sessionId : this.sessionId , workflowId : this.workflowData.id } ) ;
2021-03-10 01:50:07 -08:00
}
2021-02-08 23:59:32 -08:00
} ,
] ,
2020-11-13 14:31:27 -08:00
} ;
}
2021-02-20 04:51:06 -08:00
2019-12-19 14:07:55 -08:00
/ * *
* Returns hook functions to save workflow execution and call error workflow
*
* @returns { IWorkflowExecuteHooks }
* /
function hookFunctionsSave ( parentProcessMode? : string ) : IWorkflowExecuteHooks {
return {
nodeExecuteBefore : [ ] ,
nodeExecuteAfter : [ ] ,
workflowExecuteBefore : [ ] ,
workflowExecuteAfter : [
async function ( this : WorkflowHooks , fullRunData : IRun , newStaticData : IDataObject ) : Promise < void > {
2021-05-01 20:43:01 -07:00
Logger . debug ( ` Executing hook (hookFunctionsSave) ` , { executionId : this.executionId , workflowId : this.workflowData.id } ) ;
2019-12-19 14:07:55 -08:00
2020-07-12 03:48:32 -07:00
// Prune old execution data
if ( config . get ( 'executions.pruneData' ) ) {
pruneExecutionData ( ) ;
}
2019-12-19 14:07:55 -08:00
const isManualMode = [ this . mode , parentProcessMode ] . includes ( 'manual' ) ;
2019-06-23 03:35:23 -07:00
try {
2019-12-19 14:07:55 -08:00
if ( ! isManualMode && WorkflowHelpers . isWorkflowIdValid ( this . workflowData . id as string ) === true && newStaticData ) {
2019-08-08 11:38:25 -07:00
// Workflow is saved so update in database
try {
2019-12-19 14:07:55 -08:00
await WorkflowHelpers . saveStaticDataById ( this . workflowData . id as string , newStaticData ) ;
2019-08-08 11:38:25 -07:00
} catch ( e ) {
2021-05-01 20:43:01 -07:00
Logger . error ( ` There was a problem saving the workflow with id " ${ this . workflowData . id } " to save changed staticData: " ${ e . message } " (hookFunctionsSave) ` , { executionId : this.executionId , workflowId : this.workflowData.id } ) ;
2019-08-08 11:38:25 -07:00
}
}
2019-06-23 03:35:23 -07:00
2019-07-21 10:47:41 -07:00
let saveManualExecutions = config . get ( 'executions.saveDataManualExecutions' ) as boolean ;
2019-12-19 14:07:55 -08:00
if ( this . workflowData . settings !== undefined && this . workflowData . settings . saveManualExecutions !== undefined ) {
2019-06-23 03:35:23 -07:00
// Apply to workflow override
2019-12-19 14:07:55 -08:00
saveManualExecutions = this . workflowData . settings . saveManualExecutions as boolean ;
2019-06-23 03:35:23 -07:00
}
2019-12-19 14:07:55 -08:00
if ( isManualMode && saveManualExecutions === false ) {
2021-02-08 23:59:32 -08:00
// Data is always saved, so we remove from database
2021-02-20 04:51:06 -08:00
await Db . collections . Execution ! . delete ( this . executionId ) ;
2019-07-10 11:53:13 -07:00
return ;
}
// Check config to know if execution should be saved or not
2019-07-21 10:47:41 -07:00
let saveDataErrorExecution = config . get ( 'executions.saveDataOnError' ) as string ;
let saveDataSuccessExecution = config . get ( 'executions.saveDataOnSuccess' ) as string ;
2019-12-19 14:07:55 -08:00
if ( this . workflowData . settings !== undefined ) {
saveDataErrorExecution = ( this . workflowData . settings . saveDataErrorExecution as string ) || saveDataErrorExecution ;
saveDataSuccessExecution = ( this . workflowData . settings . saveDataSuccessExecution as string ) || saveDataSuccessExecution ;
2019-07-10 11:53:13 -07:00
}
2019-06-23 03:35:23 -07:00
2019-07-10 11:53:13 -07:00
const workflowDidSucceed = ! fullRunData . data . resultData . error ;
if ( workflowDidSucceed === true && saveDataSuccessExecution === 'none' ||
workflowDidSucceed === false && saveDataErrorExecution === 'none'
) {
2019-12-19 14:07:55 -08:00
if ( ! isManualMode ) {
executeErrorWorkflow ( this . workflowData , fullRunData , this . mode , undefined , this . retryOf ) ;
}
2021-02-08 23:59:32 -08:00
// Data is always saved, so we remove from database
2021-03-10 06:51:18 -08:00
await Db . collections . Execution ! . delete ( this . executionId ) ;
2019-06-23 03:35:23 -07:00
return ;
}
const fullExecutionData : IExecutionDb = {
data : fullRunData.data ,
mode : fullRunData.mode ,
finished : fullRunData.finished ? fullRunData.finished : false ,
startedAt : fullRunData.startedAt ,
stoppedAt : fullRunData.stoppedAt ,
2019-12-19 14:07:55 -08:00
workflowData : this.workflowData ,
2019-06-23 03:35:23 -07:00
} ;
2019-12-19 14:07:55 -08:00
if ( this . retryOf !== undefined ) {
fullExecutionData . retryOf = this . retryOf . toString ( ) ;
2019-06-23 03:35:23 -07:00
}
2019-12-19 14:07:55 -08:00
if ( this . workflowData . id !== undefined && WorkflowHelpers . isWorkflowIdValid ( this . workflowData . id . toString ( ) ) === true ) {
fullExecutionData . workflowId = this . workflowData . id . toString ( ) ;
2019-06-23 03:35:23 -07:00
}
2021-05-01 20:43:01 -07:00
// Leave log message before flatten as that operation increased memory usage a lot and the chance of a crash is highest here
2021-06-15 03:32:13 -07:00
Logger . debug ( ` Save execution data to database for execution ID ${ this . executionId } ` , {
executionId : this.executionId ,
workflowId : this.workflowData.id ,
finished : fullExecutionData.finished ,
stoppedAt : fullExecutionData.stoppedAt ,
} ) ;
2021-05-01 20:43:01 -07:00
2019-06-23 03:35:23 -07:00
const executionData = ResponseHelper . flattenExecutionData ( fullExecutionData ) ;
// Save the Execution in DB
2021-02-08 23:59:32 -08:00
await Db . collections . Execution ! . update ( this . executionId , executionData as IExecutionFlattedDb ) ;
2019-06-23 03:35:23 -07:00
2019-12-19 14:07:55 -08:00
if ( fullRunData . finished === true && this . retryOf !== undefined ) {
2019-06-23 03:35:23 -07:00
// If the retry was successful save the reference it on the original execution
// await Db.collections.Execution!.save(executionData as IExecutionFlattedDb);
2021-02-08 23:59:32 -08:00
await Db . collections . Execution ! . update ( this . retryOf , { retrySuccessId : this.executionId } ) ;
2019-06-23 03:35:23 -07:00
}
2019-12-19 14:07:55 -08:00
if ( ! isManualMode ) {
2021-02-08 23:59:32 -08:00
executeErrorWorkflow ( this . workflowData , fullRunData , this . mode , this . executionId , this . retryOf ) ;
2019-12-19 14:07:55 -08:00
}
2019-06-23 03:35:23 -07:00
} catch ( error ) {
2021-06-15 03:32:13 -07:00
Logger . error ( ` Failed saving execution data to DB on execution ID ${ this . executionId } ` , {
executionId : this.executionId ,
workflowId : this.workflowData.id ,
error ,
} ) ;
2019-12-19 14:07:55 -08:00
if ( ! isManualMode ) {
executeErrorWorkflow ( this . workflowData , fullRunData , this . mode , undefined , this . retryOf ) ;
}
2019-06-23 03:35:23 -07:00
}
} ,
2020-10-22 06:46:03 -07:00
] ,
2019-06-23 03:35:23 -07:00
} ;
2019-12-19 14:07:55 -08:00
}
2021-03-10 06:51:18 -08:00
/ * *
* Returns hook functions to save workflow execution and call error workflow
* for running with queues . Manual executions should never run on queues as
* they are always executed in the main process .
*
* @returns { IWorkflowExecuteHooks }
* /
function hookFunctionsSaveWorker ( ) : IWorkflowExecuteHooks {
return {
nodeExecuteBefore : [ ] ,
nodeExecuteAfter : [ ] ,
workflowExecuteBefore : [ ] ,
workflowExecuteAfter : [
async function ( this : WorkflowHooks , fullRunData : IRun , newStaticData : IDataObject ) : Promise < void > {
try {
if ( WorkflowHelpers . isWorkflowIdValid ( this . workflowData . id as string ) === true && newStaticData ) {
// Workflow is saved so update in database
try {
await WorkflowHelpers . saveStaticDataById ( this . workflowData . id as string , newStaticData ) ;
} catch ( e ) {
2021-05-01 20:43:01 -07:00
Logger . error ( ` There was a problem saving the workflow with id " ${ this . workflowData . id } " to save changed staticData: " ${ e . message } " (workflowExecuteAfter) ` , { sessionId : this.sessionId , workflowId : this.workflowData.id } ) ;
2021-03-10 06:51:18 -08:00
}
}
const workflowDidSucceed = ! fullRunData . data . resultData . error ;
2021-06-03 16:41:12 -07:00
if ( workflowDidSucceed === false ) {
2021-03-10 06:51:18 -08:00
executeErrorWorkflow ( this . workflowData , fullRunData , this . mode , undefined , this . retryOf ) ;
}
const fullExecutionData : IExecutionDb = {
data : fullRunData.data ,
mode : fullRunData.mode ,
finished : fullRunData.finished ? fullRunData.finished : false ,
startedAt : fullRunData.startedAt ,
stoppedAt : fullRunData.stoppedAt ,
workflowData : this.workflowData ,
} ;
if ( this . retryOf !== undefined ) {
fullExecutionData . retryOf = this . retryOf . toString ( ) ;
}
if ( this . workflowData . id !== undefined && WorkflowHelpers . isWorkflowIdValid ( this . workflowData . id . toString ( ) ) === true ) {
fullExecutionData . workflowId = this . workflowData . id . toString ( ) ;
}
const executionData = ResponseHelper . flattenExecutionData ( fullExecutionData ) ;
// Save the Execution in DB
await Db . collections . Execution ! . update ( this . executionId , executionData as IExecutionFlattedDb ) ;
if ( fullRunData . finished === true && this . retryOf !== undefined ) {
// If the retry was successful save the reference it on the original execution
await Db . collections . Execution ! . update ( this . retryOf , { retrySuccessId : this.executionId } ) ;
}
} catch ( error ) {
executeErrorWorkflow ( this . workflowData , fullRunData , this . mode , undefined , this . retryOf ) ;
}
} ,
] ,
} ;
}
2021-02-20 04:51:06 -08:00
export async function getRunData ( workflowData : IWorkflowBase , inputData? : INodeExecutionData [ ] ) : Promise < IWorkflowExecutionDataProcess > {
2019-12-19 14:07:55 -08:00
const mode = 'integrated' ;
// Find Start-Node
const requiredNodeTypes = [ 'n8n-nodes-base.start' ] ;
let startNode : INode | undefined ;
for ( const node of workflowData ! . nodes ) {
if ( requiredNodeTypes . includes ( node . type ) ) {
startNode = node ;
break ;
}
}
if ( startNode === undefined ) {
// If the workflow does not contain a start-node we can not know what
// should be executed and with what data to start.
throw new Error ( ` The workflow does not contain a "Start" node and can so not be executed. ` ) ;
}
// Always start with empty data if no inputData got supplied
inputData = inputData || [
{
2020-10-22 06:46:03 -07:00
json : { } ,
} ,
2019-12-19 14:07:55 -08:00
] ;
// Initialize the incoming data
const nodeExecutionStack : IExecuteData [ ] = [ ] ;
nodeExecutionStack . push (
{
node : startNode ,
data : {
main : [ inputData ] ,
} ,
2020-10-22 06:46:03 -07:00
}
2019-12-19 14:07:55 -08:00
) ;
const runExecutionData : IRunExecutionData = {
startData : {
} ,
resultData : {
runData : { } ,
} ,
executionData : {
contextData : { } ,
nodeExecutionStack ,
waitingExecution : { } ,
} ,
} ;
2021-02-20 04:51:06 -08:00
// Get the needed credentials for the current workflow as they will differ to the ones of the
// calling workflow.
const credentials = await WorkflowCredentials ( workflowData ! . nodes ) ;
const runData : IWorkflowExecutionDataProcess = {
credentials ,
executionMode : mode ,
executionData : runExecutionData ,
// @ts-ignore
workflowData ,
} ;
return runData ;
}
export async function getWorkflowData ( workflowInfo : IExecuteWorkflowInfo ) : Promise < IWorkflowBase > {
if ( workflowInfo . id === undefined && workflowInfo . code === undefined ) {
throw new Error ( ` No information about the workflow to execute found. Please provide either the "id" or "code"! ` ) ;
}
if ( Db . collections ! . Workflow === null ) {
// The first time executeWorkflow gets called the Database has
// to get initialized first
await Db . init ( ) ;
}
let workflowData : IWorkflowBase | undefined ;
if ( workflowInfo . id !== undefined ) {
workflowData = await Db . collections ! . Workflow ! . findOne ( workflowInfo . id ) ;
if ( workflowData === undefined ) {
throw new Error ( ` The workflow with the id " ${ workflowInfo . id } " does not exist. ` ) ;
}
} else {
workflowData = workflowInfo . code ;
}
return workflowData ! ;
}
/ * *
* Executes the workflow with the given ID
*
* @export
* @param { string } workflowId The id of the workflow to execute
* @param { IWorkflowExecuteAdditionalData } additionalData
* @param { INodeExecutionData [ ] } [ inputData ]
* @returns { ( Promise < Array < INodeExecutionData [ ] | null > > ) }
* /
2021-04-17 07:44:07 -07:00
export async function executeWorkflow ( workflowInfo : IExecuteWorkflowInfo , additionalData : IWorkflowExecuteAdditionalData , inputData? : INodeExecutionData [ ] , parentExecutionId? : string , loadedWorkflowData? : IWorkflowBase , loadedRunData? : IWorkflowExecutionDataProcess ) : Promise < Array < INodeExecutionData [ ] | null > | IWorkflowExecuteProcess > {
2021-02-20 04:51:06 -08:00
const externalHooks = ExternalHooks ( ) ;
await externalHooks . init ( ) ;
const nodeTypes = NodeTypes ( ) ;
const workflowData = loadedWorkflowData !== undefined ? loadedWorkflowData : await getWorkflowData ( workflowInfo ) ;
const workflowName = workflowData ? workflowData.name : undefined ;
const workflow = new Workflow ( { id : workflowInfo.id , name : workflowName , nodes : workflowData ! . nodes , connections : workflowData ! . connections , active : workflowData ! . active , nodeTypes , staticData : workflowData ! . staticData } ) ;
const runData = loadedRunData !== undefined ? loadedRunData : await getRunData ( workflowData , inputData ) ;
let executionId ;
if ( parentExecutionId !== undefined ) {
executionId = parentExecutionId ;
} else {
executionId = parentExecutionId !== undefined ? parentExecutionId : await ActiveExecutions . getInstance ( ) . add ( runData ) ;
}
const runExecutionData = runData . executionData as IRunExecutionData ;
// Get the needed credentials for the current workflow as they will differ to the ones of the
// calling workflow.
const credentials = await WorkflowCredentials ( workflowData ! . nodes ) ;
// Create new additionalData to have different workflow loaded and to call
// different webooks
const additionalDataIntegrated = await getBase ( credentials ) ;
additionalDataIntegrated . hooks = getWorkflowHooksIntegrated ( runData . executionMode , executionId , workflowData ! , { parentProcessMode : additionalData.hooks ! . mode } ) ;
2021-03-02 23:31:55 -08:00
// Make sure we pass on the original executeWorkflow function we received
// This one already contains changes to talk to parent process
// and get executionID from `activeExecutions` running on main process
additionalDataIntegrated . executeWorkflow = additionalData . executeWorkflow ;
2021-05-07 17:06:26 -07:00
let subworkflowTimeout = additionalData . executionTimeoutTimestamp ;
if ( workflowData . settings ? . executionTimeout !== undefined && workflowData . settings . executionTimeout > 0 ) {
// We might have received a max timeout timestamp from the parent workflow
// If we did, then we get the minimum time between the two timeouts
// If no timeout was given from the parent, then we use our timeout.
subworkflowTimeout = Math . min ( additionalData . executionTimeoutTimestamp || Number . MAX_SAFE_INTEGER , Date . now ( ) + ( workflowData . settings . executionTimeout as number * 1000 ) ) ;
}
2021-05-29 11:41:25 -07:00
2021-05-07 17:06:26 -07:00
additionalDataIntegrated . executionTimeoutTimestamp = subworkflowTimeout ;
2021-02-20 04:51:06 -08:00
2019-12-19 14:07:55 -08:00
// Execute the workflow
2021-02-20 04:51:06 -08:00
const workflowExecute = new WorkflowExecute ( additionalDataIntegrated , runData . executionMode , runExecutionData ) ;
2021-04-17 07:44:07 -07:00
if ( parentExecutionId !== undefined ) {
// Must be changed to become typed
return {
startedAt : new Date ( ) ,
workflow ,
workflowExecute ,
} ;
}
2019-12-19 14:07:55 -08:00
const data = await workflowExecute . processRunExecutionData ( workflow ) ;
2020-10-20 10:01:40 -07:00
await externalHooks . run ( 'workflow.postExecute' , [ data , workflowData ] ) ;
2019-12-19 14:07:55 -08:00
if ( data . finished === true ) {
// Workflow did finish successfully
2021-02-20 04:51:06 -08:00
2021-04-17 07:44:07 -07:00
await ActiveExecutions . getInstance ( ) . remove ( executionId , data ) ;
const returnData = WorkflowHelpers . getDataLastExecutedNodeData ( data ) ;
return returnData ! . data ! . main ;
2019-12-19 14:07:55 -08:00
} else {
2021-03-19 07:18:35 -07:00
await ActiveExecutions . getInstance ( ) . remove ( executionId , data ) ;
2019-12-19 14:07:55 -08:00
// Workflow did fail
2021-04-16 09:33:36 -07:00
const { error } = data . data . resultData ;
throw {
. . . error ,
stack : error ! . stack ,
} ;
2019-12-19 14:07:55 -08:00
}
}
2019-06-23 03:35:23 -07:00
2021-06-12 11:22:55 -07:00
export function sendMessageToUI ( source : string , message : any ) { // tslint:disable-line:no-any
2021-05-29 11:41:25 -07:00
if ( this . sessionId === undefined ) {
return ;
}
// Push data to session which started workflow
try {
const pushInstance = Push . getInstance ( ) ;
pushInstance . send ( 'sendConsoleMessage' , {
source : ` Node: " ${ source } " ` ,
message ,
} , this . sessionId ) ;
} catch ( error ) {
Logger . warn ( ` There was a problem sending messsage to UI: ${ error . message } ` ) ;
}
}
2019-08-08 11:38:25 -07:00
/ * *
* Returns the base additional data without webhooks
*
* @export
* @param { IWorkflowCredentials } credentials
2020-05-04 21:07:19 -07:00
* @param { INodeParameters } currentNodeParameters
2019-08-08 11:38:25 -07:00
* @returns { Promise < IWorkflowExecuteAdditionalData > }
* /
2021-04-17 07:44:07 -07:00
export async function getBase ( credentials : IWorkflowCredentials , currentNodeParameters? : INodeParameters , executionTimeoutTimestamp? : number ) : Promise < IWorkflowExecuteAdditionalData > {
2019-06-23 03:35:23 -07:00
const urlBaseWebhook = WebhookHelpers . getWebhookBaseUrl ( ) ;
2019-07-21 10:47:41 -07:00
const timezone = config . get ( 'generic.timezone' ) as string ;
const webhookBaseUrl = urlBaseWebhook + config . get ( 'endpoints.webhook' ) as string ;
const webhookTestBaseUrl = urlBaseWebhook + config . get ( 'endpoints.webhookTest' ) as string ;
2019-06-23 03:35:23 -07:00
const encryptionKey = await UserSettings . getEncryptionKey ( ) ;
if ( encryptionKey === undefined ) {
throw new Error ( 'No encryption key got found to decrypt the credentials!' ) ;
}
return {
2019-08-08 11:38:25 -07:00
credentials ,
2020-01-25 23:48:38 -08:00
credentialsHelper : new CredentialsHelper ( credentials , encryptionKey ) ,
2019-06-23 03:35:23 -07:00
encryptionKey ,
2019-12-19 14:07:55 -08:00
executeWorkflow ,
restApiUrl : urlBaseWebhook + config . get ( 'endpoints.rest' ) as string ,
2019-06-23 03:35:23 -07:00
timezone ,
webhookBaseUrl ,
webhookTestBaseUrl ,
2019-10-20 12:42:34 -07:00
currentNodeParameters ,
2021-04-17 07:44:07 -07:00
executionTimeoutTimestamp ,
2019-06-23 03:35:23 -07:00
} ;
}
2019-08-08 11:38:25 -07:00
/ * *
2019-12-19 14:07:55 -08:00
* Returns WorkflowHooks instance for running integrated workflows
* ( Workflows which get started inside of another workflow )
* /
export function getWorkflowHooksIntegrated ( mode : WorkflowExecuteMode , executionId : string , workflowData : IWorkflowBase , optionalParameters? : IWorkflowHooksOptionalParameters ) : WorkflowHooks {
optionalParameters = optionalParameters || { } ;
const hookFunctions = hookFunctionsSave ( optionalParameters . parentProcessMode ) ;
2020-11-13 14:31:27 -08:00
const preExecuteFunctions = hookFunctionsPreExecute ( optionalParameters . parentProcessMode ) ;
for ( const key of Object . keys ( preExecuteFunctions ) ) {
2021-02-08 23:59:32 -08:00
if ( hookFunctions [ key ] === undefined ) {
hookFunctions [ key ] = [ ] ;
}
2020-11-13 14:31:27 -08:00
hookFunctions [ key ] ! . push . apply ( hookFunctions [ key ] , preExecuteFunctions [ key ] ) ;
}
2019-12-19 14:07:55 -08:00
return new WorkflowHooks ( hookFunctions , mode , executionId , workflowData , optionalParameters ) ;
}
2021-03-10 06:51:18 -08:00
/ * *
* Returns WorkflowHooks instance for running integrated workflows
* ( Workflows which get started inside of another workflow )
* /
export function getWorkflowHooksWorkerExecuter ( mode : WorkflowExecuteMode , executionId : string , workflowData : IWorkflowBase , optionalParameters? : IWorkflowHooksOptionalParameters ) : WorkflowHooks {
optionalParameters = optionalParameters || { } ;
const hookFunctions = hookFunctionsSaveWorker ( ) ;
const preExecuteFunctions = hookFunctionsPreExecute ( optionalParameters . parentProcessMode ) ;
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 , mode , executionId , workflowData , optionalParameters ) ;
}
2019-12-19 14:07:55 -08:00
2021-02-08 23:59:32 -08:00
/ * *
* Returns WorkflowHooks instance for main process if workflow runs via worker
* /
export function getWorkflowHooksWorkerMain ( mode : WorkflowExecuteMode , executionId : string , workflowData : IWorkflowBase , optionalParameters? : IWorkflowHooksOptionalParameters ) : WorkflowHooks {
optionalParameters = optionalParameters || { } ;
const hookFunctions = hookFunctionsPush ( ) ;
const preExecuteFunctions = hookFunctionsPreExecute ( optionalParameters . parentProcessMode ) ;
for ( const key of Object . keys ( preExecuteFunctions ) ) {
if ( hookFunctions [ key ] === undefined ) {
hookFunctions [ key ] = [ ] ;
}
hookFunctions [ key ] ! . push . apply ( hookFunctions [ key ] , preExecuteFunctions [ key ] ) ;
}
// When running with worker mode, main process executes
// Only workflowExecuteBefore + workflowExecuteAfter
// So to avoid confusion, we are removing other hooks.
hookFunctions . nodeExecuteBefore = [ ] ;
hookFunctions . nodeExecuteAfter = [ ] ;
return new WorkflowHooks ( hookFunctions , mode , executionId , workflowData , optionalParameters ) ;
}
2019-12-19 14:07:55 -08:00
/ * *
* Returns WorkflowHooks instance for running the main workflow
2019-08-08 11:38:25 -07:00
*
* @export
* @param { IWorkflowExecutionDataProcess } data
* @param { string } executionId
2019-12-19 14:07:55 -08:00
* @returns { WorkflowHooks }
2019-08-08 11:38:25 -07:00
* /
2020-11-13 14:31:27 -08:00
export function getWorkflowHooksMain ( data : IWorkflowExecutionDataProcess , executionId : string , isMainProcess = false ) : WorkflowHooks {
2019-12-19 14:07:55 -08:00
const hookFunctions = hookFunctionsSave ( ) ;
const pushFunctions = hookFunctionsPush ( ) ;
for ( const key of Object . keys ( pushFunctions ) ) {
2021-02-08 23:59:32 -08:00
if ( hookFunctions [ key ] === undefined ) {
hookFunctions [ key ] = [ ] ;
}
2019-12-19 14:07:55 -08:00
hookFunctions [ key ] ! . push . apply ( hookFunctions [ key ] , pushFunctions [ key ] ) ;
}
2020-11-13 14:31:27 -08:00
if ( isMainProcess ) {
const preExecuteFunctions = hookFunctionsPreExecute ( ) ;
for ( const key of Object . keys ( preExecuteFunctions ) ) {
2021-02-08 23:59:32 -08:00
if ( hookFunctions [ key ] === undefined ) {
hookFunctions [ key ] = [ ] ;
}
2020-11-13 14:31:27 -08:00
hookFunctions [ key ] ! . push . apply ( hookFunctions [ key ] , preExecuteFunctions [ key ] ) ;
}
}
2021-02-20 04:51:06 -08:00
return new WorkflowHooks ( hookFunctions , data . executionMode , executionId , data . workflowData , { sessionId : data.sessionId , retryOf : data.retryOf as string } ) ;
2019-08-08 11:38:25 -07:00
}