refactor(plugin-workflow): split transaction for collection trigger (#1080)

* refactor(plugin-workflow): split transaction for collection trigger

* fix(plugin-workflow): fix delay test case

* refactor(plugin-workflow): use micro queue to dispatch executions

* fix(plugin-workflow): fix usages of private api

* fix(plugin-workflow): fix wrong variable

packages/plugins/users/src/server.ts

@@ -21,14 +21,11 @@ export interface UserPluginConfig {
 export default class UsersPlugin extends Plugin<UserPluginConfig> {
   public jwtService: JwtService;
 
-  public tokenMiddleware: Middleware;
-
   public authenticators: Registry<HandlerType> = new Registry();
 
   constructor(app, options) {
     super(app, options);
     this.jwtService = new JwtService(options?.jwt || {});
-    this.tokenMiddleware = new Middleware(parseToken);
   }
 
   async beforeLoad() {

packages/plugins/workflow/src/server/Plugin.ts

@@ -22,6 +22,8 @@ export default class WorkflowPlugin extends Plugin {
   triggers: Registry<Trigger> = new Registry();
   calculators = calculators;
   extensions = extensions;
+  executing: Promise<any> = null;
+  pending: [ExecutionModel, JobModel][] = [];
 
   onBeforeSave = async (instance: WorkflowModel, options) => {
     const Model = <typeof WorkflowModel>instance.constructor;
@@ -127,39 +129,38 @@ export default class WorkflowPlugin extends Plugin {
     }
   }
 
-  async trigger(workflow: WorkflowModel, context: Object, options: Transactionable & { context?: any } = {}): Promise<ExecutionModel | null> {
+  public async trigger(workflow: WorkflowModel, context: Object, options: Transactionable & { context?: any } = {}): Promise<void> {
     // `null` means not to trigger
-    if (context === null) {
+    if (context == null) {
       return null;
     }
 
-    let transaction = null;
+    // @ts-ignore
-
+    const transaction = options.transaction && !options.transaction.finished
-    if (workflow.useTransaction) {
+      ? options.transaction
-      // @ts-ignore
+      : await (<typeof WorkflowModel>workflow.constructor).database.sequelize.transaction();
-      transaction = options.transaction && !options.transaction.finished
-        ? options.transaction
-        : await (<typeof WorkflowModel>workflow.constructor).database.sequelize.transaction();
 
+    if (options.context?.transaction) {
       const existed = await workflow.countExecutions({
         where: {
-          transaction: transaction.id
+          transaction: options.context.transaction
         },
         transaction
       });
 
       if (existed) {
-        console.warn(`workflow ${workflow.id} has already been triggered in same execution (${transaction.id}), and newly triggering will be skipped.`);
+        console.warn(`workflow ${workflow.id} has already been triggered in same execution (${options.context.transaction}), and newly triggering will be skipped.`);
-        return null;
+        return;
       }
     }
 
     const execution = await workflow.createExecution({
       context,
       key: workflow.key,
-      status: EXECUTION_STATUS.STARTED,
+      status: EXECUTION_STATUS.CREATED,
       useTransaction: workflow.useTransaction,
-      transaction: transaction?.id
+      // @ts-ignore
+      transaction: options.context?.transaction ?? transaction?.id
     }, { transaction });
 
     console.log('workflow triggered:', new Date(), workflow.id, execution.id);
@@ -187,27 +188,61 @@ export default class WorkflowPlugin extends Plugin {
 
     execution.workflow = workflow;
 
-    const processor = this.createProcessor(execution, { transaction, _context: options.context });
-
-    await processor.start();
-
     // @ts-ignore
     if (transaction && (!options.transaction || options.transaction.finished)) {
       await transaction.commit();
     }
 
-    return execution;
+    setTimeout(() => this.dispatch(execution), 0);
   }
 
-  async resume(job) {
+  public async resume(job) {
     if (!job.execution) {
       job.execution = await job.getExecution();
     }
-    const processor = this.createProcessor(job.execution);
+
-    return processor.resume(job);
+    setTimeout(() => this.dispatch(job.execution, job), 0);
   }
 
-  createProcessor(execution: ExecutionModel, options = {}): Processor {
+  private async dispatch(execution?: ExecutionModel, job?: JobModel) {
+    if (this.executing) {
+      if (job) {
+        this.pending.push([execution, job]);
+      }
+      return;
+    }
+
+    if (!execution) {
+      execution = await this.db.getRepository('executions').findOne({
+        filter: {
+          status: EXECUTION_STATUS.CREATED
+        },
+        sort: 'createdAt'
+      }) as ExecutionModel;
+      if (!execution) {
+        return;
+      }
+    }
+
+    if (execution.status === EXECUTION_STATUS.CREATED) {
+      await execution.update({ status: EXECUTION_STATUS.STARTED });
+    }
+
+    const processor = this.createProcessor(execution, { _context: { transaction: execution.transaction} });
+
+    this.executing = job ? processor.resume(job) : processor.start();
+
+    await this.executing;
+
+    this.executing = null;
+
+    setTimeout(() => {
+      const args = this.pending.length ? this.pending.shift() : [];
+      this.dispatch(...args);
+    });
+  }
+
+  private createProcessor(execution: ExecutionModel, options = {}): Processor {
     return new Processor(execution, { ...options, plugin: this });
   }
 }

packages/plugins/workflow/src/server/Processor.ts

@@ -75,17 +75,9 @@ export default class Processor {
     const { sequelize } = (<typeof ExecutionModel>this.execution.constructor).database;
 
     // @ts-ignore
-    const transaction = options.transaction && !options.transaction.finished
+    return options.transaction && !options.transaction.finished
       ? options.transaction
       : await sequelize.transaction();
-
-    // @ts-ignore
-    if (this.execution.transaction !== transaction.id) {
-
-    // @ts-ignore
-      await this.execution.update({ transaction: transaction.id }, { transaction });
-    }
-    return transaction;
   }
 
   async prepare(commit?: boolean) {

packages/plugins/workflow/src/server/__tests__/Plugin.test.ts

@@ -1,6 +1,6 @@
 import { MockServer } from '@nocobase/test';
 import Database from '@nocobase/database';
-import { getApp } from '.';
+import { getApp, sleep } from '.';
 
 
 
@@ -61,6 +61,9 @@ describe('workflow > Plugin', () => {
       expect(workflow.current).toBe(true);
 
       const p1 = await PostRepo.create({ values: { title: 't1' } });
+
+      await sleep(500);
+
       const count = await workflow.countExecutions();
       expect(count).toBe(0);
 
@@ -68,6 +71,9 @@ describe('workflow > Plugin', () => {
       expect(workflow.current).toBe(true);
 
       const p2 = await PostRepo.create({ values: { title: 't2' } });
+
+      await sleep(500);
+
       const executions = await workflow.getExecutions();
       expect(executions.length).toBe(1);
     });
@@ -82,7 +88,11 @@ describe('workflow > Plugin', () => {
         }
       });
       expect(workflow.current).toBe(true);
+
       const p1 = await PostRepo.create({ values: { title: 't1' } });
+
+      await sleep(500);
+
       const c1 = await workflow.countExecutions();
       expect(c1).toBe(1);
 
@@ -90,6 +100,9 @@ describe('workflow > Plugin', () => {
       expect(workflow.current).toBe(true);
 
       const p2 = await PostRepo.create({ values: { title: 't2' } });
+
+      await sleep(500);
+
       const c2 = await workflow.countExecutions();
       expect(c2).toBe(1);
     });
@@ -105,6 +118,9 @@ describe('workflow > Plugin', () => {
       });
 
       const p1 = await PostRepo.create({ values: { title: 't1' } });
+
+      await sleep(500);
+
       const c1 = await workflow.countExecutions();
       expect(c1).toBe(1);
 
@@ -114,6 +130,9 @@ describe('workflow > Plugin', () => {
       expect(workflow.current).toBe(true);
 
       const p2 = await PostRepo.create({ values: { title: 't2' } });
+
+      await sleep(500);
+
       const c2 = await workflow.countExecutions();
       expect(c2).toBe(1);
 
@@ -123,6 +142,9 @@ describe('workflow > Plugin', () => {
       expect(workflow.current).toBe(true);
 
       const p3 = await PostRepo.create({ values: { title: 't3' } });
+
+      await sleep(500);
+
       const c3 = await workflow.countExecutions();
       expect(c3).toBe(2);
     });
@@ -138,6 +160,9 @@ describe('workflow > Plugin', () => {
       });
 
       const p1 = await PostRepo.create({ values: { title: 't1' } });
+
+      await sleep(500);
+
       const c1 = await workflow.countExecutions();
       expect(c1).toBe(1);
 
@@ -149,6 +174,9 @@ describe('workflow > Plugin', () => {
       });
 
       const p2 = await PostRepo.create({ values: { title: 't2' } });
+
+      await sleep(500);
+
       const c2 = await workflow.countExecutions();
       expect(c2).toBe(1);
     });

packages/plugins/workflow/src/server/__tests__/Processor.test.ts

@@ -1,6 +1,6 @@
 import Database from '@nocobase/database';
 import { Application } from '@nocobase/server';
-import { getApp } from '.';
+import { getApp, sleep } from '.';
 import { BRANCH_INDEX, EXECUTION_STATUS, JOB_STATUS } from '../constants';
 
 
@@ -37,6 +37,8 @@ describe('workflow > Processor', () => {
     it('empty workflow without any nodes', async () => {
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       expect(execution.context.data.title).toEqual(post.title);
       expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED);
@@ -49,6 +51,8 @@ describe('workflow > Processor', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED);
 
@@ -65,6 +69,8 @@ describe('workflow > Processor', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       expect(execution.context.data.title).toEqual(post.title);
       expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED);
@@ -92,6 +98,8 @@ describe('workflow > Processor', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       expect(execution.context.data.title).toEqual(post.title);
       expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED);
@@ -110,6 +118,8 @@ describe('workflow > Processor', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       expect(execution.status).toEqual(EXECUTION_STATUS.REJECTED);
 
@@ -136,6 +146,8 @@ describe('workflow > Processor', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       expect(execution.status).toEqual(EXECUTION_STATUS.STARTED);
       const [pending] = await execution.getJobs();
@@ -143,8 +155,11 @@ describe('workflow > Processor', () => {
       expect(pending.result).toEqual(null);
 
       pending.set('result', 123);
-      const processor = plugin.createProcessor(execution);
+      pending.execution = execution;
-      await processor.resume(pending);
+      await plugin.resume(pending);
+
+      await sleep(500);
+
       expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED);
 
       const jobs = await execution.getJobs({ order: [['id', 'ASC']] });
@@ -167,6 +182,8 @@ describe('workflow > Processor', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       expect(execution.status).toEqual(EXECUTION_STATUS.STARTED);
       const [pending] = await execution.getJobs();
@@ -174,8 +191,11 @@ describe('workflow > Processor', () => {
       expect(pending.result).toEqual(null);
 
       pending.set('result', 123);
-      const processor = plugin.createProcessor(execution);
+      pending.execution = execution;
-      await processor.resume(pending);
+      await plugin.resume(pending);
+
+      await sleep(500);
+
       expect(execution.status).toEqual(EXECUTION_STATUS.REJECTED);
 
       const jobs = await execution.getJobs();
@@ -206,6 +226,8 @@ describe('workflow > Processor', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED);
 
@@ -237,13 +259,17 @@ describe('workflow > Processor', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       expect(execution.status).toEqual(EXECUTION_STATUS.STARTED);
 
       const [pending] = await execution.getJobs({ where: { nodeId: n2.id } });
       pending.set('result', 123);
-      const processor = plugin.createProcessor(execution);
+      pending.execution = execution;
-      await processor.resume(pending);
+      await plugin.resume(pending);
+
+      await sleep(500);
 
       const jobs = await execution.getJobs();
       expect(jobs.length).toEqual(3);
@@ -270,13 +296,18 @@ describe('workflow > Processor', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       expect(execution.status).toEqual(EXECUTION_STATUS.STARTED);
 
       const [pending] = await execution.getJobs({ where: { nodeId: n2.id } });
       pending.set('result', 123);
-      const processor = plugin.createProcessor(execution);
+      pending.execution = execution;
-      await processor.resume(pending);
+      await plugin.resume(pending);
+
+      await sleep(500);
+
       expect(execution.status).toEqual(EXECUTION_STATUS.REJECTED);
 
       const jobs = await execution.getJobs();
@@ -319,6 +350,8 @@ describe('workflow > Processor', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       expect(execution.status).toEqual(EXECUTION_STATUS.STARTED);
 
@@ -327,8 +360,10 @@ describe('workflow > Processor', () => {
 
       const pending = pendingJobs.find(item => item.nodeId === n3.id );
       pending.set('result', 123);
-      const processor = plugin.createProcessor(execution);
+      pending.execution = execution;
-      await processor.resume(pending);
+      await plugin.resume(pending);
+
+      await sleep(500);
 
       expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED);
       const jobs = await execution.getJobs({ order: [['id', 'ASC']] });
@@ -369,6 +404,8 @@ describe('workflow > Processor', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [e1] = await workflow.getExecutions();
       expect(e1.status).toEqual(EXECUTION_STATUS.STARTED);
 
@@ -377,8 +414,10 @@ describe('workflow > Processor', () => {
 
       const pending = pendingJobs.find(item => item.nodeId === n2.id );
       pending.set('result', 123);
-      const processor = plugin.createProcessor(e1);
+      pending.execution = e1;
-      await processor.resume(pending);
+      await plugin.resume(pending);
+
+      await sleep(500);
 
       const [e2] = await workflow.getExecutions();
       expect(e2.status).toEqual(EXECUTION_STATUS.RESOLVED);
@@ -403,6 +442,8 @@ describe('workflow > Processor', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const posts = await PostRepo.find();
       expect(posts.length).toBe(2);
 

packages/plugins/workflow/src/server/__tests__/actions/workflows.test.ts

@@ -1,6 +1,6 @@
 import { MockServer } from '@nocobase/test';
 import Database from '@nocobase/database';
-import { getApp } from '..';
+import { getApp, sleep } from '..';
 
 
 
@@ -61,6 +61,9 @@ describe('workflow > actions > workflows', () => {
       });
 
       const p1 = await PostRepo.create({ values: { title: 't1' } });
+
+      await sleep(500);
+
       const c1 = await workflow.countExecutions();
       expect(c1).toBe(1);
 
@@ -87,6 +90,9 @@ describe('workflow > actions > workflows', () => {
       });
 
       const p1 = await PostRepo.create({ values: { title: 't1' } });
+
+      await sleep(500);
+
       const c1 = await workflow.countExecutions();
       expect(c1).toBe(1);
 
@@ -100,6 +106,9 @@ describe('workflow > actions > workflows', () => {
       expect(status).toBe(200);
 
       const p2 = await PostRepo.create({ values: { title: 't2' } });
+
+      await sleep(500);
+
       const c2 = await workflow.countExecutions();
       expect(c2).toBe(1);
     });
@@ -129,6 +138,8 @@ describe('workflow > actions > workflows', () => {
 
       const p1 = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       await WorkflowModel.update({
         enabled: true
       }, {
@@ -149,6 +160,8 @@ describe('workflow > actions > workflows', () => {
 
       const p2 = await PostRepo.create({ values: { title: 't2' } });
 
+      await sleep(500);
+
       const [e1] = await w1next.getExecutions();
       const [e2] = await w2next.getExecutions();
       expect(e1.key).toBe(e2.key);
@@ -230,6 +243,8 @@ describe('workflow > actions > workflows', () => {
         values: { title: 't1', read: 1 }
       });
 
+      await sleep(500);
+
       const [execution] = await w2.getExecutions();
       const [echo, calculation] = await execution.getJobs({ order: [['id', 'ASC']] });
       expect(calculation.result).toBe(2);

packages/plugins/workflow/src/server/__tests__/instructions/calculation.test.ts

@@ -1,6 +1,6 @@
 import { Application } from '@nocobase/server';
 import Database from '@nocobase/database';
-import { getApp } from '..';
+import { getApp, sleep } from '..';
 
 
 
@@ -47,6 +47,9 @@ describe('workflow > instructions > calculation', () => {
       });
 
       const post = await PostRepo.create({ values: { title: 't1' } });
+
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       const [job] = await execution.getJobs();
       expect(job.result).toBe(2);
@@ -67,6 +70,9 @@ describe('workflow > instructions > calculation', () => {
       });
 
       const post = await PostRepo.create({ values: { title: 't1' } });
+
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       const [job] = await execution.getJobs();
       expect(job.result).toBe(1);
@@ -87,6 +93,9 @@ describe('workflow > instructions > calculation', () => {
       });
 
       const post = await PostRepo.create({ values: { title: 't1' } });
+
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       const [job] = await execution.getJobs();
       expect(job.result).toBe(1);
@@ -114,6 +123,9 @@ describe('workflow > instructions > calculation', () => {
       await n1.setDownstream(n2);
 
       const post = await PostRepo.create({ values: { title: 't1' } });
+
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       const [n1Job, n2Job] = await execution.getJobs({ order: [['id', 'ASC']]});
       expect(n2Job.result).toBe(1);
@@ -141,6 +153,9 @@ describe('workflow > instructions > calculation', () => {
       await n1.setDownstream(n2);
 
       const post = await PostRepo.create({ values: { title: 't1' } });
+
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       const [n1Job, n2Job] = await execution.getJobs({ order: [['id', 'ASC']]});
       expect(n2Job.result).toBe(1);
@@ -161,6 +176,9 @@ describe('workflow > instructions > calculation', () => {
       });
 
       const post = await PostRepo.create({ values: { title: 't1' } });
+
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       const [job] = await execution.getJobs();
       expect(job.result).toBe(2);
@@ -192,6 +210,9 @@ describe('workflow > instructions > calculation', () => {
       });
 
       const post = await PostRepo.create({ values: { title: 't1' } });
+
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       const [job] = await execution.getJobs();
       expect(job.result).toBe(-1);

packages/plugins/workflow/src/server/__tests__/instructions/condition.test.ts

@@ -1,6 +1,6 @@
 import { Application } from '@nocobase/server';
 import Database from '@nocobase/database';
-import { getApp } from '..';
+import { getApp, sleep } from '..';
 import { EXECUTION_STATUS, BRANCH_INDEX } from '../../constants';
 
 
@@ -67,6 +67,8 @@ describe('workflow > instructions > condition', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED);
 
@@ -104,6 +106,8 @@ describe('workflow > instructions > condition', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED);
 
@@ -142,6 +146,8 @@ describe('workflow > instructions > condition', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       const [job] = await execution.getJobs();
       expect(job.result).toBe(true);
@@ -171,6 +177,8 @@ describe('workflow > instructions > condition', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       const [job] = await execution.getJobs();
       expect(job.result).toBe(false);
@@ -200,6 +208,8 @@ describe('workflow > instructions > condition', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions({ include: ['jobs'] });
       const [job] = execution.jobs;
       expect(job.result).toBe(true);
@@ -229,6 +239,8 @@ describe('workflow > instructions > condition', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       const [job] = await execution.getJobs();
       expect(job.result).toBe(false);
@@ -263,6 +275,8 @@ describe('workflow > instructions > condition', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       const [job] = await execution.getJobs();
       expect(job.result).toBe(false);

packages/plugins/workflow/src/server/__tests__/instructions/create.test.ts

@@ -1,6 +1,6 @@
 import { Application } from '@nocobase/server';
 import Database from '@nocobase/database';
-import { getApp } from '..';
+import { getApp, sleep } from '..';
 
 
 
@@ -47,6 +47,8 @@ describe('workflow > instructions > create', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       const [job] = await execution.getJobs();
       expect(job.result.postId).toBe(post.id);

packages/plugins/workflow/src/server/__tests__/instructions/delay.test.ts

@@ -36,13 +36,15 @@ describe('workflow > instructions > delay', () => {
       const n1 = await workflow.createNode({
         type: 'delay',
         config: {
-          duration: 1000,
+          duration: 2000,
           endStatus: JOB_STATUS.RESOLVED
         }
       });
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [e1] = await workflow.getExecutions();
       expect(e1.status).toEqual(EXECUTION_STATUS.STARTED);
       const [j1] = await e1.getJobs();
@@ -60,13 +62,15 @@ describe('workflow > instructions > delay', () => {
       const n1 = await workflow.createNode({
         type: 'delay',
         config: {
-          duration: 1000,
+          duration: 2000,
           endStatus: JOB_STATUS.REJECTED
         }
       });
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [e1] = await workflow.getExecutions();
       expect(e1.status).toEqual(EXECUTION_STATUS.STARTED);
       const [j1] = await e1.getJobs();
@@ -84,7 +88,7 @@ describe('workflow > instructions > delay', () => {
       const n1 = await workflow.createNode({
         type: 'delay',
         config: {
-          duration: 1000,
+          duration: 2000,
           endStatus: JOB_STATUS.RESOLVED
         }
       });
@@ -104,6 +108,8 @@ describe('workflow > instructions > delay', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [e1] = await workflow.getExecutions();
       expect(e1.status).toEqual(EXECUTION_STATUS.STARTED);
       const [j1] = await e1.getJobs();
@@ -133,6 +139,8 @@ describe('workflow > instructions > delay', () => {
     it('restart app should trigger delayed job', async () => {
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [e1] = await workflow.getExecutions();
       expect(e1.status).toEqual(EXECUTION_STATUS.STARTED);
       const [j1] = await e1.getJobs();
@@ -153,6 +161,8 @@ describe('workflow > instructions > delay', () => {
     it('restart app should trigger missed delayed job', async () => {
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [e1] = await workflow.getExecutions();
       expect(e1.status).toEqual(EXECUTION_STATUS.STARTED);
       const [j1] = await e1.getJobs();

packages/plugins/workflow/src/server/__tests__/instructions/destroy.test.ts

@@ -1,6 +1,6 @@
 import { Application } from '@nocobase/server';
 import Database from '@nocobase/database';
-import { getApp } from '..';
+import { getApp, sleep } from '..';
 
 
 
@@ -47,6 +47,8 @@ describe('workflow > instructions > destroy', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       const [job] = await execution.getJobs();
       expect(job.result).toBe(1);

packages/plugins/workflow/src/server/__tests__/instructions/parallel.test.ts

@@ -1,6 +1,6 @@
 import Database from '@nocobase/database';
 import { Application } from '@nocobase/server';
-import { getApp } from '..';
+import { getApp, sleep } from '..';
 import { EXECUTION_STATUS } from '../../constants';
 
 
@@ -51,6 +51,8 @@ describe('workflow > instructions > parallel', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       expect(execution.status).toBe(EXECUTION_STATUS.RESOLVED);
       const jobs = await execution.getJobs({ order: [['id', 'ASC']] });
@@ -74,6 +76,8 @@ describe('workflow > instructions > parallel', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       expect(execution.status).toBe(EXECUTION_STATUS.REJECTED);
       const jobs = await execution.getJobs({ order: [['id', 'ASC']] });
@@ -99,6 +103,8 @@ describe('workflow > instructions > parallel', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       expect(execution.status).toBe(EXECUTION_STATUS.REJECTED);
       const jobs = await execution.getJobs({ order: [['id', 'ASC']] });
@@ -127,6 +133,8 @@ describe('workflow > instructions > parallel', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       expect(execution.status).toBe(EXECUTION_STATUS.RESOLVED);
       const jobs = await execution.getJobs({ order: [['id', 'ASC']] });
@@ -153,6 +161,8 @@ describe('workflow > instructions > parallel', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       expect(execution.status).toBe(EXECUTION_STATUS.RESOLVED);
       const jobs = await execution.getJobs({ order: [['id', 'ASC']] });
@@ -179,6 +189,8 @@ describe('workflow > instructions > parallel', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       expect(execution.status).toBe(EXECUTION_STATUS.REJECTED);
       const jobs = await execution.getJobs({ order: [['id', 'ASC']] });
@@ -207,6 +219,8 @@ describe('workflow > instructions > parallel', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       expect(execution.status).toBe(EXECUTION_STATUS.RESOLVED);
       const jobs = await execution.getJobs({ order: [['id', 'ASC']] });
@@ -233,6 +247,8 @@ describe('workflow > instructions > parallel', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       expect(execution.status).toBe(EXECUTION_STATUS.REJECTED);
       const jobs = await execution.getJobs({ order: [['id', 'ASC']] });
@@ -263,6 +279,8 @@ describe('workflow > instructions > parallel', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       expect(execution.status).toBe(EXECUTION_STATUS.RESOLVED);
       const jobs = await execution.getJobs({ order: [['id', 'ASC']] });
@@ -298,6 +316,8 @@ describe('workflow > instructions > parallel', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       expect(execution.status).toBe(EXECUTION_STATUS.RESOLVED);
       const jobs = await execution.getJobs({ order: [['id', 'ASC']] });
@@ -325,6 +345,8 @@ describe('workflow > instructions > parallel', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       expect(execution.status).toBe(EXECUTION_STATUS.RESOLVED);
       const jobs = await execution.getJobs({ order: [['id', 'ASC']] });
@@ -360,13 +382,17 @@ describe('workflow > instructions > parallel', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       expect(execution.status).toBe(EXECUTION_STATUS.STARTED);
 
       const [pending] = await execution.getJobs({ where: { nodeId: n2.id } });
       pending.set('result', 123);
-      const processor = plugin.createProcessor(execution);
+      pending.execution = execution;
-      await processor.resume(pending);
+      await plugin.resume(pending);
+
+      await sleep(500);
 
       expect(execution.status).toBe(EXECUTION_STATUS.RESOLVED);
       const jobs = await execution.getJobs({ order: [['id', 'ASC']] });
@@ -406,6 +432,8 @@ describe('workflow > instructions > parallel', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       expect(execution.status).toBe(EXECUTION_STATUS.RESOLVED);
       const jobs = await execution.getJobs({ order: [['id', 'ASC']] });

packages/plugins/workflow/src/server/__tests__/instructions/prompt.test.ts

@@ -47,6 +47,8 @@ describe.skip('workflow > instructions > prompt', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [pending] = await workflow.getExecutions();
       expect(pending.status).toBe(EXECUTION_STATUS.STARTED);
       const [j1] = await pending.getJobs();
@@ -130,6 +132,8 @@ describe.skip('workflow > instructions > prompt', () => {
 
         const post = await PostRepo.create({ values: { title: 't1' } });
 
+        await sleep(500);
+
         const [pending] = await workflow.getExecutions();
         expect(pending.status).toBe(EXECUTION_STATUS.STARTED);
         const [j1] = await pending.getJobs();
@@ -194,6 +198,8 @@ describe.skip('workflow > instructions > prompt', () => {
 
         const post = await PostRepo.create({ values: { title: 't1' } });
 
+        await sleep(500);
+
         const [pending] = await workflow.getExecutions();
         expect(pending.status).toBe(EXECUTION_STATUS.STARTED);
         const [j1] = await pending.getJobs();
@@ -234,6 +240,8 @@ describe.skip('workflow > instructions > prompt', () => {
 
         const post = await PostRepo.create({ values: { title: 't1' } });
 
+        await sleep(500);
+
         const UserJobModel = db.getModel('users_jobs');
         const usersJobs = await UserJobModel.findAll();
         expect(usersJobs.length).toBe(1);
@@ -271,6 +279,8 @@ describe.skip('workflow > instructions > prompt', () => {
 
         const post = await PostRepo.create({ values: { title: 't1' } });
 
+        await sleep(500);
+
         const UserJobModel = db.getModel('users_jobs');
         const pendingJobs = await UserJobModel.findAll({
           order: [[ 'userId', 'ASC' ]]
@@ -327,6 +337,8 @@ describe.skip('workflow > instructions > prompt', () => {
 
         const post = await PostRepo.create({ values: { title: 't1' } });
 
+        await sleep(500);
+
         const UserJobModel = db.getModel('users_jobs');
         const pendingJobs = await UserJobModel.findAll({
           order: [[ 'userId', 'ASC' ]]
@@ -374,6 +386,8 @@ describe.skip('workflow > instructions > prompt', () => {
 
         const post = await PostRepo.create({ values: { title: 't1' } });
 
+        await sleep(500);
+
         const UserJobModel = db.getModel('users_jobs');
         const pendingJobs = await UserJobModel.findAll({
           order: [[ 'userId', 'ASC' ]]
@@ -431,6 +445,8 @@ describe.skip('workflow > instructions > prompt', () => {
 
         const post = await PostRepo.create({ values: { title: 't1' } });
 
+        await sleep(500);
+
         const UserJobModel = db.getModel('users_jobs');
         const pendingJobs = await UserJobModel.findAll({
           order: [[ 'userId', 'ASC' ]]
@@ -473,6 +489,8 @@ describe.skip('workflow > instructions > prompt', () => {
 
         const post = await PostRepo.create({ values: { title: 't1' } });
 
+        await sleep(500);
+
         const UserJobModel = db.getModel('users_jobs');
         const pendingJobs = await UserJobModel.findAll({
           order: [[ 'userId', 'ASC' ]]
@@ -523,6 +541,8 @@ describe.skip('workflow > instructions > prompt', () => {
 
         const post = await PostRepo.create({ values: { title: 't1' } });
 
+        await sleep(500);
+
         const UserJobModel = db.getModel('users_jobs');
         const pendingJobs = await UserJobModel.findAll({
           order: [[ 'userId', 'ASC' ]]

packages/plugins/workflow/src/server/__tests__/instructions/query.test.ts

@@ -1,6 +1,6 @@
 import { Application } from '@nocobase/server';
 import Database from '@nocobase/database';
-import { getApp } from '..';
+import { getApp, sleep } from '..';
 
 
 
@@ -46,6 +46,8 @@ describe('workflow > instructions > query', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       const [job] = await execution.getJobs();
       expect(job.result.title).toBe(post.title);
@@ -66,6 +68,8 @@ describe('workflow > instructions > query', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       const [job] = await execution.getJobs();
       expect(job.result.title).toBe(post.title);
@@ -86,6 +90,8 @@ describe('workflow > instructions > query', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       const [job] = await execution.getJobs();
       expect(job.result).toBe(null);
@@ -106,6 +112,8 @@ describe('workflow > instructions > query', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       const [job] = await execution.getJobs();
       expect(job.result.title).toBe(post.title);
@@ -131,6 +139,8 @@ describe('workflow > instructions > query', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       const jobs = await execution.getJobs({ order: [['id', 'ASC']] });
       expect(jobs[1].result.title).toBe(post.title);
@@ -154,6 +164,8 @@ describe('workflow > instructions > query', () => {
         values: { title: 't1', tags: [tag.id] }
       });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       const [job] = await execution.getJobs();
       expect(job.result.id).toBe(tag.id);
@@ -175,6 +187,8 @@ describe('workflow > instructions > query', () => {
         values: { title: 't1', tags: [tag.id] }
       });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       const [job] = await execution.getJobs();
       expect(job.result.posts.length).toBe(1);
@@ -193,8 +207,13 @@ describe('workflow > instructions > query', () => {
       });
 
       const p1 = await PostRepo.create({ values: { title: 't1' } });
+
+      await sleep(500);
+
       const p2 = await PostRepo.create({ values: { title: 't2' } });
 
+      await sleep(500);
+
       // get the 2nd execution
       const [execution] = await workflow.getExecutions({ order: [['id', 'DESC']] });
       expect(execution.context.data.title).toBe(p2.title);
@@ -215,6 +234,8 @@ describe('workflow > instructions > query', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       const [job] = await execution.getJobs();
       expect(job.result.length).toBe(1);
@@ -237,6 +258,8 @@ describe('workflow > instructions > query', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       const [job] = await execution.getJobs();
       expect(job.result.length).toBe(1);
@@ -259,6 +282,8 @@ describe('workflow > instructions > query', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       const [job] = await execution.getJobs();
       expect(job.result.length).toBe(0);

packages/plugins/workflow/src/server/__tests__/instructions/update.test.ts

@@ -1,6 +1,6 @@
 import { Application } from '@nocobase/server';
 import Database from '@nocobase/database';
-import { getApp } from '..';
+import { getApp, sleep } from '..';
 
 
 
@@ -51,6 +51,8 @@ describe('workflow > instructions > update', () => {
       const post = await PostRepo.create({ values: { title: 't1' } });
       expect(post.published).toBe(false);
 
+      await sleep(500);
+
       const [execution] = await workflow.getExecutions();
       const [job] = await execution.getJobs();
       expect(job.result.published).toBe(true);
@@ -93,6 +95,9 @@ describe('workflow > instructions > update', () => {
 
     // NOTE: the result of post immediately created will not be changed by workflow
     const { id } = await PostRepo.create({ values: { title: 'test' } });
+
+    await sleep(500);
+
     // should get from db
     const post = await PostRepo.findById(id);
     expect(post.title).toBe('changed');

packages/plugins/workflow/src/server/__tests__/triggers/collection.test.ts

@@ -41,6 +41,8 @@ describe('workflow > triggers > collection', () => {
 
       const post = await PostRepo.create({ values: { title: 't1' } });
 
+      await sleep(500);
+
       const executions = await workflow.getExecutions();
       expect(executions.length).toBe(0);
     });
@@ -60,6 +62,8 @@ describe('workflow > triggers > collection', () => {
       const post = await PostRepo.create({ values: { title: 't1' } });
       await PostRepo.update({ filterByTk: post.id, values: { title: 't2' } });
 
+      await sleep(500);
+
       const executions = await workflow.getExecutions();
       expect(executions.length).toBe(1);
       expect(executions[0].context.data.title).toBe('t2');
@@ -79,6 +83,8 @@ describe('workflow > triggers > collection', () => {
       const post = await PostRepo.create({ values: { title: 't1' } });
       await PostRepo.update({ filterByTk: post.id, values: { title: 't2' } });
 
+      await sleep(500);
+
       const executions = await workflow.getExecutions();
       expect(executions.length).toBe(1);
       expect(executions[0].status).toBe(EXECUTION_STATUS.RESOLVED);
@@ -99,6 +105,8 @@ describe('workflow > triggers > collection', () => {
       const post = await PostRepo.create({ values: { title: 't1' } });
       await PostRepo.update({ filterByTk: post.id, values: { title: 't2' } });
 
+      await sleep(500);
+
       const executions = await workflow.getExecutions();
       expect(executions.length).toBe(0);
     });

packages/plugins/workflow/src/server/constants.ts

@@ -1,5 +1,5 @@
 export const EXECUTION_STATUS = {
-  CRETEAD: null,
+  CREATED: null,
   STARTED: 0,
   RESOLVED: 1,
   REJECTED: -1,

packages/plugins/workflow/src/server/instructions/delay.ts

@@ -47,7 +47,7 @@ export default class implements Instruction {
     }) as JobModel[];
 
     jobs.forEach(job => {
-      this.schedule(job, job.node.config.duration);
+      this.schedule(job, job.node!.config.duration);
     });
   }
 
@@ -70,8 +70,8 @@ export default class implements Instruction {
   async trigger(job) {
     const execution = await job.getExecution() as ExecutionModel;
     if (execution.status === EXECUTION_STATUS.STARTED) {
-      const processor = this.plugin.createProcessor(execution);
+      job.execution = execution;
-      await processor.resume(job);
+      await this.plugin.resume(job);
     }
     if (this.timers.get(job.id)) {
       this.timers.delete(job.id);

packages/plugins/workflow/src/server/triggers/collection.ts

@@ -38,6 +38,7 @@ function getFieldRawName(collection: Collection, name: string) {
 async function handler(this: CollectionTrigger, workflow: WorkflowModel, data: Model, options) {
   const { collection: collectionName, condition, changed } = workflow.config;
   const collection = (<typeof Model>data.constructor).database.getCollection(collectionName);
+  const { transaction, context } = options;
 
   // NOTE: if no configured fields changed, do not trigger
   if (changed
@@ -46,7 +47,6 @@ async function handler(this: CollectionTrigger, workflow: WorkflowModel, data: M
         .filter(name => !['linkTo', 'hasOne', 'hasMany', 'belongsToMany'].includes(collection.getField(name).type))
         .every(name => !data.changedWithAssociations(getFieldRawName(collection, name)))
   ) {
-    // TODO: temp comment out
     return;
   }
   // NOTE: if no configured condition match, do not trigger
@@ -54,7 +54,6 @@ async function handler(this: CollectionTrigger, workflow: WorkflowModel, data: M
     // TODO: change to map filter format to calculation format
     // const calculation = toCalculation(condition);
     const { repository, model } = collection;
-    const { transaction, context } = options;
     const count = await repository.count({
       filter: {
         $and: [
@@ -71,9 +70,16 @@ async function handler(this: CollectionTrigger, workflow: WorkflowModel, data: M
     }
   }
 
-  return this.plugin.trigger(workflow, { data: data.get() }, {
+  // TODO(bug): use setTimeout here test case will not exit (only work in SQLite)?
-    context: options.context,
+  // setTimeout(() => {
-    transaction: options.transaction
+  //   this.plugin.trigger(workflow, { data: data.get() }, {
+  //     context
+  //   });
+  // }, 0);
+
+  await this.plugin.trigger(workflow, { data: data.get() }, {
+    context,
+    transaction
   });
 }