fix(plugin-workflow): break cycling trigger through transaction id (#341)

packages/plugins/workflow/src/__tests__/execution.test.ts

@@ -509,4 +509,28 @@ describe('execution', () => {
       expect(jobs.length).toEqual(5);
     });
   });
+
+  describe('cycling trigger', () => {
+    it('trigger should not be triggered more than once in same execution', async () => {
+      const n1 = await workflow.createNode({
+        type: 'create',
+        config: {
+          collection: 'posts',
+          params: {
+            values: {
+              title: 't2'
+            }
+          }
+        }
+      });
+
+      const post = await PostModel.create({ title: 't1' });
+
+      const posts = await PostModel.findAll();
+      expect(posts.length).toBe(2);
+
+      const [execution] = await workflow.getExecutions();
+      expect(execution.status).toBe(EXECUTION_STATUS.RESOLVED);
+    });
+  });
 });

packages/plugins/workflow/src/collections/executions.ts

@@ -17,6 +17,11 @@ export default {
       title: '使用事务',
       defaultValue: false
     },
+    {
+      type: 'uuid',
+      name: 'transaction',
+      defaultValue: null
+    },
     {
       interface: 'linkTo',
       type: 'hasMany',

packages/plugins/workflow/src/models/Execution.ts

@@ -73,9 +73,9 @@ export default class ExecutionModel extends Model {
     });
   }
 
-  getTransaction() {
+  async getTransaction() {
-    const { sequelize } = (<typeof WorkflowModel>this.constructor).database;
+    const { sequelize } = (<typeof ExecutionModel>this.constructor).database;
-    // @ts-ignore
+
     if (!this.useTransaction) {
       return undefined;
     }
@@ -83,15 +83,21 @@ export default class ExecutionModel extends Model {
     const { options } = this;
 
     // @ts-ignore
-    return options.transaction && !options.transaction.finished
+    const transaction = options.transaction && !options.transaction.finished
       ? options.transaction
       : sequelize.transaction();
+
+    // @ts-ignore
+    if (this.transaction !== transaction.id) {
+      // @ts-ignore
+      await this.update({ transaction: transaction.id }, { transaction });
+    }
+    return transaction;
   }
 
   async prepare(options, commit = false) {
     this.options = options || {};
-    // @ts-ignore
+    const transaction = await this.getTransaction();
-    const transaction = await this.getTransaction()
     this.transaction = transaction;
 
     if (!this.workflow) {
@@ -230,7 +236,7 @@ export default class ExecutionModel extends Model {
 
   // TODO(optimize)
   async saveJob(payload) {
-    const { database } = <typeof WorkflowModel>this.constructor;
+    const { database } = <typeof ExecutionModel>this.constructor;
     const { model } = database.getCollection('jobs');
     const [job] = (await model.upsert(
       {

packages/plugins/workflow/src/models/Workflow.ts

@@ -1,5 +1,5 @@
 import { Database, Model } from '@nocobase/database';
-import { HasManyCreateAssociationMixin, HasManyGetAssociationsMixin } from 'sequelize';
+import { HasManyCountAssociationsMixin, HasManyCreateAssociationMixin, HasManyGetAssociationsMixin } from 'sequelize';
 
 import triggers from '../triggers';
 import { EXECUTION_STATUS } from '../constants';
@@ -25,6 +25,7 @@ export default class WorkflowModel extends Model {
   declare createNode: HasManyCreateAssociationMixin<FlowNodeModel>;
 
   declare executions: ExecutionModel[];
+  declare countExecutions: HasManyCountAssociationsMixin;
   declare getExecutions: HasManyGetAssociationsMixin<ExecutionModel>;
   declare createExecution: HasManyCreateAssociationMixin<ExecutionModel>;
 
@@ -75,10 +76,25 @@ export default class WorkflowModel extends Model {
 
     const transaction = await this.getTransaction(options);
 
+    if (this.useTransaction) {
+      const existed = await this.countExecutions({
+        where: {
+          transaction: transaction.id
+        },
+        transaction
+      });
+
+      if (existed) {
+        console.warn(`workflow ${this.id} has already been triggered in same execution (${transaction.id}), and newly triggering will be skipped.`);
+        return;
+      }
+    }
+
     const execution = await this.createExecution({
       context,
       status: EXECUTION_STATUS.STARTED,
-      useTransaction: this.useTransaction
+      useTransaction: this.useTransaction,
+      transaction: transaction.id
     }, { transaction });
 
     execution.workflow = this;