Feat: plugin workflow (#210)

* feat(plugin-workflow): refactor calculator and some api

* fix(plugin-workflow): comments

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": {

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;

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,7 +18,8 @@ export async function getApp(options = {}): Promise<MockServer> {
   app.plugin(plugin);
 
   // for test only
-  registerInstruction('echo', {
+  if (!instructions.get('echo')) {
+    instructions.register('echo', {
       run(this, { result }, execution) {
         return {
           status: JOB_STATUS.RESOLVED,
@@ -25,14 +27,18 @@ export async function getApp(options = {}): Promise<MockServer> {
         };
       }
     });
+  }
 
-  registerInstruction('error', {
+  if (!instructions.get('error')) {
+    instructions.register('error', {
       run(this, input, execution) {
         throw new Error('definite error');
       }
     });
+  }
 
-  registerInstruction('prompt->error', {
+  if (!instructions.get('prompt->error')) {
+    instructions.register('prompt->error', {
       run(this, input, execution) {
         return {
           status: JOB_STATUS.PENDING
@@ -42,6 +48,11 @@ export async function getApp(options = {}): Promise<MockServer> {
         throw new Error('input failed');
       }
     });
+  }
+
+  if (!calculators.get('no1')) {
+    calculators.register('no1', () => 1);
+  }
 
   await app.load();
 

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);
+    });
+  });
+});

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);
+    });
   });
 });

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;

packages/plugin-workflow/src/__tests__/utils/getter.test.ts

@@ -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);
-    });
-  });
-});

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<Function>();
+
+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<string>) {
+  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

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
+    };
+  }
+}

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);
   }
-}
+};

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<string, Instruction>();
+export const instructions = new Registry<Instruction>();
 
-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;

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<any>, branch) => promise.then(() => execution.exec(branch, job)), Promise.resolve());
+    await branches.reduce((promise: Promise<any>, branch) => promise.then(() => execution.run(branch, job)), Promise.resolve());
 
     return execution.end(this, job);
   },
@@ -89,4 +89,4 @@ export default {
 
     return job;
   }
-}
+};

packages/plugin-workflow/src/instructions/prompt.ts

@@ -11,4 +11,4 @@ export default {
     job.set('status', JOB_STATUS.RESOLVED);
     return job;
   }
-}
+};

packages/plugin-workflow/src/instructions/query.ts

@@ -21,4 +21,4 @@ export default {
       status: JOB_STATUS.RESOLVED
     };
   }
-}
+};

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
     });
   }
 }

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 {

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;

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<string, Trigger>();
+export const triggers = new Registry<Trigger>();
 
-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);

packages/plugin-workflow/src/utils/calculators.ts

@@ -1,52 +0,0 @@
-type Calculator = (...args: any[]) => any;
-
-const calculators = new Map<string, Calculator>();
-
-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
-});

packages/plugin-workflow/src/utils/getter.ts

@@ -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<string>) {
-  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;
-  }
-}

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';

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<T> {
+  private map = new Map<string, T>();
+  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<T> {
+    return this.map.values();
+  }
+
+  getEntities(): Iterable<[string, T]> {
+    return this.map.entries();
+  }
+}
+
+export default Registry;