diff --git a/packages/plugins/workflow/src/__tests__/execution.test.ts b/packages/plugins/workflow/src/__tests__/execution.test.ts index 7b121d789..615210c2d 100644 --- a/packages/plugins/workflow/src/__tests__/execution.test.ts +++ b/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); + }); + }); }); diff --git a/packages/plugins/workflow/src/collections/executions.ts b/packages/plugins/workflow/src/collections/executions.ts index 8cf0e49ad..6df4a317f 100644 --- a/packages/plugins/workflow/src/collections/executions.ts +++ b/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', diff --git a/packages/plugins/workflow/src/models/Execution.ts b/packages/plugins/workflow/src/models/Execution.ts index 55f92f949..9d0e15435 100644 --- a/packages/plugins/workflow/src/models/Execution.ts +++ b/packages/plugins/workflow/src/models/Execution.ts @@ -73,9 +73,9 @@ export default class ExecutionModel extends Model { }); } - getTransaction() { - const { sequelize } = (this.constructor).database; - // @ts-ignore + async getTransaction() { + const { sequelize } = (this.constructor).database; + 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 } = this.constructor; + const { database } = this.constructor; const { model } = database.getCollection('jobs'); const [job] = (await model.upsert( { diff --git a/packages/plugins/workflow/src/models/Workflow.ts b/packages/plugins/workflow/src/models/Workflow.ts index b8424808d..dcf7137b3 100644 --- a/packages/plugins/workflow/src/models/Workflow.ts +++ b/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; declare executions: ExecutionModel[]; + declare countExecutions: HasManyCountAssociationsMixin; declare getExecutions: HasManyGetAssociationsMixin; declare createExecution: HasManyCreateAssociationMixin; @@ -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;