refactor(plugin-workflow): add sync option for trigger (#3383)

* refactor(plugin-workflow): add sync option for trigger

* feat(plugin-workflow-request): support sync call in request

* fix(plugin-workflow-request): fix request async call

* refactor(plugin-workflow): add method to check if workflow is sync

* fix(plugin-workflow): fix useAvailable in nodes

* fix(plugin-workflow): fix node.isAvailable check

* test(plugin-workflow): limit mysql version to pass test

* fix(plugin-workflow-delay): fix test case

* fix(plugin-workflow-delay): fix test case

* feat(plugin-workflow): add sync field for workflow

* refactor(plugin-workflow): adjust end node logic

* fix(plugin-workflow): support sync mode in form trigger

* feat(plugin-workflow): add end instruction

* fix(plugin-workflow-form-trigger): fix test cases

* fix(plugin-workflow): fix transaction for sync event
This commit is contained in:
Junyi 2024-01-25 22:10:03 +08:00 committed by GitHub
parent be7a4845ab
commit 4b8915b616
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
41 changed files with 593 additions and 116 deletions

View File

@ -505,7 +505,7 @@ function WorkflowConfig() {
const workflowTypes = workflowPlugin.getTriggersOptions().filter((item) => {
return typeof item.options.useActionTriggerable === 'function'
? item.options.useActionTriggerable()
: item.options.useActionTriggerable;
: Boolean(item.options.useActionTriggerable);
});
const description = {
submit: t('Workflow will be triggered after submitting succeeded.', { ns: 'workflow' }),

View File

@ -1,7 +1,7 @@
import React from 'react';
import { InputNumber, Select } from 'antd';
import { css, useCompile } from '@nocobase/client';
import { Instruction, JOB_STATUS } from '@nocobase/plugin-workflow/client';
import { css, useCompile, usePlugin } from '@nocobase/client';
import WorkflowPlugin, { Instruction, JOB_STATUS } from '@nocobase/plugin-workflow/client';
import { NAMESPACE } from '../locale';
@ -70,7 +70,7 @@ export default class extends Instruction {
},
endStatus: {
type: 'number',
title: `{{t("End Status", { ns: "${NAMESPACE}" })}}`,
title: `{{t("End status", { ns: "${NAMESPACE}" })}}`,
'x-decorator': 'FormItem',
'x-component': 'Radio.Group',
enum: [
@ -84,4 +84,7 @@ export default class extends Instruction {
components = {
Duration,
};
isAvailable({ engine, workflow, upstream, branchIndex }) {
return !engine.isWorkflowSync(workflow);
}
}

View File

@ -2,7 +2,7 @@
"Delay": "Delay",
"Delay a period of time and then continue or exit the process. Can be used to set wait or timeout times in parallel branches.": "Delay a period of time and then continue or exit the process. Can be used to set wait or timeout times in parallel branches.",
"Duration": "Duration",
"End Status": "End Status",
"End status": "End status",
"Select status": "Select status",
"Succeed and continue": "Succeed and continue",
"Fail and exit": "Fail and exit"

View File

@ -1,7 +1,7 @@
{
"Delay": "Retraso",
"Duration": "Duración",
"End Status": "Estado Final",
"End status": "Estado final",
"Select status": "Seleccionar estado",
"Succeed and continue": "Éxito y continuar",
"Fail and exit": "Falla y sale"

View File

@ -1,7 +1,7 @@
{
"Delay": "Délai",
"Duration": "Durée",
"End Status": "Statut de fin",
"End status": "Statut de fin",
"Select status": "Sélectionner un statut",
"Succeed and continue": "Réussir et continuer",
"Fail and exit": "Échouer et quitter"

View File

@ -1,7 +1,7 @@
{
"Delay": "Atraso",
"Duration": "Duração",
"End Status": "Status final",
"End status": "Status final",
"Select status": "Selecionar status",
"Succeed and continue": "Ter sucesso e continuar",
"Fail and exit": "Falhar e sair"

View File

@ -2,7 +2,7 @@
"Delay": "延时",
"Delay a period of time and then continue or exit the process. Can be used to set wait or timeout times in parallel branches.": "延时一段时间,然后继续或退出流程。可以用于并行分支中等待其他分支或设置超时时间。",
"Duration": "时长",
"End Status": "到时状态",
"End status": "到时状态",
"Select status": "选择状态",
"Succeed and continue": "通过并继续",
"Fail and exit": "失败并退出"

View File

@ -45,7 +45,7 @@ export default class extends Trigger {
return;
}
this.trigger(context);
return this.trigger(context);
};
private async trigger(context) {
@ -67,11 +67,17 @@ export default class extends Trigger {
enabled: true,
},
});
workflows.forEach((workflow) => {
const syncGroup = [];
const asyncGroup = [];
for (const workflow of workflows) {
const trigger = triggers.find((trigger) => trigger[0] == workflow.key);
if (context.body?.data) {
const { data } = context.body;
(Array.isArray(data) ? data : [data]).forEach(async (row: Model) => {
const event = [workflow];
if (context.action.resourceName !== 'workflows') {
if (!context.body) {
continue;
}
const { body: data } = context;
for (const row of Array.isArray(data) ? data : [data]) {
let payload = row;
if (trigger[1]) {
const paths = trigger[1].split('.');
@ -87,7 +93,7 @@ export default class extends Trigger {
const { collection, appends = [] } = workflow.config;
const model = <typeof Model>payload.constructor;
if (collection !== model.collection.name) {
return;
continue;
}
if (appends.length) {
payload = await model.collection.repository.findOne({
@ -95,16 +101,34 @@ export default class extends Trigger {
appends,
});
}
this.workflow.trigger(workflow, { data: toJSON(payload), ...userInfo });
});
// this.workflow.trigger(workflow, { data: toJSON(payload), ...userInfo });
event.push({ data: toJSON(payload), ...userInfo });
}
} else {
const data = trigger[1] ? get(values, trigger[1]) : values;
this.workflow.trigger(workflow, {
data,
...userInfo,
});
// this.workflow.trigger(workflow, {
// data,
// ...userInfo,
// });
event.push({ data, ...userInfo });
}
});
(workflow.sync ? syncGroup : asyncGroup).push(event);
}
for (const event of syncGroup) {
await this.workflow.trigger(event[0], event[1]);
// if (processor.execution.status < EXECUTION_STATUS.STARTED) {
// // error handling
// return context.throw(
// 500,
// 'Your data saved, but some workflow on your action failed, please contact the administrator.',
// );
// }
}
for (const event of asyncGroup) {
this.workflow.trigger(event[0], event[1]);
}
}
on(workflow: WorkflowModel) {}

View File

@ -5,7 +5,7 @@ import { MockServer } from '@nocobase/test';
import Plugin from '..';
describe('workflow > instructions > sql', () => {
describe('workflow > form-trigge[r', () => {
let app: MockServer;
let db: Database;
let agent;
@ -400,4 +400,65 @@ describe('workflow > instructions > sql', () => {
expect(e3[0].context.data).toMatchObject({ title: 't2' });
});
});
describe('sync', () => {
it('sync form trigger', async () => {
const workflow = await WorkflowModel.create({
enabled: true,
type: 'form',
sync: true,
config: {
collection: 'posts',
},
});
const res1 = await userAgents[0].resource('posts').create({
values: { title: 't1' },
triggerWorkflows: `${workflow.key}`,
});
expect(res1.status).toBe(200);
const executions = await workflow.getExecutions();
expect(executions.length).toBe(1);
expect(executions[0].status).toBe(EXECUTION_STATUS.RESOLVED);
});
it('sync and async will all be triggered in one action', async () => {
const w1 = await WorkflowModel.create({
enabled: true,
type: 'form',
config: {
collection: 'posts',
},
});
const w2 = await WorkflowModel.create({
enabled: true,
type: 'form',
sync: true,
config: {
collection: 'posts',
},
});
const res1 = await userAgents[0].resource('posts').create({
values: { title: 't1' },
triggerWorkflows: [w1.key, w2.key].join(),
});
expect(res1.status).toBe(200);
const e1s = await w1.getExecutions();
expect(e1s.length).toBe(0);
const e2s = await w2.getExecutions();
expect(e2s.length).toBe(1);
expect(e2s[0].status).toBe(EXECUTION_STATUS.RESOLVED);
await sleep(500);
const e3s = await w1.getExecutions();
expect(e3s.length).toBe(1);
expect(e3s[0].status).toBe(EXECUTION_STATUS.RESOLVED);
});
});
});

View File

@ -20,8 +20,7 @@ export default class extends Plugin {
// this.app.addProvider(Provider);
const workflow = this.app.pm.get('workflow') as WorkflowPlugin;
const manualInstruction = new Manual();
workflow.instructions.register(manualInstruction.type, manualInstruction);
workflow.registerInstruction('manual', Manual);
this.app.schemaInitializerManager.add(addBlockButton);
this.app.schemaInitializerManager.add(addActionButton);

View File

@ -1,6 +1,6 @@
import { SchemaInitializerItemType, useCollectionManager, useCompile } from '@nocobase/client';
import { SchemaInitializerItemType, useCollectionManager, useCompile, usePlugin } from '@nocobase/client';
import {
import WorkflowPlugin, {
defaultFieldNames,
getCollectionFieldOptions,
CollectionBlockInitializer,
@ -154,4 +154,7 @@ export default class extends Instruction {
}
: null;
}
isAvailable({ engine, workflow, upstream, branchIndex }) {
return !engine.isWorkflowSync(workflow);
}
}

View File

@ -41,6 +41,26 @@ async function request(config) {
export default class extends Instruction {
async run(node: FlowNodeModel, prevJob, processor: Processor) {
const config = processor.getParsedValue(node.config, node.id) as RequestConfig;
const { workflow } = processor.execution;
const sync = this.workflow.isWorkflowSync(workflow);
if (sync) {
try {
const response = await request(config);
return {
status: JOB_STATUS.RESOLVED,
result: response.data,
};
} catch (error) {
return {
status: JOB_STATUS.FAILED,
result: error.isAxiosError ? error.toJSON() : error.message,
};
}
}
const job = await processor.saveJob({
status: JOB_STATUS.PENDING,
nodeId: node.id,
@ -48,8 +68,6 @@ export default class extends Instruction {
upstreamId: prevJob?.id ?? null,
});
const config = processor.getParsedValue(node.config, node.id) as RequestConfig;
// eslint-disable-next-line promise/catch-or-return
request(config)
.then((response) => {

View File

@ -6,7 +6,7 @@ import bodyParser from 'koa-bodyparser';
import Database from '@nocobase/database';
import { MockServer } from '@nocobase/test';
import { EXECUTION_STATUS, JOB_STATUS } from '@nocobase/plugin-workflow';
import PluginWorkflow, { Processor, EXECUTION_STATUS, JOB_STATUS } from '@nocobase/plugin-workflow';
import { getApp, sleep } from '@nocobase/plugin-workflow-test';
import { RequestConfig } from '../RequestInstruction';
@ -349,4 +349,31 @@ describe('workflow > instructions > request', () => {
server.close();
});
});
describe('sync request', () => {
it('sync trigger', async () => {
const syncFlow = await WorkflowModel.create({
type: 'syncTrigger',
enabled: true,
});
await syncFlow.createNode({
type: 'request',
config: {
url: api.URL_DATA,
method: 'GET',
} as RequestConfig,
});
const workflowPlugin = app.pm.get(PluginWorkflow) as PluginWorkflow;
const processor = (await workflowPlugin.trigger(syncFlow, { data: { title: 't1' } })) as Processor;
const [execution] = await syncFlow.getExecutions();
expect(processor.execution.id).toEqual(execution.id);
expect(processor.execution.status).toEqual(execution.status);
expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED);
const [job] = await execution.getJobs();
expect(job.status).toEqual(JOB_STATUS.RESOLVED);
expect(job.result).toEqual({ meta: {}, data: {} });
});
});
});

View File

@ -4,6 +4,7 @@ import { ApplicationOptions, Plugin } from '@nocobase/server';
import { MockServer, createMockServer } from '@nocobase/test';
import functions from './functions';
import triggers from './triggers';
import instructions from './instructions';
export interface MockServerOptions extends ApplicationOptions {
@ -38,6 +39,7 @@ export async function getApp(options: MockServerOptions = {}): Promise<MockServe
[
'workflow',
{
triggers,
instructions,
functions,
},

View File

@ -13,7 +13,6 @@ export default {
error: {
run(node, input, processor) {
throw new Error('definite error');
return null;
},
},

View File

@ -0,0 +1,13 @@
export default {
syncTrigger: class {
constructor(public readonly workflow) {}
on() {}
off() {}
sync = true;
},
asyncTrigger: class {
constructor(public readonly workflow) {}
on() {}
off() {}
},
};

View File

@ -18,11 +18,11 @@ interface AddButtonProps {
export function AddButton(props: AddButtonProps) {
const { upstream, branchIndex = null } = props;
const { instructions } = usePlugin(WorkflowPlugin);
const engine = usePlugin(WorkflowPlugin);
const compile = useCompile();
const api = useAPIClient();
const { workflow, refresh } = useFlowContext() ?? {};
const instructionList = Array.from(instructions.getValues()) as Instruction[];
const instructionList = Array.from(engine.instructions.getValues()) as Instruction[];
const { styles } = useStyles();
const groups = useMemo(() => {
@ -32,12 +32,11 @@ export function AddButton(props: AddButtonProps) {
{ key: 'manual', label: `{{t("Manual", { ns: "${NAMESPACE}" })}}` },
{ key: 'extended', label: `{{t("Extended types", { ns: "${NAMESPACE}" })}}` },
]
.filter((group) => instructionList.filter((item) => item.group === group.key).length)
.map((group) => {
const groupInstructions = instructionList.filter(
(item) =>
item.group === group.key &&
(item.isAvailable ? item.isAvailable({ workflow, upstream, branchIndex }) : true),
(item.isAvailable ? item.isAvailable({ engine, workflow, upstream, branchIndex }) : true),
);
return {
@ -59,27 +58,23 @@ export function AddButton(props: AddButtonProps) {
: null,
})),
};
});
}, [branchIndex, instructionList, upstream, workflow]);
const resource = useMemo(() => {
if (!workflow) {
return null;
}
return api.resource('workflows.nodes', workflow.id);
}, [workflow?.id]);
})
.filter((group) => group.children.length);
}, [branchIndex, engine, instructionList, upstream, workflow]);
const onCreate = useCallback(
async ({ keyPath }) => {
const type = keyPath.pop();
const config = {};
const [optionKey] = keyPath;
const instruction = instructions.get(type);
const instruction = engine.instructions.get(type);
if (optionKey) {
const { value } = instruction.options?.find((item) => item.key === optionKey) ?? {};
Object.assign(config, typeof value === 'function' ? value() : value);
}
if (resource) {
await resource.create({
if (workflow) {
await api.resource('workflows.nodes', workflow.id).create({
values: {
type,
upstreamId: upstream?.id ?? null,
@ -91,7 +86,7 @@ export function AddButton(props: AddButtonProps) {
refresh();
}
},
[branchIndex, resource?.create, upstream?.id],
[api, branchIndex, engine.instructions, refresh, upstream?.id, workflow],
);
const menu = useMemo<MenuProps>(() => {

View File

@ -1,13 +1,51 @@
import { SchemaComponent, SchemaComponentContext, usePlugin } from '@nocobase/client';
import { SchemaComponent, SchemaComponentContext, usePlugin, useRecord } from '@nocobase/client';
import { Card } from 'antd';
import React, { useContext } from 'react';
import React, { useContext, useEffect } from 'react';
import { ExecutionLink } from './ExecutionLink';
import { ExecutionResourceProvider } from './ExecutionResourceProvider';
import { WorkflowLink } from './WorkflowLink';
import OpenDrawer from './components/OpenDrawer';
import { workflowSchema } from './schemas/workflows';
import { ExecutionStatusSelect, ExecutionStatusColumn } from './components/ExecutionStatus';
import WorkflowPlugin from '.';
import WorkflowPlugin, { RadioWithTooltip } from '.';
import { onFieldChange } from '@formily/core';
import { useField, useFormEffects } from '@formily/react';
function SyncOptionSelect(props) {
const field = useField<any>();
const record = useRecord();
const workflowPlugin = usePlugin(WorkflowPlugin);
useFormEffects((form) => {
onFieldChange('type', (f: any) => {
let disabled = record.id || !f.value;
if (f.value) {
const trigger = workflowPlugin.triggers.get(f.value);
if (trigger.sync != null) {
disabled = true;
field.setValue(trigger.sync);
} else {
field.setInitialValue(false);
}
}
field.setPattern(disabled ? 'disabled' : 'editable');
});
});
useEffect(() => {
if (record.id) {
field.setPattern('disabled');
const trigger = workflowPlugin.triggers.get(record.type);
if (trigger.sync != null) {
field.setValue(trigger.sync);
} else {
field.setInitialValue(false);
}
}
}, [record.id, field, workflowPlugin.triggers]);
return <RadioWithTooltip {...props} />;
}
export function WorkflowPane() {
const ctx = useContext(SchemaComponentContext);
@ -23,6 +61,7 @@ export function WorkflowPane() {
ExecutionLink,
OpenDrawer,
ExecutionStatusSelect,
SyncOptionSelect,
ExecutionStatusColumn,
}}
scope={{

View File

@ -25,6 +25,7 @@ import ScheduleTrigger from './triggers/schedule';
import { Instruction } from './nodes';
import CalculationInstruction from './nodes/calculation';
import ConditionInstruction from './nodes/condition';
import EndInstruction from './nodes/end';
import QueryInstruction from './nodes/query';
import CreateInstruction from './nodes/create';
import UpdateInstruction from './nodes/update';
@ -45,6 +46,10 @@ export default class PluginWorkflowClient extends Plugin {
}));
};
isWorkflowSync(workflow) {
return this.triggers.get(workflow.type).sync ?? workflow.sync;
}
registerTrigger(type: string, trigger: Trigger | { new (): Trigger }) {
if (typeof trigger === 'function') {
this.triggers.register(type, new trigger());
@ -82,6 +87,8 @@ export default class PluginWorkflowClient extends Plugin {
this.registerInstruction('calculation', CalculationInstruction);
this.registerInstruction('condition', ConditionInstruction);
this.registerInstruction('end', EndInstruction);
this.registerInstruction('query', QueryInstruction);
this.registerInstruction('create', CreateInstruction);
this.registerInstruction('update', UpdateInstruction);

View File

@ -164,6 +164,7 @@ function Calculation({ calculator, operands = [], onChange }) {
onChange={(v) => onChange({ operands, calculator: v })}
placeholder={lang('Operator')}
popupMatchSelectWidth={false}
className="auto-width"
>
{calculatorGroups
.filter((group) => Boolean(getGroupCalculators(group.value).length))

View File

@ -0,0 +1,25 @@
import { Instruction } from '.';
import { NAMESPACE } from '../locale';
import { JOB_STATUS } from '../constants';
export default class extends Instruction {
title = '{{t("End process")}}';
type = 'end';
group = 'control';
description = `{{t("End the process immediately, with set status.", { ns: "${NAMESPACE}" })}}`;
fieldset = {
endStatus: {
type: 'number',
title: `{{t("End status", { ns: "${NAMESPACE}" })}}`,
'x-decorator': 'FormItem',
'x-component': 'Radio.Group',
enum: [
{ label: `{{t("Succeeded", { ns: "${NAMESPACE}" })}}`, value: JOB_STATUS.RESOLVED },
{ label: `{{t("Failed", { ns: "${NAMESPACE}" })}}`, value: JOB_STATUS.FAILED },
],
required: true,
default: JOB_STATUS.RESOLVED,
},
};
end = true;
}

View File

@ -1,4 +1,4 @@
import { DeleteOutlined } from '@ant-design/icons';
import { CloseOutlined, DeleteOutlined } from '@ant-design/icons';
import { createForm } from '@formily/core';
import { ISchema, useForm } from '@formily/react';
import { App, Button, Dropdown, Input, Tag, Tooltip, message } from 'antd';
@ -33,6 +33,7 @@ import useStyles from '../style';
import { UseVariableOptions, VariableOption } from '../variable';
export type NodeAvailableContext = {
engine: WorkflowPlugin;
workflow: object;
upstream: object;
branchIndex: number;
@ -53,6 +54,7 @@ export abstract class Instruction {
useScopeVariables?(node, options?): VariableOption[];
useInitializers?(node): SchemaInitializerItemType | null;
isAvailable?(ctx: NodeAvailableContext): boolean;
end?: boolean | ((node) => boolean);
}
function useUpdateAction() {
@ -121,13 +123,18 @@ export function Node({ data }) {
const { styles } = useStyles();
const { getAriaLabel } = useGetAriaLabelOfAddButton(data);
const workflowPlugin = usePlugin(WorkflowPlugin);
const { Component = NodeDefaultView } = workflowPlugin.instructions.get(data.type);
const { Component = NodeDefaultView, end } = workflowPlugin.instructions.get(data.type);
return (
<NodeContext.Provider value={data}>
<div className={cx(styles.nodeBlockClass)}>
<Component data={data} />
<AddButton aria-label={getAriaLabel()} upstream={data} />
{!end || (typeof end === 'function' && !end(data)) ? (
<AddButton aria-label={getAriaLabel()} upstream={data} />
) : (
<div className="end-sign">
<CloseOutlined />
</div>
)}
</div>
</NodeContext.Provider>
);

View File

@ -89,6 +89,27 @@ const workflowFieldset = {
'x-component': 'CollectionField',
'x-decorator': 'FormItem',
},
sync: {
type: 'boolean',
title: `{{ t("Execute mode", { ns: "${NAMESPACE}" }) }}`,
description: `{{ t("Could not be changed after created.", { ns: "${NAMESPACE}" }) }}`,
'x-decorator': 'FormItem',
'x-component': 'SyncOptionSelect',
'x-component-props': {
options: [
{
label: `{{ t("Asynchronously", { ns: "${NAMESPACE}" }) }}`,
value: false,
tooltip: `{{ t("Will be executed in the background as a queued task.", { ns: "${NAMESPACE}" }) }}`,
},
{
label: `{{ t("Synchronously", { ns: "${NAMESPACE}" }) }}`,
value: true,
tooltip: `{{ t("For user actions that require immediate feedback. Can not use asynchronous nodes in such mode.", { ns: "${NAMESPACE}" }) }}`,
},
],
},
},
enabled: {
'x-component': 'CollectionField',
'x-decorator': 'FormItem',
@ -196,6 +217,7 @@ export const workflowSchema: ISchema = {
properties: {
title: workflowFieldset.title,
type: workflowFieldset.type,
sync: workflowFieldset.sync,
description: workflowFieldset.description,
options: workflowFieldset.options,
footer: {
@ -333,6 +355,7 @@ export const workflowSchema: ISchema = {
properties: {
title: workflowFieldset.title,
enabled: workflowFieldset.enabled,
sync: workflowFieldset.sync,
description: workflowFieldset.description,
options: workflowFieldset.options,
footer: {

View File

@ -54,6 +54,7 @@ function useUpdateConfigAction() {
}
export abstract class Trigger {
sync: boolean;
title: string;
description?: string;
// group: string;

View File

@ -8,6 +8,7 @@ import { ScheduleConfig } from './ScheduleConfig';
import { SCHEDULE_MODE } from './constants';
export default class extends Trigger {
sync = false;
title = `{{t("Schedule event", { ns: "${NAMESPACE}" })}}`;
description = `{{t("Event will be scheduled and triggered base on time conditions.", { ns: "${NAMESPACE}" })}}`;
fieldset = {

View File

@ -177,5 +177,10 @@
"Executed workflow cannot be modified": "已经执行过的工作流不能被修改",
"Can not delete": "无法删除",
"The result of this node has been referenced by other nodes ({{nodes}}), please remove the usage before deleting.":
"该节点的执行结果已被其他节点({{nodes}})引用,删除前请先移除引用。"
"该节点的执行结果已被其他节点({{nodes}})引用,删除前请先移除引用。",
"End process": "结束流程",
"End the process immediately, with set status.": "以设置的状态立即结束流程。",
"End status": "结束状态",
"Succeeded": "成功"
}

View File

@ -2,7 +2,7 @@ import path from 'path';
import LRUCache from 'lru-cache';
import { Op } from '@nocobase/database';
import { Op, Transactionable } from '@nocobase/database';
import { Plugin } from '@nocobase/server';
import { Registry } from '@nocobase/utils';
@ -17,6 +17,7 @@ import ScheduleTrigger from './triggers/ScheduleTrigger';
import { Instruction, InstructionInterface } from './instructions';
import CalculationInstruction from './instructions/CalculationInstruction';
import ConditionInstruction from './instructions/ConditionInstruction';
import EndInstruction from './instructions/EndInstruction';
import CreateInstruction from './instructions/CreateInstruction';
import DestroyInstruction from './instructions/DestroyInstruction';
import QueryInstruction from './instructions/QueryInstruction';
@ -63,6 +64,14 @@ export default class PluginWorkflowServer extends Plugin {
return logger;
}
isWorkflowSync(workflow: WorkflowModel) {
const trigger = this.triggers.get(workflow.type);
if (!trigger) {
throw new Error(`invalid trigger type ${workflow.type} of workflow ${workflow.id}`);
}
return trigger.sync ?? workflow.sync;
}
onBeforeSave = async (instance: WorkflowModel, options) => {
const Model = <typeof WorkflowModel>instance.constructor;
@ -141,6 +150,7 @@ export default class PluginWorkflowServer extends Plugin {
private initInstructions<T extends Instruction>(more: { [key: string]: T | { new (p: Plugin): T } } = {}) {
this.registerInstruction('calculation', CalculationInstruction);
this.registerInstruction('condition', ConditionInstruction);
this.registerInstruction('end', EndInstruction);
this.registerInstruction('create', CreateInstruction);
this.registerInstruction('destroy', DestroyInstruction);
this.registerInstruction('query', QueryInstruction);
@ -267,11 +277,15 @@ export default class PluginWorkflowServer extends Plugin {
}
}
public trigger(workflow: WorkflowModel, context: object, options: { context?: any } = {}): void {
public trigger(
workflow: WorkflowModel,
context: object,
options: { context?: any } & Transactionable = {},
): void | Promise<Processor | null> {
const logger = this.getLogger(workflow.id);
if (!this.ready) {
logger.warn(`app is not ready, event of workflow ${workflow.id} will be ignored`);
logger.debug(`ignored event data:`, { data: context });
logger.debug(`ignored event data:`, context);
return;
}
// `null` means not to trigger
@ -280,7 +294,11 @@ export default class PluginWorkflowServer extends Plugin {
return;
}
this.events.push([workflow, context, options]);
if (this.isWorkflowSync(workflow)) {
return this.triggerSync(workflow, context, options);
}
this.events.push([workflow, context, { context: options.context }]);
this.eventsCount = this.events.length;
logger.info(`new event triggered, now events: ${this.events.length}`);
@ -296,6 +314,27 @@ export default class PluginWorkflowServer extends Plugin {
setTimeout(this.prepare);
}
private async triggerSync(
workflow: WorkflowModel,
context: object,
options: { context?: any } & Transactionable = {},
): Promise<Processor | null> {
let execution;
try {
execution = await this.createExecution(workflow, context, options);
} catch (err) {
this.getLogger(workflow.id).error(`creating execution failed: ${err.message}`, err);
return null;
}
try {
return this.process(execution, null, options);
} catch (err) {
this.getLogger(execution.workflowId).error(`execution (${execution.id}) error: ${err.message}`, err);
}
return null;
}
public async resume(job) {
if (!job.execution) {
job.execution = await job.getExecution();
@ -311,15 +350,14 @@ export default class PluginWorkflowServer extends Plugin {
return new Processor(execution, { ...options, plugin: this });
}
private async createExecution(event: CachedEvent): Promise<ExecutionModel | null> {
const [workflow, context, options] = event;
private async createExecution(workflow: WorkflowModel, context, options): Promise<ExecutionModel | null> {
if (options.context?.executionId) {
// NOTE: no transaction here for read-uncommitted execution
const existed = await workflow.countExecutions({
where: {
id: options.context.executionId,
},
transaction: options.transaction,
});
if (existed) {
@ -331,40 +369,44 @@ export default class PluginWorkflowServer extends Plugin {
}
}
return this.db.sequelize.transaction(async (transaction) => {
const execution = await workflow.createExecution(
{
context,
const { transaction = await this.db.sequelize.transaction() } = options;
const execution = await workflow.createExecution(
{
context,
key: workflow.key,
status: EXECUTION_STATUS.QUEUEING,
},
{ transaction },
);
this.getLogger(workflow.id).info(`execution of workflow ${workflow.id} created as ${execution.id}`);
await workflow.increment(['executed', 'allExecuted'], { transaction });
// NOTE: https://sequelize.org/api/v6/class/src/model.js~model#instance-method-increment
if (this.db.options.dialect !== 'postgres') {
await workflow.reload({ transaction });
}
await (<typeof WorkflowModel>workflow.constructor).update(
{
allExecuted: workflow.allExecuted,
},
{
where: {
key: workflow.key,
status: EXECUTION_STATUS.QUEUEING,
},
{ transaction },
);
transaction,
},
);
this.getLogger(workflow.id).info(`execution of workflow ${workflow.id} created as ${execution.id}`);
if (!options.transaction) {
await transaction.commit();
}
await workflow.increment(['executed', 'allExecuted'], { transaction });
// NOTE: https://sequelize.org/api/v6/class/src/model.js~model#instance-method-increment
if (this.db.options.dialect !== 'postgres') {
await workflow.reload({ transaction });
}
execution.workflow = workflow;
await (<typeof WorkflowModel>workflow.constructor).update(
{
allExecuted: workflow.allExecuted,
},
{
where: {
key: workflow.key,
},
transaction,
},
);
execution.workflow = workflow;
return execution;
});
return execution;
}
private prepare = async () => {
@ -379,7 +421,7 @@ export default class PluginWorkflowServer extends Plugin {
logger.info(`preparing execution for event`);
try {
const execution = await this.createExecution(event);
const execution = await this.createExecution(...event);
// NOTE: cache first execution for most cases
if (!this.executing && !this.pending.length) {
this.pending.push([execution]);
@ -442,12 +484,16 @@ export default class PluginWorkflowServer extends Plugin {
})();
}
private async process(execution: ExecutionModel, job?: JobModel) {
private async process(
execution: ExecutionModel,
job?: JobModel,
{ transaction }: Transactionable = {},
): Promise<Processor> {
if (execution.status === EXECUTION_STATUS.QUEUEING) {
await execution.update({ status: EXECUTION_STATUS.STARTED });
await execution.update({ status: EXECUTION_STATUS.STARTED }, { transaction });
}
const processor = this.createProcessor(execution);
const processor = this.createProcessor(execution, { transaction });
this.getLogger(execution.workflowId).info(`execution (${execution.id}) ${job ? 'resuming' : 'starting'}...`);
@ -462,5 +508,7 @@ export default class PluginWorkflowServer extends Plugin {
} catch (err) {
this.getLogger(execution.workflowId).error(`execution (${execution.id}) error: ${err.message}`, err);
}
return processor;
}
}

View File

@ -24,17 +24,19 @@ export default class Processor {
};
logger: Logger;
transaction: Transaction;
nodes: FlowNodeModel[] = [];
nodesMap = new Map<number, FlowNodeModel>();
jobsMap = new Map<number, JobModel>();
jobsMapByNodeKey: { [key: string]: any } = {};
lastSavedJob: JobModel | null = null;
constructor(
public execution: ExecutionModel,
public options: ProcessorOptions,
) {
this.logger = options.plugin.getLogger(execution.workflowId);
this.transaction = options.transaction;
}
// make dual linked nodes list then cache
@ -66,17 +68,18 @@ export default class Processor {
}
public async prepare() {
const { execution } = this;
const { execution, transaction } = this;
if (!execution.workflow) {
execution.workflow = await execution.getWorkflow();
execution.workflow = await execution.getWorkflow({ transaction });
}
const nodes = await execution.workflow.getNodes();
const nodes = await execution.workflow.getNodes({ transaction });
this.makeNodes(nodes);
const jobs = await execution.getJobs({
order: [['id', 'ASC']],
transaction,
});
this.makeJobs(jobs);
@ -125,7 +128,13 @@ export default class Processor {
job = {
result:
err instanceof Error
? { message: err.message, stack: process.env.NODE_ENV === 'production' ? [] : err.stack }
? {
message: err.message,
stack:
process.env.NODE_ENV === 'production'
? 'Error stack will not be shown under "production" environment, please check logs.'
: err.stack,
}
: err,
status: JOB_STATUS.ERROR,
};
@ -199,7 +208,7 @@ export default class Processor {
async exit(s?: number) {
if (typeof s === 'number') {
const status = (<typeof Processor>this.constructor).StatusMap[s] ?? Math.sign(s);
await this.execution.update({ status });
await this.execution.update({ status }, { transaction: this.transaction });
}
this.logger.info(`execution (${this.execution.id}) exiting with status ${this.execution.status}`);
return null;
@ -208,21 +217,26 @@ export default class Processor {
// TODO(optimize)
async saveJob(payload) {
const { database } = <typeof ExecutionModel>this.execution.constructor;
const { transaction } = this;
const { model } = database.getCollection('jobs');
let job;
if (payload instanceof model) {
job = await payload.save();
job = await payload.save({ transaction });
} else if (payload.id) {
job = await model.findByPk(payload.id);
await job.update(payload);
job = await model.findByPk(payload.id, { transaction });
await job.update(payload, { transaction });
} else {
job = await model.create({
...payload,
executionId: this.execution.id,
});
job = await model.create(
{
...payload,
executionId: this.execution.id,
},
{ transaction },
);
}
this.jobsMap.set(job.id, job);
this.lastSavedJob = job;
this.jobsMapByNodeKey[job.nodeKey] = job.result;
return job;

View File

@ -1,6 +1,8 @@
import { MockServer } from '@nocobase/test';
import Database from '@nocobase/database';
import { getApp, sleep } from '@nocobase/plugin-workflow-test';
import Plugin from '..';
import { EXECUTION_STATUS } from '../constants';
describe('workflow > Plugin', () => {
@ -8,12 +10,14 @@ describe('workflow > Plugin', () => {
let db: Database;
let PostRepo;
let WorkflowModel;
let plugin;
beforeEach(async () => {
app = await getApp();
db = app.db;
WorkflowModel = db.getCollection('workflows').model;
PostRepo = db.getCollection('posts').repository;
plugin = app.pm.get(Plugin) as Plugin;
});
afterEach(() => app.destroy());
@ -504,4 +508,37 @@ describe('workflow > Plugin', () => {
expect(executions.length).toBe(0);
});
});
describe('sync', () => {
it('sync on trigger class', async () => {
const w1 = await WorkflowModel.create({
enabled: true,
type: 'syncTrigger',
});
const processor = await plugin.trigger(w1, {});
const executions = await w1.getExecutions();
expect(executions.length).toBe(1);
expect(executions[0].status).toBe(EXECUTION_STATUS.RESOLVED);
expect(processor.execution.id).toBe(executions[0].id);
expect(processor.execution.status).toBe(executions[0].status);
});
it('sync on workflow instance', async () => {
const w1 = await WorkflowModel.create({
enabled: true,
type: 'asyncTrigger',
sync: true,
});
const processor = await plugin.trigger(w1, {});
const executions = await w1.getExecutions();
expect(executions.length).toBe(1);
expect(executions[0].status).toBe(EXECUTION_STATUS.RESOLVED);
expect(processor.execution.id).toBe(executions[0].id);
expect(processor.execution.status).toBe(executions[0].status);
});
});
});

View File

@ -0,0 +1,62 @@
import { Application } from '@nocobase/server';
import Database from '@nocobase/database';
import { getApp, sleep } from '@nocobase/plugin-workflow-test';
import WorkflowPlugin from '../..';
import { EXECUTION_STATUS, JOB_STATUS } from '../../constants';
describe('workflow > instructions > end', () => {
let app: Application;
let db: Database;
let plugin: WorkflowPlugin;
let PostRepo;
let workflow;
beforeEach(async () => {
app = await getApp();
plugin = app.pm.get(WorkflowPlugin) as WorkflowPlugin;
db = app.db;
PostRepo = db.getCollection('posts').repository;
const WorkflowModel = db.getCollection('workflows').model;
workflow = await WorkflowModel.create({
enabled: true,
type: 'syncTrigger',
});
});
afterEach(() => app.destroy());
describe('end status', () => {
it('succeeded', async () => {
const n1 = await workflow.createNode({
type: 'end',
config: {
endStatus: JOB_STATUS.RESOLVED,
},
});
await plugin.trigger(workflow, {});
const [execution] = await workflow.getExecutions();
expect(execution.status).toBe(EXECUTION_STATUS.RESOLVED);
const [job] = await execution.getJobs();
expect(job.status).toBe(JOB_STATUS.RESOLVED);
});
it('failed', async () => {
const n1 = await workflow.createNode({
type: 'end',
config: {
endStatus: JOB_STATUS.FAILED,
},
});
await plugin.trigger(workflow, {});
const [execution] = await workflow.getExecutions();
expect(execution.status).toBe(EXECUTION_STATUS.FAILED);
const [job] = await execution.getJobs();
expect(job.status).toBe(JOB_STATUS.FAILED);
});
});
});

View File

@ -370,4 +370,34 @@ describe('workflow > triggers > collection', () => {
});
});
});
describe('sync', () => {
it('sync collection trigger', async () => {
const workflow = await WorkflowModel.create({
enabled: true,
type: 'collection',
sync: true,
config: {
mode: 1,
collection: 'posts',
},
});
const n1 = await workflow.createNode({
type: 'create',
config: {
collection: 'comments',
params: {
values: {},
},
},
});
await PostRepo.create({ values: { title: 't1' } });
const executions = await workflow.getExecutions();
expect(executions.length).toBe(1);
expect(executions[0].status).toBe(EXECUTION_STATUS.RESOLVED);
});
});
});

View File

@ -65,6 +65,11 @@ export default function () {
name: 'current',
defaultValue: false,
},
{
type: 'boolean',
name: 'sync',
defaultValue: false,
},
{
type: 'hasMany',
name: 'revisions',

View File

@ -15,7 +15,7 @@ export class CreateInstruction extends Instruction {
context: {
executionId: processor.execution.id,
},
// transaction: processor.transaction,
transaction: processor.transaction,
});
let result = created;
@ -28,7 +28,7 @@ export class CreateInstruction extends Instruction {
result = await repository.findOne({
filterByTk: created[model.primaryKeyAttribute],
appends: Array.from(includeFields),
// transaction: processor.transaction,
transaction: processor.transaction,
});
}

View File

@ -14,7 +14,7 @@ export class DestroyInstruction extends Instruction {
context: {
executionId: processor.execution.id,
},
// transaction: processor.transaction,
transaction: processor.transaction,
});
return {

View File

@ -0,0 +1,17 @@
import Instruction from '.';
import Processor from '../Processor';
import { JOB_STATUS } from '../constants';
import { FlowNodeModel } from '../types';
interface Config {
endStatus: number;
}
export default class extends Instruction {
async run(node: FlowNodeModel, prevJob, processor: Processor) {
const { endStatus } = <Config>node.config;
return {
status: endStatus ?? JOB_STATUS.RESOLVED,
};
}
}

View File

@ -33,7 +33,7 @@ export class QueryInstruction extends Instruction {
.filter((item) => item.field)
.map((item) => `${item.direction?.toLowerCase() === 'desc' ? '-' : ''}${item.field}`),
appends,
// transaction: processor.transaction,
transaction: processor.transaction,
});
if (failOnEmpty && (multiple ? !result.length : !result)) {

View File

@ -14,7 +14,7 @@ export class UpdateInstruction extends Instruction {
context: {
executionId: processor.execution.id,
},
// transaction: processor.transaction,
transaction: processor.transaction,
});
return {

View File

@ -85,13 +85,18 @@ async function handler(this: CollectionTrigger, workflow: WorkflowModel, data: M
// TODO: `result.toJSON()` throws error
const json = toJSON(result);
this.workflow.trigger(
const promise = this.workflow.trigger(
workflow,
{ data: json },
{
context,
transaction,
},
);
if (workflow.sync) {
await promise;
}
}
export default class CollectionTrigger extends Trigger {

View File

@ -392,6 +392,8 @@ function matchNext(this: ScheduleTrigger, workflow, now: Date, range: number = t
}
export default class ScheduleTrigger extends Trigger {
sync = false;
static CacheRules = [
({ config, allExecuted }) => (config.limit ? allExecuted < config.limit : true) && config.startsOn,
matchNext,

View File

@ -7,6 +7,7 @@ export abstract class Trigger {
abstract on(workflow: WorkflowModel): void;
abstract off(workflow: WorkflowModel): void;
duplicateConfig?(workflow: WorkflowModel, options: Transactionable): object | Promise<object>;
sync?: boolean;
}
export default Trigger;

View File

@ -19,16 +19,19 @@ export default class WorkflowModel extends Model {
declare description?: string;
declare type: string;
declare config: any;
declare options: any;
declare executed: number;
declare allExecuted: number;
declare sync: boolean;
declare createdAt: Date;
declare updatedAt: Date;
declare nodes: FlowNodeModel[];
declare nodes?: FlowNodeModel[];
declare getNodes: HasManyGetAssociationsMixin<FlowNodeModel>;
declare createNode: HasManyCreateAssociationMixin<FlowNodeModel>;
declare executions: ExecutionModel[];
declare executions?: ExecutionModel[];
declare countExecutions: HasManyCountAssociationsMixin;
declare getExecutions: HasManyGetAssociationsMixin<ExecutionModel>;
declare createExecution: HasManyCreateAssociationMixin<ExecutionModel>;