diff --git a/packages/client/src/workflow/calculators.tsx b/packages/client/src/workflow/calculators.tsx index 066469c89..a98b15069 100644 --- a/packages/client/src/workflow/calculators.tsx +++ b/packages/client/src/workflow/calculators.tsx @@ -75,25 +75,35 @@ const ConstantTypes = { string: { title: '字符串', value: 'string', - component({ onChange, type, options, ...props }) { - return onChange(ev.target.value)} />; + component({ onChange, type, options, value }) { + return ( + onChange({ value: ev.target.value, type, options })} + /> + ); }, default: '' }, number: { title: '数字', value: 'number', - component({ type, options, ...props }) { - return ; + component({ onChange, type, options, value }) { + return ( + onChange({ value: v, type, options })} + /> + ); }, default: 0 }, boolean: { title: '逻辑值', value: 'boolean', - component({ type, options, ...props }) { + component({ onChange, type, options, value }) { return ( - onChange({ value: v, type, options })}> @@ -104,8 +114,8 @@ const ConstantTypes = { // date: { // title: '日期', // value: 'date', - // component({ type, options, ...props }) { - // return ; + // component({ onChange, type, options, value }) { + // return onChange({ value: v, type, options })}/>; // }, // default: new Date() // } @@ -119,13 +129,16 @@ export const VariableTypes = { value: item.value, label: item.title })), - component({ options: { type } = { type: 'string' } }) { - return ConstantTypes[type]?.component ?? NullRender; + component({ options = { type: 'string' } }) { + return ConstantTypes[options.type]?.component ?? NullRender; }, appendTypeValue({ options = { type: 'string' } }) { return options?.type ? [options.type] : []; }, onTypeChange(old, [type, optionsType], onChange) { + if (old?.options?.type === optionsType) { + return; + } const { default: value } = ConstantTypes[optionsType]; onChange({ value, @@ -137,9 +150,9 @@ export const VariableTypes = { return { path }; } }, - job: { + $jobsMapByNodeId: { title: '节点数据', - value: 'job', + value: '$jobsMapByNodeId', options() { const node = useNodeContext(); const stack = []; @@ -182,7 +195,7 @@ export const VariableTypes = { return { nodeId, path: path.join('.') }; }, stringify({ options }) { - const stack = ['job']; + const stack = ['$jobsMapByNodeId']; if (options.nodeId) { stack.push(options.nodeId); if (options.path) { @@ -192,9 +205,9 @@ export const VariableTypes = { return `{{${stack.join('.')}}}`; } }, - context: { + $context: { title: '触发数据', - value: 'context', + value: '$context', component() { const { workflow } = useFlowContext(); const trigger = triggers.get(workflow.type); @@ -204,7 +217,7 @@ export const VariableTypes = { return { path: path.join('.') }; }, stringify({ options }) { - const stack = ['context']; + const stack = ['$context']; if (options?.path) { stack.push(options.path); } diff --git a/packages/client/src/workflow/nodes/query.tsx b/packages/client/src/workflow/nodes/query.tsx index 33c066307..4e16e45a9 100644 --- a/packages/client/src/workflow/nodes/query.tsx +++ b/packages/client/src/workflow/nodes/query.tsx @@ -1,7 +1,7 @@ -import React, { useState } from 'react'; +import React from 'react'; import { useForm } from '@formily/react'; import { action } from '@formily/reactive'; -import { Cascader, Select } from 'antd'; +import { Select } from 'antd'; import { t } from 'i18next'; import { css } from '@emotion/css'; diff --git a/packages/client/src/workflow/triggers/model.tsx b/packages/client/src/workflow/triggers/model.tsx index 8ed33e1c2..7536b553c 100644 --- a/packages/client/src/workflow/triggers/model.tsx +++ b/packages/client/src/workflow/triggers/model.tsx @@ -6,6 +6,8 @@ import { Select } from 'antd'; import { useCollectionManager } from '../../collection-manager'; import { useFlowContext } from '../WorkflowCanvas'; import { BaseTypeSet } from '../calculators'; +import { useForm } from '@formily/react'; +import { useCollectionFilterOptions } from '../../collection-manager/action-hooks'; export default { title: '数据表事件', @@ -20,11 +22,36 @@ export default { 'x-decorator': 'FormItem', 'x-component': 'Select', }, - // mode: { - // type: 'number', - // title: '触发时机', - // name: 'mode', - // } + mode: { + type: 'number', + title: '触发时机', + name: 'mode', + 'x-decorator': 'FormItem', + 'x-component': 'Select', + 'x-component-props': { + options: [ + { value: 1, label: '新增数据' }, + { value: 2, label: '更新数据' }, + { value: 3, label: '新增或更新数据' }, + { value: 4, label: '删除数据' } + ] + } + }, + filter: { + type: 'object', + title: '满足条件', + description: 'not supported by server yet', + name: 'filter', + 'x-decorator': 'FormItem', + 'x-component': 'Filter', + 'x-component-props': { + useProps() { + const { values } = useForm(); + const options = useCollectionFilterOptions(values.collection); + return { options }; + } + } + } }, scope: { useAsyncDataSource() { @@ -45,7 +72,7 @@ export default { const collection = collections.find(item => item.name === workflow.config.collection) ?? { fields: [] }; return ( - { onChange({ type, options: { ...options, path: `data.${path}` } }); }}> {collection.fields diff --git a/packages/plugin-workflow/src/__tests__/execution.test.ts b/packages/plugin-workflow/src/__tests__/execution.test.ts index c17cc47f7..7b121d789 100644 --- a/packages/plugin-workflow/src/__tests__/execution.test.ts +++ b/packages/plugin-workflow/src/__tests__/execution.test.ts @@ -46,9 +46,9 @@ describe('execution', () => { title: 'echo', type: 'echo' }); - + const post = await PostModel.create({ title: 't1' }); - + const [execution] = await workflow.getExecutions(); expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED); @@ -63,26 +63,26 @@ describe('execution', () => { title: 'echo', type: 'echo' }); - + const post = await PostModel.create({ title: 't1' }); - + const [execution] = await workflow.getExecutions(); expect(execution.context.data.title).toEqual(post.title); expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED); - + const jobs = await execution.getJobs(); expect(jobs.length).toEqual(1); const { status, result } = jobs[0].get(); expect(status).toEqual(JOB_STATUS.RESOLVED); expect(result).toMatchObject({ data: JSON.parse(JSON.stringify(post.toJSON())) }); }); - + it('workflow with multiple simple nodes', async () => { const n1 = await workflow.createNode({ title: 'echo 1', type: 'echo' }); - + const n2 = await workflow.createNode({ title: 'echo 2', type: 'echo', @@ -90,13 +90,13 @@ describe('execution', () => { }); await n1.setDownstream(n2); - + const post = await PostModel.create({ title: 't1' }); - + const [execution] = await workflow.getExecutions(); expect(execution.context.data.title).toEqual(post.title); expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED); - + const jobs = await execution.getJobs(); expect(jobs.length).toEqual(2); const { status, result } = jobs[1].get(); @@ -109,7 +109,7 @@ describe('execution', () => { title: 'error', type: 'error' }); - + const post = await PostModel.create({ title: 't1' }); const [execution] = await workflow.getExecutions(); @@ -119,7 +119,7 @@ describe('execution', () => { expect(jobs.length).toEqual(1); const { status, result } = jobs[0].get(); expect(status).toEqual(JOB_STATUS.REJECTED); - expect(result).toBe('Error: definite error'); + expect(result.message).toBe('definite error'); }); }); @@ -129,7 +129,7 @@ describe('execution', () => { title: 'prompt', type: 'prompt', }); - + const n2 = await workflow.createNode({ title: 'echo', type: 'echo', @@ -185,7 +185,7 @@ describe('execution', () => { const jobs = await execution.getJobs(); expect(jobs.length).toEqual(1); expect(jobs[0].status).toEqual(JOB_STATUS.REJECTED); - expect(jobs[0].result).toEqual('Error: input failed'); + expect(jobs[0].result.message).toEqual('input failed'); }); }); diff --git a/packages/plugin-workflow/src/__tests__/instructions/calculation.test.ts b/packages/plugin-workflow/src/__tests__/instructions/calculation.test.ts index 90487681e..24ede4d76 100644 --- a/packages/plugin-workflow/src/__tests__/instructions/calculation.test.ts +++ b/packages/plugin-workflow/src/__tests__/instructions/calculation.test.ts @@ -60,7 +60,7 @@ describe('workflow > instructions > calculation', () => { calculator: 'add', operands: [ { value: 1 }, - { type: 'context', options: { path: 'data.read' } } + { type: '$context', options: { path: 'data.read' } } ] } } @@ -104,7 +104,7 @@ describe('workflow > instructions > calculation', () => { calculator: 'add', operands: [ { value: 1 }, - { type: 'job', options: { nodeId: n1.id, path: 'data.read' } } + { type: '$jobsMapByNodeId', options: { nodeId: n1.id, path: 'data.read' } } ] } }, @@ -177,7 +177,7 @@ describe('workflow > instructions > calculation', () => { operands: [ { value: 1 }, { - type: 'calculation', + type: '$calculation', options: { calculator: 'minus', operands: [ diff --git a/packages/plugin-workflow/src/calculators/index.ts b/packages/plugin-workflow/src/calculators/index.ts index 468e9e47c..8e88ee77d 100644 --- a/packages/plugin-workflow/src/calculators/index.ts +++ b/packages/plugin-workflow/src/calculators/index.ts @@ -9,7 +9,7 @@ export const calculators = new Registry(); export default calculators; -export type OperandType = 'context' | 'input' | 'job' | 'calculation'; +export type OperandType = '$context' | '$input' | '$jobsMapByNodeId' | '$calculation'; export type ObjectGetterOptions = { path?: string @@ -30,22 +30,22 @@ export type ConstantOperand = { }; export type ContextOperand = { - type: 'context'; + type: '$context'; options: ObjectGetterOptions; }; export type InputOperand = { - type: 'input'; + type: '$input'; options: ObjectGetterOptions; }; export type JobOperand = { - type: 'job'; + type: '$jobsMapByNodeId'; options: JobGetterOptions; }; export type Calculation = { - type: 'calculation'; + type: '$calculation'; options: CalculationOptions }; @@ -68,22 +68,22 @@ export function calculate(operand: Operand, lastJob: JobModel, execution: Execut switch (operand.type) { // @Deprecated // from execution context - case 'context': + case '$context': return get(execution.context, operand.options.path); // @Deprecated // from last job (or input job) - case 'input': + case '$input': return lastJob ?? get(lastJob.result, operand.options.path); // @Deprecated // from job in execution - case 'job': + case '$jobsMapByNodeId': // assume jobs have been fetched from execution before const job = execution.jobsMapByNodeId[operand.options.nodeId]; return job && get(job, operand.options.path); - case 'calculation': + case '$calculation': const fn = calculators.get(operand.options.calculator); if (!fn) { throw new Error(`no calculator function registered for "${operand.options.calculator}"`); diff --git a/packages/plugin-workflow/src/collections/executions.ts b/packages/plugin-workflow/src/collections/executions.ts index 1053e6657..8cf0e49ad 100644 --- a/packages/plugin-workflow/src/collections/executions.ts +++ b/packages/plugin-workflow/src/collections/executions.ts @@ -11,6 +11,12 @@ export default { name: 'workflow', title: '所属工作流' }, + { + type: 'boolean', + name: 'useTransaction', + title: '使用事务', + defaultValue: false + }, { interface: 'linkTo', type: 'hasMany', diff --git a/packages/plugin-workflow/src/collections/workflows.ts b/packages/plugin-workflow/src/collections/workflows.ts index ede082b06..a6746afb1 100644 --- a/packages/plugin-workflow/src/collections/workflows.ts +++ b/packages/plugin-workflow/src/collections/workflows.ts @@ -39,6 +39,13 @@ export default { required: true, defaultValue: {} }, + { + interface: 'boolean', + type: 'boolean', + title: '使用事务', + name: 'useTransaction', + defaultValue: false + }, { interface: 'linkTo', type: 'hasMany', diff --git a/packages/plugin-workflow/src/instructions/calculation.ts b/packages/plugin-workflow/src/instructions/calculation.ts index a35c2ae8e..3654b8dbd 100644 --- a/packages/plugin-workflow/src/instructions/calculation.ts +++ b/packages/plugin-workflow/src/instructions/calculation.ts @@ -35,7 +35,7 @@ export default { const result = calculation ? calculate({ - type: 'calculation', + type: '$calculation', options: execution.getParsedValue(calculation) }, prevJob, execution) : null; diff --git a/packages/plugin-workflow/src/instructions/create.ts b/packages/plugin-workflow/src/instructions/create.ts index 2532fc556..8119450aa 100644 --- a/packages/plugin-workflow/src/instructions/create.ts +++ b/packages/plugin-workflow/src/instructions/create.ts @@ -9,8 +9,9 @@ export default { } = this.config; const repo = (this.constructor).database.getRepository(collection); + const options = execution.getParsedValue(params); const result = await repo.create({ - ...execution.getParsedValue(params), + ...options, transaction: execution.transaction }); diff --git a/packages/plugin-workflow/src/instructions/destroy.ts b/packages/plugin-workflow/src/instructions/destroy.ts index e8d39d773..7b52808b4 100644 --- a/packages/plugin-workflow/src/instructions/destroy.ts +++ b/packages/plugin-workflow/src/instructions/destroy.ts @@ -9,8 +9,9 @@ export default { } = this.config; const repo = (this.constructor).database.getRepository(collection); + const options = execution.getParsedValue(params); const result = await repo.destroy({ - ...execution.getParsedValue(params), + ...options, transaction: execution.transaction }); diff --git a/packages/plugin-workflow/src/instructions/update.ts b/packages/plugin-workflow/src/instructions/update.ts index 41b092cb4..66abbeb1b 100644 --- a/packages/plugin-workflow/src/instructions/update.ts +++ b/packages/plugin-workflow/src/instructions/update.ts @@ -10,8 +10,9 @@ export default { } = this.config; const repo = (this.constructor).database.getRepository(collection); + const options = execution.getParsedValue(params); const result = await repo.update({ - ...execution.getParsedValue(params), + ...options, transaction: execution.transaction }); diff --git a/packages/plugin-workflow/src/models/Execution.ts b/packages/plugin-workflow/src/models/Execution.ts index 2e58a5d01..fbbd99cc0 100644 --- a/packages/plugin-workflow/src/models/Execution.ts +++ b/packages/plugin-workflow/src/models/Execution.ts @@ -19,6 +19,8 @@ export default class ExecutionModel extends Model { declare title: string; declare context: any; declare status: number; + // NOTE: this duplicated column is for transaction in preparing cycle from workflow + declare useTransaction: boolean; declare createdAt: Date; declare updatedAt: Date; @@ -71,10 +73,25 @@ export default class ExecutionModel extends Model { }); } + getTransaction() { + const { sequelize } = (this.constructor).database; + // @ts-ignore + if (!this.useTransaction || sequelize.options.dialect === 'sqlite') { + return undefined; + } + + const { options } = this; + + // @ts-ignore + return options.transaction && !options.transaction.finished + ? options.transaction + : sequelize.transaction(); + } + async prepare(options, commit = false) { this.options = options || {}; - const { transaction = await (this.constructor).database.sequelize.transaction() } = - this.options; + // @ts-ignore + const transaction = await this.getTransaction() this.transaction = transaction; if (!this.workflow) { @@ -122,7 +139,8 @@ export default class ExecutionModel extends Model { } private async commit() { - if (!this.options || !this.options.transaction) { + // @ts-ignore + if (this.transaction && (!this.options.transaction || this.options.transaction.finished)) { await this.transaction.commit(); } } @@ -138,7 +156,9 @@ export default class ExecutionModel extends Model { } catch (err) { // for uncaught error, set to rejected job = { - result: err instanceof Error ? err.toString() : err, + result: err instanceof Error + ? { message: err.message, stack: process.env.NODE_ENV === 'production' ? [] : err.stack } + : err, status: JOB_STATUS.REJECTED, }; // if previous job is from resuming @@ -148,7 +168,7 @@ export default class ExecutionModel extends Model { } } - let savedJob: JobModel; + let savedJob; // TODO(optimize): many checking of resuming or new could be improved // could be implemented separately in exec() / resume() if (job instanceof Model) { @@ -162,7 +182,7 @@ export default class ExecutionModel extends Model { }); } - if (savedJob.get('status') === JOB_STATUS.RESOLVED && node.downstream) { + if (savedJob.status === JOB_STATUS.RESOLVED && node.downstream) { // run next node return this.run(node.downstream, savedJob); } diff --git a/packages/plugin-workflow/src/models/Workflow.ts b/packages/plugin-workflow/src/models/Workflow.ts index 0625d1a71..b8424808d 100644 --- a/packages/plugin-workflow/src/models/Workflow.ts +++ b/packages/plugin-workflow/src/models/Workflow.ts @@ -15,6 +15,7 @@ export default class WorkflowModel extends Model { declare description?: string; declare type: string; declare config: any; + declare useTransaction: boolean; declare createdAt: Date; declare updatedAt: Date; @@ -46,30 +47,48 @@ export default class WorkflowModel extends Model { return `workflow-${this.get('id')}`; } + getTransaction(options) { + if (!this.useTransaction) { + return undefined; + } + + return options.transaction && !options.transaction.finished + ? options.transaction + : (this.constructor).database.sequelize.transaction(); + } + async toggle(enable?: boolean) { const type = this.get('type'); const { on, off } = triggers.get(type); if (typeof enable !== 'undefined' ? enable : this.get('enabled')) { - on.call(this, this.start.bind(this)); + on.call(this, this.trigger.bind(this)); } else { off.call(this); } } - async start(context: Object, options) { + async trigger(context: Object, options) { // `null` means not to trigger if (context === null) { return; } + const transaction = await this.getTransaction(options); + const execution = await this.createExecution({ context, status: EXECUTION_STATUS.STARTED, - }); + useTransaction: this.useTransaction + }, { transaction }); execution.workflow = this; - await execution.start(options); + await execution.start({ transaction }); + + if (transaction && (!options.transaction || options.transaction.finished)) { + await transaction.commit(); + } + return execution; } } diff --git a/packages/plugin-workflow/src/triggers/model.ts b/packages/plugin-workflow/src/triggers/model.ts index 7003ee0df..993e565c7 100644 --- a/packages/plugin-workflow/src/triggers/model.ts +++ b/packages/plugin-workflow/src/triggers/model.ts @@ -22,12 +22,18 @@ export default { name: 'model', on(this: WorkflowModel, callback: Function) { const { database } = this.constructor; - const { collection, mode } = this.config; + const { collection, mode, filter } = this.config; const Collection = database.getCollection(collection); if (!Collection) { return; } - const handler = (data: any, options) => callback({ data: data.get() }, options); + // async function, should return promise + const handler = (data: any, options) => { + if (filter) { + // TODO: check all conditions in filter against data + } + return callback({ data: data.get() }, options); + }; // TODO: duplication when mode change should be considered for (let [key, event] of MODE_BITMAP_EVENTS.entries()) { if (mode & key) {