From 1f59d4bed20f522bf7cca1a9d86960cda705b0b6 Mon Sep 17 00:00:00 2001 From: Junyi Date: Fri, 29 Mar 2024 09:09:12 +0800 Subject: [PATCH] refactor(plugin-workflow): add transaction check from data source (#3857) --- .../plugin-workflow/src/server/Plugin.ts | 37 ++++++++++++++----- .../src/server/collections/executions.ts | 2 +- .../server/instructions/CreateInstruction.ts | 5 ++- .../server/instructions/DestroyInstruction.ts | 2 +- .../server/instructions/QueryInstruction.ts | 2 +- .../server/instructions/UpdateInstruction.ts | 2 +- .../src/server/triggers/CollectionTrigger.ts | 2 +- 7 files changed, 35 insertions(+), 17 deletions(-) diff --git a/packages/plugins/@nocobase/plugin-workflow/src/server/Plugin.ts b/packages/plugins/@nocobase/plugin-workflow/src/server/Plugin.ts index 16540e48f..3e19f6cba 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/server/Plugin.ts +++ b/packages/plugins/@nocobase/plugin-workflow/src/server/Plugin.ts @@ -35,6 +35,7 @@ export default class PluginWorkflowServer extends Plugin { instructions: Registry = new Registry(); triggers: Registry = new Registry(); functions: Registry = new Registry(); + enabledCache: Map = new Map(); private ready = false; private executing: Promise | null = null; @@ -72,7 +73,7 @@ export default class PluginWorkflowServer extends Plugin { return trigger.sync ?? workflow.sync; } - onBeforeSave = async (instance: WorkflowModel, options) => { + private onBeforeSave = async (instance: WorkflowModel, options) => { const Model = instance.constructor; if (instance.enabled) { @@ -273,8 +274,10 @@ export default class PluginWorkflowServer extends Plugin { trigger.off({ ...workflow.get(), ...prev }); } trigger.on(workflow); + this.enabledCache.set(workflow.id, workflow); } else { trigger.off(workflow); + this.enabledCache.delete(workflow.id); } } @@ -304,9 +307,7 @@ export default class PluginWorkflowServer extends Plugin { this.eventsCount = this.events.length; logger.info(`new event triggered, now events: ${this.events.length}`); - logger.debug(`event data:`, { - data: context, - }); + logger.debug(`event data:`, { context }); if (this.events.length > 1) { return; @@ -488,23 +489,39 @@ export default class PluginWorkflowServer extends Plugin { if (execution.status === EXECUTION_STATUS.QUEUEING) { await execution.update({ status: EXECUTION_STATUS.STARTED }, { transaction: options.transaction }); } - + const logger = this.getLogger(execution.workflowId); const processor = this.createProcessor(execution, options); - this.getLogger(execution.workflowId).info(`execution (${execution.id}) ${job ? 'resuming' : 'starting'}...`); + logger.info(`execution (${execution.id}) ${job ? 'resuming' : 'starting'}...`); + + // this.emit('beforeProcess', processor); try { await (job ? processor.resume(job) : processor.start()); - this.getLogger(execution.workflowId).info( - `execution (${execution.id}) finished with status: ${execution.status}`, - ); + logger.info(`execution (${execution.id}) finished with status: ${execution.status}`, { execution }); if (execution.status && execution.workflow.options?.deleteExecutionOnStatus?.includes(execution.status)) { await execution.destroy(); } } catch (err) { - this.getLogger(execution.workflowId).error(`execution (${execution.id}) error: ${err.message}`, err); + logger.error(`execution (${execution.id}) error: ${err.message}`, err); } + // this.emit('afterProcess', processor); + return processor; } + + useDataSourceTransaction(dataSourceName = 'main', transaction, create = false) { + // @ts-ignore + const { db } = this.app.dataSourceManager.dataSources.get(dataSourceName).collectionManager; + if (!db) { + return; + } + if (db.sequelize === transaction?.sequelize) { + return transaction; + } + if (create) { + return db.sequelize.transaction(); + } + } } diff --git a/packages/plugins/@nocobase/plugin-workflow/src/server/collections/executions.ts b/packages/plugins/@nocobase/plugin-workflow/src/server/collections/executions.ts index 33d294d62..c46d0bfe2 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/server/collections/executions.ts +++ b/packages/plugins/@nocobase/plugin-workflow/src/server/collections/executions.ts @@ -12,7 +12,7 @@ export default { name: 'workflow', }, { - type: 'uid', + type: 'string', name: 'key', }, { diff --git a/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/CreateInstruction.ts b/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/CreateInstruction.ts index 3d4de9b7d..fd5efeb73 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/CreateInstruction.ts +++ b/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/CreateInstruction.ts @@ -15,13 +15,14 @@ export class CreateInstruction extends Instruction { .get(dataSourceName) .collectionManager.getCollection(collectionName); const options = processor.getParsedValue(params, node.id); + const transaction = this.workflow.useDataSourceTransaction(dataSourceName, processor.transaction); const created = await repository.create({ ...options, context: { stack: Array.from(new Set((processor.execution.context.stack ?? []).concat(processor.execution.id))), }, - transaction: processor.transaction, + transaction, }); let result = created; @@ -34,7 +35,7 @@ export class CreateInstruction extends Instruction { result = await repository.findOne({ filterByTk: created[model.primaryKeyAttribute], appends: Array.from(includeFields), - transaction: processor.transaction, + transaction, }); } diff --git a/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/DestroyInstruction.ts b/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/DestroyInstruction.ts index 5544000d3..a93e48617 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/DestroyInstruction.ts +++ b/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/DestroyInstruction.ts @@ -20,7 +20,7 @@ export class DestroyInstruction extends Instruction { context: { stack: Array.from(new Set((processor.execution.context.stack ?? []).concat(processor.execution.id))), }, - transaction: processor.transaction, + transaction: this.workflow.useDataSourceTransaction(dataSourceName, processor.transaction), }); return { diff --git a/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/QueryInstruction.ts b/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/QueryInstruction.ts index f1888a473..332aceb1e 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/QueryInstruction.ts +++ b/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/QueryInstruction.ts @@ -38,7 +38,7 @@ export class QueryInstruction extends Instruction { .filter((item) => item.field) .map((item) => `${item.direction?.toLowerCase() === 'desc' ? '-' : ''}${item.field}`), appends, - transaction: processor.transaction, + transaction: this.workflow.useDataSourceTransaction(dataSourceName, processor.transaction), }); if (failOnEmpty && (multiple ? !result.length : !result)) { diff --git a/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/UpdateInstruction.ts b/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/UpdateInstruction.ts index f8c6b39d3..95e352d3d 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/UpdateInstruction.ts +++ b/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/UpdateInstruction.ts @@ -20,7 +20,7 @@ export class UpdateInstruction extends Instruction { context: { stack: Array.from(new Set((processor.execution.context.stack ?? []).concat(processor.execution.id))), }, - transaction: processor.transaction, + transaction: this.workflow.useDataSourceTransaction(dataSourceName, processor.transaction), }); return { diff --git a/packages/plugins/@nocobase/plugin-workflow/src/server/triggers/CollectionTrigger.ts b/packages/plugins/@nocobase/plugin-workflow/src/server/triggers/CollectionTrigger.ts index 7d8522854..7650112ca 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/server/triggers/CollectionTrigger.ts +++ b/packages/plugins/@nocobase/plugin-workflow/src/server/triggers/CollectionTrigger.ts @@ -96,7 +96,7 @@ async function handler(this: CollectionTrigger, workflow: WorkflowModel, data: M workflow, { data: json, stack: context?.stack }, { - transaction, + transaction: this.workflow.useDataSourceTransaction(dataSourceName, transaction), }, ); } else {