From 4b8915b6160a098982bb14dfc12224094a60027b Mon Sep 17 00:00:00 2001 From: Junyi <mytharcher@users.noreply.github.com> Date: Thu, 25 Jan 2024 22:10:03 +0800 Subject: [PATCH] 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 --- .../antd/action/Action.Designer.tsx | 2 +- .../src/client/DelayInstruction.tsx | 9 +- .../src/locale/en-US.json | 2 +- .../src/locale/es-ES.json | 2 +- .../src/locale/fr-FR.json | 2 +- .../src/locale/pt-BR.json | 2 +- .../src/locale/zh-CN.json | 2 +- .../src/server/FormTrigger.ts | 50 +++++-- .../src/server/__tests__/trigger.test.ts | 63 ++++++++- .../src/client/index.ts | 3 +- .../src/client/instruction/index.tsx | 7 +- .../src/server/RequestInstruction.ts | 22 ++- .../src/server/__tests__/instruction.test.ts | 29 +++- .../plugin-workflow-test/src/server/index.ts | 2 + .../src/server/instructions.ts | 1 - .../src/server/triggers.ts | 13 ++ .../plugin-workflow/src/client/AddButton.tsx | 27 ++-- .../src/client/WorkflowPane.tsx | 45 +++++- .../plugin-workflow/src/client/index.tsx | 7 + .../src/client/nodes/condition.tsx | 1 + .../plugin-workflow/src/client/nodes/end.tsx | 25 ++++ .../src/client/nodes/index.tsx | 15 +- .../src/client/schemas/workflows.ts | 23 ++++ .../src/client/triggers/index.tsx | 1 + .../src/client/triggers/schedule/index.tsx | 1 + .../plugin-workflow/src/locale/zh-CN.json | 7 +- .../plugin-workflow/src/server/Plugin.ts | 128 ++++++++++++------ .../plugin-workflow/src/server/Processor.ts | 40 ++++-- .../src/server/__tests__/Plugin.test.ts | 37 +++++ .../server/__tests__/instructions/end.test.ts | 62 +++++++++ .../__tests__/triggers/collection.test.ts | 30 ++++ .../src/server/collections/workflows.ts | 5 + .../server/instructions/CreateInstruction.ts | 4 +- .../server/instructions/DestroyInstruction.ts | 2 +- .../src/server/instructions/EndInstruction.ts | 17 +++ .../server/instructions/QueryInstruction.ts | 2 +- .../server/instructions/UpdateInstruction.ts | 2 +- .../src/server/triggers/CollectionTrigger.ts | 7 +- .../src/server/triggers/ScheduleTrigger.ts | 2 + .../src/server/triggers/index.ts | 1 + .../src/server/types/Workflow.ts | 7 +- 41 files changed, 593 insertions(+), 116 deletions(-) create mode 100644 packages/plugins/@nocobase/plugin-workflow-test/src/server/triggers.ts create mode 100644 packages/plugins/@nocobase/plugin-workflow/src/client/nodes/end.tsx create mode 100644 packages/plugins/@nocobase/plugin-workflow/src/server/__tests__/instructions/end.test.ts create mode 100644 packages/plugins/@nocobase/plugin-workflow/src/server/instructions/EndInstruction.ts diff --git a/packages/core/client/src/schema-component/antd/action/Action.Designer.tsx b/packages/core/client/src/schema-component/antd/action/Action.Designer.tsx index 836ceebda..c26425041 100644 --- a/packages/core/client/src/schema-component/antd/action/Action.Designer.tsx +++ b/packages/core/client/src/schema-component/antd/action/Action.Designer.tsx @@ -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' }), diff --git a/packages/plugins/@nocobase/plugin-workflow-delay/src/client/DelayInstruction.tsx b/packages/plugins/@nocobase/plugin-workflow-delay/src/client/DelayInstruction.tsx index a6eb137e5..e090964cf 100644 --- a/packages/plugins/@nocobase/plugin-workflow-delay/src/client/DelayInstruction.tsx +++ b/packages/plugins/@nocobase/plugin-workflow-delay/src/client/DelayInstruction.tsx @@ -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); + } } diff --git a/packages/plugins/@nocobase/plugin-workflow-delay/src/locale/en-US.json b/packages/plugins/@nocobase/plugin-workflow-delay/src/locale/en-US.json index f664b9afa..28814db47 100644 --- a/packages/plugins/@nocobase/plugin-workflow-delay/src/locale/en-US.json +++ b/packages/plugins/@nocobase/plugin-workflow-delay/src/locale/en-US.json @@ -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" diff --git a/packages/plugins/@nocobase/plugin-workflow-delay/src/locale/es-ES.json b/packages/plugins/@nocobase/plugin-workflow-delay/src/locale/es-ES.json index eec82769a..9931cf087 100644 --- a/packages/plugins/@nocobase/plugin-workflow-delay/src/locale/es-ES.json +++ b/packages/plugins/@nocobase/plugin-workflow-delay/src/locale/es-ES.json @@ -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" diff --git a/packages/plugins/@nocobase/plugin-workflow-delay/src/locale/fr-FR.json b/packages/plugins/@nocobase/plugin-workflow-delay/src/locale/fr-FR.json index c084f5852..6e5cfa59c 100644 --- a/packages/plugins/@nocobase/plugin-workflow-delay/src/locale/fr-FR.json +++ b/packages/plugins/@nocobase/plugin-workflow-delay/src/locale/fr-FR.json @@ -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" diff --git a/packages/plugins/@nocobase/plugin-workflow-delay/src/locale/pt-BR.json b/packages/plugins/@nocobase/plugin-workflow-delay/src/locale/pt-BR.json index cd73e9f56..b00baec14 100644 --- a/packages/plugins/@nocobase/plugin-workflow-delay/src/locale/pt-BR.json +++ b/packages/plugins/@nocobase/plugin-workflow-delay/src/locale/pt-BR.json @@ -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" diff --git a/packages/plugins/@nocobase/plugin-workflow-delay/src/locale/zh-CN.json b/packages/plugins/@nocobase/plugin-workflow-delay/src/locale/zh-CN.json index 04f0c3d88..0eb731514 100644 --- a/packages/plugins/@nocobase/plugin-workflow-delay/src/locale/zh-CN.json +++ b/packages/plugins/@nocobase/plugin-workflow-delay/src/locale/zh-CN.json @@ -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": "失败并退出" diff --git a/packages/plugins/@nocobase/plugin-workflow-form-trigger/src/server/FormTrigger.ts b/packages/plugins/@nocobase/plugin-workflow-form-trigger/src/server/FormTrigger.ts index b7e85539c..a55f5ac31 100644 --- a/packages/plugins/@nocobase/plugin-workflow-form-trigger/src/server/FormTrigger.ts +++ b/packages/plugins/@nocobase/plugin-workflow-form-trigger/src/server/FormTrigger.ts @@ -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) {} diff --git a/packages/plugins/@nocobase/plugin-workflow-form-trigger/src/server/__tests__/trigger.test.ts b/packages/plugins/@nocobase/plugin-workflow-form-trigger/src/server/__tests__/trigger.test.ts index 7351acc4a..8491e6240 100644 --- a/packages/plugins/@nocobase/plugin-workflow-form-trigger/src/server/__tests__/trigger.test.ts +++ b/packages/plugins/@nocobase/plugin-workflow-form-trigger/src/server/__tests__/trigger.test.ts @@ -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); + }); + }); }); diff --git a/packages/plugins/@nocobase/plugin-workflow-manual/src/client/index.ts b/packages/plugins/@nocobase/plugin-workflow-manual/src/client/index.ts index dd57574e2..95a7954c1 100644 --- a/packages/plugins/@nocobase/plugin-workflow-manual/src/client/index.ts +++ b/packages/plugins/@nocobase/plugin-workflow-manual/src/client/index.ts @@ -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); diff --git a/packages/plugins/@nocobase/plugin-workflow-manual/src/client/instruction/index.tsx b/packages/plugins/@nocobase/plugin-workflow-manual/src/client/instruction/index.tsx index e826dc6c9..de9031315 100644 --- a/packages/plugins/@nocobase/plugin-workflow-manual/src/client/instruction/index.tsx +++ b/packages/plugins/@nocobase/plugin-workflow-manual/src/client/instruction/index.tsx @@ -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); + } } diff --git a/packages/plugins/@nocobase/plugin-workflow-request/src/server/RequestInstruction.ts b/packages/plugins/@nocobase/plugin-workflow-request/src/server/RequestInstruction.ts index 9d3d215d0..d337fb977 100644 --- a/packages/plugins/@nocobase/plugin-workflow-request/src/server/RequestInstruction.ts +++ b/packages/plugins/@nocobase/plugin-workflow-request/src/server/RequestInstruction.ts @@ -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) => { diff --git a/packages/plugins/@nocobase/plugin-workflow-request/src/server/__tests__/instruction.test.ts b/packages/plugins/@nocobase/plugin-workflow-request/src/server/__tests__/instruction.test.ts index cee58705c..61f589280 100644 --- a/packages/plugins/@nocobase/plugin-workflow-request/src/server/__tests__/instruction.test.ts +++ b/packages/plugins/@nocobase/plugin-workflow-request/src/server/__tests__/instruction.test.ts @@ -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: {} }); + }); + }); }); diff --git a/packages/plugins/@nocobase/plugin-workflow-test/src/server/index.ts b/packages/plugins/@nocobase/plugin-workflow-test/src/server/index.ts index 30e504cfa..1f384d5bc 100644 --- a/packages/plugins/@nocobase/plugin-workflow-test/src/server/index.ts +++ b/packages/plugins/@nocobase/plugin-workflow-test/src/server/index.ts @@ -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, }, diff --git a/packages/plugins/@nocobase/plugin-workflow-test/src/server/instructions.ts b/packages/plugins/@nocobase/plugin-workflow-test/src/server/instructions.ts index 69fd116c8..f92f07796 100644 --- a/packages/plugins/@nocobase/plugin-workflow-test/src/server/instructions.ts +++ b/packages/plugins/@nocobase/plugin-workflow-test/src/server/instructions.ts @@ -13,7 +13,6 @@ export default { error: { run(node, input, processor) { throw new Error('definite error'); - return null; }, }, diff --git a/packages/plugins/@nocobase/plugin-workflow-test/src/server/triggers.ts b/packages/plugins/@nocobase/plugin-workflow-test/src/server/triggers.ts new file mode 100644 index 000000000..91c5ae454 --- /dev/null +++ b/packages/plugins/@nocobase/plugin-workflow-test/src/server/triggers.ts @@ -0,0 +1,13 @@ +export default { + syncTrigger: class { + constructor(public readonly workflow) {} + on() {} + off() {} + sync = true; + }, + asyncTrigger: class { + constructor(public readonly workflow) {} + on() {} + off() {} + }, +}; diff --git a/packages/plugins/@nocobase/plugin-workflow/src/client/AddButton.tsx b/packages/plugins/@nocobase/plugin-workflow/src/client/AddButton.tsx index 6443b88a0..410565939 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/client/AddButton.tsx +++ b/packages/plugins/@nocobase/plugin-workflow/src/client/AddButton.tsx @@ -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>(() => { diff --git a/packages/plugins/@nocobase/plugin-workflow/src/client/WorkflowPane.tsx b/packages/plugins/@nocobase/plugin-workflow/src/client/WorkflowPane.tsx index 517b8627f..3265fb799 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/client/WorkflowPane.tsx +++ b/packages/plugins/@nocobase/plugin-workflow/src/client/WorkflowPane.tsx @@ -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={{ diff --git a/packages/plugins/@nocobase/plugin-workflow/src/client/index.tsx b/packages/plugins/@nocobase/plugin-workflow/src/client/index.tsx index 227fb0134..de36d9ebd 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/client/index.tsx +++ b/packages/plugins/@nocobase/plugin-workflow/src/client/index.tsx @@ -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); diff --git a/packages/plugins/@nocobase/plugin-workflow/src/client/nodes/condition.tsx b/packages/plugins/@nocobase/plugin-workflow/src/client/nodes/condition.tsx index dd2e2333c..4100f86d7 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/client/nodes/condition.tsx +++ b/packages/plugins/@nocobase/plugin-workflow/src/client/nodes/condition.tsx @@ -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)) diff --git a/packages/plugins/@nocobase/plugin-workflow/src/client/nodes/end.tsx b/packages/plugins/@nocobase/plugin-workflow/src/client/nodes/end.tsx new file mode 100644 index 000000000..e5f5e2340 --- /dev/null +++ b/packages/plugins/@nocobase/plugin-workflow/src/client/nodes/end.tsx @@ -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; +} diff --git a/packages/plugins/@nocobase/plugin-workflow/src/client/nodes/index.tsx b/packages/plugins/@nocobase/plugin-workflow/src/client/nodes/index.tsx index 2ee005acb..33c96bd1c 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/client/nodes/index.tsx +++ b/packages/plugins/@nocobase/plugin-workflow/src/client/nodes/index.tsx @@ -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> ); diff --git a/packages/plugins/@nocobase/plugin-workflow/src/client/schemas/workflows.ts b/packages/plugins/@nocobase/plugin-workflow/src/client/schemas/workflows.ts index a72a1c224..b577b9b3b 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/client/schemas/workflows.ts +++ b/packages/plugins/@nocobase/plugin-workflow/src/client/schemas/workflows.ts @@ -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: { diff --git a/packages/plugins/@nocobase/plugin-workflow/src/client/triggers/index.tsx b/packages/plugins/@nocobase/plugin-workflow/src/client/triggers/index.tsx index e22f85b3c..ec9e7d171 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/client/triggers/index.tsx +++ b/packages/plugins/@nocobase/plugin-workflow/src/client/triggers/index.tsx @@ -54,6 +54,7 @@ function useUpdateConfigAction() { } export abstract class Trigger { + sync: boolean; title: string; description?: string; // group: string; diff --git a/packages/plugins/@nocobase/plugin-workflow/src/client/triggers/schedule/index.tsx b/packages/plugins/@nocobase/plugin-workflow/src/client/triggers/schedule/index.tsx index 973640d78..3f0a2cad4 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/client/triggers/schedule/index.tsx +++ b/packages/plugins/@nocobase/plugin-workflow/src/client/triggers/schedule/index.tsx @@ -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 = { diff --git a/packages/plugins/@nocobase/plugin-workflow/src/locale/zh-CN.json b/packages/plugins/@nocobase/plugin-workflow/src/locale/zh-CN.json index 446247507..9ca2f7b48 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/locale/zh-CN.json +++ b/packages/plugins/@nocobase/plugin-workflow/src/locale/zh-CN.json @@ -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": "成功" } diff --git a/packages/plugins/@nocobase/plugin-workflow/src/server/Plugin.ts b/packages/plugins/@nocobase/plugin-workflow/src/server/Plugin.ts index 0fd34ee39..1ef66b735 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/server/Plugin.ts +++ b/packages/plugins/@nocobase/plugin-workflow/src/server/Plugin.ts @@ -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; } } diff --git a/packages/plugins/@nocobase/plugin-workflow/src/server/Processor.ts b/packages/plugins/@nocobase/plugin-workflow/src/server/Processor.ts index 0cae71471..aa60555f7 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/server/Processor.ts +++ b/packages/plugins/@nocobase/plugin-workflow/src/server/Processor.ts @@ -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; diff --git a/packages/plugins/@nocobase/plugin-workflow/src/server/__tests__/Plugin.test.ts b/packages/plugins/@nocobase/plugin-workflow/src/server/__tests__/Plugin.test.ts index 85283d833..bf3f0545a 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/server/__tests__/Plugin.test.ts +++ b/packages/plugins/@nocobase/plugin-workflow/src/server/__tests__/Plugin.test.ts @@ -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); + }); + }); }); diff --git a/packages/plugins/@nocobase/plugin-workflow/src/server/__tests__/instructions/end.test.ts b/packages/plugins/@nocobase/plugin-workflow/src/server/__tests__/instructions/end.test.ts new file mode 100644 index 000000000..ace3ae4cc --- /dev/null +++ b/packages/plugins/@nocobase/plugin-workflow/src/server/__tests__/instructions/end.test.ts @@ -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); + }); + }); +}); diff --git a/packages/plugins/@nocobase/plugin-workflow/src/server/__tests__/triggers/collection.test.ts b/packages/plugins/@nocobase/plugin-workflow/src/server/__tests__/triggers/collection.test.ts index 5a7a93d43..cc74e8bd3 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/server/__tests__/triggers/collection.test.ts +++ b/packages/plugins/@nocobase/plugin-workflow/src/server/__tests__/triggers/collection.test.ts @@ -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); + }); + }); }); diff --git a/packages/plugins/@nocobase/plugin-workflow/src/server/collections/workflows.ts b/packages/plugins/@nocobase/plugin-workflow/src/server/collections/workflows.ts index 05fe626e0..64465ad95 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/server/collections/workflows.ts +++ b/packages/plugins/@nocobase/plugin-workflow/src/server/collections/workflows.ts @@ -65,6 +65,11 @@ export default function () { name: 'current', defaultValue: false, }, + { + type: 'boolean', + name: 'sync', + defaultValue: false, + }, { type: 'hasMany', name: 'revisions', diff --git a/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/CreateInstruction.ts b/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/CreateInstruction.ts index 70f682bea..edc2ca967 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/CreateInstruction.ts +++ b/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/CreateInstruction.ts @@ -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, }); } diff --git a/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/DestroyInstruction.ts b/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/DestroyInstruction.ts index abdd07ed9..f6e4be160 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/DestroyInstruction.ts +++ b/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/DestroyInstruction.ts @@ -14,7 +14,7 @@ export class DestroyInstruction extends Instruction { context: { executionId: processor.execution.id, }, - // transaction: processor.transaction, + transaction: processor.transaction, }); return { diff --git a/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/EndInstruction.ts b/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/EndInstruction.ts new file mode 100644 index 000000000..23d401d43 --- /dev/null +++ b/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/EndInstruction.ts @@ -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, + }; + } +} diff --git a/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/QueryInstruction.ts b/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/QueryInstruction.ts index c46699df5..3f13aeaf0 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/QueryInstruction.ts +++ b/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/QueryInstruction.ts @@ -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)) { diff --git a/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/UpdateInstruction.ts b/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/UpdateInstruction.ts index 8aa89b5e5..e95ef41ee 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/UpdateInstruction.ts +++ b/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/UpdateInstruction.ts @@ -14,7 +14,7 @@ export class UpdateInstruction extends Instruction { context: { executionId: processor.execution.id, }, - // transaction: processor.transaction, + transaction: processor.transaction, }); return { diff --git a/packages/plugins/@nocobase/plugin-workflow/src/server/triggers/CollectionTrigger.ts b/packages/plugins/@nocobase/plugin-workflow/src/server/triggers/CollectionTrigger.ts index da7c5cce0..bbeb82313 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/server/triggers/CollectionTrigger.ts +++ b/packages/plugins/@nocobase/plugin-workflow/src/server/triggers/CollectionTrigger.ts @@ -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 { diff --git a/packages/plugins/@nocobase/plugin-workflow/src/server/triggers/ScheduleTrigger.ts b/packages/plugins/@nocobase/plugin-workflow/src/server/triggers/ScheduleTrigger.ts index 3536fbc33..a7e06effa 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/server/triggers/ScheduleTrigger.ts +++ b/packages/plugins/@nocobase/plugin-workflow/src/server/triggers/ScheduleTrigger.ts @@ -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, diff --git a/packages/plugins/@nocobase/plugin-workflow/src/server/triggers/index.ts b/packages/plugins/@nocobase/plugin-workflow/src/server/triggers/index.ts index 408ceff87..34ec0a556 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/server/triggers/index.ts +++ b/packages/plugins/@nocobase/plugin-workflow/src/server/triggers/index.ts @@ -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; diff --git a/packages/plugins/@nocobase/plugin-workflow/src/server/types/Workflow.ts b/packages/plugins/@nocobase/plugin-workflow/src/server/types/Workflow.ts index 68c0f7f3d..2171779e4 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/server/types/Workflow.ts +++ b/packages/plugins/@nocobase/plugin-workflow/src/server/types/Workflow.ts @@ -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>;