import { createDeferredPromise , IDeferredPromise , IExecuteResponsePromiseData , IRun , } from 'n8n-workflow' ; import { ChildProcess } from 'child_process' ; import { stringify } from 'flatted' ; import * as PCancelable from 'p-cancelable' ; import { Db , IExecutingWorkflowData , IExecutionDb , IExecutionFlattedDb , IExecutionsCurrentSummary , IWorkflowExecutionDataProcess , ResponseHelper , WorkflowHelpers , } from '.' ; export class ActiveExecutions { private activeExecutions : { [ index ] : IExecutingWorkflowData ; } = { } ; param { ChildProcess } process param { IWorkflowExecutionDataProcess } executionData returns { string } memberof async add ( executionData , process ? , executionId ? ) { if ( executionId === undefined ) { const fullExecutionData = { data : executionData . executionData ! , mode : executionData . executionMode , finished : false , startedAt : new Date ( ) , workflowData : executionData . workflowData , } ; if ( executionData . retryOf !== undefined ) { fullExecutionData . retryOf = executionData . retryOf . toString ( ) ; } if ( executionData . workflowData . id !== undefined && WorkflowHelpers . isWorkflowIdValid ( executionData . workflowData . id . toString ( ) ) ) { fullExecutionData . workflowId = executionData . workflowData . id . toString ( ) ; } const execution = ResponseHelper . flattenExecutionData ( fullExecutionData ) ; const executionResult = await Db . collections . Execution ! . save ( execution as IExecutionFlattedDb ) ; executionId = typeof executionResult . id === 'object' ? executionResult . id ! . toString ( ) : executionResult . id + '' ; } else { const execution = { id : executionId , data : stringify ( executionData . executionData ! ) , waitTill : null , } ; await Db . collections . Execution ! . update ( executionId , execution ) ; } this . activeExecutions [ executionId ] = { executionData , process , startedAt : new Date ( ) , postExecutePromises : [ ] , } ; return executionId ; } param { string } executionId param { PCancelable < IRun > } workflowExecution memberof attachWorkflowExecution ( executionId , workflowExecution ) { if ( this . activeExecutions [ executionId ] === undefined ) { throw new Error ( `No active execution with id "${ executionId }" got found to attach to workflowExecution to!` ) ; } this . activeExecutions [ executionId ] . workflowExecution = workflowExecution ; } attachResponsePromise ( executionId , responsePromise ) { if ( this . activeExecutions [ executionId ] === undefined ) { throw new Error ( `No active execution with id "${ executionId }" got found to attach to workflowExecution to!` ) ; } this . activeExecutions [ executionId ] . responsePromise = responsePromise ; } resolveResponsePromise ( executionId , response ) { if ( this . activeExecutions [ executionId ] === undefined ) { return ; } this . activeExecutions [ executionId ] . responsePromise ?. resolve ( response ) ; } param { string } executionId param { IRun } fullRunData returns { void } memberof remove ( executionId , fullRunData ? ) { if ( this . activeExecutions [ executionId ] === undefined ) { return ; } for ( const promise of this . activeExecutions [ executionId ] . postExecutePromises ) { promise . resolve ( fullRunData ) ; } delete this . activeExecutions [ executionId ] ; } param { string } executionId param { string } timeout returns { ( Promise < IRun | undefined > ) } memberof async stopExecution ( executionId , timeout ? ) { if ( this . activeExecutions [ executionId ] === undefined ) { return ; } if ( this . activeExecutions [ executionId ] . process !== undefined ) { if ( this . activeExecutions [ executionId ] . process ! . connected ) { setTimeout ( ( ) => { this . activeExecutions [ executionId ] . process ! . send ( { type : timeout || 'stopExecution' , } ) ; } , 1 ) ; } } else { this . activeExecutions [ executionId ] . workflowExecution ! . cancel ( ) ; } return this . getPostExecutePromise ( executionId ) ; } param { string } executionId returns { Promise < IRun > } memberof async getPostExecutePromise ( executionId ) { const waitPromise = await createDeferredPromise < IRun | undefined > ( ) ; if ( this . activeExecutions [ executionId ] === undefined ) { throw new Error ( `There is no active execution with id "${ executionId }".` ) ; } this . activeExecutions [ executionId ] . postExecutePromises . push ( waitPromise ) ; return waitPromise . promise ( ) ; } returns { IExecutionsCurrentSummary [ ] } memberof getActiveExecutions ( ) { const returnData = [ ] ; let data ; for ( const id of Object . keys ( this . activeExecutions ) ) { data = this . activeExecutions [ id ] ; returnData . push ( { id , retryOf : data . executionData . retryOf as string | undefined , startedAt : data . startedAt , mode : data . executionData . executionMode , workflowId : data . executionData . workflowData . id ! as string , } ) ; } return returnData ; } } let activeExecutionsInstance ; export function getInstance ( ) { if ( activeExecutionsInstance === undefined ) { activeExecutionsInstance = new ActiveExecutions ( ) ; } return activeExecutionsInstance ; }