diff --git a/packages/plugins/workflow/src/client/triggers/index.tsx b/packages/plugins/workflow/src/client/triggers/index.tsx index 0d73f9237..76a686a03 100644 --- a/packages/plugins/workflow/src/client/triggers/index.tsx +++ b/packages/plugins/workflow/src/client/triggers/index.tsx @@ -48,8 +48,8 @@ export interface Trigger { view?: ISchema; scope?: { [key: string]: any }; components?: { [key: string]: any }; - render?(props): React.ReactElement; - getter?(node: any): React.ReactElement; + render?(props): React.ReactNode; + getter?(node: any): React.ReactNode; }; export const triggers = new Registry(); @@ -70,56 +70,52 @@ function TriggerExecution() { , + shape: 'circle', + className: 'workflow-node-job-button', + type: 'primary' + }, properties: { - trigger: { + [execution.id]: { type: 'void', - 'x-component': 'Action', - 'x-component-props': { - title: , - shape: 'circle', - className: 'workflow-node-job-button', - type: 'primary' + 'x-decorator': 'Form', + 'x-decorator-props': { + initialValue: execution }, + 'x-component': 'Action.Modal', + title: ( +
+ {compile(trigger.title)} + {workflow.title} + #{execution.id} +
+ ), properties: { - [execution.id]: { - type: 'void', - 'x-decorator': 'Form', - 'x-decorator-props': { - initialValue: execution + createdAt: { + type: 'string', + title: `{{t("Triggered at", { ns: "${NAMESPACE}" })}}`, + 'x-decorator': 'FormItem', + 'x-component': 'DatePicker', + 'x-component-props': { + showTime: true }, - 'x-component': 'Action.Modal', - title: ( -
- {compile(trigger.title)} - {workflow.title} - #{execution.id} -
- ), - properties: { - createdAt: { - type: 'string', - title: `{{t("Triggered at", { ns: "${NAMESPACE}" })}}`, - 'x-decorator': 'FormItem', - 'x-component': 'DatePicker', - 'x-component-props': { - showTime: true - }, - 'x-read-pretty': true, - }, - context: { - type: 'object', - title: `{{t("Trigger variables", { ns: "${NAMESPACE}" })}}`, - 'x-decorator': 'FormItem', - 'x-component': 'Input.JSON', - 'x-component-props': { - className: css` - padding: 1em; - background-color: #eee; - ` - }, - 'x-read-pretty': true, - } - } + 'x-read-pretty': true, + }, + context: { + type: 'object', + title: `{{t("Trigger variables", { ns: "${NAMESPACE}" })}}`, + 'x-decorator': 'FormItem', + 'x-component': 'Input.JSON', + 'x-component-props': { + className: css` + padding: 1em; + background-color: #eee; + ` + }, + 'x-read-pretty': true, } } } diff --git a/packages/plugins/workflow/src/server/Plugin.ts b/packages/plugins/workflow/src/server/Plugin.ts index a964d9369..a6b66b47f 100644 --- a/packages/plugins/workflow/src/server/Plugin.ts +++ b/packages/plugins/workflow/src/server/Plugin.ts @@ -16,14 +16,15 @@ import initTriggers, { Trigger } from './triggers'; import JobModel from './models/Job'; - +type Pending = [ExecutionModel, JobModel?]; export default class WorkflowPlugin extends Plugin { instructions: Registry = new Registry(); triggers: Registry = new Registry(); calculators = calculators; extensions = extensions; - executing: Promise = null; - pending: [ExecutionModel, JobModel][] = []; + executing: ExecutionModel = null; + pending: Pending[] = []; + events: [WorkflowModel, any, { context?: any }][] = []; onBeforeSave = async (instance: WorkflowModel, options) => { const Model = instance.constructor; @@ -102,7 +103,7 @@ export default class WorkflowPlugin extends Plugin { }); // check for not started executions - await this.dispatch(); + this.dispatch(); }); this.app.on('beforeStop', async () => { @@ -132,12 +133,25 @@ export default class WorkflowPlugin extends Plugin { } } - public async trigger(workflow: WorkflowModel, context: Object, options: Transactionable & { context?: any } = {}): Promise { + public trigger(workflow: WorkflowModel, context: Object, options: { context?: any } = {}): Promise { // `null` means not to trigger if (context == null) { - return null; + return; } + this.events.push([workflow, context, options]); + + if (this.events.length > 1) { + return; + } + + // NOTE: no await for quick return + setTimeout(this.prepare); + } + + private prepare = async () => { + const [workflow, context, options] = this.events[0]; + if (options.context?.executionId) { // NOTE: no transaction here for read-uncommitted execution const existed = await workflow.countExecutions({ @@ -152,43 +166,47 @@ export default class WorkflowPlugin extends Plugin { } } - const transaction = await (workflow.constructor).database.sequelize.transaction(); + await this.db.sequelize.transaction(async transaction => { + const execution = await workflow.createExecution({ + context, + key: workflow.key, + status: EXECUTION_STATUS.CREATED, + useTransaction: workflow.useTransaction, + }, { transaction }); - const execution = await workflow.createExecution({ - context, - key: workflow.key, - status: EXECUTION_STATUS.CREATED, - useTransaction: workflow.useTransaction, - }, { transaction }); + const executed = await workflow.countExecutions({ transaction }); - console.log('workflow triggered:', new Date(), workflow.id, execution.id); + // NOTE: not to trigger afterUpdate hook here + await workflow.update({ executed }, { transaction, hooks: false }); - const executed = await workflow.countExecutions({ transaction }); + const allExecuted = await (execution.constructor).count({ + where: { + key: workflow.key + }, + transaction + }); + await (workflow.constructor).update({ + allExecuted + }, { + where: { + key: workflow.key + }, + individualHooks: true, + transaction + }); - // NOTE: not to trigger afterUpdate hook here - await workflow.update({ executed }, { transaction, hooks: false }); + execution.workflow = workflow; - const allExecuted = await (execution.constructor).count({ - where: { - key: workflow.key - }, - transaction - }); - await (workflow.constructor).update({ - allExecuted - }, { - where: { - key: workflow.key - }, - individualHooks: true, - transaction + return execution; }); - execution.workflow = workflow; + this.events.shift(); - await transaction.commit(); - - setTimeout(() => this.dispatch(execution)); + if (this.events.length) { + await this.prepare(); + } else { + this.dispatch(); + } } public async resume(job) { @@ -196,28 +214,37 @@ export default class WorkflowPlugin extends Plugin { job.execution = await job.getExecution(); } - setTimeout(() => this.dispatch(job.execution, job)); + this.pending.push([job.execution, job]); + this.dispatch(); } - private async dispatch(execution?: ExecutionModel, job?: JobModel) { + private async dispatch() { if (this.executing) { - if (job) { - this.pending.push([execution, job]); - } return; } - if (!execution) { - execution = await this.db.getRepository('executions').findOne({ + let next: Pending; + // resuming has high priority + if (this.pending.length) { + next = this.pending.shift(); + } else { + const execution = await this.db.getRepository('executions').findOne({ filter: { status: EXECUTION_STATUS.CREATED }, sort: 'createdAt' }) as ExecutionModel; - if (!execution) { - return; + if (execution) { + next = [execution]; } + }; + if (next) { + this.process(...next); } + } + + private async process(execution: ExecutionModel, job?: JobModel) { + this.executing = execution; if (execution.status === EXECUTION_STATUS.CREATED) { await execution.update({ status: EXECUTION_STATUS.STARTED }); @@ -225,16 +252,13 @@ export default class WorkflowPlugin extends Plugin { const processor = this.createProcessor(execution); - this.executing = job ? processor.resume(job) : processor.start(); + console.log('workflow processing:', new Date(), execution.workflowId, execution.id); - await this.executing; + await (job ? processor.resume(job) : processor.start()); this.executing = null; - setTimeout(() => { - const args = this.pending.length ? this.pending.shift() : []; - this.dispatch(...args); - }); + this.dispatch(); } private createProcessor(execution: ExecutionModel, options = {}): Processor { diff --git a/packages/plugins/workflow/src/server/Processor.ts b/packages/plugins/workflow/src/server/Processor.ts index 627ddefe0..725340125 100644 --- a/packages/plugins/workflow/src/server/Processor.ts +++ b/packages/plugins/workflow/src/server/Processor.ts @@ -70,15 +70,13 @@ export default class Processor { const { options } = this; - const { sequelize } = (this.execution.constructor).database; - // @ts-ignore return options.transaction && !options.transaction.finished ? options.transaction - : await sequelize.transaction(); + : await options.plugin.db.sequelize.transaction(); } - async prepare(commit?: boolean) { + private async prepare() { const transaction = await this.getTransaction(); this.transaction = transaction; @@ -97,10 +95,6 @@ export default class Processor { }); this.makeJobs(jobs); - - if (commit) { - await this.commit(); - } } public async start() { diff --git a/packages/plugins/workflow/src/server/__tests__/Plugin.test.ts b/packages/plugins/workflow/src/server/__tests__/Plugin.test.ts index aacb3f324..9956b0af8 100644 --- a/packages/plugins/workflow/src/server/__tests__/Plugin.test.ts +++ b/packages/plugins/workflow/src/server/__tests__/Plugin.test.ts @@ -183,6 +183,41 @@ describe('workflow > Plugin', () => { }); }); + describe('cycling trigger', () => { + it('trigger should not be triggered more than once in same execution', async () => { + const workflow = await WorkflowModel.create({ + enabled: true, + type: 'collection', + config: { + mode: 1, + collection: 'posts' + } + }); + + const n1 = await workflow.createNode({ + type: 'create', + config: { + collection: 'posts', + params: { + values: { + title: 't2' + } + } + } + }); + + const post = await PostRepo.create({ values: { title: 't1' } }); + + await sleep(500); + + const posts = await PostRepo.find(); + expect(posts.length).toBe(2); + + const [execution] = await workflow.getExecutions(); + expect(execution.status).toBe(EXECUTION_STATUS.RESOLVED); + }); + }); + describe('dispatcher', () => { it.skip('multiple triggers in same event', async () => { const w1 = await WorkflowModel.create({ @@ -203,15 +238,27 @@ describe('workflow > Plugin', () => { } }); + const w3 = await WorkflowModel.create({ + enabled: true, + type: 'collection', + config: { + mode: 1, + collection: 'posts' + } + }); + const p1 = await PostRepo.create({ values: { title: 't1' } }); - await sleep(500); + await sleep(1000); const [e1] = await w1.getExecutions(); expect(e1.status).toBe(EXECUTION_STATUS.RESOLVED); const [e2] = await w2.getExecutions(); expect(e2.status).toBe(EXECUTION_STATUS.RESOLVED); + + const [e3] = await w3.getExecutions(); + expect(e3.status).toBe(EXECUTION_STATUS.RESOLVED); }); it('when server starts, process all created executions', async () => { diff --git a/packages/plugins/workflow/src/server/__tests__/Processor.test.ts b/packages/plugins/workflow/src/server/__tests__/Processor.test.ts index a1c09867d..d12700e2c 100644 --- a/packages/plugins/workflow/src/server/__tests__/Processor.test.ts +++ b/packages/plugins/workflow/src/server/__tests__/Processor.test.ts @@ -425,30 +425,4 @@ describe('workflow > Processor', () => { 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 PostRepo.create({ values: { title: 't1' } }); - - await sleep(500); - - const posts = await PostRepo.find(); - expect(posts.length).toBe(2); - - const [execution] = await workflow.getExecutions(); - expect(execution.status).toBe(EXECUTION_STATUS.RESOLVED); - }); - }); }); diff --git a/packages/plugins/workflow/src/server/triggers/collection.ts b/packages/plugins/workflow/src/server/triggers/collection.ts index aeb6605df..7563de978 100644 --- a/packages/plugins/workflow/src/server/triggers/collection.ts +++ b/packages/plugins/workflow/src/server/triggers/collection.ts @@ -42,10 +42,10 @@ async function handler(this: CollectionTrigger, workflow: WorkflowModel, data: M // NOTE: if no configured fields changed, do not trigger if (changed - && changed.length - && changed - .filter(name => !['linkTo', 'hasOne', 'hasMany', 'belongsToMany'].includes(collection.getField(name).type)) - .every(name => !data.changedWithAssociations(getFieldRawName(collection, name))) + && changed.length + && changed + .filter(name => !['linkTo', 'hasOne', 'hasMany', 'belongsToMany'].includes(collection.getField(name).type)) + .every(name => !data.changedWithAssociations(getFieldRawName(collection, name))) ) { return; } @@ -70,10 +70,8 @@ async function handler(this: CollectionTrigger, workflow: WorkflowModel, data: M } } - setTimeout(() => { - this.plugin.trigger(workflow, { data: data.get() }, { - context - }); + this.plugin.trigger(workflow, { data: data.get() }, { + context }); }