import { CronJob } from 'cron' ; import { IGetExecutePollFunctions , IGetExecuteTriggerFunctions , INode , IPollResponse , ITriggerResponse , IWorkflowExecuteAdditionalData , LoggerProxy as Logger , Workflow , WorkflowActivateMode , WorkflowExecuteMode , } from 'n8n-workflow' ; import { ITriggerTime , IWorkflowData } from '.' ; export class ActiveWorkflows { private workflowData : { [ key ] : IWorkflowData ; } = { } ; param { string } id returns { boolean } memberof isActive ( id ) { return this . workflowData . hasOwnProperty ( id ) ; } returns { string [ ] } memberof allActiveWorkflows ( ) { return Object . keys ( this . workflowData ) ; } param { string } id returns { ( WorkflowData | undefined ) } memberof get ( id ) { return this . workflowData [ id ] ; } param { string } id param { Workflow } workflow param { IWorkflowExecuteAdditionalData } additionalData returns { Promise < void > } memberof async add ( id , workflow , additionalData , mode , activation , getTriggerFunctions , getPollFunctions ) { this . workflowData [ id ] = { } ; const triggerNodes = workflow . getTriggerNodes ( ) ; let triggerResponse ; this . workflowData [ id ] . triggerResponses = [ ] ; for ( const triggerNode of triggerNodes ) { triggerResponse = await workflow . runTrigger ( triggerNode , getTriggerFunctions , additionalData , mode , activation ) ; if ( triggerResponse !== undefined ) { this . workflowData [ id ] . triggerResponses ! . push ( triggerResponse ) ; } } const pollNodes = workflow . getPollNodes ( ) ; if ( pollNodes . length ) { this . workflowData [ id ] . pollResponses = [ ] ; for ( const pollNode of pollNodes ) { this . workflowData [ id ] . pollResponses ! . push ( await this . activatePolling ( pollNode , workflow , additionalData , getPollFunctions , mode , activation ) ) ; } } } param { INode } node param { Workflow } workflow param { IWorkflowExecuteAdditionalData } additionalData param { IGetExecutePollFunctions } getPollFunctions returns { Promise < IPollResponse > } memberof async activatePolling ( node , workflow , additionalData , getPollFunctions , mode , activation ) { const pollFunctions = getPollFunctions ( workflow , node , additionalData , mode , activation ) ; const pollTimes = pollFunctions . getNodeParameter ( 'pollTimes' ) as unknown as { item : ITriggerTime [ ] ; } ; const parameterOrder = [ 'second' , 'minute' , 'hour' , 'dayOfMonth' , 'month' , 'weekday' , ] ; const cronTimes = [ ] ; let cronTime ; let parameterName ; if ( pollTimes . item !== undefined ) { for ( const item of pollTimes . item ) { cronTime = [ ] ; if ( item . mode === 'custom' ) { cronTimes . push ( ( item . cronExpression as string ) . trim ( ) ) ; continue ; } if ( item . mode === 'everyMinute' ) { cronTimes . push ( `${ Math . floor ( Math . random ( ) * 60 ) . toString ( ) } * * * * *` ) ; continue ; } if ( item . mode === 'everyX' ) { if ( item . unit === 'minutes' ) { cronTimes . push ( `${ Math . floor ( Math . random ( ) * 60 ) . toString ( ) } */${ item . value } * * * *` ) ; } else if ( item . unit === 'hours' ) { cronTimes . push ( `${ Math . floor ( Math . random ( ) * 60 ) . toString ( ) } 0 */${ item . value } * * *` ) ; } continue ; } for ( parameterName of parameterOrder ) { if ( item [ parameterName ] !== undefined ) { cronTime . push ( item [ parameterName ] as string ) ; } else if ( parameterName === 'second' ) { cronTime . push ( Math . floor ( Math . random ( ) * 60 ) . toString ( ) ) ; } else { cronTime . push ( '*' ) ; } } cronTimes . push ( cronTime . join ( ' ' ) ) ; } } const executeTrigger = async ( ) => { Logger . debug ( `Polling trigger initiated for workflow "${ workflow . name }"` , { workflowName : workflow . name , workflowId : workflow . id , } ) ; const pollResponse = await workflow . runPoll ( node , pollFunctions ) ; if ( pollResponse !== null ) { pollFunctions . __emit ( pollResponse ) ; } } ; await executeTrigger ( ) ; const timezone = pollFunctions . getTimezone ( ) ; const cronJobs = [ ] ; for ( const cronTime of cronTimes ) { const cronTimeParts = cronTime . split ( ' ' ) ; if ( cronTimeParts . length > 0 && cronTimeParts [ 0 ] . includes ( '*' ) ) { throw new Error ( 'The polling interval is too short. It has to be at least a minute!' ) ; } cronJobs . push ( new CronJob ( cronTime , executeTrigger , undefined , true , timezone ) ) ; } async function closeFunction ( ) { for ( const cronJob of cronJobs ) { cronJob . stop ( ) ; } } return { closeFunction , } ; } param { string } id returns { Promise < void > } memberof async remove ( id ) { if ( ! this . isActive ( id ) ) { throw new Error ( `The workflow with the id "${ id }" is currently not active and can so not be removed` ) ; } const workflowData = this . workflowData [ id ] ; if ( workflowData . triggerResponses ) { for ( const triggerResponse of workflowData . triggerResponses ) { if ( triggerResponse . closeFunction ) { await triggerResponse . closeFunction ( ) ; } } } if ( workflowData . pollResponses ) { for ( const pollResponse of workflowData . pollResponses ) { if ( pollResponse . closeFunction ) { await pollResponse . closeFunction ( ) ; } } } delete this . workflowData [ id ] ; } }