Merge branch 'main' into next

packages/plugins/@nocobase/plugin-workflow-delay/src/server/DelayInstruction.ts

@@ -23,7 +23,7 @@ interface DelayConfig {
 }
 
 export default class extends Instruction {
-  timers: Map<number, NodeJS.Timeout> = new Map();
+  timers: Map<string, NodeJS.Timeout> = new Map();
 
   constructor(public workflow: WorkflowPlugin) {
     super(workflow);
@@ -72,26 +72,32 @@ export default class extends Instruction {
   };
 
   schedule(job) {
-    const now = new Date();
-    const createdAt = Date.parse(job.createdAt);
-    const delay = createdAt + job.result - now.getTime();
+    const createdAt = new Date(job.createdAt).getTime();
+    const delay = createdAt + job.result - Date.now();
     if (delay > 0) {
-      const trigger = this.trigger.bind(this, job);
-      this.timers.set(job.id, setTimeout(trigger, delay));
+      const trigger = this.trigger.bind(this, job.id);
+      this.timers.set(job.id.toString(), setTimeout(trigger, delay));
     } else {
       this.trigger(job);
     }
   }
 
-  async trigger(job) {
+  async trigger(jobOrId: JobModel | string) {
+    const { model } = this.workflow.app.db.getCollection('jobs');
+    const job =
+      jobOrId instanceof model
+        ? jobOrId
+        : await this.workflow.app.db.getRepository('jobs').findOne({ filterByTk: jobOrId });
     if (!job.execution) {
       job.execution = await job.getExecution();
     }
     if (job.execution.status === EXECUTION_STATUS.STARTED) {
       this.workflow.resume(job);
     }
-    if (this.timers.get(job.id)) {
-      this.timers.delete(job.id);
+    const idStr = job.id.toString();
+    if (this.timers.get(idStr)) {
+      clearTimeout(this.timers.get(idStr));
+      this.timers.delete(idStr);
     }
   }
 
@@ -104,10 +110,10 @@ export default class extends Instruction {
       nodeKey: node.key,
       upstreamId: prevJob?.id ?? null,
     });
-    job.node = node;
 
     // add to schedule
     this.schedule(job);
+    processor.logger.debug(`delay node (${node.id}) will resume after ${duration}ms`);
 
     return null;
   }

packages/plugins/@nocobase/plugin-workflow-request/src/server/RequestInstruction.ts

@@ -10,7 +10,7 @@
 import axios, { AxiosRequestConfig } from 'axios';
 import { trim } from 'lodash';
 
-import { Processor, Instruction, JOB_STATUS, FlowNodeModel } from '@nocobase/plugin-workflow';
+import { Processor, Instruction, JOB_STATUS, FlowNodeModel, IJob } from '@nocobase/plugin-workflow';
 import PluginFileManagerServer, { AttachmentModel } from '@nocobase/plugin-file-manager';
 import { Application } from '@nocobase/server';
 import { Readable } from 'stream';
@@ -183,22 +183,24 @@ export default class extends Instruction {
       }
     }
 
-    const job = processor.saveJob({
+    const { id } = processor.saveJob({
       status: JOB_STATUS.PENDING,
       nodeId: node.id,
       nodeKey: node.key,
       upstreamId: prevJob?.id ?? null,
     });
 
+    const jobDone: IJob = {
+      status: JOB_STATUS.PENDING,
+    };
+
     // eslint-disable-next-line promise/catch-or-return
     request(config, this.workflow.app)
       .then((response) => {
         processor.logger.info(`request (#${node.id}) response success, status: ${response.status}`);
 
-        job.set({
-          status: JOB_STATUS.RESOLVED,
-          result: responseSuccess(response, config.onlyData),
-        });
+        jobDone.status = JOB_STATUS.RESOLVED;
+        jobDone.result = responseSuccess(response, config.onlyData);
       })
       .catch((error) => {
         if (error.isAxiosError) {
@@ -213,14 +215,22 @@ export default class extends Instruction {
           processor.logger.error(`request (#${node.id}) failed unexpectedly: ${error.message}`);
         }
 
-        job.set({
-          status: JOB_STATUS.FAILED,
-          result: responseFailure(error),
-        });
+        jobDone.status = JOB_STATUS.FAILED;
+        jobDone.result = responseFailure(error);
       })
       .finally(() => {
         processor.logger.debug(`request (#${node.id}) ended, resume workflow...`);
-        setTimeout(() => {
+        setTimeout(async () => {
+          const job = await this.workflow.app.db.getRepository('jobs').findOne({
+            filterByTk: id,
+          });
+          if (!job) {
+            processor.logger.error(
+              `request job (${id}) not found, execution (${processor.execution.id}) cannot be resumed.`,
+            );
+            return;
+          }
+          job.set(jobDone);
           job.execution = processor.execution;
           this.workflow.resume(job);
         });

packages/plugins/@nocobase/plugin-workflow/src/server/Processor.ts

@@ -16,7 +16,7 @@ import set from 'lodash/set';
 import type Plugin from './Plugin';
 import { EXECUTION_STATUS, JOB_STATUS } from './constants';
 import { Runner } from './instructions';
-import { ExecutionModel, FlowNodeModel, JobModel } from './types';
+import type { ExecutionModel, FlowNodeModel, JobModel } from './types';
 
 export interface ProcessorOptions extends Transactionable {
   plugin: Plugin;
@@ -55,7 +55,7 @@ export default class Processor {
   /**
    * @experimental
    */
-  nodesMap = new Map<number, FlowNodeModel>();
+  nodesMap = new Map<number | string, FlowNodeModel>();
 
   private jobsMapByNodeKey: { [key: string]: JobModel } = {};
   private jobResultsMapByNodeKey: { [key: string]: any } = {};
@@ -283,7 +283,25 @@ export default class Processor {
         if (job.isNewRecord) {
           newJobs.push(job);
         } else {
-          await job.save({ transaction: this.mainTransaction });
+          const JobCollection = this.options.plugin.db.getCollection('jobs');
+          const changes = [];
+          if (job.changed('status')) {
+            changes.push([`status`, job.status]);
+            job.changed('status', false);
+          }
+          if (job.changed('result')) {
+            changes.push([`result`, JSON.stringify(job.result)]);
+            job.changed('result', false);
+          }
+          if (changes.length) {
+            await this.options.plugin.db.sequelize.query(
+              `UPDATE ${JobCollection.quotedTableName()} SET ${changes.map(([key]) => `${key} = ?`)} WHERE id='${
+                job.id
+              }'`,
+              { replacements: changes.map(([, value]) => value), transaction: this.mainTransaction },
+            );
+          }
+          // await job.save({ transaction: this.mainTransaction });
         }
       }
       if (newJobs.length) {
@@ -323,13 +341,18 @@ export default class Processor {
       job = payload;
       job.set('updatedAt', new Date());
     } else {
-      job = model.build({
-        ...payload,
-        id: this.options.plugin.snowflake.getUniqueID().toString(),
-        createdAt: new Date(),
-        updatedAt: new Date(),
-        executionId: this.execution.id,
-      });
+      job = model.build(
+        {
+          ...payload,
+          id: this.options.plugin.snowflake.getUniqueID().toString(),
+          createdAt: new Date(),
+          updatedAt: new Date(),
+          executionId: this.execution.id,
+        },
+        {
+          isNewRecord: true,
+        },
+      );
     }
     this.jobsToSave.set(job.id, job);
 
@@ -417,7 +440,7 @@ export default class Processor {
   /**
    * @experimental
    */
-  public getScope(sourceNodeId: number, includeSelfScope = false) {
+  public getScope(sourceNodeId: number | string, includeSelfScope = false) {
     const node = this.nodesMap.get(sourceNodeId);
     const systemFns = {};
     const scope = {
@@ -448,7 +471,7 @@ export default class Processor {
   /**
    * @experimental
    */
-  public getParsedValue(value, sourceNodeId: number, { additionalScope = {}, includeSelfScope = false } = {}) {
+  public getParsedValue(value, sourceNodeId: number | string, { additionalScope = {}, includeSelfScope = false } = {}) {
     const template = parse(value);
     const scope = Object.assign(this.getScope(sourceNodeId, includeSelfScope), additionalScope);
     template.parameters.forEach(({ key }) => {