diff --git a/packages/plugin-workflow/package.json b/packages/plugin-workflow/package.json index f099e50a6..37a4fd6c6 100644 --- a/packages/plugin-workflow/package.json +++ b/packages/plugin-workflow/package.json @@ -12,6 +12,7 @@ "dependencies": { "@nocobase/database": "^0.6.0-alpha.0", "@nocobase/server": "^0.6.0-alpha.0", + "@nocobase/utils": "^0.6.0-alpha.0", "json-templates": "^4.2.0" }, "devDependencies": { diff --git a/packages/plugin-workflow/src/__tests__/collections/posts.ts b/packages/plugin-workflow/src/__tests__/collections/posts.ts index 6da100450..59c564c83 100644 --- a/packages/plugin-workflow/src/__tests__/collections/posts.ts +++ b/packages/plugin-workflow/src/__tests__/collections/posts.ts @@ -11,6 +11,11 @@ export default { type: 'boolean', name: 'published', defaultValue: false + }, + { + type: 'integer', + name: 'read', + defaultValue: 0 } ] } as CollectionOptions; diff --git a/packages/plugin-workflow/src/__tests__/index.ts b/packages/plugin-workflow/src/__tests__/index.ts index 65b026537..3a0792c94 100644 --- a/packages/plugin-workflow/src/__tests__/index.ts +++ b/packages/plugin-workflow/src/__tests__/index.ts @@ -2,8 +2,9 @@ import path from 'path'; import { MockServer, mockServer } from '@nocobase/test'; import plugin from '../server'; -import { registerInstruction } from '../instructions'; +import instructions from '../instructions'; import { JOB_STATUS } from '../constants'; +import calculators from '../calculators'; export function sleep(ms: number) { return new Promise(resolve => { @@ -17,34 +18,44 @@ export async function getApp(options = {}): Promise { app.plugin(plugin); // for test only - registerInstruction('echo', { - run(this, { result }, execution) { - return { - status: JOB_STATUS.RESOLVED, - result - }; - } - }); + if (!instructions.get('echo')) { + instructions.register('echo', { + run(this, { result }, execution) { + return { + status: JOB_STATUS.RESOLVED, + result + }; + } + }); + } - registerInstruction('error', { - run(this, input, execution) { - throw new Error('definite error'); - } - }); + if (!instructions.get('error')) { + instructions.register('error', { + run(this, input, execution) { + throw new Error('definite error'); + } + }); + } + + if (!instructions.get('prompt->error')) { + instructions.register('prompt->error', { + run(this, input, execution) { + return { + status: JOB_STATUS.PENDING + }; + }, + resume(this, input, execution) { + throw new Error('input failed'); + } + }); + } + + if (!calculators.get('no1')) { + calculators.register('no1', () => 1); + } - registerInstruction('prompt->error', { - run(this, input, execution) { - return { - status: JOB_STATUS.PENDING - }; - }, - resume(this, input, execution) { - throw new Error('input failed'); - } - }); - await app.load(); - + await app.db.import({ directory: path.resolve(__dirname, './collections') }); @@ -56,6 +67,6 @@ export async function getApp(options = {}): Promise { } // 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/calculation.test.ts b/packages/plugin-workflow/src/__tests__/instructions/calculation.test.ts new file mode 100644 index 000000000..90487681e --- /dev/null +++ b/packages/plugin-workflow/src/__tests__/instructions/calculation.test.ts @@ -0,0 +1,200 @@ +import { Application } from '@nocobase/server'; +import Database from '@nocobase/database'; +import { getApp } from '..'; + + + +describe('workflow > instructions > calculation', () => { + let app: Application; + let db: Database; + let PostModel; + let WorkflowModel; + let workflow; + + beforeEach(async () => { + app = await getApp(); + + db = app.db; + WorkflowModel = db.getCollection('workflows').model; + PostModel = db.getCollection('posts').model; + + workflow = await WorkflowModel.create({ + title: 'test workflow', + enabled: true, + type: 'model', + config: { + mode: 1, + collection: 'posts' + } + }); + }); + + afterEach(() => db.close()); + + describe('operand types', () => { + it('constant', async () => { + const n1 = await workflow.createNode({ + type: 'calculation', + config: { + calculation: { + calculator: 'add', + operands: [ + { value: 1 }, + { value: 1 } + ] + } + } + }); + + const post = await PostModel.create({ title: 't1' }); + const [execution] = await workflow.getExecutions(); + const [job] = await execution.getJobs(); + expect(job.result).toBe(2); + }); + + it('context (legacy)', async () => { + const n1 = await workflow.createNode({ + type: 'calculation', + config: { + calculation: { + calculator: 'add', + operands: [ + { value: 1 }, + { type: 'context', options: { path: 'data.read' } } + ] + } + } + }); + + const post = await PostModel.create({ title: 't1' }); + const [execution] = await workflow.getExecutions(); + const [job] = await execution.getJobs(); + expect(job.result).toBe(1); + }); + + it('context by json-template', async () => { + const n1 = await workflow.createNode({ + type: 'calculation', + config: { + calculation: { + calculator: 'add', + operands: [ + { value: 1 }, + { value: '{{$context.data.read}}' } + ] + } + } + }); + + const post = await PostModel.create({ title: 't1' }); + const [execution] = await workflow.getExecutions(); + const [job] = await execution.getJobs(); + expect(job.result).toBe(1); + }); + + it('job result (legacy)', async () => { + const n1 = await workflow.createNode({ + type: 'echo' + }); + + const n2 = await workflow.createNode({ + type: 'calculation', + config: { + calculation: { + calculator: 'add', + operands: [ + { value: 1 }, + { type: 'job', options: { nodeId: n1.id, path: 'data.read' } } + ] + } + }, + upstreamId: n1.id + }); + + await n1.setDownstream(n2); + + const post = await PostModel.create({ title: 't1' }); + const [execution] = await workflow.getExecutions(); + const [n1Job, n2Job] = await execution.getJobs({ order: [['id', 'ASC']]}); + expect(n2Job.result).toBe(1); + }); + + it('job result by json-template', async () => { + const n1 = await workflow.createNode({ + type: 'echo' + }); + + const n2 = await workflow.createNode({ + type: 'calculation', + config: { + calculation: { + calculator: 'add', + operands: [ + { value: 1 }, + { value: `{{$jobsMapByNodeId.${n1.id}.data.read}}` } + ] + } + }, + upstreamId: n1.id + }); + + await n1.setDownstream(n2); + + const post = await PostModel.create({ title: 't1' }); + const [execution] = await workflow.getExecutions(); + const [n1Job, n2Job] = await execution.getJobs({ order: [['id', 'ASC']]}); + expect(n2Job.result).toBe(1); + }); + + it('function', async () => { + const n1 = await workflow.createNode({ + type: 'calculation', + config: { + calculation: { + calculator: 'add', + operands: [ + { value: 1 }, + { value: '{{$fn.no1}}' } + ] + } + } + }); + + const post = await PostModel.create({ title: 't1' }); + const [execution] = await workflow.getExecutions(); + const [job] = await execution.getJobs(); + expect(job.result).toBe(2); + }); + }); + + describe('nested operands', () => { + it('1 + ( 0 - 2 )', async () => { + const n1 = await workflow.createNode({ + type: 'calculation', + config: { + calculation: { + calculator: 'add', + operands: [ + { value: 1 }, + { + type: 'calculation', + options: { + calculator: 'minus', + operands: [ + { value: '{{$context.data.read}}' }, + { value: 2 } + ] + } + } + ] + } + } + }); + + const post = await PostModel.create({ title: 't1' }); + const [execution] = await workflow.getExecutions(); + const [job] = await execution.getJobs(); + expect(job.result).toBe(-1); + }); + }); +}); diff --git a/packages/plugin-workflow/src/__tests__/instructions/condition.test.ts b/packages/plugin-workflow/src/__tests__/instructions/condition.test.ts index 5fbdcbb17..1e6594277 100644 --- a/packages/plugin-workflow/src/__tests__/instructions/condition.test.ts +++ b/packages/plugin-workflow/src/__tests__/instructions/condition.test.ts @@ -111,9 +111,161 @@ describe('workflow > instructions > condition', () => { expect(jobs.length).toEqual(2); expect(jobs[1].result).toEqual(false); }); + + it('not', async () => { + + }); }); describe('group calculation', () => { + it('and true', async () => { + const n1 = workflow.createNode({ + type: 'condition', + config: { + calculation: { + group: { + type: 'and', + calculations: [ + { + calculator: 'equal', + operands: [{ value: 1 }, { value: 1 }] + }, + { + calculator: 'equal', + operands: [{ value: 1 }, { value: 1 }] + } + ] + } + } + } + }); + const post = await PostModel.create({ title: 't1' }); + + const [execution] = await workflow.getExecutions(); + const [job] = await execution.getJobs(); + expect(job.result).toBe(true); + }); + + it('and false', async () => { + const n1 = workflow.createNode({ + type: 'condition', + config: { + calculation: { + group: { + type: 'and', + calculations: [ + { + calculator: 'equal', + operands: [{ value: 1 }, { value: 1 }] + }, + { + calculator: 'equal', + operands: [{ value: 0 }, { value: 1 }] + } + ] + } + } + } + }); + + const post = await PostModel.create({ title: 't1' }); + + const [execution] = await workflow.getExecutions(); + const [job] = await execution.getJobs(); + expect(job.result).toBe(false); + }); + + it('or true', async () => { + const n1 = workflow.createNode({ + type: 'condition', + config: { + calculation: { + group: { + type: 'or', + calculations: [ + { + calculator: 'equal', + operands: [{ value: 1 }, { value: 1 }] + }, + { + calculator: 'equal', + operands: [{ value: 0 }, { value: 1 }] + } + ] + } + } + } + }); + + const post = await PostModel.create({ title: 't1' }); + + const [execution] = await workflow.getExecutions(); + const [job] = await execution.getJobs(); + expect(job.result).toBe(true); + }); + + it('or false', async () => { + const n1 = workflow.createNode({ + type: 'condition', + config: { + calculation: { + group: { + type: 'and', + calculations: [ + { + calculator: 'equal', + operands: [{ value: 0 }, { value: 1 }] + }, + { + calculator: 'equal', + operands: [{ value: 0 }, { value: 1 }] + } + ] + } + } + } + }); + + const post = await PostModel.create({ title: 't1' }); + + const [execution] = await workflow.getExecutions(); + const [job] = await execution.getJobs(); + expect(job.result).toBe(false); + }); + + it('nested', async () => { + const n1 = workflow.createNode({ + type: 'condition', + config: { + calculation: { + group: { + type: 'and', + calculations: [ + { + calculator: 'equal', + operands: [{ value: 1 }, { value: 1 }] + }, + { + group: { + type: 'or', + calculations: [ + { calculator: 'equal', operands: [{ value: 0 }, { value: 1 }] }, + { calculator: 'equal', operands: [{ value: 0 }, { value: 1 }] } + ] + } + } + ] + } + } + } + }); + + const post = await PostModel.create({ title: 't1' }); + + const [execution] = await workflow.getExecutions(); + const [job] = await execution.getJobs(); + expect(job.result).toBe(false); + }); }); }); diff --git a/packages/plugin-workflow/src/__tests__/instructions/destroy.test.ts b/packages/plugin-workflow/src/__tests__/instructions/destroy.test.ts index 1730feef7..3019fabde 100644 --- a/packages/plugin-workflow/src/__tests__/instructions/destroy.test.ts +++ b/packages/plugin-workflow/src/__tests__/instructions/destroy.test.ts @@ -4,7 +4,7 @@ import { getApp } from '..'; -describe('workflow > instructions > create', () => { +describe('workflow > instructions > destroy', () => { let app: Application; let db: Database; let PostModel; diff --git a/packages/plugin-workflow/src/__tests__/utils/getter.test.ts b/packages/plugin-workflow/src/__tests__/utils/getter.test.ts deleted file mode 100644 index 70f580534..000000000 --- a/packages/plugin-workflow/src/__tests__/utils/getter.test.ts +++ /dev/null @@ -1,142 +0,0 @@ -import { Application } from '@nocobase/server'; -import Database from '@nocobase/database'; -import { getApp } from '..'; - -import { get } from 'lodash'; - -import { getValue } from '../../utils/getter'; -import { BRANCH_INDEX } from '../../constants'; - -describe('value getter', () => { - let app: Application; - let db: Database; - let JobModel; - let WorkflowModel; - let ExecutionModel; - let PostModel; - let workflow; - - beforeEach(async () => { - app = await getApp(); - - db = app.db; - WorkflowModel = db.getCollection('workflows').model; - JobModel = db.getCollection('jobs').model; - ExecutionModel = db.getCollection('executions').model; - PostModel = db.getCollection('posts').model; - - workflow = await WorkflowModel.create({ - title: 'test workflow', - enabled: true, - type: 'model', - config: { - mode: 1, - collection: 'posts' - } - }); - }); - - afterEach(() => db.close()); - - describe('get from constants', () => { - it('null', () => { - const v1 = getValue({ - value: null - }, null, null); - expect(v1).toBe(null); - }); - - it('number', () => { - const v1 = getValue({ - value: 1 - }, null, null); - expect(v1).toBe(1); - }); - }); - - describe('get from context', () => { - it('paths', async () => { - const post = await PostModel.create({ title: 't1' }); - const [execution] = await workflow.getExecutions(); - - const v1 = getValue({ - type: 'context', - options: {} - }, null, execution); - expect(v1).toMatchObject({ data: { title: 't1' } }); - - const v2 = getValue({ - type: 'context', - options: { path: 'data' } - }, null, execution); - expect(v2).toMatchObject({ title: 't1' }); - - const v3 = getValue({ - type: 'context', - options: { path: 'data.title' } - }, null, execution); - expect(v3).toBe(post.title); - }); - }); - - describe('get from job by id', () => { - it('base getting from executed job', async () => { - const n1 = await workflow.createNode({ - type: 'echo' - }); - const post = await PostModel.create({ title: 't1' }); - const [execution] = await workflow.getExecutions(); - await execution.prepare({}, true); - - const v1 = getValue({ - type: 'job', - options: { - nodeId: n1.id, - path: 'data.title' - } - }, null, execution); - - expect(v1).toBe(post.title); - }); - - it('result of unexecuted job could not be got', async () => { - const n1 = await workflow.createNode({ - type: 'condition' - }); - - const n2 = await workflow.createNode({ - type: 'echo', - branchIndex: BRANCH_INDEX.ON_TRUE, - upstreamId: n1.id - }); - - const n3 = await workflow.createNode({ - type: 'echo', - branchIndex: BRANCH_INDEX.ON_FALSE, - upstreamId: n1.id - }); - - const post = await PostModel.create({ title: 't1' }); - const [execution] = await workflow.getExecutions(); - await execution.prepare({}, true); - - const v1 = getValue({ - type: 'job', - options: { - nodeId: n3.id - } - }, null, execution); - - expect(v1).toBeUndefined(); - - const v2 = getValue({ - type: 'job', - options: { - nodeId: n2.id - } - }, null, execution); - - expect(v2).toBe(true); - }); - }); -}); diff --git a/packages/plugin-workflow/src/calculators/index.ts b/packages/plugin-workflow/src/calculators/index.ts new file mode 100644 index 000000000..97b96da1d --- /dev/null +++ b/packages/plugin-workflow/src/calculators/index.ts @@ -0,0 +1,171 @@ +import { get as getWithPath } from 'lodash'; +import { Registry } from "@nocobase/utils"; + +import ExecutionModel from '../models/Execution'; +import JobModel from '../models/Job'; + +export const calculators = new Registry(); + +export default calculators; + + +export type OperandType = 'context' | 'input' | 'job' | 'calculation'; + +export type ObjectGetterOptions = { + path?: string +}; + +export type JobGetterOptions = ObjectGetterOptions & { + nodeId: number +}; + +export type CalculationOptions = { + calculator: string, + operands: Operand[] +}; + +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 Calculation = { + type: 'calculation'; + options: CalculationOptions +}; + +// TODO(type): union type here is wrong +export type Operand = ContextOperand | InputOperand | JobOperand | ConstantOperand | Calculation; + +// @deprecated +// HACK: if no path provided, return self +// @see https://github.com/lodash/lodash/pull/1270 +// TODO(question): should add default value as lodash? +function get(object, path?: string | Array) { + return path == null || !path.length ? object : getWithPath(object, path); +} + +// NOTE: +// this method could only be used in executing nodes. +// because type of 'job' need loaded jobs in runtime execution. +// or the execution should be prepared first. +export function calculate(operand: Operand, lastJob: JobModel, execution: ExecutionModel) { + switch (operand.type) { + // @Deprecated + // from execution context + case 'context': + return get(execution.context, operand.options.path); + + // @Deprecated + // from last job (or input job) + case 'input': + return lastJob ?? get(lastJob.result, operand.options.path); + + // @Deprecated + // from job in execution + case 'job': + // assume jobs have been fetched from execution before + const job = execution.jobsMapByNodeId[operand.options.nodeId]; + return job && get(job, operand.options.path); + + case 'calculation': + const fn = calculators.get(operand.options.calculator); + if (!fn) { + throw new Error(`no calculator function registered for "${operand.options.calculator}"`); + } + return fn(...operand.options.operands.map(item => calculate(item, lastJob, execution))); + + // constant + default: + return operand.value; + } +} + + + +// built-in functions + +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; +} + +calculators.register('equal', equal); +calculators.register('gt', gt); +calculators.register('gte', gte); +calculators.register('lt', lt); +calculators.register('lte', lte); + +calculators.register('===', equal); +calculators.register('>', gt); +calculators.register('>=', gte); +calculators.register('<', lt); +calculators.register('<=', lte); + + + +function add(...args) { + return args.reduce((sum, a) => sum + a, 0); +} + +function minus(a, b) { + return a - b; +} + +function multipe(...args) { + return args.reduce((result, a) => result * a, 1); +} + +function divide(a, b) { + return a / b; +} + +function mod(a, b) { + return a % b; +} + +calculators.register('add', add); +calculators.register('minus', minus); +calculators.register('multipe', multipe); +calculators.register('divide', divide); +calculators.register('mod', mod); + +calculators.register('+', add); +calculators.register('-', minus); +calculators.register('*', multipe); +calculators.register('/', divide); +calculators.register('%', mod); + +calculators.register('now', () => new Date()); + +// TODO: add more common calculators diff --git a/packages/plugin-workflow/src/instructions/calculation.ts b/packages/plugin-workflow/src/instructions/calculation.ts new file mode 100644 index 000000000..a35c2ae8e --- /dev/null +++ b/packages/plugin-workflow/src/instructions/calculation.ts @@ -0,0 +1,48 @@ +import { JOB_STATUS } from "../constants"; +import FlowNodeModel from "../models/FlowNode"; +import { calculate } from "../calculators"; + +// @calculation: { +// calculator: 'concat', +// operands: [ +// { +// type: 'calculation', +// options: { +// calculator: 'add', +// operands: [{ value: 1 }, { value: 2 }] +// } +// }, +// { +// type: 'constant', +// value: '{{$context.data.title}}' +// }, +// { +// type: 'context', +// options: { +// path: 'data.title' +// } +// }, +// { +// type: 'constant', +// value: 1 +// } +// ] +// } + +export default { + async run(this: FlowNodeModel, prevJob, execution) { + const { calculation } = this.config || {}; + + const result = calculation + ? calculate({ + type: 'calculation', + options: execution.getParsedValue(calculation) + }, prevJob, execution) + : null; + + return { + result, + status: JOB_STATUS.RESOLVED + }; + } +} diff --git a/packages/plugin-workflow/src/instructions/condition.ts b/packages/plugin-workflow/src/instructions/condition.ts index 8d77a620d..2f8bcbe4e 100644 --- a/packages/plugin-workflow/src/instructions/condition.ts +++ b/packages/plugin-workflow/src/instructions/condition.ts @@ -1,23 +1,5 @@ -// 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 Sequelize from 'sequelize'; -import { getValue, Operand } from "../utils/getter"; -import { getCalculator } from "../utils/calculators"; +import { calculate, Operand } from "../calculators"; +import calculators from "../calculators"; import { JOB_STATUS } from "../constants"; type BaseCalculation = { @@ -41,21 +23,48 @@ type GroupCalculation = BaseCalculation & { // TODO(type) type Calculation = SingleCalculation | GroupCalculation; -function calculate(config, input, execution) { - if (!config) { +// @calculation: { +// not: false, +// group: { +// type: 'and', +// calculations: [ +// { +// calculator: 'time.equal', +// operands: [{ value: '{{$context.time}}' }, { value: '{{$fn.now}}' }] +// }, +// { +// calculator: 'value.equal', +// operands: [{ value: '{{$jobsMapByNodeId.213}}' }, { value: 1 }] +// }, +// { +// group: { +// type: 'or', +// calculations: [ +// { +// calculator: 'value.equal', +// operands: [{ value: '{{$jobsMapByNodeId.213}}' }, { value: 1 }] +// } +// ] +// } +// } +// ] +// } +// } +function logicCalculate(calculation, input, execution) { + if (!calculation) { return true; } - const { not, group } = config; + const { not, group } = calculation; let result; if (group) { const method = group.type === 'and' ? 'every' : 'some'; - result = group.calculations[method](calculation => calculate(calculation, input, execution)); + result = group.calculations[method](item => logicCalculate(item, input, execution)); } else { - const args = config.operands.map(operand => getValue(operand, input, execution)); - const fn = getCalculator(config.calculator); + const args = calculation.operands.map(operand => calculate(operand, input, execution)); + const fn = calculators.get(calculation.calculator); if (!fn) { - throw new Error(`no calculator function registered for "${config.calculator}"`); + throw new Error(`no calculator function registered for "${calculation.calculator}"`); } result = fn(...args); } @@ -68,10 +77,10 @@ export default { async run(this, prevJob, execution) { // TODO(optimize): loading of jobs could be reduced and turned into incrementally in execution // const jobs = await execution.getJobs(); - const { calculation } = this.config || {}; - const result = calculate(calculation, prevJob, execution); + const { calculation, rejectOnFalse } = this.config || {}; + const result = logicCalculate(calculation, prevJob, execution); - if (!result && this.config.rejectOnFalse) { + if (!result && rejectOnFalse) { return { status: JOB_STATUS.REJECTED, result @@ -83,7 +92,7 @@ export default { result, // TODO(optimize): try unify the building of job nodeId: this.id, - upstreamId: prevJob instanceof Sequelize.Model ? prevJob.get('id') : null + upstreamId: prevJob && prevJob.id || null }; const branchNode = execution.nodes @@ -95,7 +104,7 @@ export default { const savedJob = await execution.saveJob(job); - return execution.exec(branchNode, savedJob); + return execution.run(branchNode, savedJob); }, async resume(this, branchJob, execution) { @@ -107,4 +116,4 @@ export default { // pass control to upper scope by ending current scope return execution.end(this, branchJob); } -} +}; diff --git a/packages/plugin-workflow/src/instructions/index.ts b/packages/plugin-workflow/src/instructions/index.ts index 263b30f78..e3e1ed7a9 100644 --- a/packages/plugin-workflow/src/instructions/index.ts +++ b/packages/plugin-workflow/src/instructions/index.ts @@ -1,17 +1,20 @@ -import ExecutionModel from "../models/Execution"; -import FlowNodeModel from "../models/FlowNode"; +import { Registry } from '@nocobase/utils'; + +import ExecutionModel from '../models/Execution'; +import FlowNodeModel from '../models/FlowNode'; import prompt from './prompt'; +import calculation from './calculation'; import condition from './condition'; import parallel from './parallel'; -import query from "./query"; -import create from "./create"; -import update from "./update"; -import destroy from "./destroy"; +import query from './query'; +import create from './create'; +import update from './update'; +import destroy from './destroy'; export interface Job { status: number; - result: unknown; + result?: unknown; [key: string]: unknown; } @@ -29,24 +32,24 @@ export interface Instruction { // - could be the workflow execution object (containing context data) execution: ExecutionModel ): InstructionResult; + // for start node in main flow (or branch) to resume when manual sub branch triggered - resume?(): InstructionResult + resume?( + this: FlowNodeModel, + input: any, + execution: ExecutionModel + ): InstructionResult } -const registery = new Map(); +export const instructions = new Registry(); -export function getInstruction(key: string): Instruction { - return registery.get(key); -} +instructions.register('prompt', prompt); +instructions.register('calculation', calculation); +instructions.register('condition', condition); +instructions.register('parallel', parallel); +instructions.register('query', query); +instructions.register('create', create); +instructions.register('update', update); +instructions.register('destroy', destroy); -export function registerInstruction(key: string, instruction: any) { - registery.set(key, instruction); -} - -registerInstruction('prompt', prompt); -registerInstruction('condition', condition); -registerInstruction('parallel', parallel); -registerInstruction('query', query); -registerInstruction('create', create); -registerInstruction('update', update); -registerInstruction('destroy', destroy); +export default instructions; diff --git a/packages/plugin-workflow/src/instructions/parallel.ts b/packages/plugin-workflow/src/instructions/parallel.ts index 9c3a775f4..31b0795b5 100644 --- a/packages/plugin-workflow/src/instructions/parallel.ts +++ b/packages/plugin-workflow/src/instructions/parallel.ts @@ -57,7 +57,7 @@ export default { // 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()); + await branches.reduce((promise: Promise, branch) => promise.then(() => execution.run(branch, job)), Promise.resolve()); return execution.end(this, job); }, @@ -89,4 +89,4 @@ export default { return job; } -} +}; diff --git a/packages/plugin-workflow/src/instructions/prompt.ts b/packages/plugin-workflow/src/instructions/prompt.ts index ec3455eb4..70161531e 100644 --- a/packages/plugin-workflow/src/instructions/prompt.ts +++ b/packages/plugin-workflow/src/instructions/prompt.ts @@ -11,4 +11,4 @@ export default { job.set('status', JOB_STATUS.RESOLVED); return job; } -} +}; diff --git a/packages/plugin-workflow/src/instructions/query.ts b/packages/plugin-workflow/src/instructions/query.ts index e641b8239..6a5cdff8f 100644 --- a/packages/plugin-workflow/src/instructions/query.ts +++ b/packages/plugin-workflow/src/instructions/query.ts @@ -21,4 +21,4 @@ export default { status: JOB_STATUS.RESOLVED }; } -} +}; diff --git a/packages/plugin-workflow/src/models/Execution.ts b/packages/plugin-workflow/src/models/Execution.ts index 0587c0abf..2e58a5d01 100644 --- a/packages/plugin-workflow/src/models/Execution.ts +++ b/packages/plugin-workflow/src/models/Execution.ts @@ -2,11 +2,11 @@ import { Database, Model } from '@nocobase/database'; import parse from 'json-templates'; import { BelongsToGetAssociationMixin, HasManyGetAssociationsMixin, Transaction } from 'sequelize'; import { EXECUTION_STATUS, JOB_STATUS } from '../constants'; -import { getInstruction } from '../instructions'; +import instructions from '../instructions'; +import WorkflowModel from './Workflow'; import FlowNodeModel from './FlowNode'; import JobModel from './Job'; -import WorkflowModel from './Workflow'; - +import calculators from '../calculators'; export interface ExecutionOptions { transaction?: Transaction; @@ -97,21 +97,21 @@ export default class ExecutionModel extends Model { } } - async start(options: ExecutionOptions) { + public async start(options: ExecutionOptions) { if (this.status !== EXECUTION_STATUS.STARTED) { throw new Error(`execution was ended with status ${this.status}`); } await this.prepare(options); if (this.nodes.length) { - const head = this.nodes.find((item) => !item.upstream); - await this.exec(head, { result: this.context }); + const head = this.nodes.find(item => !item.upstream); + await this.run(head, { result: this.context }); } else { await this.exit(null); } await this.commit(); } - async resume(job: JobModel, options: ExecutionOptions) { + public async resume(job: JobModel, options: ExecutionOptions) { if (this.status !== EXECUTION_STATUS.STARTED) { throw new Error(`execution was ended with status ${this.status}`); } @@ -127,7 +127,7 @@ export default class ExecutionModel extends Model { } } - private async run(instruction, node: FlowNodeModel, prevJob) { + private async exec(instruction: Function, node: FlowNodeModel, prevJob) { let job; try { // call instruction to get result and status @@ -164,21 +164,24 @@ export default class ExecutionModel extends Model { if (savedJob.get('status') === JOB_STATUS.RESOLVED && node.downstream) { // run next node - return this.exec(node.downstream, savedJob); + return this.run(node.downstream, savedJob); } // all nodes in scope have been executed return this.end(node, savedJob); } - async exec(node, input?) { - const { run } = getInstruction(node.type); + public async run(node, input?) { + const { run } = instructions.get(node.type); + if (typeof run !== 'function') { + return Promise.reject(new Error('`run` should be implemented for customized execution of the node')); + } - return this.run(run, node, input); + return this.exec(run, node, input); } // parent node should take over the control - end(node, job) { + public end(node, job) { const parentNode = this.findBranchParentNode(node); // no parent, means on main flow if (parentNode) { @@ -191,12 +194,12 @@ export default class ExecutionModel extends Model { } async recall(node, job) { - const { resume } = getInstruction(node.type); - if (!resume) { + const { resume } = instructions.get(node.type); + if (typeof resume !== 'function') { return Promise.reject(new Error('`resume` should be implemented because the node made branch')); } - return this.run(resume, node, job); + return this.exec(resume, node, job); } async exit(job: JobModel | null) { @@ -251,10 +254,20 @@ export default class ExecutionModel extends Model { return null; } - getParsedValue(value) { + public getParsedValue(value, node?) { + const injectedFns = {}; + const scope = { + execution: this, + node + }; + for (let [name, fn] of calculators.getEntities()) { + injectedFns[name] = fn.bind(scope); + } + return parse(value)({ $context: this.context, $jobsMapByNodeId: this.jobsMapByNodeId, + $fn: injectedFns }); } } diff --git a/packages/plugin-workflow/src/models/Workflow.ts b/packages/plugin-workflow/src/models/Workflow.ts index 578081d08..0625d1a71 100644 --- a/packages/plugin-workflow/src/models/Workflow.ts +++ b/packages/plugin-workflow/src/models/Workflow.ts @@ -1,7 +1,8 @@ import { Database, Model } from '@nocobase/database'; import { HasManyCreateAssociationMixin, HasManyGetAssociationsMixin } from 'sequelize'; + +import triggers from '../triggers'; import { EXECUTION_STATUS } from '../constants'; -import { get as getTrigger } from '../triggers'; import ExecutionModel from './Execution'; import FlowNodeModel from './FlowNode'; @@ -47,7 +48,7 @@ export default class WorkflowModel extends Model { async toggle(enable?: boolean) { const type = this.get('type'); - const { on, off } = getTrigger(type); + const { on, off } = triggers.get(type); if (typeof enable !== 'undefined' ? enable : this.get('enabled')) { on.call(this, this.start.bind(this)); } else { diff --git a/packages/plugin-workflow/src/server.ts b/packages/plugin-workflow/src/server.ts index 5548a3930..8316c3eae 100644 --- a/packages/plugin-workflow/src/server.ts +++ b/packages/plugin-workflow/src/server.ts @@ -5,6 +5,10 @@ import { Plugin } from '@nocobase/server'; import WorkflowModel from './models/Workflow'; import ExecutionModel from './models/Execution'; +export * from './calculators'; +export * from './triggers'; +export * from './instructions'; + export default class WorkflowPlugin extends Plugin { async load(options = {}) { const { db } = this.app; diff --git a/packages/plugin-workflow/src/triggers/index.ts b/packages/plugin-workflow/src/triggers/index.ts index a48982e81..ca26d4bd2 100644 --- a/packages/plugin-workflow/src/triggers/index.ts +++ b/packages/plugin-workflow/src/triggers/index.ts @@ -1,3 +1,4 @@ +import { Registry } from '@nocobase/utils'; import WorkflowModel from '../models/Workflow'; import modelTrigger from './model'; @@ -7,14 +8,8 @@ export interface Trigger { off(this: WorkflowModel): void; } -const triggers = new Map(); +export const triggers = new Registry(); -export function register(type: string, trigger: Trigger): void { - triggers.set(type, trigger); -} +export default triggers; -export function get(type: string): Trigger | undefined { - return triggers.get(type); -} - -register(modelTrigger.name, modelTrigger); +triggers.register(modelTrigger.name, modelTrigger); diff --git a/packages/plugin-workflow/src/utils/calculators.ts b/packages/plugin-workflow/src/utils/calculators.ts deleted file mode 100644 index 33cfd16fe..000000000 --- a/packages/plugin-workflow/src/utils/calculators.ts +++ /dev/null @@ -1,52 +0,0 @@ -type Calculator = (...args: any[]) => any; - -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/utils/getter.ts b/packages/plugin-workflow/src/utils/getter.ts deleted file mode 100644 index a27459876..000000000 --- a/packages/plugin-workflow/src/utils/getter.ts +++ /dev/null @@ -1,68 +0,0 @@ -import { get as getWithPath } from 'lodash'; - -import ExecutionModel from '../models/Execution'; -import JobModel from '../models/Job'; - -export type OperandType = 'context' | 'input' | 'job'; - -export type ObjectGetterOptions = { - path?: string -}; - -export type JobGetterOptions = ObjectGetterOptions & { - nodeId: 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; -}; - -// TODO(type): union type here is wrong -export type Operand = ContextOperand | InputOperand | JobOperand | ConstantOperand; - -// HACK: if no path provided, return self -// @see https://github.com/lodash/lodash/pull/1270 -// TODO(question): should add default value as lodash? -function get(object, path?: string | Array) { - return path == null || !path.length ? object : getWithPath(object, path); -} - - -// NOTE: -// this method could only be used in executing nodes. -// because type of 'job' need loaded jobs in runtime execution. -// or the execution should be prepared first. -export function getValue(operand: Operand, lastJob: JobModel, execution: ExecutionModel) { - switch (operand.type) { - // from execution context - case 'context': - return get(execution.context, operand.options.path); - // from last job (or input job) - case 'input': - return lastJob ?? get(lastJob.result, operand.options.path); - // from job in execution - case 'job': - // assume jobs have been fetched from execution before - const job = execution.jobsMapByNodeId[operand.options.nodeId]; - return job && get(job, operand.options.path); - // constant - default: - return operand.value; - } -} diff --git a/packages/utils/src/index.ts b/packages/utils/src/index.ts index d490a7cc4..02f1a11ac 100644 --- a/packages/utils/src/index.ts +++ b/packages/utils/src/index.ts @@ -2,3 +2,4 @@ export * from './merge'; export * from './mixin'; export * from './mixin/AsyncEmitter'; export * from './uid'; +export * from './registry'; diff --git a/packages/utils/src/registry.ts b/packages/utils/src/registry.ts new file mode 100644 index 000000000..d506e1ee0 --- /dev/null +++ b/packages/utils/src/registry.ts @@ -0,0 +1,42 @@ +import path from 'path'; +import { promises as fs } from 'fs'; + +export interface RegistryOptions { + override: boolean; +} + +export class Registry { + private map = new Map(); + options: RegistryOptions; + + constructor(options: RegistryOptions = { override: false }) { + this.options = options; + } + + register(key: string, value: T): void { + if (!this.options.override && this.map.has(key)) { + throw new Error(`this registry does not allow to override existing keys: "${key}"`); + } + + this.map.set(key, value); + } + + // async import({ directory, extensions = ['.js', '.ts', '.json'] }) { + // const files = await fs.readdir(directory); + // return files.filter(file => extensions.includes(path.extname(file))) + // } + + get(key: string): T { + return this.map.get(key); + } + + getValues(): Iterable { + return this.map.values(); + } + + getEntities(): Iterable<[string, T]> { + return this.map.entries(); + } +} + +export default Registry;