Add workflow runner (#6458)
- add workflow runner module
- add an endpoint to trigger a workflow via api
- improve error handling for serverless functions
## Testing
- create 2 serverless functions
- create a workflow
- create this workflow Version
```
{
"type": "MANUAL",
"input": {"b": "bb"},
"nextAction": {
"name": "step_1",
"displayName": "Code",
"type": "CODE",
"valid": true,
"settings": {
"serverlessFunctionId": "Serverless function 1 Id",
"errorHandlingOptions": {
"retryOnFailure": {
"value": false
},
"continueOnFailure": {
"value": false
}
}
},
"nextAction": {
"name": "step_1",
"displayName": "Code",
"type": "CODE",
"valid": true,
"settings": {
"serverlessFunctionId": "Serverless function 1 Id",
"errorHandlingOptions": {
"retryOnFailure": {
"value": false
},
"continueOnFailure": {
"value": false
}
}
},
"nextAction": {
"name": "step_1",
"displayName": "Code",
"type": "CODE",
"valid": true,
"settings": {
"serverlessFunctionId": "Serverless function 2 Id",
"errorHandlingOptions": {
"retryOnFailure": {
"value": false
},
"continueOnFailure": {
"value": false
}
}
}
}
}
}
}
`
``
- call
```
mutation Trigger {
triggerWorkflow(workflowVersionId: "WORKFLOW_VERSION_ID") {
result
}
}
```
- try when errors are injected in serverless function
This commit is contained in:
@ -0,0 +1,32 @@
|
||||
import { Processor } from 'src/engine/integrations/message-queue/decorators/processor.decorator';
|
||||
import { MessageQueue } from 'src/engine/integrations/message-queue/message-queue.constants';
|
||||
import { Process } from 'src/engine/integrations/message-queue/decorators/process.decorator';
|
||||
import { WorkflowRunnerService } from 'src/modules/workflow/workflow-runner/workflow-runner.service';
|
||||
import { WorkflowCommonService } from 'src/modules/workflow/common/workflow-common.services';
|
||||
|
||||
type RunWorkflowJobData = { workspaceId: string; workflowVersionId: string };
|
||||
|
||||
@Processor(MessageQueue.workflowQueue)
|
||||
export class WorkflowRunnerJob {
|
||||
constructor(
|
||||
private readonly workflowCommonService: WorkflowCommonService,
|
||||
private readonly workflowRunnerService: WorkflowRunnerService,
|
||||
) {}
|
||||
|
||||
@Process(WorkflowRunnerJob.name)
|
||||
async handle({
|
||||
workspaceId,
|
||||
workflowVersionId,
|
||||
}: RunWorkflowJobData): Promise<void> {
|
||||
const workflowVersion = await this.workflowCommonService.getWorkflowVersion(
|
||||
workspaceId,
|
||||
workflowVersionId,
|
||||
);
|
||||
|
||||
await this.workflowRunnerService.run({
|
||||
action: workflowVersion.trigger.nextAction,
|
||||
workspaceId,
|
||||
payload: workflowVersion.trigger.input,
|
||||
});
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,13 @@
|
||||
import { Module } from '@nestjs/common';
|
||||
|
||||
import { WorkflowRunnerService } from 'src/modules/workflow/workflow-runner/workflow-runner.service';
|
||||
import { WorkflowRunnerJob } from 'src/modules/workflow/workflow-runner/workflow-runner.job';
|
||||
import { WorkflowCommonModule } from 'src/modules/workflow/common/workflow-common.module';
|
||||
import { ServerlessFunctionModule } from 'src/engine/metadata-modules/serverless-function/serverless-function.module';
|
||||
|
||||
@Module({
|
||||
imports: [WorkflowCommonModule, ServerlessFunctionModule],
|
||||
providers: [WorkflowRunnerService, WorkflowRunnerJob],
|
||||
exports: [WorkflowRunnerService],
|
||||
})
|
||||
export class WorkflowRunnerModule {}
|
||||
@ -0,0 +1,82 @@
|
||||
import { Injectable } from '@nestjs/common';
|
||||
|
||||
import { ServerlessFunctionService } from 'src/engine/metadata-modules/serverless-function/serverless-function.service';
|
||||
import {
|
||||
WorkflowTriggerException,
|
||||
WorkflowTriggerExceptionCode,
|
||||
} from 'src/modules/workflow/workflow-trigger/workflow-trigger.exception';
|
||||
import {
|
||||
WorkflowAction,
|
||||
WorkflowActionType,
|
||||
} from 'src/modules/workflow/common/types/workflow-action.type';
|
||||
|
||||
const MAX_RETRIES_ON_FAILURE = 3;
|
||||
|
||||
@Injectable()
|
||||
export class WorkflowRunnerService {
|
||||
constructor(
|
||||
private readonly serverlessFunctionService: ServerlessFunctionService,
|
||||
) {}
|
||||
|
||||
async run({
|
||||
action,
|
||||
workspaceId,
|
||||
payload,
|
||||
attemptCount = 1,
|
||||
}: {
|
||||
action?: WorkflowAction;
|
||||
workspaceId: string;
|
||||
payload?: object;
|
||||
attemptCount?: number;
|
||||
}) {
|
||||
if (!action) {
|
||||
return payload;
|
||||
}
|
||||
|
||||
let result: object | undefined = undefined;
|
||||
|
||||
switch (action.type) {
|
||||
case WorkflowActionType.CODE: {
|
||||
const executionResult = await this.serverlessFunctionService.executeOne(
|
||||
action.settings.serverlessFunctionId,
|
||||
workspaceId,
|
||||
payload,
|
||||
);
|
||||
|
||||
if (executionResult.data) {
|
||||
result = executionResult.data;
|
||||
}
|
||||
if (!executionResult.error) {
|
||||
throw new Error('Execution result error, no data or error');
|
||||
}
|
||||
if (action.settings.errorHandlingOptions.continueOnFailure.value) {
|
||||
result = payload;
|
||||
break;
|
||||
} else if (
|
||||
action.settings.errorHandlingOptions.retryOnFailure.value &&
|
||||
attemptCount < MAX_RETRIES_ON_FAILURE
|
||||
) {
|
||||
return await this.run({
|
||||
action,
|
||||
workspaceId,
|
||||
payload,
|
||||
attemptCount: attemptCount + 1,
|
||||
});
|
||||
} else {
|
||||
return executionResult.error;
|
||||
}
|
||||
}
|
||||
default:
|
||||
throw new WorkflowTriggerException(
|
||||
`Unknown action type '${action.type}'`,
|
||||
WorkflowTriggerExceptionCode.INVALID_ACTION_TYPE,
|
||||
);
|
||||
}
|
||||
|
||||
return await this.run({
|
||||
action: action.nextAction,
|
||||
workspaceId,
|
||||
payload: result,
|
||||
});
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user