From c470209ecd116e44ed4fc279e14baf1a97a31b84 Mon Sep 17 00:00:00 2001 From: Junyi Date: Mon, 14 Nov 2022 20:05:01 +0800 Subject: [PATCH] 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 | 3 - .../plugins/workflow/src/server/Plugin.ts | 81 +++++++++++++------ .../plugins/workflow/src/server/Processor.ts | 10 +-- .../src/server/__tests__/Plugin.test.ts | 30 ++++++- .../src/server/__tests__/Processor.test.ts | 67 ++++++++++++--- .../__tests__/actions/workflows.test.ts | 17 +++- .../instructions/calculation.test.ts | 23 +++++- .../__tests__/instructions/condition.test.ts | 16 +++- .../__tests__/instructions/create.test.ts | 4 +- .../__tests__/instructions/delay.test.ts | 16 +++- .../__tests__/instructions/destroy.test.ts | 4 +- .../__tests__/instructions/parallel.test.ts | 34 +++++++- .../__tests__/instructions/prompt.test.ts | 20 +++++ .../__tests__/instructions/query.test.ts | 27 ++++++- .../__tests__/instructions/update.test.ts | 7 +- .../__tests__/triggers/collection.test.ts | 8 ++ .../plugins/workflow/src/server/constants.ts | 2 +- .../workflow/src/server/instructions/delay.ts | 6 +- .../src/server/triggers/collection.ts | 16 ++-- 19 files changed, 320 insertions(+), 71 deletions(-) diff --git a/packages/plugins/users/src/server.ts b/packages/plugins/users/src/server.ts index dc6b00153..aa1c07422 100644 --- a/packages/plugins/users/src/server.ts +++ b/packages/plugins/users/src/server.ts @@ -21,14 +21,11 @@ export interface UserPluginConfig { export default class UsersPlugin extends Plugin { public jwtService: JwtService; - public tokenMiddleware: Middleware; - public authenticators: Registry = new Registry(); constructor(app, options) { super(app, options); this.jwtService = new JwtService(options?.jwt || {}); - this.tokenMiddleware = new Middleware(parseToken); } async beforeLoad() { diff --git a/packages/plugins/workflow/src/server/Plugin.ts b/packages/plugins/workflow/src/server/Plugin.ts index 97713d00a..5b98e6456 100644 --- a/packages/plugins/workflow/src/server/Plugin.ts +++ b/packages/plugins/workflow/src/server/Plugin.ts @@ -22,6 +22,8 @@ export default class WorkflowPlugin extends Plugin { triggers: Registry = new Registry(); calculators = calculators; extensions = extensions; + executing: Promise = null; + pending: [ExecutionModel, JobModel][] = []; onBeforeSave = async (instance: WorkflowModel, options) => { const Model = instance.constructor; @@ -127,39 +129,38 @@ export default class WorkflowPlugin extends Plugin { } } - async trigger(workflow: WorkflowModel, context: Object, options: Transactionable & { context?: any } = {}): Promise { + public async trigger(workflow: WorkflowModel, context: Object, options: Transactionable & { context?: any } = {}): Promise { // `null` means not to trigger - if (context === null) { + if (context == null) { return null; } - let transaction = null; - - if (workflow.useTransaction) { - // @ts-ignore - transaction = options.transaction && !options.transaction.finished - ? options.transaction - : await (workflow.constructor).database.sequelize.transaction(); + // @ts-ignore + const transaction = options.transaction && !options.transaction.finished + ? options.transaction + : await (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.`); - return null; + console.warn(`workflow ${workflow.id} has already been triggered in same execution (${options.context.transaction}), and newly triggering will be skipped.`); + 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 }); } } diff --git a/packages/plugins/workflow/src/server/Processor.ts b/packages/plugins/workflow/src/server/Processor.ts index 2e1b5b547..b514ed637 100644 --- a/packages/plugins/workflow/src/server/Processor.ts +++ b/packages/plugins/workflow/src/server/Processor.ts @@ -75,17 +75,9 @@ export default class Processor { const { sequelize } = (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) { diff --git a/packages/plugins/workflow/src/server/__tests__/Plugin.test.ts b/packages/plugins/workflow/src/server/__tests__/Plugin.test.ts index 2831930cc..2606fab0a 100644 --- a/packages/plugins/workflow/src/server/__tests__/Plugin.test.ts +++ b/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); }); diff --git a/packages/plugins/workflow/src/server/__tests__/Processor.test.ts b/packages/plugins/workflow/src/server/__tests__/Processor.test.ts index 08b25b669..a1c09867d 100644 --- a/packages/plugins/workflow/src/server/__tests__/Processor.test.ts +++ b/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); - await processor.resume(pending); + pending.execution = execution; + 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); - await processor.resume(pending); + pending.execution = execution; + 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); - await processor.resume(pending); + pending.execution = execution; + 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); - await processor.resume(pending); + pending.execution = execution; + 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); - await processor.resume(pending); + pending.execution = execution; + 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); - await processor.resume(pending); + pending.execution = e1; + 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); diff --git a/packages/plugins/workflow/src/server/__tests__/actions/workflows.test.ts b/packages/plugins/workflow/src/server/__tests__/actions/workflows.test.ts index aa1936bb1..d9967f14a 100644 --- a/packages/plugins/workflow/src/server/__tests__/actions/workflows.test.ts +++ b/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); diff --git a/packages/plugins/workflow/src/server/__tests__/instructions/calculation.test.ts b/packages/plugins/workflow/src/server/__tests__/instructions/calculation.test.ts index ea5966021..2e9dcabad 100644 --- a/packages/plugins/workflow/src/server/__tests__/instructions/calculation.test.ts +++ b/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); diff --git a/packages/plugins/workflow/src/server/__tests__/instructions/condition.test.ts b/packages/plugins/workflow/src/server/__tests__/instructions/condition.test.ts index 7499b5e94..bd7a9e79c 100644 --- a/packages/plugins/workflow/src/server/__tests__/instructions/condition.test.ts +++ b/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); diff --git a/packages/plugins/workflow/src/server/__tests__/instructions/create.test.ts b/packages/plugins/workflow/src/server/__tests__/instructions/create.test.ts index b2ddf77d8..726d825c7 100644 --- a/packages/plugins/workflow/src/server/__tests__/instructions/create.test.ts +++ b/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); diff --git a/packages/plugins/workflow/src/server/__tests__/instructions/delay.test.ts b/packages/plugins/workflow/src/server/__tests__/instructions/delay.test.ts index d8351e849..b87bfc2b2 100644 --- a/packages/plugins/workflow/src/server/__tests__/instructions/delay.test.ts +++ b/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(); diff --git a/packages/plugins/workflow/src/server/__tests__/instructions/destroy.test.ts b/packages/plugins/workflow/src/server/__tests__/instructions/destroy.test.ts index 2a86b650a..d210652e6 100644 --- a/packages/plugins/workflow/src/server/__tests__/instructions/destroy.test.ts +++ b/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); diff --git a/packages/plugins/workflow/src/server/__tests__/instructions/parallel.test.ts b/packages/plugins/workflow/src/server/__tests__/instructions/parallel.test.ts index 58379b42e..28c05e2c1 100644 --- a/packages/plugins/workflow/src/server/__tests__/instructions/parallel.test.ts +++ b/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); - await processor.resume(pending); + pending.execution = execution; + 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']] }); diff --git a/packages/plugins/workflow/src/server/__tests__/instructions/prompt.test.ts b/packages/plugins/workflow/src/server/__tests__/instructions/prompt.test.ts index a38b29deb..c13a7e29c 100644 --- a/packages/plugins/workflow/src/server/__tests__/instructions/prompt.test.ts +++ b/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' ]] diff --git a/packages/plugins/workflow/src/server/__tests__/instructions/query.test.ts b/packages/plugins/workflow/src/server/__tests__/instructions/query.test.ts index ef52cb10f..6f04d725b 100644 --- a/packages/plugins/workflow/src/server/__tests__/instructions/query.test.ts +++ b/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); diff --git a/packages/plugins/workflow/src/server/__tests__/instructions/update.test.ts b/packages/plugins/workflow/src/server/__tests__/instructions/update.test.ts index 61880d8d6..b3883fcfb 100644 --- a/packages/plugins/workflow/src/server/__tests__/instructions/update.test.ts +++ b/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'); diff --git a/packages/plugins/workflow/src/server/__tests__/triggers/collection.test.ts b/packages/plugins/workflow/src/server/__tests__/triggers/collection.test.ts index 4ec8b5e14..124fa9be8 100644 --- a/packages/plugins/workflow/src/server/__tests__/triggers/collection.test.ts +++ b/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); }); diff --git a/packages/plugins/workflow/src/server/constants.ts b/packages/plugins/workflow/src/server/constants.ts index 920cc24b9..0d746bea6 100644 --- a/packages/plugins/workflow/src/server/constants.ts +++ b/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, diff --git a/packages/plugins/workflow/src/server/instructions/delay.ts b/packages/plugins/workflow/src/server/instructions/delay.ts index 35e52c9fc..633318a81 100644 --- a/packages/plugins/workflow/src/server/instructions/delay.ts +++ b/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); - await processor.resume(job); + job.execution = execution; + await this.plugin.resume(job); } if (this.timers.get(job.id)) { this.timers.delete(job.id); diff --git a/packages/plugins/workflow/src/server/triggers/collection.ts b/packages/plugins/workflow/src/server/triggers/collection.ts index 584286039..c1ce70522 100644 --- a/packages/plugins/workflow/src/server/triggers/collection.ts +++ b/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 = (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() }, { - context: options.context, - transaction: options.transaction + // TODO(bug): use setTimeout here test case will not exit (only work in SQLite)? + // setTimeout(() => { + // this.plugin.trigger(workflow, { data: data.get() }, { + // context + // }); + // }, 0); + + await this.plugin.trigger(workflow, { data: data.get() }, { + context, + transaction }); }