From 1cce3bf1648633f28849dab01ecdeda351915fde Mon Sep 17 00:00:00 2001 From: mytharcher Date: Sun, 9 Jan 2022 22:22:26 +0800 Subject: [PATCH 1/6] feat: server mvp for configurable workflow with nodes --- packages/plugin-workflow/.npmignore | 7 + packages/plugin-workflow/package.json | 16 ++ .../src/__tests__/collections/posts.ts | 15 ++ .../src/__tests__/collections/targets.ts | 15 ++ .../plugin-workflow/src/__tests__/index.ts | 26 +++ .../__tests__/instructions/condition.test.ts | 114 ++++++++++ .../src/__tests__/workflow.test.ts | 213 ++++++++++++++++++ .../src/collections/executions.ts | 33 +++ .../src/collections/flow_nodes.ts | 59 +++++ .../plugin-workflow/src/collections/jobs.ts | 48 ++++ .../src/collections/workflows.ts | 56 +++++ packages/plugin-workflow/src/constants.ts | 11 + .../src/instructions/condition/calculators.ts | 52 +++++ .../src/instructions/condition/getter.ts | 56 +++++ .../src/instructions/condition/index.ts | 72 ++++++ .../plugin-workflow/src/instructions/echo.ts | 6 + .../plugin-workflow/src/instructions/index.ts | 37 +++ .../src/instructions/prompt.ts | 6 + .../plugin-workflow/src/models/Execution.ts | 140 ++++++++++++ .../plugin-workflow/src/models/Workflow.ts | 49 ++++ packages/plugin-workflow/src/server.ts | 61 +++++ .../src/triggers/data-change.ts | 10 + .../plugin-workflow/src/triggers/index.ts | 21 ++ 23 files changed, 1123 insertions(+) create mode 100644 packages/plugin-workflow/.npmignore create mode 100644 packages/plugin-workflow/package.json create mode 100644 packages/plugin-workflow/src/__tests__/collections/posts.ts create mode 100644 packages/plugin-workflow/src/__tests__/collections/targets.ts create mode 100644 packages/plugin-workflow/src/__tests__/index.ts create mode 100644 packages/plugin-workflow/src/__tests__/instructions/condition.test.ts create mode 100644 packages/plugin-workflow/src/__tests__/workflow.test.ts create mode 100644 packages/plugin-workflow/src/collections/executions.ts create mode 100644 packages/plugin-workflow/src/collections/flow_nodes.ts create mode 100644 packages/plugin-workflow/src/collections/jobs.ts create mode 100644 packages/plugin-workflow/src/collections/workflows.ts create mode 100644 packages/plugin-workflow/src/constants.ts create mode 100644 packages/plugin-workflow/src/instructions/condition/calculators.ts create mode 100644 packages/plugin-workflow/src/instructions/condition/getter.ts create mode 100644 packages/plugin-workflow/src/instructions/condition/index.ts create mode 100644 packages/plugin-workflow/src/instructions/echo.ts create mode 100644 packages/plugin-workflow/src/instructions/index.ts create mode 100644 packages/plugin-workflow/src/instructions/prompt.ts create mode 100644 packages/plugin-workflow/src/models/Execution.ts create mode 100644 packages/plugin-workflow/src/models/Workflow.ts create mode 100644 packages/plugin-workflow/src/server.ts create mode 100644 packages/plugin-workflow/src/triggers/data-change.ts create mode 100644 packages/plugin-workflow/src/triggers/index.ts diff --git a/packages/plugin-workflow/.npmignore b/packages/plugin-workflow/.npmignore new file mode 100644 index 000000000..461574b2f --- /dev/null +++ b/packages/plugin-workflow/.npmignore @@ -0,0 +1,7 @@ +node_modules +*.log +docs +__tests__ +tsconfig.json +src +.fatherrc.ts \ No newline at end of file diff --git a/packages/plugin-workflow/package.json b/packages/plugin-workflow/package.json new file mode 100644 index 000000000..70927033c --- /dev/null +++ b/packages/plugin-workflow/package.json @@ -0,0 +1,16 @@ +{ + "name": "@nocobase/plugin-workflow", + "version": "0.5.0-alpha.37", + "main": "lib/index.js", + "private": true, + "license": "MIT", + "dependencies": { + "@nocobase/server": "^0.5.0-alpha.37", + "json-templates": "^4.1.0", + "node-schedule": "^2.0.0" + }, + "devDependencies": { + "@types/node-schedule": "^1.3.1" + }, + "gitHead": "f0b335ac30f29f25c95d7d137655fa64d8d67f1e" +} diff --git a/packages/plugin-workflow/src/__tests__/collections/posts.ts b/packages/plugin-workflow/src/__tests__/collections/posts.ts new file mode 100644 index 000000000..efa6cb228 --- /dev/null +++ b/packages/plugin-workflow/src/__tests__/collections/posts.ts @@ -0,0 +1,15 @@ +import { TableOptions } from '@nocobase/database'; + +export default { + name: 'posts', + fields: [ + { + type: 'string', + name: 'title', + }, + { + type: 'boolean', + name: 'published', + } + ] +} as TableOptions; diff --git a/packages/plugin-workflow/src/__tests__/collections/targets.ts b/packages/plugin-workflow/src/__tests__/collections/targets.ts new file mode 100644 index 000000000..d1b6ed448 --- /dev/null +++ b/packages/plugin-workflow/src/__tests__/collections/targets.ts @@ -0,0 +1,15 @@ +import { TableOptions } from '@nocobase/database'; + +export default { + name: 'targets', + fields: [ + { + type: 'string', + name: 'col1', + }, + { + type: 'string', + name: 'col2', + } + ], +} as TableOptions; diff --git a/packages/plugin-workflow/src/__tests__/index.ts b/packages/plugin-workflow/src/__tests__/index.ts new file mode 100644 index 000000000..9aa44c18e --- /dev/null +++ b/packages/plugin-workflow/src/__tests__/index.ts @@ -0,0 +1,26 @@ +import path from 'path'; +import { MockServer, mockServer } from '@nocobase/test'; + +import plugin from '../server'; + +export async function getApp(options = {}): Promise { + const app = mockServer(options); + + app.plugin(plugin); + + await app.load(); + + app.db.import({ + directory: path.resolve(__dirname, './collections') + }); + + try { + await app.db.sync(); + } catch (error) { + console.error(error); + } + // TODO: need a better life cycle event than manually trigger + await app.emitAsync('beforeStart'); + + return app; +} diff --git a/packages/plugin-workflow/src/__tests__/instructions/condition.test.ts b/packages/plugin-workflow/src/__tests__/instructions/condition.test.ts new file mode 100644 index 000000000..d91afdba0 --- /dev/null +++ b/packages/plugin-workflow/src/__tests__/instructions/condition.test.ts @@ -0,0 +1,114 @@ +import { Application } from '@nocobase/server'; +import Database, { Model, ModelCtor } from '@nocobase/database'; +import { getApp } from '..'; +import { WorkflowModel } from '../../models/Workflow'; +import { EXECUTION_STATUS, JOB_STATUS } from '../../constants'; + + + +describe('workflow > instructions > condition', () => { + let app: Application; + let db: Database; + let PostModel: ModelCtor; + let WorkflowModel: ModelCtor; + + beforeEach(async () => { + app = await getApp(); + + db = app.db; + WorkflowModel = db.getModel('workflows') as any; + PostModel = db.getModel('posts'); + }); + + afterEach(() => db.close()); + + describe('single calculation', () => { + it('calculation to true downstream', async () => { + const workflow = await WorkflowModel.create({ + title: 'condition workflow', + enabled: true, + type: 'afterCreate', + config: { + collection: 'posts' + } + }); + + const n1 = await workflow.createNode({ + title: 'condition', + type: 'condition', + // (1 === 1): true + config: { + calculator: 'equal', + operands: [{ value: 1 }, { value: 1 }] + } + }); + + await workflow.createNode({ + title: 'true to echo', + type: 'echo', + when: true, + upstream_id: n1.id + }); + + await workflow.createNode({ + title: 'false to echo', + type: 'echo', + when: false, + upstream_id: n1.id + }); + + const post = await PostModel.create({ title: 't1' }); + + const [execution] = await workflow.getExecutions(); + expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED); + + const jobs = await execution.getJobs(); + expect(jobs.length).toEqual(2); + expect(jobs[1].result).toEqual(true); + }); + + it('calculation to false downstream', async () => { + const workflow = await WorkflowModel.create({ + title: 'condition workflow', + enabled: true, + type: 'afterCreate', + config: { + collection: 'posts' + } + }); + + const n1 = await workflow.createNode({ + title: 'condition', + type: 'condition', + // (0 === 1): false + config: { + calculator: 'equal', + operands: [{ value: 0 }, { value: 1 }] + } + }); + + await workflow.createNode({ + title: 'true to echo', + type: 'echo', + when: true, + upstream_id: n1.id + }); + + await workflow.createNode({ + title: 'false to echo', + type: 'echo', + when: false, + upstream_id: n1.id + }); + + const post = await PostModel.create({ title: 't1' }); + + const [execution] = await workflow.getExecutions(); + expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED); + + const jobs = await execution.getJobs(); + expect(jobs.length).toEqual(2); + expect(jobs[1].result).toEqual(false); + }); + }); +}); diff --git a/packages/plugin-workflow/src/__tests__/workflow.test.ts b/packages/plugin-workflow/src/__tests__/workflow.test.ts new file mode 100644 index 000000000..a3a27a2c9 --- /dev/null +++ b/packages/plugin-workflow/src/__tests__/workflow.test.ts @@ -0,0 +1,213 @@ +import { Application } from '@nocobase/server'; +import Database, { Model, ModelCtor } from '@nocobase/database'; +import { getApp } from '.'; +import { WorkflowModel } from '../models/Workflow'; +import { EXECUTION_STATUS, JOB_STATUS } from '../constants'; + +jest.setTimeout(300000); + +describe('workflow', () => { + let app: Application; + let db: Database; + let PostModel: ModelCtor; + // let Target: ModelCtor; + let WorkflowModel: ModelCtor; + + beforeEach(async () => { + app = await getApp(); + + db = app.db; + WorkflowModel = db.getModel('workflows') as any; + PostModel = db.getModel('posts'); + // Target = db.getModel('targets'); + }); + + afterEach(() => db.close()); + + describe('base', () => { + it('empty workflow without any nodes', async () => { + const workflow = await WorkflowModel.create({ + title: 'empty workflow', + enabled: true, + type: 'afterCreate', + config: { + collection: 'posts' + } + }); + + 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); + }); + + it('workflow with single simple node', async () => { + const workflow = await WorkflowModel.create({ + title: 'simple workflow', + enabled: true, + type: 'afterCreate', + config: { + collection: 'posts' + } + }); + + await workflow.createNode({ + 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 workflow = await WorkflowModel.create({ + title: 'simple workflow', + enabled: true, + type: 'afterCreate', + config: { + collection: 'posts' + } + }); + + const n1 = await workflow.createNode({ + title: 'echo 1', + type: 'echo' + }); + + await workflow.createNode({ + title: 'echo 2', + type: 'echo', + upstream_id: n1.id + }); + + 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(); + expect(status).toEqual(JOB_STATUS.RESOLVED); + expect(result).toMatchObject({ data: JSON.parse(JSON.stringify(post.toJSON())) }); + }); + + // TODO: or should throw error? + it('execute resolved workflow', async () => { + const workflow = await WorkflowModel.create({ + title: 'simple workflow', + enabled: true, + type: 'afterCreate', + config: { + collection: 'posts' + } + }); + + const post = await PostModel.create({ title: 't1' }); + + const [execution] = await workflow.getExecutions(); + expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED); + + await execution.exec(123); + expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED); + const jobs = await execution.getJobs(); + expect(jobs.length).toEqual(0); + }); + }); + + describe('manual nodes', () => { + it('manual node should pause execution, and could be manually resume', async () => { + const workflow = await WorkflowModel.create({ + title: 'manual workflow', + enabled: true, + type: 'afterCreate', + config: { + collection: 'posts' + } + }); + + const n1 = await workflow.createNode({ + title: 'prompt', + type: 'prompt', + }); + + await workflow.createNode({ + title: 'echo', + type: 'echo', + upstream_id: n1.id + }); + + const post = await PostModel.create({ title: 't1' }); + + const [execution] = await workflow.getExecutions(); + expect(execution.status).toEqual(EXECUTION_STATUS.STARTED); + const [pending] = await execution.getJobs(); + expect(pending.status).toEqual(JOB_STATUS.PENDING); + expect(pending.result).toEqual(null); + + await execution.exec(123, null, {}); + expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED); + + const jobs = await execution.getJobs(); + expect(jobs.length).toEqual(2); + expect(jobs[0].status).toEqual(JOB_STATUS.RESOLVED); + expect(jobs[0].result).toEqual(123); + expect(jobs[1].status).toEqual(JOB_STATUS.RESOLVED); + expect(jobs[1].result).toEqual(123); + }); + }); + + describe('condition node', () => { + it('condition node link to different downstreams', async () => { + const workflow = await WorkflowModel.create({ + title: 'condition workflow', + enabled: true, + type: 'afterCreate', + config: { + collection: 'posts' + } + }); + + const n1 = await workflow.createNode({ + title: 'condition', + type: 'condition', + // no config means always true + }); + + await workflow.createNode({ + title: 'true to echo', + type: 'echo', + when: true, + upstream_id: n1.id + }); + + await workflow.createNode({ + title: 'false to echo', + type: 'echo', + when: false, + upstream_id: n1.id + }); + + const post = await PostModel.create({ title: 't1' }); + + const [execution] = await workflow.getExecutions(); + expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED); + + const jobs = await execution.getJobs(); + expect(jobs.length).toEqual(2); + expect(jobs[1].result).toEqual(true); + }); + }); +}); diff --git a/packages/plugin-workflow/src/collections/executions.ts b/packages/plugin-workflow/src/collections/executions.ts new file mode 100644 index 000000000..23e49b367 --- /dev/null +++ b/packages/plugin-workflow/src/collections/executions.ts @@ -0,0 +1,33 @@ +import { TableOptions } from '@nocobase/database'; + +export default { + name: 'executions', + model: 'ExecutionModel', + title: '执行流程', + fields: [ + { + interface: 'linkTo', + type: 'belongsTo', + name: 'workflow', + title: '所属工作流' + }, + { + interface: 'linkTo', + type: 'hasMany', + name: 'jobs', + title: '流程记录' + }, + { + interface: 'json', + type: 'jsonb', + name: 'context', + title: '上下文数据' + }, + { + interface: 'select', + type: 'integer', + name: 'status', + title: '状态' + } + ] +} as TableOptions; diff --git a/packages/plugin-workflow/src/collections/flow_nodes.ts b/packages/plugin-workflow/src/collections/flow_nodes.ts new file mode 100644 index 000000000..b6e1a057a --- /dev/null +++ b/packages/plugin-workflow/src/collections/flow_nodes.ts @@ -0,0 +1,59 @@ +import { TableOptions } from '@nocobase/database'; + +export default { + name: 'flow_nodes', + // model: 'FlowNodeModel', + title: 'Workflow Nodes', + fields: [ + { + interface: 'string', + type: 'string', + name: 'title', + title: '名称', + component: { + showInTable: true, + showInDetail: true, + showInForm: true, + }, + }, + // which workflow belongs to + { + interface: 'linkTo', + name: 'workflow', + type: 'belongsTo', + }, + { + interface: 'linkTo', + name: 'upstream', + type: 'belongsTo', + target: 'flow_nodes' + }, + // only works when upstream node is condition type. + // put here because the design of flow-links model is not really necessary for now. + // or it should be put into flow-links model. + { + name: 'when', + type: 'boolean', + // defaultValue: null + }, + { + interface: 'select', + type: 'string', + name: 'type', + title: '类型', + dataSource: [ + { label: '无处理', value: 'echo' }, + { label: '数据处理', value: 'data' }, + { label: '数据查询', value: 'query' }, + { label: '等待人工输入', value: 'prompt' }, + { label: '条件判断', value: 'condition' }, + ] + }, + { + interface: 'json', + type: 'jsonb', + name: 'config', + title: '配置' + } + ] +} as TableOptions; diff --git a/packages/plugin-workflow/src/collections/jobs.ts b/packages/plugin-workflow/src/collections/jobs.ts new file mode 100644 index 000000000..5ac5e4bae --- /dev/null +++ b/packages/plugin-workflow/src/collections/jobs.ts @@ -0,0 +1,48 @@ +import { TableOptions } from '@nocobase/database'; + +export default { + name: 'jobs', + title: '流程记录', + fields: [ + { + interface: 'linkTo', + type: 'belongsTo', + name: 'execution', + title: '所属流程' + }, + { + interface: 'linkTo', + type: 'belongsTo', + name: 'node', + target: 'flow_nodes', + title: '所属节点' + }, + { + interface: 'linkTo', + type: 'belongsTo', + name: 'upstream', + target: 'jobs', + title: '上游记录' + }, + // pending / resolved / rejected + { + interface: 'status', + type: 'integer', + name: 'status', + title: '处理状态' + }, + { + interface: 'json', + type: 'jsonb', + name: 'result', + title: '处理结果' + }, + // TODO: possibly need node snapshot in case if node has been changed + // { + // interface: 'json', + // type: 'jsonb', + // name: 'nodeSnapshot', + // title: 'node snapshot' + // } + ] +} as TableOptions; diff --git a/packages/plugin-workflow/src/collections/workflows.ts b/packages/plugin-workflow/src/collections/workflows.ts new file mode 100644 index 000000000..83fd9e0da --- /dev/null +++ b/packages/plugin-workflow/src/collections/workflows.ts @@ -0,0 +1,56 @@ +import { TableOptions } from '@nocobase/database'; + +export default { + name: 'workflows', + model: 'WorkflowModel', + title: '自动化', + fields: [ + { + interface: 'string', + type: 'string', + name: 'title', + title: '自动化名称', + required: true + }, + { + interface: 'boolean', + type: 'boolean', + name: 'enabled', + title: '启用' + }, + { + interface: 'textarea', + type: 'text', + name: 'description', + title: '描述' + }, + { + interface: 'select', + type: 'string', + title: '触发方式', + name: 'type', + required: true + }, + { + interface: 'json', + type: 'jsonb', + title: '触发配置', + name: 'config', + required: true + }, + { + interface: 'linkTo', + type: 'hasMany', + name: 'nodes', + target: 'flow_nodes', + title: '流程节点' + }, + { + interface: 'linkTo', + type: 'hasMany', + name: 'executions', + target: 'executions', + title: '触发执行' + } + ] +} as TableOptions; diff --git a/packages/plugin-workflow/src/constants.ts b/packages/plugin-workflow/src/constants.ts new file mode 100644 index 000000000..4f140c0ea --- /dev/null +++ b/packages/plugin-workflow/src/constants.ts @@ -0,0 +1,11 @@ +export const EXECUTION_STATUS = { + STARTED: 0, + RESOLVED: 1, + REJECTED: -1 +}; + +export const JOB_STATUS = { + PENDING: 0, + RESOLVED: 1, + REJECTED: -1 +}; diff --git a/packages/plugin-workflow/src/instructions/condition/calculators.ts b/packages/plugin-workflow/src/instructions/condition/calculators.ts new file mode 100644 index 000000000..d15d14a8f --- /dev/null +++ b/packages/plugin-workflow/src/instructions/condition/calculators.ts @@ -0,0 +1,52 @@ +type Calculator = (...args: any[]) => boolean; + +const calculators = new Map(); + +export function getCalculator(type: string): Calculator { + return calculators.get(type); +} + +export function registerCalculator(type: string, fn: Calculator) { + calculators.set(type, fn); +} + +export function registerCalculators(calculators) { + Object.keys(calculators).forEach(key => { + registerCalculator(key, calculators[key]); + }); +} + +function equal(a, b) { + return a === b; +} + +function gt(a, b) { + return a > b; +} + +function gte(a, b) { + return a >= b; +} + +function lt(a, b) { + return a < b; +} + +function lte(a, b) { + return a <= b; +} + +// TODO: add more common calculators + +registerCalculators({ + equal, + gt, + gte, + lt, + lte, + '===': equal, + '>': gt, + '>=': gte, + '<': lt, + '<=': lte +}); diff --git a/packages/plugin-workflow/src/instructions/condition/getter.ts b/packages/plugin-workflow/src/instructions/condition/getter.ts new file mode 100644 index 000000000..11969f3ef --- /dev/null +++ b/packages/plugin-workflow/src/instructions/condition/getter.ts @@ -0,0 +1,56 @@ +import { get } from 'lodash'; + +import { ModelCtor } from '@nocobase/database'; +import { ExecutionModel } from '../../models/Execution'; + +export type OperandType = 'context' | 'input' | 'job'; + +export type ObjectGetterOptions = { + path?: string +}; + +export type JobGetterOptions = ObjectGetterOptions & { + id: number +}; + +export type ConstantOperand = { + type?: 'constant'; + value: any +}; + +export type ContextOperand = { + type: 'context'; + options: ObjectGetterOptions; +}; + +export type InputOperand = { + type: 'input'; + options: ObjectGetterOptions; +}; + +export type JobOperand = { + type: 'job'; + options: JobGetterOptions; +}; + +export type Operand = ContextOperand | InputOperand | JobOperand | ConstantOperand; + +// TODO: other instructions may also use this method, could be moved to utils. +export function getValue(operand: Operand, input: any, execution: ModelCtor) { + switch (operand.type) { + // from execution context + case 'context': + return get(execution, operand.options.path); + // from input from last job or manual + case 'input': + return get(input, operand.options.path); + // from job in execution + case 'job': + // assume jobs have been fetched from execution before + const job = execution.jobs.find(item => item.id === operand.options.id); + return get(job, operand.options.path); + // constant + default: + return operand.value; + } +} diff --git a/packages/plugin-workflow/src/instructions/condition/index.ts b/packages/plugin-workflow/src/instructions/condition/index.ts new file mode 100644 index 000000000..f2b0be198 --- /dev/null +++ b/packages/plugin-workflow/src/instructions/condition/index.ts @@ -0,0 +1,72 @@ +// config: { +// not: false, +// group: { +// type: 'and', +// calculations: [ +// { +// calculator: 'time.equal', +// operands: [{ type: 'context', options: { path: 'time' } }, { type: 'fn', options: { name: 'newDate', args: [] } }] +// }, +// { +// calculator: 'value.equal', +// operands: [{ type: 'job.result', options: { id: 213, path: '' } }, { type: 'constant', value: { a: 1 } }] +// } +// ] +// } +// } + +import { getValue, Operand } from "./getter"; +import { getCalculator } from "./calculators"; + +type BaseCalculation = { + not?: boolean; +}; + +type SingleCalculation = BaseCalculation & { + calculation: string; + operands?: Operand[]; +}; + +type GroupCalculationOptions = { + type: 'and' | 'or'; + calculations: Calculation[] +}; + +type GroupCalculation = BaseCalculation & { + group: GroupCalculationOptions +}; + +// TODO(type) +type Calculation = SingleCalculation | GroupCalculation; + +function calculate(config, input, execution) { + if (!config) { + return true; + } + + const { not, group } = config; + let result; + if (group) { + const method = group.type === 'and' ? 'every' : 'some'; + result = group.calculations[method](calculation => calculate(calculation, input, execution)); + } else { + const args = config.operands.map(operand => getValue(operand, input, execution)); + const fn = getCalculator(config.calculator); + if (!fn) { + throw new Error(`no calculator function registered for "${config.calculator}"`); + } + result = fn(...args); + } + + return not ? !result : result; +} + + +export default { + manual: false, + async run(this, input, execution) { + // TODO(optimize): loading of jobs could be reduced and turned into incrementally in execution + const jobs = await execution.getJobs(); + return calculate(this.config as Calculation, input, execution); + } +} diff --git a/packages/plugin-workflow/src/instructions/echo.ts b/packages/plugin-workflow/src/instructions/echo.ts new file mode 100644 index 000000000..bd307bce3 --- /dev/null +++ b/packages/plugin-workflow/src/instructions/echo.ts @@ -0,0 +1,6 @@ +export default { + manual: false, + run(this, input, context) { + return input; + } +}; diff --git a/packages/plugin-workflow/src/instructions/index.ts b/packages/plugin-workflow/src/instructions/index.ts new file mode 100644 index 000000000..53df2583c --- /dev/null +++ b/packages/plugin-workflow/src/instructions/index.ts @@ -0,0 +1,37 @@ +// something like template for type of nodes + +import { ModelCtor, Model } from "@nocobase/database"; +import { ExecutionModel } from "../models/Execution"; + +import echo from './echo'; +import prompt from './prompt'; +import condition from './condition'; + +// what should a instruction do? +// - base on input and context, do any calculations or system call (io), and produce a result or pending. +// what should input to be? +// - just use previously output result for convenience? +// what should context to be? +// - could be the workflow execution object (containing context data) +export type Instruction = { + manual: boolean; + run( + this: ModelCtor, + input: any, + execution: ModelCtor + ): any +} + +const registery = new Map(); + +export function getInstruction(key: string): Instruction { + return registery.get(key); +} + +export function registerInstruction(key: string, fn: Instruction) { + registery.set(key, fn); +} + +registerInstruction('echo', echo); +registerInstruction('prompt', prompt); +registerInstruction('condition', condition); diff --git a/packages/plugin-workflow/src/instructions/prompt.ts b/packages/plugin-workflow/src/instructions/prompt.ts new file mode 100644 index 000000000..2d3114675 --- /dev/null +++ b/packages/plugin-workflow/src/instructions/prompt.ts @@ -0,0 +1,6 @@ +export default { + manual: true, + run(this, input, context) { + return input; + } +} diff --git a/packages/plugin-workflow/src/models/Execution.ts b/packages/plugin-workflow/src/models/Execution.ts new file mode 100644 index 000000000..0a0f5fc89 --- /dev/null +++ b/packages/plugin-workflow/src/models/Execution.ts @@ -0,0 +1,140 @@ +import { Model } from '@nocobase/database'; + +import { EXECUTION_STATUS, JOB_STATUS } from '../constants'; +import { getInstruction } from '../instructions'; + +export class ExecutionModel extends Model { + async exec(input, previousJob = null, options = {}) { + // check execution status for quick out + if (this.get('status') !== EXECUTION_STATUS.STARTED) { + return; + } + + let lastJob = previousJob || await this.getLastJob(options); + const node = await this.getNextNode(lastJob); + // if not found any node + if (!node) { + // set execution as resolved + await this.update({ + status: EXECUTION_STATUS.RESOLVED + }); + + return; + } + + // got node.id and node.type + // find node instruction by type from registered node types in memory (program defined) + const instruction = getInstruction(node.type); + + let result = null; + let status = JOB_STATUS.PENDING; + // check if manual or node is on current job + if (!instruction.manual || (lastJob && lastJob.node_id === node.id)) { + // execute instruction of next node and get status + try { + result = await instruction.run.call(node, input ?? lastJob?.result, this); + status = JOB_STATUS.RESOLVED; + } catch(err) { + result = err; + status = JOB_STATUS.REJECTED; + } + } + + // manually exec pending job + if (lastJob && lastJob.node_id === node.id) { + if (lastJob.status !== JOB_STATUS.PENDING) { + // not allow to retry resolved or rejected job for now + // TODO: based on retry config + return; + } + // RUN instruction + // should update the record based on input + lastJob.update({ + status, + result + }); + } else { + // RUN instruction + lastJob = await this.createJob({ + status, + node_id: node.id, + upstream_id: lastJob ? lastJob.id : null, + // TODO: how to presentation error? + result + }); + } + + switch(status) { + case JOB_STATUS.PENDING: + case JOB_STATUS.REJECTED: + // TODO: should handle rejected when configured + return; + default: + // should return chained promise to run any nodes as many as possible, + // till end (pending/rejected/no more) + return this.exec(result, lastJob, options); + } + } + + async getLastJob(options) { + const jobs = await this.getJobs(); + + if (!jobs.length) { + return null; + } + + // find last job, last means no any other jobs set upstream to + const lastJobIds = new Set(jobs.map(item => item.id)); + jobs.forEach(item => { + if (item.upstream_id) { + lastJobIds.delete(item.upstream_id); + } + }); + // TODO(feature): + // if has multiple jobs? which one or some should be run next? + // if has determined flowNodeId, run that one. + // else not supported for now (multiple race pendings) + const [jobId] = Array.from(lastJobIds); + return jobs.find(item => item.id === jobId) || null; + } + + async getNextNode(lastJob) { + if (!this.get('workflow')) { + // cache workflow + this.setDataValue('workflow', await this.getWorkflow()); + } + const workflow = this.get('workflow'); + + // if has not any job, means initial execution + if (!lastJob) { + // find first node for this workflow + // first one is the one has no upstream + const [firstNode = null] = await workflow.getNodes({ + where: { + upstream_id: null + } + }); + + // put firstNode as next node to be execute + return firstNode; + } + + const lastNode = await lastJob.getNode(); + + if (lastJob.status === JOB_STATUS.PENDING) { + return lastNode; + } + + const [nextNode = null] = await workflow.getNodes({ + where: { + upstream_id: lastJob.node_id, + // TODO: need better design + ...(lastNode.type === 'condition' ? { + when: lastJob.result + } : {}) + } + }); + + return nextNode; + } +} diff --git a/packages/plugin-workflow/src/models/Workflow.ts b/packages/plugin-workflow/src/models/Workflow.ts new file mode 100644 index 000000000..79319d02d --- /dev/null +++ b/packages/plugin-workflow/src/models/Workflow.ts @@ -0,0 +1,49 @@ +import { Model } from '@nocobase/database'; + +import { get as getTrigger } from '../triggers'; +import { EXECUTION_STATUS } from '../constants'; + +export class WorkflowModel extends Model { + static async mount() { + const workflows = await this.findAll({ + where: { enabled: true } + }); + + workflows.forEach(workflow => { + workflow.mount(); + }); + + this.addHook('afterCreate', (model: WorkflowModel) => model.mount()); + // TODO: afterUpdate, afterDestroy + } + + async mount() { + if (!this.get('enabled')) { + return; + } + const type = this.get('type'); + const config = this.get('config'); + const trigger = getTrigger(type); + trigger.call(this, config, this.start.bind(this)); + } + + // TODO + async unmount() { + + } + + async start(context: Object, options) { + // `null` means not to trigger + if (context === null) { + return; + } + + const execution = await this.createExecution({ + context, + status: EXECUTION_STATUS.STARTED + }); + execution.setDataValue('workflow', this); + await execution.exec(context, null, options); + return execution; + } +} diff --git a/packages/plugin-workflow/src/server.ts b/packages/plugin-workflow/src/server.ts new file mode 100644 index 000000000..56276d86e --- /dev/null +++ b/packages/plugin-workflow/src/server.ts @@ -0,0 +1,61 @@ +import path from 'path'; + +import { registerModels } from '@nocobase/database'; + +import { WorkflowModel } from './models/Workflow'; +import { ExecutionModel } from './models/Execution'; + +export default { + name: 'workflow', + async load(options = {}) { + const { db } = this.app; + + registerModels({ + WorkflowModel, + ExecutionModel, + }); + + db.import({ + directory: path.resolve(__dirname, 'collections'), + }); + + // [Life Cycle]: + // * load all workflows in db + // * add all hooks for enabled workflows + // * add hooks for create/update[enabled]/delete workflow to add/remove specific hooks + this.app.on('beforeStart', async () => { + const Workflow = db.getModel('workflows'); + await Workflow.mount(); + }) + + // [Life Cycle]: initialize all necessary seed data + this.app.on('db.init', async () => { + + }); + + // const [Automation, AutomationJob] = database.getModels(['automations', 'automations_jobs']); + + // Automation.addHook('afterCreate', async (model: AutomationModel) => { + // model.get('enabled') && await model.loadJobs(); + // }); + + // Automation.addHook('afterUpdate', async (model: AutomationModel) => { + // if (!model.changed('enabled' as any)) { + // return; + // } + // model.get('enabled') ? await model.loadJobs() : await model.cancelJobs(); + // }); + + // Automation.addHook('beforeDestroy', async (model: AutomationModel) => { + // await model.cancelJobs(); + // }); + + // AutomationJob.addHook('afterCreate', async (model: AutomationJobModel) => { + // await model.bootstrap(); + // }); + + // AutomationJob.addHook('beforeDestroy', async (model: AutomationJobModel) => { + // await model.cancel(); + // }); + } +} diff --git a/packages/plugin-workflow/src/triggers/data-change.ts b/packages/plugin-workflow/src/triggers/data-change.ts new file mode 100644 index 000000000..ffc076a4a --- /dev/null +++ b/packages/plugin-workflow/src/triggers/data-change.ts @@ -0,0 +1,10 @@ +export interface IDataChangeTriggerConfig { + collection: string; + // TODO: ICondition + filter: any; +} + +export function afterCreate(config: IDataChangeTriggerConfig, callback: Function) { + const Model = this.database.getModel(config.collection); + Model.addHook('afterCreate', `workflow-${this.get('id')}`, (data: typeof Model, options) => callback({ data }, options)); +} diff --git a/packages/plugin-workflow/src/triggers/index.ts b/packages/plugin-workflow/src/triggers/index.ts new file mode 100644 index 000000000..8b7da7e64 --- /dev/null +++ b/packages/plugin-workflow/src/triggers/index.ts @@ -0,0 +1,21 @@ +import * as dataChangeTriggers from './data-change'; + +export interface ITrigger { + (config: any): void +} + +const triggers = new Map(); + +export function register(type: string, trigger: ITrigger): void { + triggers.set(type, trigger); +} + +export function get(type: string): ITrigger | undefined { + return triggers.get(type); +} + +for (const key in dataChangeTriggers) { + if (dataChangeTriggers.hasOwnProperty(key)) { + register(key, dataChangeTriggers[key]); + } +} From 6018013195ade5eadb816a6ad8cd589e9997f58b Mon Sep 17 00:00:00 2001 From: mytharcher Date: Wed, 26 Jan 2022 02:27:23 +0800 Subject: [PATCH 2/6] feat(plugin-workflow): execution life cycle with branch and join --- .../{workflow.test.ts => execution.test.ts} | 81 +++++- .../plugin-workflow/src/__tests__/index.ts | 18 ++ .../__tests__/instructions/condition.test.ts | 38 ++- .../src/collections/flow_nodes.ts | 39 ++- packages/plugin-workflow/src/constants.ts | 13 +- .../{condition/index.ts => condition.ts} | 59 +++- .../plugin-workflow/src/instructions/echo.ts | 6 - .../plugin-workflow/src/instructions/index.ts | 31 ++- .../src/instructions/prompt.ts | 14 +- .../plugin-workflow/src/models/Execution.ts | 259 ++++++++++-------- .../plugin-workflow/src/models/Workflow.ts | 4 +- .../plugin-workflow/src/triggers/index.ts | 4 +- .../condition => utils}/calculators.ts | 2 +- .../condition => utils}/getter.ts | 2 +- 14 files changed, 394 insertions(+), 176 deletions(-) rename packages/plugin-workflow/src/__tests__/{workflow.test.ts => execution.test.ts} (74%) rename packages/plugin-workflow/src/instructions/{condition/index.ts => condition.ts} (50%) delete mode 100644 packages/plugin-workflow/src/instructions/echo.ts rename packages/plugin-workflow/src/{instructions/condition => utils}/calculators.ts (94%) rename packages/plugin-workflow/src/{instructions/condition => utils}/getter.ts (95%) diff --git a/packages/plugin-workflow/src/__tests__/workflow.test.ts b/packages/plugin-workflow/src/__tests__/execution.test.ts similarity index 74% rename from packages/plugin-workflow/src/__tests__/workflow.test.ts rename to packages/plugin-workflow/src/__tests__/execution.test.ts index a3a27a2c9..1d9e57ff5 100644 --- a/packages/plugin-workflow/src/__tests__/workflow.test.ts +++ b/packages/plugin-workflow/src/__tests__/execution.test.ts @@ -2,11 +2,11 @@ import { Application } from '@nocobase/server'; import Database, { Model, ModelCtor } from '@nocobase/database'; import { getApp } from '.'; import { WorkflowModel } from '../models/Workflow'; -import { EXECUTION_STATUS, JOB_STATUS } from '../constants'; +import { EXECUTION_STATUS, JOB_STATUS, LINK_TYPE } from '../constants'; jest.setTimeout(300000); -describe('workflow', () => { +describe('execution', () => { let app: Application; let db: Database; let PostModel: ModelCtor; @@ -85,11 +85,13 @@ describe('workflow', () => { type: 'echo' }); - await workflow.createNode({ + const n2 = await workflow.createNode({ title: 'echo 2', type: 'echo', upstream_id: n1.id }); + + await n1.setDownstream(n2); const post = await PostModel.create({ title: 't1' }); @@ -104,7 +106,6 @@ describe('workflow', () => { expect(result).toMatchObject({ data: JSON.parse(JSON.stringify(post.toJSON())) }); }); - // TODO: or should throw error? it('execute resolved workflow', async () => { const workflow = await WorkflowModel.create({ title: 'simple workflow', @@ -115,15 +116,20 @@ describe('workflow', () => { } }); + await workflow.createNode({ + title: 'echo', + type: 'echo' + }); + const post = await PostModel.create({ title: 't1' }); const [execution] = await workflow.getExecutions(); expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED); - await execution.exec(123); + expect(execution.start()).rejects.toThrow(); expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED); const jobs = await execution.getJobs(); - expect(jobs.length).toEqual(0); + expect(jobs.length).toEqual(1); }); }); @@ -143,12 +149,14 @@ describe('workflow', () => { type: 'prompt', }); - await workflow.createNode({ + const n2 = await workflow.createNode({ title: 'echo', type: 'echo', upstream_id: n1.id }); + await n1.setDownstream(n2); + const post = await PostModel.create({ title: 't1' }); const [execution] = await workflow.getExecutions(); @@ -157,10 +165,11 @@ describe('workflow', () => { expect(pending.status).toEqual(JOB_STATUS.PENDING); expect(pending.result).toEqual(null); - await execution.exec(123, null, {}); + pending.set('result', 123); + await execution.resume(pending); expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED); - const jobs = await execution.getJobs(); + const jobs = await execution.getJobs({ order: [['id', 'ASC']] }); expect(jobs.length).toEqual(2); expect(jobs[0].status).toEqual(JOB_STATUS.RESOLVED); expect(jobs[0].result).toEqual(123); @@ -186,17 +195,17 @@ describe('workflow', () => { // no config means always true }); - await workflow.createNode({ + const n2 = await workflow.createNode({ title: 'true to echo', type: 'echo', - when: true, + linkType: LINK_TYPE.ON_TRUE, upstream_id: n1.id }); await workflow.createNode({ title: 'false to echo', type: 'echo', - when: false, + linkType: LINK_TYPE.ON_FALSE, upstream_id: n1.id }); @@ -205,9 +214,55 @@ describe('workflow', () => { const [execution] = await workflow.getExecutions(); expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED); - const jobs = await execution.getJobs(); + const jobs = await execution.getJobs({ order: [['id', 'ASC']] }); expect(jobs.length).toEqual(2); + expect(jobs[0].node_id).toEqual(n1.id); + expect(jobs[1].node_id).toEqual(n2.id); expect(jobs[1].result).toEqual(true); }); + + it('suspend downstream in condition branch, then go on', async () => { + const workflow = await WorkflowModel.create({ + title: 'condition workflow', + enabled: true, + type: 'afterCreate', + config: { + collection: 'posts' + } + }); + + const n1 = await workflow.createNode({ + title: 'condition', + type: 'condition', + // no config means always true + }); + + const n2 = await workflow.createNode({ + title: 'manual', + type: 'prompt', + linkType: LINK_TYPE.ON_TRUE, + upstream_id: n1.id + }); + + const n3 = await workflow.createNode({ + title: 'echo input value', + type: 'echo', + upstream_id: n1.id + }); + + await n1.setDownstream(n3); + + const post = await PostModel.create({ title: 't1' }); + + const [execution] = await workflow.getExecutions(); + expect(execution.status).toEqual(EXECUTION_STATUS.STARTED); + + const [pending] = await execution.getJobs({ node_id: n2.id }); + pending.set('result', 123); + await execution.resume(pending); + + const jobs = await execution.getJobs(); + expect(jobs.length).toEqual(3); + }); }); }); diff --git a/packages/plugin-workflow/src/__tests__/index.ts b/packages/plugin-workflow/src/__tests__/index.ts index 9aa44c18e..15e40b047 100644 --- a/packages/plugin-workflow/src/__tests__/index.ts +++ b/packages/plugin-workflow/src/__tests__/index.ts @@ -2,12 +2,30 @@ import path from 'path'; import { MockServer, mockServer } from '@nocobase/test'; import plugin from '../server'; +import { InstructionResult, registerInstruction } from '../instructions'; +import { JOB_STATUS } from '../constants'; export async function getApp(options = {}): Promise { const app = mockServer(options); app.plugin(plugin); + // for test only + registerInstruction('echo', { + run(this, { result }, execution) { + return { + status: JOB_STATUS.RESOLVED, + result + }; + } + }); + + registerInstruction('error', { + run(this, input, execution) { + throw new Error('definite error'); + } + }); + await app.load(); app.db.import({ diff --git a/packages/plugin-workflow/src/__tests__/instructions/condition.test.ts b/packages/plugin-workflow/src/__tests__/instructions/condition.test.ts index d91afdba0..1581bfb12 100644 --- a/packages/plugin-workflow/src/__tests__/instructions/condition.test.ts +++ b/packages/plugin-workflow/src/__tests__/instructions/condition.test.ts @@ -2,7 +2,7 @@ import { Application } from '@nocobase/server'; import Database, { Model, ModelCtor } from '@nocobase/database'; import { getApp } from '..'; import { WorkflowModel } from '../../models/Workflow'; -import { EXECUTION_STATUS, JOB_STATUS } from '../../constants'; +import { EXECUTION_STATUS, JOB_STATUS, LINK_TYPE } from '../../constants'; @@ -22,6 +22,10 @@ describe('workflow > instructions > condition', () => { afterEach(() => db.close()); + describe('config.rejectOnFalse', () => { + + }); + describe('single calculation', () => { it('calculation to true downstream', async () => { const workflow = await WorkflowModel.create({ @@ -36,24 +40,26 @@ describe('workflow > instructions > condition', () => { const n1 = await workflow.createNode({ title: 'condition', type: 'condition', - // (1 === 1): true config: { - calculator: 'equal', - operands: [{ value: 1 }, { value: 1 }] + // (1 === 1): true + calculation: { + calculator: 'equal', + operands: [{ value: 1 }, { value: 1 }] + } } }); - await workflow.createNode({ + const n2 = await workflow.createNode({ title: 'true to echo', type: 'echo', - when: true, + linkType: LINK_TYPE.ON_TRUE, upstream_id: n1.id }); - await workflow.createNode({ + const n3 = await workflow.createNode({ title: 'false to echo', type: 'echo', - when: false, + linkType: LINK_TYPE.ON_FALSE, upstream_id: n1.id }); @@ -80,24 +86,26 @@ describe('workflow > instructions > condition', () => { const n1 = await workflow.createNode({ title: 'condition', type: 'condition', - // (0 === 1): false config: { - calculator: 'equal', - operands: [{ value: 0 }, { value: 1 }] + // (0 === 1): false + calculation: { + calculator: 'equal', + operands: [{ value: 0 }, { value: 1 }] + } } }); await workflow.createNode({ title: 'true to echo', type: 'echo', - when: true, + linkType: LINK_TYPE.ON_TRUE, upstream_id: n1.id }); await workflow.createNode({ title: 'false to echo', type: 'echo', - when: false, + linkType: LINK_TYPE.ON_FALSE, upstream_id: n1.id }); @@ -111,4 +119,8 @@ describe('workflow > instructions > condition', () => { expect(jobs[1].result).toEqual(false); }); }); + + describe('group calculation', () => { + + }); }); diff --git a/packages/plugin-workflow/src/collections/flow_nodes.ts b/packages/plugin-workflow/src/collections/flow_nodes.ts index b6e1a057a..8ebf1f766 100644 --- a/packages/plugin-workflow/src/collections/flow_nodes.ts +++ b/packages/plugin-workflow/src/collections/flow_nodes.ts @@ -1,4 +1,5 @@ import { TableOptions } from '@nocobase/database'; +import { LINK_TYPE } from '../constants'; export default { name: 'flow_nodes', @@ -28,21 +29,46 @@ export default { type: 'belongsTo', target: 'flow_nodes' }, - // only works when upstream node is condition type. + { + interface: 'linkTo', + name: 'branches', + type: 'hasMany', + target: 'flow_nodes', + sourceKey: 'id', + foreignKey: 'upstream_id', + }, + // only works when upstream node is branching type, like condition and parallel. // put here because the design of flow-links model is not really necessary for now. // or it should be put into flow-links model. + // if keeps 1:n relactionship, cannot support cycle flow. { - name: 'when', - type: 'boolean', - // defaultValue: null + interface: 'select', + name: 'linkType', + type: 'smallint', + title: 'Link Type', + dataSource: [ + { label: 'Default', value: LINK_TYPE.DEFAULT }, + { label: 'Branched, on true', value: LINK_TYPE.ON_TRUE }, + { label: 'Branched, on false', value: LINK_TYPE.ON_FALSE }, + { label: 'Branched, no limit', value: LINK_TYPE.NO_LIMIT } + ] + }, + // for reasons: + // 1. redirect type node to solve cycle flow. + // 2. recognize as true next node after branches. + { + interface: 'linkTo', + name: 'downstream', + type: 'belongsTo', + target: 'flow_nodes' }, { interface: 'select', type: 'string', name: 'type', title: '类型', + // TODO: data for test only now dataSource: [ - { label: '无处理', value: 'echo' }, { label: '数据处理', value: 'data' }, { label: '数据查询', value: 'query' }, { label: '等待人工输入', value: 'prompt' }, @@ -53,7 +79,8 @@ export default { interface: 'json', type: 'jsonb', name: 'config', - title: '配置' + title: '配置', + defaultValue: {} } ] } as TableOptions; diff --git a/packages/plugin-workflow/src/constants.ts b/packages/plugin-workflow/src/constants.ts index 4f140c0ea..ecc2ed6fc 100644 --- a/packages/plugin-workflow/src/constants.ts +++ b/packages/plugin-workflow/src/constants.ts @@ -1,11 +1,20 @@ export const EXECUTION_STATUS = { STARTED: 0, RESOLVED: 1, - REJECTED: -1 + REJECTED: -1, + CANCELLED: -2 }; export const JOB_STATUS = { PENDING: 0, RESOLVED: 1, - REJECTED: -1 + REJECTED: -1, + CANCELLED: -2 +}; + +export const LINK_TYPE = { + DEFAULT: null, + ON_TRUE: 1, + ON_FALSE: 0, + NO_LIMIT: -1 }; diff --git a/packages/plugin-workflow/src/instructions/condition/index.ts b/packages/plugin-workflow/src/instructions/condition.ts similarity index 50% rename from packages/plugin-workflow/src/instructions/condition/index.ts rename to packages/plugin-workflow/src/instructions/condition.ts index f2b0be198..832ded4f3 100644 --- a/packages/plugin-workflow/src/instructions/condition/index.ts +++ b/packages/plugin-workflow/src/instructions/condition.ts @@ -15,8 +15,10 @@ // } // } -import { getValue, Operand } from "./getter"; -import { getCalculator } from "./calculators"; +import Sequelize = require('sequelize'); +import { getValue, Operand } from "../utils/getter"; +import { getCalculator } from "../utils/calculators"; +import { JOB_STATUS } from "../constants"; type BaseCalculation = { not?: boolean; @@ -63,10 +65,55 @@ function calculate(config, input, execution) { export default { - manual: false, - async run(this, input, execution) { + async run(this, prevJob, execution) { // TODO(optimize): loading of jobs could be reduced and turned into incrementally in execution - const jobs = await execution.getJobs(); - return calculate(this.config as Calculation, input, execution); + // const jobs = await execution.getJobs(); + const { calculation } = this.config || {}; + const result = calculate(calculation, prevJob, execution); + + if (!result && this.config.rejectOnFalse) { + return { + status: JOB_STATUS.REJECTED, + result + }; + } + + const job = { + status: JOB_STATUS.RESOLVED, + result, + // TODO(optimize): try unify the building of job + node_id: this.id, + upstream_id: prevJob instanceof Sequelize.Model ? prevJob.get('id') : null + }; + + const branchNode = execution.nodes + .find(item => item.upstream === this && item.linkType === Number(result)); + + if (!branchNode) { + return job; + } + + const savedJob = await execution.saveJob(job); + + // return execution.exec(branchNode, savedJob); + const tailJob = await execution.exec(branchNode, savedJob); + + if (tailJob.status === JOB_STATUS.PENDING) { + savedJob.set('status', JOB_STATUS.PENDING); + return savedJob; + } + + return tailJob; + }, + + async resume(this, branchJob, execution) { + if (branchJob.status === JOB_STATUS.RESOLVED) { + const job = execution.findBranchParentJob(branchJob, this); + job.set('status', JOB_STATUS.RESOLVED); + return job; + } + + // pass control to upper scope by ending current scope + return execution.end(this, branchJob); } } diff --git a/packages/plugin-workflow/src/instructions/echo.ts b/packages/plugin-workflow/src/instructions/echo.ts deleted file mode 100644 index bd307bce3..000000000 --- a/packages/plugin-workflow/src/instructions/echo.ts +++ /dev/null @@ -1,6 +0,0 @@ -export default { - manual: false, - run(this, input, context) { - return input; - } -}; diff --git a/packages/plugin-workflow/src/instructions/index.ts b/packages/plugin-workflow/src/instructions/index.ts index 53df2583c..1c9722a54 100644 --- a/packages/plugin-workflow/src/instructions/index.ts +++ b/packages/plugin-workflow/src/instructions/index.ts @@ -3,23 +3,32 @@ import { ModelCtor, Model } from "@nocobase/database"; import { ExecutionModel } from "../models/Execution"; -import echo from './echo'; import prompt from './prompt'; import condition from './condition'; +// import parallel from './parallel'; + +export interface Job { + status: number; + result: unknown; + [key: string]: unknown; +} + +export type InstructionResult = Job | Promise; // what should a instruction do? // - base on input and context, do any calculations or system call (io), and produce a result or pending. -// what should input to be? -// - just use previously output result for convenience? -// what should context to be? -// - could be the workflow execution object (containing context data) -export type Instruction = { - manual: boolean; +export interface Instruction { run( this: ModelCtor, + // what should input to be? + // - just use previously output result for convenience? input: any, + // what should context to be? + // - could be the workflow execution object (containing context data) execution: ModelCtor - ): any + ): InstructionResult; + // for start node in main flow (or branch) to resume when manual sub branch triggered + resume?(): InstructionResult } const registery = new Map(); @@ -28,10 +37,10 @@ export function getInstruction(key: string): Instruction { return registery.get(key); } -export function registerInstruction(key: string, fn: Instruction) { - registery.set(key, fn); +export function registerInstruction(key: string, instruction: any) { + registery.set(key, instruction); } -registerInstruction('echo', echo); registerInstruction('prompt', prompt); registerInstruction('condition', condition); +// registerInstruction('parallel', parallel); diff --git a/packages/plugin-workflow/src/instructions/prompt.ts b/packages/plugin-workflow/src/instructions/prompt.ts index 2d3114675..ec3455eb4 100644 --- a/packages/plugin-workflow/src/instructions/prompt.ts +++ b/packages/plugin-workflow/src/instructions/prompt.ts @@ -1,6 +1,14 @@ +import { JOB_STATUS } from "../constants"; + export default { - manual: true, - run(this, input, context) { - return input; + run(this, input, execution) { + return { + status: JOB_STATUS.PENDING + }; + }, + + resume(this, job, execution) { + job.set('status', JOB_STATUS.RESOLVED); + return job; } } diff --git a/packages/plugin-workflow/src/models/Execution.ts b/packages/plugin-workflow/src/models/Execution.ts index 0a0f5fc89..78ee0f29a 100644 --- a/packages/plugin-workflow/src/models/Execution.ts +++ b/packages/plugin-workflow/src/models/Execution.ts @@ -1,140 +1,175 @@ -import { Model } from '@nocobase/database'; +import Sequelize from 'sequelize'; +import { Model, ModelCtor } from '@nocobase/database'; import { EXECUTION_STATUS, JOB_STATUS } from '../constants'; import { getInstruction } from '../instructions'; export class ExecutionModel extends Model { - async exec(input, previousJob = null, options = {}) { - // check execution status for quick out - if (this.get('status') !== EXECUTION_STATUS.STARTED) { - return; - } + nodes: Array = []; + nodesMap = new Map(); + jobsMap = new Map(); - let lastJob = previousJob || await this.getLastJob(options); - const node = await this.getNextNode(lastJob); - // if not found any node - if (!node) { - // set execution as resolved - await this.update({ - status: EXECUTION_STATUS.RESOLVED - }); + // make dual linked nodes list then cache + makeNodes(nodes = []) { + this.nodes = nodes; - return; - } + nodes.forEach(node => { + this.nodesMap.set(node.id, node); + }); - // got node.id and node.type - // find node instruction by type from registered node types in memory (program defined) - const instruction = getInstruction(node.type); - - let result = null; - let status = JOB_STATUS.PENDING; - // check if manual or node is on current job - if (!instruction.manual || (lastJob && lastJob.node_id === node.id)) { - // execute instruction of next node and get status - try { - result = await instruction.run.call(node, input ?? lastJob?.result, this); - status = JOB_STATUS.RESOLVED; - } catch(err) { - result = err; - status = JOB_STATUS.REJECTED; + nodes.forEach(node => { + if (node.upstream_id) { + node.upstream = this.nodesMap.get(node.upstream_id); } - } - // manually exec pending job - if (lastJob && lastJob.node_id === node.id) { - if (lastJob.status !== JOB_STATUS.PENDING) { - // not allow to retry resolved or rejected job for now - // TODO: based on retry config - return; + if (node.downstream_id) { + node.downstream = this.nodesMap.get(node.downstream_id); } - // RUN instruction - // should update the record based on input - lastJob.update({ - status, - result - }); - } else { - // RUN instruction - lastJob = await this.createJob({ - status, - node_id: node.id, - upstream_id: lastJob ? lastJob.id : null, - // TODO: how to presentation error? - result - }); - } - - switch(status) { - case JOB_STATUS.PENDING: - case JOB_STATUS.REJECTED: - // TODO: should handle rejected when configured - return; - default: - // should return chained promise to run any nodes as many as possible, - // till end (pending/rejected/no more) - return this.exec(result, lastJob, options); - } + }); } - async getLastJob(options) { + makeJobs(jobs: Array>) { + jobs.forEach(job => { + this.jobsMap.set(job.id, job); + }); + } + + async prepare() { + if (this.status !== EXECUTION_STATUS.STARTED) { + throw new Error(`execution was ended with status ${this.status}`); + } + + if (!this.workflow) { + this.workflow = await this.getWorkflow(); + } + + const nodes = await this.workflow.getNodes(); + + this.makeNodes(nodes); + const jobs = await this.getJobs(); - - if (!jobs.length) { - return null; - } - // find last job, last means no any other jobs set upstream to - const lastJobIds = new Set(jobs.map(item => item.id)); - jobs.forEach(item => { - if (item.upstream_id) { - lastJobIds.delete(item.upstream_id); - } - }); - // TODO(feature): - // if has multiple jobs? which one or some should be run next? - // if has determined flowNodeId, run that one. - // else not supported for now (multiple race pendings) - const [jobId] = Array.from(lastJobIds); - return jobs.find(item => item.id === jobId) || null; + this.makeJobs(jobs); } - async getNextNode(lastJob) { - if (!this.get('workflow')) { - // cache workflow - this.setDataValue('workflow', await this.getWorkflow()); + async start(options) { + await this.prepare(); + if (!this.nodes.length) { + return this.exit(null); } - const workflow = this.get('workflow'); + const head = this.nodes.find(item => !item.upstream); + return this.exec(head, { result: this.context }); + } - // if has not any job, means initial execution - if (!lastJob) { - // find first node for this workflow - // first one is the one has no upstream - const [firstNode = null] = await workflow.getNodes({ - where: { - upstream_id: null - } + async resume(job, options) { + await this.prepare(); + const node = this.nodesMap.get(job.node_id); + return this.recall(node, job); + } + + private async run(instruction, node, prevJob) { + let job; + try { + // call instruction to get result and status + job = await instruction.call(node, prevJob, this); + } catch (err) { + console.error(err); + // for uncaught error, set to rejected + job = { + result: err, + status: JOB_STATUS.REJECTED + }; + } + + let savedJob; + if (job instanceof Sequelize.Model) { + savedJob = await job.save(); + } else { + const upstream_id = prevJob instanceof Sequelize.Model ? prevJob.get('id') : null; + savedJob = await this.saveJob({ + node_id: node.id, + upstream_id, + ...job }); - - // put firstNode as next node to be execute - return firstNode; } - const lastNode = await lastJob.getNode(); - - if (lastJob.status === JOB_STATUS.PENDING) { - return lastNode; + if (savedJob.get('status') === JOB_STATUS.RESOLVED && node.downstream) { + // run next node + return this.exec(node.downstream, savedJob); } - const [nextNode = null] = await workflow.getNodes({ - where: { - upstream_id: lastJob.node_id, - // TODO: need better design - ...(lastNode.type === 'condition' ? { - when: lastJob.result - } : {}) - } + // all nodes in scope have been executed + return this.end(node, savedJob); + } + + async exec(node, input?) { + const { run } = getInstruction(node.type); + + return this.run(run, node, input); + } + + // parent node should take over the control + end(node, job) { + const parentNode = this.findBranchParentNode(node); + // no parent, means on main flow + if (parentNode) { + return this.recall(parentNode, job); + } + + // really done for all nodes + // * should mark execution as done with last job status + return this.exit(job); + } + + async recall(node, job) { + const { resume } = getInstruction(node.type); + if (!resume) { + return Promise.reject(new Error('`resume` should be implemented because the node made branch')); + } + + return this.run(resume, node, job); + } + + async exit(job) { + const executionStatusMap = { + [JOB_STATUS.PENDING]: EXECUTION_STATUS.STARTED, + [JOB_STATUS.RESOLVED]: EXECUTION_STATUS.RESOLVED, + [JOB_STATUS.REJECTED]: EXECUTION_STATUS.REJECTED, + [JOB_STATUS.CANCELLED]: EXECUTION_STATUS.CANCELLED, + }; + const status = job ? executionStatusMap[job.status] : EXECUTION_STATUS.RESOLVED; + await this.update({ status }); + return job; + } + + // TODO(optimize) + async saveJob(payload) { + const JobModel = this.database.getModel('jobs'); + const [result] = await JobModel.upsert({ + ...payload, + execution_id: this.id }); - return nextNode; + this.jobsMap.set(result.id, result); + + return result; + } + + findBranchParentNode(node): any { + for (let n = node; n; n = n.upstream) { + if (n.linkType !== null) { + return n.upstream; + } + } + return null; + } + + findBranchParentJob(job, node) { + for (let j = job; j; j = this.jobsMap.get(j.upstream_id)) { + if (j.node_id === node.id) { + return j; + } + } + return null; } } diff --git a/packages/plugin-workflow/src/models/Workflow.ts b/packages/plugin-workflow/src/models/Workflow.ts index 79319d02d..ebc7aef4e 100644 --- a/packages/plugin-workflow/src/models/Workflow.ts +++ b/packages/plugin-workflow/src/models/Workflow.ts @@ -43,7 +43,9 @@ export class WorkflowModel extends Model { status: EXECUTION_STATUS.STARTED }); execution.setDataValue('workflow', this); - await execution.exec(context, null, options); + execution.workflow = this; + + await execution.start(null, null, options); return execution; } } diff --git a/packages/plugin-workflow/src/triggers/index.ts b/packages/plugin-workflow/src/triggers/index.ts index 8b7da7e64..99751e308 100644 --- a/packages/plugin-workflow/src/triggers/index.ts +++ b/packages/plugin-workflow/src/triggers/index.ts @@ -1,7 +1,9 @@ +import { ModelCtor } from '@nocobase/database'; +import { WorkflowModel } from '../models/Workflow'; import * as dataChangeTriggers from './data-change'; export interface ITrigger { - (config: any): void + (this: ModelCtor, config: any): void } const triggers = new Map(); diff --git a/packages/plugin-workflow/src/instructions/condition/calculators.ts b/packages/plugin-workflow/src/utils/calculators.ts similarity index 94% rename from packages/plugin-workflow/src/instructions/condition/calculators.ts rename to packages/plugin-workflow/src/utils/calculators.ts index d15d14a8f..33cfd16fe 100644 --- a/packages/plugin-workflow/src/instructions/condition/calculators.ts +++ b/packages/plugin-workflow/src/utils/calculators.ts @@ -1,4 +1,4 @@ -type Calculator = (...args: any[]) => boolean; +type Calculator = (...args: any[]) => any; const calculators = new Map(); diff --git a/packages/plugin-workflow/src/instructions/condition/getter.ts b/packages/plugin-workflow/src/utils/getter.ts similarity index 95% rename from packages/plugin-workflow/src/instructions/condition/getter.ts rename to packages/plugin-workflow/src/utils/getter.ts index 11969f3ef..77a442291 100644 --- a/packages/plugin-workflow/src/instructions/condition/getter.ts +++ b/packages/plugin-workflow/src/utils/getter.ts @@ -1,7 +1,7 @@ import { get } from 'lodash'; import { ModelCtor } from '@nocobase/database'; -import { ExecutionModel } from '../../models/Execution'; +import { ExecutionModel } from '../models/Execution'; export type OperandType = 'context' | 'input' | 'job'; From 42490473189028e428261f25597d7a5e758564d5 Mon Sep 17 00:00:00 2001 From: mytharcher Date: Wed, 26 Jan 2022 17:46:22 +0800 Subject: [PATCH 3/6] fix(plugin-workflow): test for error job --- .../src/__tests__/execution.test.ts | 185 ++++++++++-------- .../plugin-workflow/src/__tests__/index.ts | 13 +- .../__tests__/instructions/condition.test.ts | 27 +-- .../plugin-workflow/src/models/Execution.ts | 10 +- 4 files changed, 141 insertions(+), 94 deletions(-) diff --git a/packages/plugin-workflow/src/__tests__/execution.test.ts b/packages/plugin-workflow/src/__tests__/execution.test.ts index 1d9e57ff5..ae26107cb 100644 --- a/packages/plugin-workflow/src/__tests__/execution.test.ts +++ b/packages/plugin-workflow/src/__tests__/execution.test.ts @@ -1,7 +1,6 @@ import { Application } from '@nocobase/server'; -import Database, { Model, ModelCtor } from '@nocobase/database'; +import Database from '@nocobase/database'; import { getApp } from '.'; -import { WorkflowModel } from '../models/Workflow'; import { EXECUTION_STATUS, JOB_STATUS, LINK_TYPE } from '../constants'; jest.setTimeout(300000); @@ -9,32 +8,32 @@ jest.setTimeout(300000); describe('execution', () => { let app: Application; let db: Database; - let PostModel: ModelCtor; - // let Target: ModelCtor; - let WorkflowModel: ModelCtor; + let PostModel; + let WorkflowModel; + let workflow; beforeEach(async () => { app = await getApp(); db = app.db; - WorkflowModel = db.getModel('workflows') as any; + WorkflowModel = db.getModel('workflows'); PostModel = db.getModel('posts'); // Target = db.getModel('targets'); + + workflow = await WorkflowModel.create({ + title: 'test workflow', + enabled: true, + type: 'afterCreate', + config: { + collection: 'posts' + } + }); }); afterEach(() => db.close()); describe('base', () => { it('empty workflow without any nodes', async () => { - const workflow = await WorkflowModel.create({ - title: 'empty workflow', - enabled: true, - type: 'afterCreate', - config: { - collection: 'posts' - } - }); - const post = await PostModel.create({ title: 't1' }); const [execution] = await workflow.getExecutions(); @@ -42,16 +41,24 @@ describe('execution', () => { expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED); }); - it('workflow with single simple node', async () => { - const workflow = await WorkflowModel.create({ - title: 'simple workflow', - enabled: true, - type: 'afterCreate', - config: { - collection: 'posts' - } + it('execute resolved workflow', async () => { + await workflow.createNode({ + title: 'echo', + type: 'echo' }); + const post = await PostModel.create({ title: 't1' }); + + const [execution] = await workflow.getExecutions(); + expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED); + + expect(execution.start()).rejects.toThrow(); + expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED); + const jobs = await execution.getJobs(); + expect(jobs.length).toEqual(1); + }); + + it('workflow with single simple node', async () => { await workflow.createNode({ title: 'echo', type: 'echo' @@ -71,15 +78,6 @@ describe('execution', () => { }); it('workflow with multiple simple nodes', async () => { - const workflow = await WorkflowModel.create({ - title: 'simple workflow', - enabled: true, - type: 'afterCreate', - config: { - collection: 'posts' - } - }); - const n1 = await workflow.createNode({ title: 'echo 1', type: 'echo' @@ -106,44 +104,27 @@ describe('execution', () => { expect(result).toMatchObject({ data: JSON.parse(JSON.stringify(post.toJSON())) }); }); - it('execute resolved workflow', async () => { - const workflow = await WorkflowModel.create({ - title: 'simple workflow', - enabled: true, - type: 'afterCreate', - config: { - collection: 'posts' - } - }); - + it('workflow with error node', async () => { await workflow.createNode({ - title: 'echo', - type: 'echo' + title: 'error', + type: 'error' }); const post = await PostModel.create({ title: 't1' }); - - const [execution] = await workflow.getExecutions(); - expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED); - expect(execution.start()).rejects.toThrow(); - expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED); + const [execution] = await workflow.getExecutions(); + expect(execution.status).toEqual(EXECUTION_STATUS.REJECTED); + const jobs = await execution.getJobs(); expect(jobs.length).toEqual(1); + const { status, result } = jobs[0].get(); + expect(status).toEqual(JOB_STATUS.REJECTED); + expect(result).toBe('Error: definite error'); }); }); describe('manual nodes', () => { - it('manual node should pause execution, and could be manually resume', async () => { - const workflow = await WorkflowModel.create({ - title: 'manual workflow', - enabled: true, - type: 'afterCreate', - config: { - collection: 'posts' - } - }); - + it('manual node should suspend execution, and could be manually resume', async () => { const n1 = await workflow.createNode({ title: 'prompt', type: 'prompt', @@ -176,19 +157,40 @@ describe('execution', () => { expect(jobs[1].status).toEqual(JOB_STATUS.RESOLVED); expect(jobs[1].result).toEqual(123); }); + + it('manual node should suspend execution, resuming with error should end execution', async () => { + const n1 = await workflow.createNode({ + title: 'prompt error', + type: 'prompt->error', + }); + const n2 = await workflow.createNode({ + title: 'echo', + type: 'echo', + upstream_id: n1.id + }); + await n1.setDownstream(n2); + + const post = await PostModel.create({ title: 't1' }); + + const [execution] = await workflow.getExecutions(); + expect(execution.status).toEqual(EXECUTION_STATUS.STARTED); + const [pending] = await execution.getJobs(); + expect(pending.status).toEqual(JOB_STATUS.PENDING); + expect(pending.result).toEqual(null); + + pending.set('result', 123); + await execution.resume(pending); + expect(execution.status).toEqual(EXECUTION_STATUS.REJECTED); + + 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'); + }); }); - describe('condition node', () => { + describe('branch: condition', () => { it('condition node link to different downstreams', async () => { - const workflow = await WorkflowModel.create({ - title: 'condition workflow', - enabled: true, - type: 'afterCreate', - config: { - collection: 'posts' - } - }); - const n1 = await workflow.createNode({ title: 'condition', type: 'condition', @@ -222,15 +224,6 @@ describe('execution', () => { }); it('suspend downstream in condition branch, then go on', async () => { - const workflow = await WorkflowModel.create({ - title: 'condition workflow', - enabled: true, - type: 'afterCreate', - config: { - collection: 'posts' - } - }); - const n1 = await workflow.createNode({ title: 'condition', type: 'condition', @@ -264,5 +257,41 @@ describe('execution', () => { const jobs = await execution.getJobs(); expect(jobs.length).toEqual(3); }); + + it('resume error downstream in condition branch, should reject', async () => { + const n1 = await workflow.createNode({ + title: 'condition', + type: 'condition', + // no config means always true + }); + + const n2 = await workflow.createNode({ + title: 'manual', + type: 'prompt->error', + linkType: LINK_TYPE.ON_TRUE, + upstream_id: n1.id + }); + + const n3 = await workflow.createNode({ + title: 'echo input value', + type: 'echo', + upstream_id: n1.id + }); + + await n1.setDownstream(n3); + + const post = await PostModel.create({ title: 't1' }); + + const [execution] = await workflow.getExecutions(); + expect(execution.status).toEqual(EXECUTION_STATUS.STARTED); + + const [pending] = await execution.getJobs({ node_id: n2.id }); + pending.set('result', 123); + await execution.resume(pending); + expect(execution.status).toEqual(EXECUTION_STATUS.REJECTED); + + const jobs = await execution.getJobs(); + expect(jobs.length).toEqual(2); + }); }); }); diff --git a/packages/plugin-workflow/src/__tests__/index.ts b/packages/plugin-workflow/src/__tests__/index.ts index 15e40b047..d799652b9 100644 --- a/packages/plugin-workflow/src/__tests__/index.ts +++ b/packages/plugin-workflow/src/__tests__/index.ts @@ -2,7 +2,7 @@ import path from 'path'; import { MockServer, mockServer } from '@nocobase/test'; import plugin from '../server'; -import { InstructionResult, registerInstruction } from '../instructions'; +import { registerInstruction } from '../instructions'; import { JOB_STATUS } from '../constants'; export async function getApp(options = {}): Promise { @@ -26,6 +26,17 @@ export async function getApp(options = {}): Promise { } }); + registerInstruction('prompt->error', { + run(this, input, execution) { + return { + status: JOB_STATUS.PENDING + }; + }, + resume(this, input, execution) { + throw new Error('input failed'); + } + }); + await app.load(); app.db.import({ diff --git a/packages/plugin-workflow/src/__tests__/instructions/condition.test.ts b/packages/plugin-workflow/src/__tests__/instructions/condition.test.ts index 1581bfb12..b9a1c1c4a 100644 --- a/packages/plugin-workflow/src/__tests__/instructions/condition.test.ts +++ b/packages/plugin-workflow/src/__tests__/instructions/condition.test.ts @@ -1,7 +1,6 @@ import { Application } from '@nocobase/server'; -import Database, { Model, ModelCtor } from '@nocobase/database'; +import Database from '@nocobase/database'; import { getApp } from '..'; -import { WorkflowModel } from '../../models/Workflow'; import { EXECUTION_STATUS, JOB_STATUS, LINK_TYPE } from '../../constants'; @@ -9,15 +8,25 @@ import { EXECUTION_STATUS, JOB_STATUS, LINK_TYPE } from '../../constants'; describe('workflow > instructions > condition', () => { let app: Application; let db: Database; - let PostModel: ModelCtor; - let WorkflowModel: ModelCtor; + let PostModel; + let WorkflowModel; + let workflow; beforeEach(async () => { app = await getApp(); db = app.db; - WorkflowModel = db.getModel('workflows') as any; + WorkflowModel = db.getModel('workflows'); PostModel = db.getModel('posts'); + + workflow = await WorkflowModel.create({ + title: 'condition workflow', + enabled: true, + type: 'afterCreate', + config: { + collection: 'posts' + } + }); }); afterEach(() => db.close()); @@ -28,14 +37,6 @@ describe('workflow > instructions > condition', () => { describe('single calculation', () => { it('calculation to true downstream', async () => { - const workflow = await WorkflowModel.create({ - title: 'condition workflow', - enabled: true, - type: 'afterCreate', - config: { - collection: 'posts' - } - }); const n1 = await workflow.createNode({ title: 'condition', diff --git a/packages/plugin-workflow/src/models/Execution.ts b/packages/plugin-workflow/src/models/Execution.ts index 78ee0f29a..2398b8b57 100644 --- a/packages/plugin-workflow/src/models/Execution.ts +++ b/packages/plugin-workflow/src/models/Execution.ts @@ -73,15 +73,21 @@ export class ExecutionModel extends Model { // call instruction to get result and status job = await instruction.call(node, prevJob, this); } catch (err) { - console.error(err); // for uncaught error, set to rejected job = { - result: err, + result: err instanceof Error ? err.toString() : err, status: JOB_STATUS.REJECTED }; + // if previous job is from resuming + if (prevJob && prevJob.node_id === node.id) { + prevJob.set(job); + job = prevJob; + } } let savedJob; + // TODO(optimize): many checking of resuming or new could be improved + // could be implemented separately in exec() / resume() if (job instanceof Sequelize.Model) { savedJob = await job.save(); } else { From e592d03f1816c98dbcf8a50d6dcde876ecfca1b6 Mon Sep 17 00:00:00 2001 From: mytharcher Date: Fri, 28 Jan 2022 00:25:26 +0800 Subject: [PATCH 4/6] chore(plugin-workflow): migrate from 0.5 to 0.6 --- packages/plugin-workflow/package.json | 6 +- .../src/__tests__/collections/posts.ts | 4 +- .../src/__tests__/collections/targets.ts | 4 +- .../src/__tests__/execution.test.ts | 60 ++++++------ .../plugin-workflow/src/__tests__/index.ts | 6 +- .../__tests__/instructions/condition.test.ts | 38 ++++---- .../src/collections/executions.ts | 4 +- .../src/collections/flow_nodes.ts | 18 +--- .../plugin-workflow/src/collections/jobs.ts | 4 +- .../src/collections/workflows.ts | 4 +- packages/plugin-workflow/src/constants.ts | 5 +- .../src/instructions/condition.ts | 6 +- .../plugin-workflow/src/instructions/index.ts | 10 +- .../plugin-workflow/src/models/Execution.ts | 96 +++++++++++++------ .../plugin-workflow/src/models/FlowNode.ts | 19 ++++ packages/plugin-workflow/src/models/Job.ts | 18 ++++ .../plugin-workflow/src/models/Workflow.ts | 38 ++++++-- packages/plugin-workflow/src/server.ts | 14 ++- .../src/triggers/data-change.ts | 10 +- .../plugin-workflow/src/triggers/index.ts | 5 +- packages/plugin-workflow/src/utils/getter.ts | 7 +- 21 files changed, 236 insertions(+), 140 deletions(-) create mode 100644 packages/plugin-workflow/src/models/FlowNode.ts create mode 100644 packages/plugin-workflow/src/models/Job.ts diff --git a/packages/plugin-workflow/package.json b/packages/plugin-workflow/package.json index 70927033c..e44e1e36f 100644 --- a/packages/plugin-workflow/package.json +++ b/packages/plugin-workflow/package.json @@ -1,16 +1,12 @@ { "name": "@nocobase/plugin-workflow", - "version": "0.5.0-alpha.37", + "version": "0.6.0-alpha.0", "main": "lib/index.js", "private": true, "license": "MIT", "dependencies": { - "@nocobase/server": "^0.5.0-alpha.37", - "json-templates": "^4.1.0", - "node-schedule": "^2.0.0" }, "devDependencies": { - "@types/node-schedule": "^1.3.1" }, "gitHead": "f0b335ac30f29f25c95d7d137655fa64d8d67f1e" } diff --git a/packages/plugin-workflow/src/__tests__/collections/posts.ts b/packages/plugin-workflow/src/__tests__/collections/posts.ts index efa6cb228..021dd12ae 100644 --- a/packages/plugin-workflow/src/__tests__/collections/posts.ts +++ b/packages/plugin-workflow/src/__tests__/collections/posts.ts @@ -1,4 +1,4 @@ -import { TableOptions } from '@nocobase/database'; +import { CollectionOptions } from '@nocobase/database'; export default { name: 'posts', @@ -12,4 +12,4 @@ export default { name: 'published', } ] -} as TableOptions; +} as CollectionOptions; diff --git a/packages/plugin-workflow/src/__tests__/collections/targets.ts b/packages/plugin-workflow/src/__tests__/collections/targets.ts index d1b6ed448..525e091c6 100644 --- a/packages/plugin-workflow/src/__tests__/collections/targets.ts +++ b/packages/plugin-workflow/src/__tests__/collections/targets.ts @@ -1,4 +1,4 @@ -import { TableOptions } from '@nocobase/database'; +import { CollectionOptions } from '@nocobase/database'; export default { name: 'targets', @@ -12,4 +12,4 @@ export default { name: 'col2', } ], -} as TableOptions; +} as CollectionOptions; diff --git a/packages/plugin-workflow/src/__tests__/execution.test.ts b/packages/plugin-workflow/src/__tests__/execution.test.ts index ae26107cb..dbddc1457 100644 --- a/packages/plugin-workflow/src/__tests__/execution.test.ts +++ b/packages/plugin-workflow/src/__tests__/execution.test.ts @@ -1,7 +1,7 @@ import { Application } from '@nocobase/server'; import Database from '@nocobase/database'; import { getApp } from '.'; -import { EXECUTION_STATUS, JOB_STATUS, LINK_TYPE } from '../constants'; +import { BRANCH_INDEX, EXECUTION_STATUS, JOB_STATUS } from '../constants'; jest.setTimeout(300000); @@ -10,22 +10,28 @@ describe('execution', () => { let db: Database; let PostModel; let WorkflowModel; + let WorkflowRepository; let workflow; beforeEach(async () => { app = await getApp(); db = app.db; - WorkflowModel = db.getModel('workflows'); - PostModel = db.getModel('posts'); - // Target = db.getModel('targets'); + WorkflowRepository = db.getCollection('workflows').repository; + WorkflowModel = db.getCollection('workflows').model; + PostModel = db.getCollection('posts').model; - workflow = await WorkflowModel.create({ - title: 'test workflow', - enabled: true, - type: 'afterCreate', - config: { - collection: 'posts' + // TODO(question): why the hooks of creating workflow won't run by using `WorkflowModel.create()`? + // maybe the model is not the original defined one which hooks have been added. + // @see database/../collections.ts@L99: `this.model = class extends M {};` + workflow = await WorkflowRepository.create({ + values: { + title: 'condition workflow', + enabled: true, + type: 'afterCreate', + config: { + collection: 'posts' + } } }); }); @@ -86,7 +92,7 @@ describe('execution', () => { const n2 = await workflow.createNode({ title: 'echo 2', type: 'echo', - upstream_id: n1.id + upstreamId: n1.id }); await n1.setDownstream(n2); @@ -133,7 +139,7 @@ describe('execution', () => { const n2 = await workflow.createNode({ title: 'echo', type: 'echo', - upstream_id: n1.id + upstreamId: n1.id }); await n1.setDownstream(n2); @@ -166,7 +172,7 @@ describe('execution', () => { const n2 = await workflow.createNode({ title: 'echo', type: 'echo', - upstream_id: n1.id + upstreamId: n1.id }); await n1.setDownstream(n2); @@ -200,15 +206,15 @@ describe('execution', () => { const n2 = await workflow.createNode({ title: 'true to echo', type: 'echo', - linkType: LINK_TYPE.ON_TRUE, - upstream_id: n1.id + branchIndex: BRANCH_INDEX.ON_TRUE, + upstreamId: n1.id }); await workflow.createNode({ title: 'false to echo', type: 'echo', - linkType: LINK_TYPE.ON_FALSE, - upstream_id: n1.id + branchIndex: BRANCH_INDEX.ON_FALSE, + upstreamId: n1.id }); const post = await PostModel.create({ title: 't1' }); @@ -218,8 +224,8 @@ describe('execution', () => { const jobs = await execution.getJobs({ order: [['id', 'ASC']] }); expect(jobs.length).toEqual(2); - expect(jobs[0].node_id).toEqual(n1.id); - expect(jobs[1].node_id).toEqual(n2.id); + expect(jobs[0].nodeId).toEqual(n1.id); + expect(jobs[1].nodeId).toEqual(n2.id); expect(jobs[1].result).toEqual(true); }); @@ -233,14 +239,14 @@ describe('execution', () => { const n2 = await workflow.createNode({ title: 'manual', type: 'prompt', - linkType: LINK_TYPE.ON_TRUE, - upstream_id: n1.id + branchIndex: BRANCH_INDEX.ON_TRUE, + upstreamId: n1.id }); const n3 = await workflow.createNode({ title: 'echo input value', type: 'echo', - upstream_id: n1.id + upstreamId: n1.id }); await n1.setDownstream(n3); @@ -250,7 +256,7 @@ describe('execution', () => { const [execution] = await workflow.getExecutions(); expect(execution.status).toEqual(EXECUTION_STATUS.STARTED); - const [pending] = await execution.getJobs({ node_id: n2.id }); + const [pending] = await execution.getJobs({ nodeId: n2.id }); pending.set('result', 123); await execution.resume(pending); @@ -268,14 +274,14 @@ describe('execution', () => { const n2 = await workflow.createNode({ title: 'manual', type: 'prompt->error', - linkType: LINK_TYPE.ON_TRUE, - upstream_id: n1.id + branchIndex: BRANCH_INDEX.ON_TRUE, + upstreamId: n1.id }); const n3 = await workflow.createNode({ title: 'echo input value', type: 'echo', - upstream_id: n1.id + upstreamId: n1.id }); await n1.setDownstream(n3); @@ -285,7 +291,7 @@ describe('execution', () => { const [execution] = await workflow.getExecutions(); expect(execution.status).toEqual(EXECUTION_STATUS.STARTED); - const [pending] = await execution.getJobs({ node_id: n2.id }); + const [pending] = await execution.getJobs({ nodeId: n2.id }); pending.set('result', 123); await execution.resume(pending); expect(execution.status).toEqual(EXECUTION_STATUS.REJECTED); diff --git a/packages/plugin-workflow/src/__tests__/index.ts b/packages/plugin-workflow/src/__tests__/index.ts index d799652b9..b097af4a5 100644 --- a/packages/plugin-workflow/src/__tests__/index.ts +++ b/packages/plugin-workflow/src/__tests__/index.ts @@ -36,10 +36,10 @@ export async function getApp(options = {}): Promise { throw new Error('input failed'); } }); - + await app.load(); - - app.db.import({ + + await app.db.import({ directory: path.resolve(__dirname, './collections') }); diff --git a/packages/plugin-workflow/src/__tests__/instructions/condition.test.ts b/packages/plugin-workflow/src/__tests__/instructions/condition.test.ts index b9a1c1c4a..cd87a8e19 100644 --- a/packages/plugin-workflow/src/__tests__/instructions/condition.test.ts +++ b/packages/plugin-workflow/src/__tests__/instructions/condition.test.ts @@ -1,7 +1,7 @@ import { Application } from '@nocobase/server'; import Database from '@nocobase/database'; import { getApp } from '..'; -import { EXECUTION_STATUS, JOB_STATUS, LINK_TYPE } from '../../constants'; +import { EXECUTION_STATUS, BRANCH_INDEX } from '../../constants'; @@ -10,21 +10,25 @@ describe('workflow > instructions > condition', () => { let db: Database; let PostModel; let WorkflowModel; + let WorkflowRepository; let workflow; beforeEach(async () => { app = await getApp(); db = app.db; - WorkflowModel = db.getModel('workflows'); - PostModel = db.getModel('posts'); + WorkflowRepository = db.getCollection('workflows').repository; + WorkflowModel = db.getCollection('workflows').model; + PostModel = db.getCollection('posts').model; - workflow = await WorkflowModel.create({ - title: 'condition workflow', - enabled: true, - type: 'afterCreate', - config: { - collection: 'posts' + workflow = await WorkflowRepository.create({ + values: { + title: 'condition workflow', + enabled: true, + type: 'afterCreate', + config: { + collection: 'posts' + } } }); }); @@ -53,15 +57,15 @@ describe('workflow > instructions > condition', () => { const n2 = await workflow.createNode({ title: 'true to echo', type: 'echo', - linkType: LINK_TYPE.ON_TRUE, - upstream_id: n1.id + branchIndex: BRANCH_INDEX.ON_TRUE, + upstreamId: n1.id }); const n3 = await workflow.createNode({ title: 'false to echo', type: 'echo', - linkType: LINK_TYPE.ON_FALSE, - upstream_id: n1.id + branchIndex: BRANCH_INDEX.ON_FALSE, + upstreamId: n1.id }); const post = await PostModel.create({ title: 't1' }); @@ -99,15 +103,15 @@ describe('workflow > instructions > condition', () => { await workflow.createNode({ title: 'true to echo', type: 'echo', - linkType: LINK_TYPE.ON_TRUE, - upstream_id: n1.id + branchIndex: BRANCH_INDEX.ON_TRUE, + upstreamId: n1.id }); await workflow.createNode({ title: 'false to echo', type: 'echo', - linkType: LINK_TYPE.ON_FALSE, - upstream_id: n1.id + branchIndex: BRANCH_INDEX.ON_FALSE, + upstreamId: n1.id }); const post = await PostModel.create({ title: 't1' }); diff --git a/packages/plugin-workflow/src/collections/executions.ts b/packages/plugin-workflow/src/collections/executions.ts index 23e49b367..1053e6657 100644 --- a/packages/plugin-workflow/src/collections/executions.ts +++ b/packages/plugin-workflow/src/collections/executions.ts @@ -1,4 +1,4 @@ -import { TableOptions } from '@nocobase/database'; +import { CollectionOptions } from '@nocobase/database'; export default { name: 'executions', @@ -30,4 +30,4 @@ export default { title: '状态' } ] -} as TableOptions; +} as CollectionOptions; diff --git a/packages/plugin-workflow/src/collections/flow_nodes.ts b/packages/plugin-workflow/src/collections/flow_nodes.ts index 8ebf1f766..0fb32fbbf 100644 --- a/packages/plugin-workflow/src/collections/flow_nodes.ts +++ b/packages/plugin-workflow/src/collections/flow_nodes.ts @@ -1,5 +1,4 @@ -import { TableOptions } from '@nocobase/database'; -import { LINK_TYPE } from '../constants'; +import { CollectionOptions } from '@nocobase/database'; export default { name: 'flow_nodes', @@ -40,18 +39,11 @@ export default { // only works when upstream node is branching type, like condition and parallel. // put here because the design of flow-links model is not really necessary for now. // or it should be put into flow-links model. - // if keeps 1:n relactionship, cannot support cycle flow. { interface: 'select', - name: 'linkType', - type: 'smallint', - title: 'Link Type', - dataSource: [ - { label: 'Default', value: LINK_TYPE.DEFAULT }, - { label: 'Branched, on true', value: LINK_TYPE.ON_TRUE }, - { label: 'Branched, on false', value: LINK_TYPE.ON_FALSE }, - { label: 'Branched, no limit', value: LINK_TYPE.NO_LIMIT } - ] + name: 'branchIndex', + type: 'integer', + title: 'branch index' }, // for reasons: // 1. redirect type node to solve cycle flow. @@ -83,4 +75,4 @@ export default { defaultValue: {} } ] -} as TableOptions; +} as CollectionOptions; diff --git a/packages/plugin-workflow/src/collections/jobs.ts b/packages/plugin-workflow/src/collections/jobs.ts index 5ac5e4bae..18485ca63 100644 --- a/packages/plugin-workflow/src/collections/jobs.ts +++ b/packages/plugin-workflow/src/collections/jobs.ts @@ -1,4 +1,4 @@ -import { TableOptions } from '@nocobase/database'; +import { CollectionOptions } from '@nocobase/database'; export default { name: 'jobs', @@ -45,4 +45,4 @@ export default { // title: 'node snapshot' // } ] -} as TableOptions; +} as CollectionOptions; diff --git a/packages/plugin-workflow/src/collections/workflows.ts b/packages/plugin-workflow/src/collections/workflows.ts index 83fd9e0da..02deea2cb 100644 --- a/packages/plugin-workflow/src/collections/workflows.ts +++ b/packages/plugin-workflow/src/collections/workflows.ts @@ -1,4 +1,4 @@ -import { TableOptions } from '@nocobase/database'; +import { CollectionOptions } from '@nocobase/database'; export default { name: 'workflows', @@ -53,4 +53,4 @@ export default { title: '触发执行' } ] -} as TableOptions; +} as CollectionOptions; diff --git a/packages/plugin-workflow/src/constants.ts b/packages/plugin-workflow/src/constants.ts index ecc2ed6fc..8cc682187 100644 --- a/packages/plugin-workflow/src/constants.ts +++ b/packages/plugin-workflow/src/constants.ts @@ -12,9 +12,8 @@ export const JOB_STATUS = { CANCELLED: -2 }; -export const LINK_TYPE = { +export const BRANCH_INDEX = { DEFAULT: null, ON_TRUE: 1, - ON_FALSE: 0, - NO_LIMIT: -1 + ON_FALSE: 0 }; diff --git a/packages/plugin-workflow/src/instructions/condition.ts b/packages/plugin-workflow/src/instructions/condition.ts index 832ded4f3..db15e5342 100644 --- a/packages/plugin-workflow/src/instructions/condition.ts +++ b/packages/plugin-workflow/src/instructions/condition.ts @@ -82,12 +82,12 @@ export default { status: JOB_STATUS.RESOLVED, result, // TODO(optimize): try unify the building of job - node_id: this.id, - upstream_id: prevJob instanceof Sequelize.Model ? prevJob.get('id') : null + nodeId: this.id, + upstreamId: prevJob instanceof Sequelize.Model ? prevJob.get('id') : null }; const branchNode = execution.nodes - .find(item => item.upstream === this && item.linkType === Number(result)); + .find(item => item.upstream === this && Boolean(item.branchIndex) === result); if (!branchNode) { return job; diff --git a/packages/plugin-workflow/src/instructions/index.ts b/packages/plugin-workflow/src/instructions/index.ts index 1c9722a54..6f4026d20 100644 --- a/packages/plugin-workflow/src/instructions/index.ts +++ b/packages/plugin-workflow/src/instructions/index.ts @@ -1,7 +1,5 @@ -// something like template for type of nodes - -import { ModelCtor, Model } from "@nocobase/database"; -import { ExecutionModel } from "../models/Execution"; +import ExecutionModel from "../models/Execution"; +import FlowNodeModel from "../models/FlowNode"; import prompt from './prompt'; import condition from './condition'; @@ -19,13 +17,13 @@ export type InstructionResult = Job | Promise; // - base on input and context, do any calculations or system call (io), and produce a result or pending. export interface Instruction { run( - this: ModelCtor, + this: FlowNodeModel, // what should input to be? // - just use previously output result for convenience? input: any, // what should context to be? // - could be the workflow execution object (containing context data) - execution: ModelCtor + execution: ExecutionModel ): InstructionResult; // for start node in main flow (or branch) to resume when manual sub branch triggered resume?(): InstructionResult diff --git a/packages/plugin-workflow/src/models/Execution.ts b/packages/plugin-workflow/src/models/Execution.ts index 2398b8b57..62d3dc839 100644 --- a/packages/plugin-workflow/src/models/Execution.ts +++ b/packages/plugin-workflow/src/models/Execution.ts @@ -1,13 +1,50 @@ -import Sequelize from 'sequelize'; -import { Model, ModelCtor } from '@nocobase/database'; +import { + Model, + BelongsToGetAssociationMixin, + Optional, + HasManyGetAssociationsMixin +} from 'sequelize'; + +import Database from '@nocobase/database'; import { EXECUTION_STATUS, JOB_STATUS } from '../constants'; import { getInstruction } from '../instructions'; +import WorkflowModel from './Workflow'; +import FlowNodeModel from './FlowNode'; +import JobModel from './Job'; -export class ExecutionModel extends Model { - nodes: Array = []; - nodesMap = new Map(); - jobsMap = new Map(); +interface ExecutionAttributes { + id: number; + title: string; + context: any; + status: number; +} + +interface ExecutionCreationAttributes extends Optional {} + +export default class ExecutionModel + extends Model + implements ExecutionAttributes { + + declare readonly database: Database; + + declare id: number; + declare title: string; + declare context: any; + declare status: number; + + declare createdAt: Date; + declare updatedAt: Date; + + declare workflow?: WorkflowModel; + declare getWorkflow: BelongsToGetAssociationMixin; + + declare jobs?: JobModel[]; + declare getJobs: HasManyGetAssociationsMixin; + + nodes: Array = []; + nodesMap = new Map(); + jobsMap = new Map(); // make dual linked nodes list then cache makeNodes(nodes = []) { @@ -18,17 +55,17 @@ export class ExecutionModel extends Model { }); nodes.forEach(node => { - if (node.upstream_id) { - node.upstream = this.nodesMap.get(node.upstream_id); + if (node.upstreamId) { + node.upstream = this.nodesMap.get(node.upstreamId); } - if (node.downstream_id) { - node.downstream = this.nodesMap.get(node.downstream_id); + if (node.downstreamId) { + node.downstream = this.nodesMap.get(node.downstreamId); } }); } - makeJobs(jobs: Array>) { + makeJobs(jobs: Array) { jobs.forEach(job => { this.jobsMap.set(job.id, job); }); @@ -55,7 +92,7 @@ export class ExecutionModel extends Model { async start(options) { await this.prepare(); if (!this.nodes.length) { - return this.exit(null); + return this.exit(); } const head = this.nodes.find(item => !item.upstream); return this.exec(head, { result: this.context }); @@ -63,7 +100,7 @@ export class ExecutionModel extends Model { async resume(job, options) { await this.prepare(); - const node = this.nodesMap.get(job.node_id); + const node = this.nodesMap.get(job.nodeId); return this.recall(node, job); } @@ -79,7 +116,7 @@ export class ExecutionModel extends Model { status: JOB_STATUS.REJECTED }; // if previous job is from resuming - if (prevJob && prevJob.node_id === node.id) { + if (prevJob && prevJob.nodeId === node.id) { prevJob.set(job); job = prevJob; } @@ -88,13 +125,13 @@ export class ExecutionModel extends Model { let savedJob; // TODO(optimize): many checking of resuming or new could be improved // could be implemented separately in exec() / resume() - if (job instanceof Sequelize.Model) { + if (job instanceof Model) { savedJob = await job.save(); } else { - const upstream_id = prevJob instanceof Sequelize.Model ? prevJob.get('id') : null; + const upstreamId = prevJob instanceof Model ? prevJob.get('id') : null; savedJob = await this.saveJob({ - node_id: node.id, - upstream_id, + nodeId: node.id, + upstreamId, ...job }); } @@ -136,7 +173,7 @@ export class ExecutionModel extends Model { return this.run(resume, node, job); } - async exit(job) { + async exit(job?: JobModel) { const executionStatusMap = { [JOB_STATUS.PENDING]: EXECUTION_STATUS.STARTED, [JOB_STATUS.RESOLVED]: EXECUTION_STATUS.RESOLVED, @@ -150,29 +187,30 @@ export class ExecutionModel extends Model { // TODO(optimize) async saveJob(payload) { - const JobModel = this.database.getModel('jobs'); - const [result] = await JobModel.upsert({ + // @ts-ignore + const { database } = this.constructor; + const { model } = database.getCollection('jobs'); + const [result] = await model.upsert({ ...payload, - execution_id: this.id - }); - + executionId: this.id + }) as [JobModel, boolean | null]; this.jobsMap.set(result.id, result); return result; } - findBranchParentNode(node): any { + findBranchParentNode(node: FlowNodeModel): FlowNodeModel | null { for (let n = node; n; n = n.upstream) { - if (n.linkType !== null) { + if (n.branchIndex !== null) { return n.upstream; } } return null; } - findBranchParentJob(job, node) { - for (let j = job; j; j = this.jobsMap.get(j.upstream_id)) { - if (j.node_id === node.id) { + findBranchParentJob(job: JobModel, node: FlowNodeModel): JobModel | null { + for (let j = job; j; j = this.jobsMap.get(j.upstreamId)) { + if (j.nodeId === node.id) { return j; } } diff --git a/packages/plugin-workflow/src/models/FlowNode.ts b/packages/plugin-workflow/src/models/FlowNode.ts new file mode 100644 index 000000000..4b9bf8db2 --- /dev/null +++ b/packages/plugin-workflow/src/models/FlowNode.ts @@ -0,0 +1,19 @@ +import { Model, BelongsToGetAssociationMixin } from 'sequelize'; +import WorkflowModel from './Workflow'; + +export default class FlowNodeModel extends Model { + declare id: number; + declare title: string; + declare branchIndex: null | number; + declare type: string; + declare config: any; + + declare createdAt: Date; + declare updatedAt: Date; + + declare upstream: FlowNodeModel; + declare downstream: FlowNodeModel; + + declare workflow?: WorkflowModel; + declare getWorkflow: BelongsToGetAssociationMixin; +} \ No newline at end of file diff --git a/packages/plugin-workflow/src/models/Job.ts b/packages/plugin-workflow/src/models/Job.ts new file mode 100644 index 000000000..fc92306a4 --- /dev/null +++ b/packages/plugin-workflow/src/models/Job.ts @@ -0,0 +1,18 @@ +import { Model, BelongsToGetAssociationMixin } from 'sequelize'; +import FlowNodeModel from './FlowNode'; + +export default class JobModel extends Model { + declare id: number; + declare status: number; + declare result: any; + + declare createdAt: Date; + declare updatedAt: Date; + + declare upstreamId: number; + declare upstream: JobModel; + + declare nodeId: number; + declare node?: FlowNodeModel; + declare getNode: BelongsToGetAssociationMixin; +} \ No newline at end of file diff --git a/packages/plugin-workflow/src/models/Workflow.ts b/packages/plugin-workflow/src/models/Workflow.ts index ebc7aef4e..18682b94e 100644 --- a/packages/plugin-workflow/src/models/Workflow.ts +++ b/packages/plugin-workflow/src/models/Workflow.ts @@ -1,15 +1,41 @@ -import { Model } from '@nocobase/database'; +import { Model, HasManyGetAssociationsMixin, HasManyCreateAssociationMixin } from 'sequelize'; + +import Database from '@nocobase/database'; import { get as getTrigger } from '../triggers'; import { EXECUTION_STATUS } from '../constants'; +import ExecutionModel from './Execution'; +import FlowNodeModel from './FlowNode'; + +export default class WorkflowModel extends Model { + declare static database: Database; + + declare id: number; + declare title: string; + declare enabled: boolean; + declare description?: string; + declare type: string; + declare config: any; + + declare createdAt: Date; + declare updatedAt: Date; + + declare nodes: FlowNodeModel[]; + declare getNodes: HasManyGetAssociationsMixin; + declare createNode: HasManyCreateAssociationMixin; + + declare executions: ExecutionModel[]; + declare getExecutions: HasManyGetAssociationsMixin; + declare createExecution: HasManyCreateAssociationMixin; -export class WorkflowModel extends Model { static async mount() { - const workflows = await this.findAll({ - where: { enabled: true } + const collection = this.database.getCollection('workflows'); + const workflows = await collection.repository.find({ + filter: { enabled: true } }); workflows.forEach(workflow => { + // @ts-ignore workflow.mount(); }); @@ -42,10 +68,10 @@ export class WorkflowModel extends Model { context, status: EXECUTION_STATUS.STARTED }); - execution.setDataValue('workflow', this); + execution.workflow = this; - await execution.start(null, null, options); + await execution.start(options); return execution; } } diff --git a/packages/plugin-workflow/src/server.ts b/packages/plugin-workflow/src/server.ts index 56276d86e..6a8c4ce83 100644 --- a/packages/plugin-workflow/src/server.ts +++ b/packages/plugin-workflow/src/server.ts @@ -1,21 +1,19 @@ import path from 'path'; -import { registerModels } from '@nocobase/database'; - -import { WorkflowModel } from './models/Workflow'; -import { ExecutionModel } from './models/Execution'; +import WorkflowModel from './models/Workflow'; +import ExecutionModel from './models/Execution'; export default { name: 'workflow', async load(options = {}) { const { db } = this.app; - registerModels({ + db.registerModels({ WorkflowModel, ExecutionModel, }); - db.import({ + await db.import({ directory: path.resolve(__dirname, 'collections'), }); @@ -24,8 +22,8 @@ export default { // * add all hooks for enabled workflows // * add hooks for create/update[enabled]/delete workflow to add/remove specific hooks this.app.on('beforeStart', async () => { - const Workflow = db.getModel('workflows'); - await Workflow.mount(); + const { model } = db.getCollection('workflows'); + await model.mount(); }) // [Life Cycle]: initialize all necessary seed data diff --git a/packages/plugin-workflow/src/triggers/data-change.ts b/packages/plugin-workflow/src/triggers/data-change.ts index ffc076a4a..611df3351 100644 --- a/packages/plugin-workflow/src/triggers/data-change.ts +++ b/packages/plugin-workflow/src/triggers/data-change.ts @@ -1,10 +1,14 @@ +import WorkflowModel from "../models/Workflow"; + export interface IDataChangeTriggerConfig { collection: string; // TODO: ICondition filter: any; } -export function afterCreate(config: IDataChangeTriggerConfig, callback: Function) { - const Model = this.database.getModel(config.collection); - Model.addHook('afterCreate', `workflow-${this.get('id')}`, (data: typeof Model, options) => callback({ data }, options)); +export function afterCreate(this: WorkflowModel, config: IDataChangeTriggerConfig, callback: Function) { + // @ts-ignore + const { database } = this.constructor; + const { model } = database.getCollection(config.collection); + model.addHook('afterCreate', `workflow-${this.get('id')}`, (data: any, options) => callback({ data }, options)); } diff --git a/packages/plugin-workflow/src/triggers/index.ts b/packages/plugin-workflow/src/triggers/index.ts index 99751e308..a823993a8 100644 --- a/packages/plugin-workflow/src/triggers/index.ts +++ b/packages/plugin-workflow/src/triggers/index.ts @@ -1,9 +1,8 @@ -import { ModelCtor } from '@nocobase/database'; -import { WorkflowModel } from '../models/Workflow'; +import WorkflowModel from '../models/Workflow'; import * as dataChangeTriggers from './data-change'; export interface ITrigger { - (this: ModelCtor, config: any): void + (this: WorkflowModel, config: any): void } const triggers = new Map(); diff --git a/packages/plugin-workflow/src/utils/getter.ts b/packages/plugin-workflow/src/utils/getter.ts index 77a442291..13b8d746c 100644 --- a/packages/plugin-workflow/src/utils/getter.ts +++ b/packages/plugin-workflow/src/utils/getter.ts @@ -1,7 +1,6 @@ import { get } from 'lodash'; -import { ModelCtor } from '@nocobase/database'; -import { ExecutionModel } from '../models/Execution'; +import ExecutionModel from '../models/Execution'; export type OperandType = 'context' | 'input' | 'job'; @@ -36,7 +35,7 @@ export type JobOperand = { export type Operand = ContextOperand | InputOperand | JobOperand | ConstantOperand; // TODO: other instructions may also use this method, could be moved to utils. -export function getValue(operand: Operand, input: any, execution: ModelCtor) { +export function getValue(operand: Operand, input: any, execution: ExecutionModel) { switch (operand.type) { // from execution context case 'context': @@ -47,7 +46,7 @@ export function getValue(operand: Operand, input: any, execution: ModelCtor item.id === operand.options.id); + const job = execution.jobsMap.get(operand.options.id); return get(job, operand.options.path); // constant default: From f9182c4004d88b895766433db130f72391fee56a Mon Sep 17 00:00:00 2001 From: mytharcher Date: Fri, 28 Jan 2022 21:02:32 +0800 Subject: [PATCH 5/6] feat(plugin-workflow): use toggle instead of mount and unmount --- .../src/__tests__/execution.test.ts | 20 +++------ .../__tests__/instructions/condition.test.ts | 16 +++---- .../plugin-workflow/src/models/Workflow.ts | 33 +++++++------- .../src/triggers/data-change.ts | 14 ------ .../plugin-workflow/src/triggers/index.ts | 20 ++++----- .../plugin-workflow/src/triggers/model.ts | 44 +++++++++++++++++++ 6 files changed, 82 insertions(+), 65 deletions(-) delete mode 100644 packages/plugin-workflow/src/triggers/data-change.ts create mode 100644 packages/plugin-workflow/src/triggers/model.ts diff --git a/packages/plugin-workflow/src/__tests__/execution.test.ts b/packages/plugin-workflow/src/__tests__/execution.test.ts index dbddc1457..2cb7de359 100644 --- a/packages/plugin-workflow/src/__tests__/execution.test.ts +++ b/packages/plugin-workflow/src/__tests__/execution.test.ts @@ -10,28 +10,22 @@ describe('execution', () => { let db: Database; let PostModel; let WorkflowModel; - let WorkflowRepository; let workflow; beforeEach(async () => { app = await getApp(); db = app.db; - WorkflowRepository = db.getCollection('workflows').repository; WorkflowModel = db.getCollection('workflows').model; PostModel = db.getCollection('posts').model; - // TODO(question): why the hooks of creating workflow won't run by using `WorkflowModel.create()`? - // maybe the model is not the original defined one which hooks have been added. - // @see database/../collections.ts@L99: `this.model = class extends M {};` - workflow = await WorkflowRepository.create({ - values: { - title: 'condition workflow', - enabled: true, - type: 'afterCreate', - config: { - collection: 'posts' - } + workflow = await WorkflowModel.create({ + title: 'test workflow', + enabled: true, + type: 'model', + config: { + mode: 1, + collection: 'posts' } }); }); diff --git a/packages/plugin-workflow/src/__tests__/instructions/condition.test.ts b/packages/plugin-workflow/src/__tests__/instructions/condition.test.ts index cd87a8e19..9fe7f9bf7 100644 --- a/packages/plugin-workflow/src/__tests__/instructions/condition.test.ts +++ b/packages/plugin-workflow/src/__tests__/instructions/condition.test.ts @@ -10,25 +10,21 @@ describe('workflow > instructions > condition', () => { let db: Database; let PostModel; let WorkflowModel; - let WorkflowRepository; let workflow; beforeEach(async () => { app = await getApp(); db = app.db; - WorkflowRepository = db.getCollection('workflows').repository; WorkflowModel = db.getCollection('workflows').model; PostModel = db.getCollection('posts').model; - workflow = await WorkflowRepository.create({ - values: { - title: 'condition workflow', - enabled: true, - type: 'afterCreate', - config: { - collection: 'posts' - } + workflow = await WorkflowModel.create({ + title: 'test workflow', + enabled: true, + type: 'afterCreate', + config: { + collection: 'posts' } }); }); diff --git a/packages/plugin-workflow/src/models/Workflow.ts b/packages/plugin-workflow/src/models/Workflow.ts index 18682b94e..0af20d334 100644 --- a/packages/plugin-workflow/src/models/Workflow.ts +++ b/packages/plugin-workflow/src/models/Workflow.ts @@ -34,28 +34,27 @@ export default class WorkflowModel extends Model { filter: { enabled: true } }); - workflows.forEach(workflow => { - // @ts-ignore - workflow.mount(); + workflows.forEach((workflow: WorkflowModel) => { + workflow.toggle(); }); - this.addHook('afterCreate', (model: WorkflowModel) => model.mount()); - // TODO: afterUpdate, afterDestroy + this.addHook('afterCreate', (model: WorkflowModel) => model.toggle()); + this.addHook('afterUpdate', (model: WorkflowModel) => model.toggle()); + this.addHook('afterDestroy', (model: WorkflowModel) => model.toggle(false)); } - async mount() { - if (!this.get('enabled')) { - return; - } + getHookId() { + return `workflow-${this.get('id')}`; + } + + async toggle(enable?: boolean) { const type = this.get('type'); - const config = this.get('config'); - const trigger = getTrigger(type); - trigger.call(this, config, this.start.bind(this)); - } - - // TODO - async unmount() { - + const { on, off } = getTrigger(type); + if (typeof enable !== 'undefined' ? enable : this.get('enabled')) { + on.call(this, this.start.bind(this)); + } else { + off.call(this); + } } async start(context: Object, options) { diff --git a/packages/plugin-workflow/src/triggers/data-change.ts b/packages/plugin-workflow/src/triggers/data-change.ts deleted file mode 100644 index 611df3351..000000000 --- a/packages/plugin-workflow/src/triggers/data-change.ts +++ /dev/null @@ -1,14 +0,0 @@ -import WorkflowModel from "../models/Workflow"; - -export interface IDataChangeTriggerConfig { - collection: string; - // TODO: ICondition - filter: any; -} - -export function afterCreate(this: WorkflowModel, config: IDataChangeTriggerConfig, callback: Function) { - // @ts-ignore - const { database } = this.constructor; - const { model } = database.getCollection(config.collection); - model.addHook('afterCreate', `workflow-${this.get('id')}`, (data: any, options) => callback({ data }, options)); -} diff --git a/packages/plugin-workflow/src/triggers/index.ts b/packages/plugin-workflow/src/triggers/index.ts index a823993a8..a48982e81 100644 --- a/packages/plugin-workflow/src/triggers/index.ts +++ b/packages/plugin-workflow/src/triggers/index.ts @@ -1,22 +1,20 @@ import WorkflowModel from '../models/Workflow'; -import * as dataChangeTriggers from './data-change'; +import modelTrigger from './model'; -export interface ITrigger { - (this: WorkflowModel, config: any): void +export interface Trigger { + name: string; + on(this: WorkflowModel, callback: Function): void; + off(this: WorkflowModel): void; } -const triggers = new Map(); +const triggers = new Map(); -export function register(type: string, trigger: ITrigger): void { +export function register(type: string, trigger: Trigger): void { triggers.set(type, trigger); } -export function get(type: string): ITrigger | undefined { +export function get(type: string): Trigger | undefined { return triggers.get(type); } -for (const key in dataChangeTriggers) { - if (dataChangeTriggers.hasOwnProperty(key)) { - register(key, dataChangeTriggers[key]); - } -} +register(modelTrigger.name, modelTrigger); diff --git a/packages/plugin-workflow/src/triggers/model.ts b/packages/plugin-workflow/src/triggers/model.ts new file mode 100644 index 000000000..2d848e4ea --- /dev/null +++ b/packages/plugin-workflow/src/triggers/model.ts @@ -0,0 +1,44 @@ +import WorkflowModel from "../models/Workflow"; + +export interface ModelChangeTriggerConfig { + collection: string; + // TODO: ICondition + filter: any; +} + +const MODE_BITMAP = { + CREATE: 1, + UPDATE: 2, + DESTROY: 4 +}; + +const MODE_BITMAP_EVENTS = new Map(); +MODE_BITMAP_EVENTS.set(MODE_BITMAP.CREATE, 'afterCreate'); +MODE_BITMAP_EVENTS.set(MODE_BITMAP.UPDATE, 'afterUpdate'); +MODE_BITMAP_EVENTS.set(MODE_BITMAP.DESTROY, 'afterDestroy'); + +export default { + name: 'model', + on(this: WorkflowModel, callback: Function) { + const { database } = this.constructor; + const { collection, mode } = this.config; + const { model } = database.getCollection(collection); + const handler = (data: any, options) => callback({ data }, options); + // TODO: duplication when mode change should be considered + for (let [key, event] of MODE_BITMAP_EVENTS.entries()) { + if (mode & key) { + model.addHook(event, this.getHookId(), handler); + } + } + }, + off(this: WorkflowModel) { + const { database } = this.constructor; + const { collection, mode } = this.config; + const { model } = database.getCollection(collection); + for (let [key, event] of MODE_BITMAP_EVENTS.entries()) { + if (mode & key) { + model.removeHook(event, this.getHookId()); + } + } + } +} \ No newline at end of file From 2f584b40bd224df0624bc0fd4a3cede5d0dac6a3 Mon Sep 17 00:00:00 2001 From: mytharcher Date: Tue, 1 Feb 2022 12:04:08 +0800 Subject: [PATCH 6/6] feat(plugin-workflow): add parallel branch and mixed test cases --- .../src/__tests__/execution.test.ts | 219 +++++++++++++++++- .../plugin-workflow/src/__tests__/index.ts | 6 + .../__tests__/instructions/condition.test.ts | 12 +- .../src/instructions/condition.ts | 15 +- .../plugin-workflow/src/instructions/index.ts | 4 +- .../src/instructions/parallel.ts | 92 ++++++++ .../plugin-workflow/src/models/Execution.ts | 97 +++++--- 7 files changed, 388 insertions(+), 57 deletions(-) create mode 100644 packages/plugin-workflow/src/instructions/parallel.ts diff --git a/packages/plugin-workflow/src/__tests__/execution.test.ts b/packages/plugin-workflow/src/__tests__/execution.test.ts index 2cb7de359..00ca0d710 100644 --- a/packages/plugin-workflow/src/__tests__/execution.test.ts +++ b/packages/plugin-workflow/src/__tests__/execution.test.ts @@ -250,7 +250,7 @@ describe('execution', () => { const [execution] = await workflow.getExecutions(); expect(execution.status).toEqual(EXECUTION_STATUS.STARTED); - const [pending] = await execution.getJobs({ nodeId: n2.id }); + const [pending] = await execution.getJobs({ where: { nodeId: n2.id } }); pending.set('result', 123); await execution.resume(pending); @@ -285,7 +285,7 @@ describe('execution', () => { const [execution] = await workflow.getExecutions(); expect(execution.status).toEqual(EXECUTION_STATUS.STARTED); - const [pending] = await execution.getJobs({ nodeId: n2.id }); + const [pending] = await execution.getJobs({ where: { nodeId: n2.id } }); pending.set('result', 123); await execution.resume(pending); expect(execution.status).toEqual(EXECUTION_STATUS.REJECTED); @@ -294,4 +294,219 @@ describe('execution', () => { expect(jobs.length).toEqual(2); }); }); + + describe('branch: parallel node', () => { + it('link to single branch', async () => { + const n1 = await workflow.createNode({ + title: 'parallel', + type: 'parallel' + }); + + const n2 = await workflow.createNode({ + title: 'echo1', + type: 'echo', + upstreamId: n1.id, + branchIndex: 0 + }); + + const n3 = await workflow.createNode({ + title: 'echo2', + type: 'echo', + upstreamId: n1.id + }); + + await n1.setDownstream(n3); + + const post = await PostModel.create({ title: 't1' }); + + const [execution] = await workflow.getExecutions(); + expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED); + const jobs = await execution.getJobs({ order: [['id', 'ASC']] }); + expect(jobs.length).toEqual(3); + }); + + it('link to multipe branches', async () => { + const n1 = await workflow.createNode({ + title: 'parallel', + type: 'parallel' + }); + + const n2 = await workflow.createNode({ + title: 'echo1', + type: 'echo', + upstreamId: n1.id, + branchIndex: 0 + }); + + const n3 = await workflow.createNode({ + title: 'echo2', + type: 'echo', + upstreamId: n1.id, + branchIndex: 1 + }); + + const n4 = await workflow.createNode({ + title: 'echo on end', + type: 'echo', + upstreamId: n1.id + }); + + await n1.setDownstream(n4); + + const post = await PostModel.create({ title: 't1' }); + + const [execution] = await workflow.getExecutions(); + expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED); + const jobs = await execution.getJobs({ order: [['id', 'ASC']] }); + expect(jobs.length).toEqual(4); + }); + + it('downstream has manual node', async () => { + const n1 = await workflow.createNode({ + title: 'parallel', + type: 'parallel' + }); + + const n2 = await workflow.createNode({ + title: 'prompt', + type: 'prompt', + upstreamId: n1.id, + branchIndex: 0 + }); + + const n3 = await workflow.createNode({ + title: 'echo', + type: 'echo', + upstreamId: n1.id, + branchIndex: 1 + }); + + const n4 = await workflow.createNode({ + title: 'echo on end', + type: 'echo', + upstreamId: n1.id + }); + + await n1.setDownstream(n4); + + const post = await PostModel.create({ title: 't1' }); + + const [execution] = await workflow.getExecutions(); + expect(execution.status).toEqual(EXECUTION_STATUS.STARTED); + + const [pending] = await execution.getJobs({ nodeId: n2.id }); + pending.set('result', 123); + await execution.resume(pending); + + expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED); + const jobs = await execution.getJobs({ order: [['id', 'ASC']] }); + expect(jobs.length).toEqual(4); + }); + }); + + describe('branch: mixed', () => { + it('condition branches contains parallel', async () => { + const n1 = await workflow.createNode({ + title: 'condition', + type: 'condition' + }); + + const n2 = await workflow.createNode({ + title: 'parallel', + type: 'parallel', + branchIndex: BRANCH_INDEX.ON_TRUE, + upstreamId: n1.id + }); + + const n3 = await workflow.createNode({ + title: 'prompt', + type: 'prompt', + upstreamId: n2.id, + branchIndex: 0 + }); + + const n4 = await workflow.createNode({ + title: 'parallel echo', + type: 'echo', + upstreamId: n2.id, + branchIndex: 1 + }); + + const n5 = await workflow.createNode({ + title: 'last echo', + type: 'echo', + upstreamId: n1.id + }); + + await n1.setDownstream(n5); + + const post = await PostModel.create({ title: 't1' }); + + const [execution] = await workflow.getExecutions(); + expect(execution.status).toEqual(EXECUTION_STATUS.STARTED); + + const pendingJobs = await execution.getJobs(); + expect(pendingJobs.length).toBe(4); + + const pending = pendingJobs.find(item => item.nodeId === n3.id ); + pending.set('result', 123); + await execution.resume(pending); + + expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED); + const jobs = await execution.getJobs({ order: [['id', 'ASC']] }); + expect(jobs.length).toEqual(5); + }); + + it('parallel branches contains condition', async () => { + const n1 = await workflow.createNode({ + title: 'parallel', + type: 'parallel' + }); + + const n2 = await workflow.createNode({ + title: 'prompt', + type: 'prompt', + upstreamId: n1.id, + branchIndex: 0 + }); + + const n3 = await workflow.createNode({ + title: 'condition', + type: 'condition', + upstreamId: n1.id, + branchIndex: 1 + }); + + const n4 = await workflow.createNode({ + title: 'condition echo', + type: 'echo', + upstreamId: n3.id, + branchIndex: BRANCH_INDEX.ON_TRUE + }); + + const n5 = await workflow.createNode({ + title: 'last echo', + type: 'echo', + upstreamId: n1.id + }); + + await n1.setDownstream(n5); + + const post = await PostModel.create({ title: 't1' }); + + const [execution] = await workflow.getExecutions(); + expect(execution.status).toEqual(EXECUTION_STATUS.STARTED); + + const pendingJobs = await execution.getJobs(); + expect(pendingJobs.length).toBe(4); + + const pending = pendingJobs.find(item => item.nodeId === n2.id ); + pending.set('result', 123); + await execution.resume(pending); + + expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED); + const jobs = await execution.getJobs({ order: [['id', 'ASC']] }); + expect(jobs.length).toEqual(5); + }); + }); }); diff --git a/packages/plugin-workflow/src/__tests__/index.ts b/packages/plugin-workflow/src/__tests__/index.ts index b097af4a5..65b026537 100644 --- a/packages/plugin-workflow/src/__tests__/index.ts +++ b/packages/plugin-workflow/src/__tests__/index.ts @@ -5,6 +5,12 @@ import plugin from '../server'; import { registerInstruction } from '../instructions'; import { JOB_STATUS } from '../constants'; +export function sleep(ms: number) { + return new Promise(resolve => { + setTimeout(resolve, ms); + }); +} + export async function getApp(options = {}): Promise { const app = mockServer(options); diff --git a/packages/plugin-workflow/src/__tests__/instructions/condition.test.ts b/packages/plugin-workflow/src/__tests__/instructions/condition.test.ts index 9fe7f9bf7..5fbdcbb17 100644 --- a/packages/plugin-workflow/src/__tests__/instructions/condition.test.ts +++ b/packages/plugin-workflow/src/__tests__/instructions/condition.test.ts @@ -22,8 +22,9 @@ describe('workflow > instructions > condition', () => { workflow = await WorkflowModel.create({ title: 'test workflow', enabled: true, - type: 'afterCreate', + type: 'model', config: { + mode: 1, collection: 'posts' } }); @@ -75,15 +76,6 @@ describe('workflow > instructions > condition', () => { }); it('calculation to false downstream', async () => { - const workflow = await WorkflowModel.create({ - title: 'condition workflow', - enabled: true, - type: 'afterCreate', - config: { - collection: 'posts' - } - }); - const n1 = await workflow.createNode({ title: 'condition', type: 'condition', diff --git a/packages/plugin-workflow/src/instructions/condition.ts b/packages/plugin-workflow/src/instructions/condition.ts index db15e5342..b38df688c 100644 --- a/packages/plugin-workflow/src/instructions/condition.ts +++ b/packages/plugin-workflow/src/instructions/condition.ts @@ -95,22 +95,13 @@ export default { const savedJob = await execution.saveJob(job); - // return execution.exec(branchNode, savedJob); - const tailJob = await execution.exec(branchNode, savedJob); - - if (tailJob.status === JOB_STATUS.PENDING) { - savedJob.set('status', JOB_STATUS.PENDING); - return savedJob; - } - - return tailJob; + return execution.exec(branchNode, savedJob); }, async resume(this, branchJob, execution) { if (branchJob.status === JOB_STATUS.RESOLVED) { - const job = execution.findBranchParentJob(branchJob, this); - job.set('status', JOB_STATUS.RESOLVED); - return job; + // return to continue this.downstream + return branchJob; } // pass control to upper scope by ending current scope diff --git a/packages/plugin-workflow/src/instructions/index.ts b/packages/plugin-workflow/src/instructions/index.ts index 6f4026d20..65a97e483 100644 --- a/packages/plugin-workflow/src/instructions/index.ts +++ b/packages/plugin-workflow/src/instructions/index.ts @@ -3,7 +3,7 @@ import FlowNodeModel from "../models/FlowNode"; import prompt from './prompt'; import condition from './condition'; -// import parallel from './parallel'; +import parallel from './parallel'; export interface Job { status: number; @@ -41,4 +41,4 @@ export function registerInstruction(key: string, instruction: any) { registerInstruction('prompt', prompt); registerInstruction('condition', condition); -// registerInstruction('parallel', parallel); +registerInstruction('parallel', parallel); diff --git a/packages/plugin-workflow/src/instructions/parallel.ts b/packages/plugin-workflow/src/instructions/parallel.ts new file mode 100644 index 000000000..9c3a775f4 --- /dev/null +++ b/packages/plugin-workflow/src/instructions/parallel.ts @@ -0,0 +1,92 @@ +import { JOB_STATUS } from "../constants"; +import ExecutionModel from "../models/Execution"; +import FlowNodeModel from "../models/FlowNode"; +import JobModel from "../models/Job"; + +export const PARALLEL_MODE = { + ALL: 'all', + ANY: 'any', + RACE: 'race' +} as const; + +const StatusGetters = { + [PARALLEL_MODE.ALL](result) { + if (result.some(j => j && j.status === JOB_STATUS.REJECTED)) { + return JOB_STATUS.REJECTED; + } + if (result.every(j => j && j.status === JOB_STATUS.RESOLVED)) { + return JOB_STATUS.RESOLVED; + } + return JOB_STATUS.PENDING; + }, + [PARALLEL_MODE.ANY](result) { + return result.some(j => j && j.status === JOB_STATUS.RESOLVED) + ? JOB_STATUS.RESOLVED + : ( + result.some(j => j && j.status === JOB_STATUS.PENDING) + ? JOB_STATUS.PENDING + : JOB_STATUS.REJECTED + ) + }, + [PARALLEL_MODE.RACE](result) { + return result.some(j => j && j.status === JOB_STATUS.RESOLVED) + ? JOB_STATUS.RESOLVED + : ( + result.some(j => j && j.status === JOB_STATUS.REJECTED) + ? JOB_STATUS.REJECTED + : JOB_STATUS.PENDING + ) + } +}; + +export default { + async run(this: FlowNodeModel, prevJob: JobModel, execution: ExecutionModel) { + const branches = execution.nodes + .filter(item => item.upstream === this && item.branchIndex !== null) + .sort((a, b) => a.branchIndex - b.branchIndex); + + const job = await execution.saveJob({ + status: JOB_STATUS.PENDING, + result: Array(branches.length).fill(null), + nodeId: this.id, + upstreamId: prevJob?.id ?? null + }); + + // NOTE: + // use `reduce` but not `Promise.all` here to avoid racing manupulating db. + // for users, this is almost equivalent to `Promise.all`, + // because of the delay is not significant sensible. + // another better aspect of this is, it could handle sequenced branches in future. + await branches.reduce((promise: Promise, branch) => promise.then(() => execution.exec(branch, job)), Promise.resolve()); + + return execution.end(this, job); + }, + + async resume(this, branchJob, execution: ExecutionModel) { + const job = execution.findBranchParentJob(branchJob, this); + + const { result, status } = job; + // if parallel has been done (resolved / rejected), do not care newly executed branch jobs. + if (status !== JOB_STATUS.PENDING) { + return null; + } + + // find the index of the node which start the branch + const jobNode = execution.nodesMap.get(branchJob.nodeId); + const { branchIndex } = execution.findBranchStartNode(jobNode); + const { mode = PARALLEL_MODE.ALL } = this.config || {}; + + const newResult = [...result.slice(0, branchIndex), branchJob.get(), ...result.slice(branchIndex + 1)]; + job.set({ + result: newResult, + status: StatusGetters[mode](newResult) + }); + + if (job.status === JOB_STATUS.PENDING) { + await job.save({ transaction: execution.transaction }); + return execution.end(this, job); + } + + return job; + } +} diff --git a/packages/plugin-workflow/src/models/Execution.ts b/packages/plugin-workflow/src/models/Execution.ts index 62d3dc839..26dc50dc1 100644 --- a/packages/plugin-workflow/src/models/Execution.ts +++ b/packages/plugin-workflow/src/models/Execution.ts @@ -2,7 +2,8 @@ import { Model, BelongsToGetAssociationMixin, Optional, - HasManyGetAssociationsMixin + HasManyGetAssociationsMixin, + Transaction } from 'sequelize'; import Database from '@nocobase/database'; @@ -22,11 +23,15 @@ interface ExecutionAttributes { interface ExecutionCreationAttributes extends Optional {} +export interface ExecutionOptions { + transaction?: Transaction; +} + export default class ExecutionModel extends Model implements ExecutionAttributes { - declare readonly database: Database; + declare static readonly database: Database; declare id: number; declare title: string; @@ -42,10 +47,20 @@ export default class ExecutionModel declare jobs?: JobModel[]; declare getJobs: HasManyGetAssociationsMixin; + options: ExecutionOptions; + transaction: Transaction; + nodes: Array = []; nodesMap = new Map(); jobsMap = new Map(); + static StatusMap = { + [JOB_STATUS.PENDING]: EXECUTION_STATUS.STARTED, + [JOB_STATUS.RESOLVED]: EXECUTION_STATUS.RESOLVED, + [JOB_STATUS.REJECTED]: EXECUTION_STATUS.REJECTED, + [JOB_STATUS.CANCELLED]: EXECUTION_STATUS.CANCELLED, + }; + // make dual linked nodes list then cache makeNodes(nodes = []) { this.nodes = nodes; @@ -71,44 +86,60 @@ export default class ExecutionModel }); } - async prepare() { + async prepare(options) { if (this.status !== EXECUTION_STATUS.STARTED) { throw new Error(`execution was ended with status ${this.status}`); } + this.options = options || {}; + const { transaction = await (this.constructor).database.sequelize.transaction() } = this.options; + this.transaction = transaction; + if (!this.workflow) { - this.workflow = await this.getWorkflow(); + this.workflow = await this.getWorkflow({ transaction }); } - const nodes = await this.workflow.getNodes(); + const nodes = await this.workflow.getNodes({ transaction }); this.makeNodes(nodes); - const jobs = await this.getJobs(); + const jobs = await this.getJobs({ transaction }); this.makeJobs(jobs); } - async start(options) { - await this.prepare(); - if (!this.nodes.length) { - return this.exit(); + async start(options: ExecutionOptions) { + await this.prepare(options); + if (this.nodes.length) { + const head = this.nodes.find(item => !item.upstream); + await this.exec(head, { result: this.context }); + } else { + await this.exit(null); } - const head = this.nodes.find(item => !item.upstream); - return this.exec(head, { result: this.context }); + await this.commit(); } - async resume(job, options) { - await this.prepare(); + async resume(job: JobModel, options: ExecutionOptions) { + await this.prepare(options); const node = this.nodesMap.get(job.nodeId); - return this.recall(node, job); + await this.recall(node, job); + await this.commit(); } - private async run(instruction, node, prevJob) { + private async commit() { + if (!this.options || !this.options.transaction) { + await this.transaction.commit(); + } + } + + private async run(instruction, node: FlowNodeModel, prevJob) { let job; try { // call instruction to get result and status job = await instruction.call(node, prevJob, this); + if (!job) { + return null; + } } catch (err) { // for uncaught error, set to rejected job = { @@ -122,11 +153,11 @@ export default class ExecutionModel } } - let savedJob; + let savedJob: JobModel; // TODO(optimize): many checking of resuming or new could be improved // could be implemented separately in exec() / resume() if (job instanceof Model) { - savedJob = await job.save(); + savedJob = await job.save({ transaction: this.transaction }) as JobModel; } else { const upstreamId = prevJob instanceof Model ? prevJob.get('id') : null; savedJob = await this.saveJob({ @@ -173,32 +204,36 @@ export default class ExecutionModel return this.run(resume, node, job); } - async exit(job?: JobModel) { - const executionStatusMap = { - [JOB_STATUS.PENDING]: EXECUTION_STATUS.STARTED, - [JOB_STATUS.RESOLVED]: EXECUTION_STATUS.RESOLVED, - [JOB_STATUS.REJECTED]: EXECUTION_STATUS.REJECTED, - [JOB_STATUS.CANCELLED]: EXECUTION_STATUS.CANCELLED, - }; - const status = job ? executionStatusMap[job.status] : EXECUTION_STATUS.RESOLVED; - await this.update({ status }); - return job; + async exit(job: JobModel | null) { + const status = job ? ExecutionModel.StatusMap[job.status] : EXECUTION_STATUS.RESOLVED; + await this.update({ status }, { transaction: this.transaction }); + return null; } // TODO(optimize) async saveJob(payload) { - // @ts-ignore - const { database } = this.constructor; + const { database } = this.constructor; const { model } = database.getCollection('jobs'); const [result] = await model.upsert({ ...payload, executionId: this.id - }) as [JobModel, boolean | null]; + }, { transaction: this.transaction }) as [JobModel, boolean | null]; this.jobsMap.set(result.id, result); return result; } + // find the first node in current branch + findBranchStartNode(node: FlowNodeModel): FlowNodeModel | null { + for (let n = node; n; n = n.upstream) { + if (n.branchIndex !== null) { + return n; + } + } + return null; + } + + // find the node start current branch findBranchParentNode(node: FlowNodeModel): FlowNodeModel | null { for (let n = node; n; n = n.upstream) { if (n.branchIndex !== null) {