import { IRun , LoggerProxy as Logger , WorkflowOperationError } from 'n8n-workflow' ; import { FindManyOptions , LessThanOrEqual , ObjectLiteral } from 'typeorm' ; import { DateUtils } from 'typeorm/util/DateUtils' ; import { ActiveExecutions , DatabaseType , Db , GenericHelpers , IExecutionFlattedDb , IExecutionsStopData , IWorkflowExecutionDataProcess , ResponseHelper , WorkflowCredentials , WorkflowRunner , } from '.' ; export class WaitTrackerClass { activeExecutionsInstance : ActiveExecutions . ActiveExecutions ; private waitingExecutions : { [ key ] : { executionId : string ; timer : NodeJS . Timeout ; } ; } = { } ; mainTimer : NodeJS . Timeout ; constructor ( ) { this . activeExecutionsInstance = ActiveExecutions . getInstance ( ) ; this . mainTimer = setInterval ( ( ) => { this . getwaitingExecutions ( ) ; } , 60000 ) ; this . getwaitingExecutions ( ) ; } async getwaitingExecutions ( ) { Logger . debug ( 'Wait tracker querying database for waiting executions' ) ; const findQuery = { select : [ 'id' , 'waitTill' ] , where : { waitTill : LessThanOrEqual ( new Date ( Date . now ( ) + 70000 ) ) , } , order : { waitTill : 'ASC' , } , } ; const dbType = ( await GenericHelpers . getConfigValue ( 'database.type' ) ) as DatabaseType ; if ( dbType === 'sqlite' ) { ( findQuery . where ! as ObjectLiteral ) . waitTill = LessThanOrEqual ( DateUtils . mixedDateToUtcDatetimeString ( new Date ( Date . now ( ) + 70000 ) ) ) ; } const executions = await Db . collections . Execution ! . find ( findQuery ) ; if ( executions . length === 0 ) { return ; } const executionIds = executions . map ( ( execution ) => execution . id . toString ( ) ) . join ( ', ' ) ; Logger . debug ( `Wait tracker found ${ executions . length } executions. Setting timer for IDs: ${ executionIds }` ) ; for ( const execution of executions ) { const executionId = execution . id . toString ( ) ; if ( this . waitingExecutions [ executionId ] === undefined ) { const triggerTime = execution . waitTill ! . getTime ( ) - new Date ( ) . getTime ( ) ; this . waitingExecutions [ executionId ] = { executionId , timer : setTimeout ( ( ) => { this . startExecution ( executionId ) ; } , triggerTime ) , } ; } } } async stopExecution ( executionId ) { if ( this . waitingExecutions [ executionId ] !== undefined ) { clearTimeout ( this . waitingExecutions [ executionId ] . timer ) ; delete this . waitingExecutions [ executionId ] ; } const execution = await Db . collections . Execution ! . findOne ( executionId ) ; if ( execution === undefined || ! execution . waitTill ) { throw new Error ( `The execution ID "${ executionId }" could not be found.` ) ; } const fullExecutionData = ResponseHelper . unflattenExecutionData ( execution ) ; const error = new WorkflowOperationError ( 'Workflow-Execution has been canceled!' ) ; fullExecutionData . data . resultData . error = { ... error , message : error . message , stack : error . stack , } ; fullExecutionData . stoppedAt = new Date ( ) ; fullExecutionData . waitTill = undefined ; await Db . collections . Execution ! . update ( executionId , ResponseHelper . flattenExecutionData ( fullExecutionData ) ) ; return { mode : fullExecutionData . mode , startedAt : new Date ( fullExecutionData . startedAt ) , stoppedAt : fullExecutionData . stoppedAt ? new Date ( fullExecutionData . stoppedAt ) : undefined , finished : fullExecutionData . finished , } ; } startExecution ( executionId ) { Logger . debug ( `Wait tracker resuming execution ${ executionId }` , { executionId } ) ; delete this . waitingExecutions [ executionId ] ; ( async ( ) => { const fullExecutionDataFlatted = await Db . collections . Execution ! . findOne ( executionId ) ; if ( fullExecutionDataFlatted === undefined ) { throw new Error ( `The execution with the id "${ executionId }" does not exist.` ) ; } const fullExecutionData = ResponseHelper . unflattenExecutionData ( fullExecutionDataFlatted ) ; if ( fullExecutionData . finished ) { throw new Error ( 'The execution did succeed and can so not be started again.' ) ; } const data = { executionMode : fullExecutionData . mode , executionData : fullExecutionData . data , workflowData : fullExecutionData . workflowData , } ; const workflowRunner = new WorkflowRunner ( ) ; await workflowRunner . run ( data , false , false , executionId ) ; } ) ( ) . catch ( ( error ) => { Logger . error ( `There was a problem starting the waiting execution with id "${ executionId }": "${ error . message }"` , { executionId } ) ; } ) ; } } let waitTrackerInstance ; export function WaitTracker ( ) { if ( waitTrackerInstance === undefined ) { waitTrackerInstance = new WaitTrackerClass ( ) ; } return waitTrackerInstance ; }