fix(plugin-workflow): fix schedule trigger (#3096)

* fix(plugin-workflow): fix schedule trigger

* fix(plugin-workflow): fix increment bug based on dialect

packages/plugins/@nocobase/plugin-workflow/src/locale/zh-CN.ts

@@ -60,7 +60,7 @@ export default {
   'Please select the associated fields that need to be accessed in subsequent nodes. With more than two levels of to-many associations may cause performance issue, please use with caution.':
     '请选中需要在后续节点中被访问的关系字段。超过两层的对多关联可能会导致性能问题，请谨慎使用。',
   'Schedule event': '定时任务',
-  'Scheduled job base on time conditions.': '基于时间条件的计划任务',
+  'Event will be scheduled and triggered base on time conditions.': '基于时间条件进行定时触发的事件。',
   'Trigger mode': '触发模式',
   'Based on certain date': '自定义时间',
   'Based on date field of collection': '根据数据表时间字段',

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

@@ -280,14 +280,23 @@ export default class WorkflowPlugin extends Plugin {
           { transaction },
         );
 
-        await workflow.increment('executed', { transaction });
+        await workflow.increment(['executed', 'allExecuted'], { transaction });
+        // NOTE: https://sequelize.org/api/v6/class/src/model.js~model#instance-method-increment
+        if (this.db.options.dialect !== 'postgres') {
+          await workflow.reload({ transaction });
+        }
 
-        await (<typeof WorkflowModel>workflow.constructor).increment('allExecuted', {
+        await (<typeof WorkflowModel>workflow.constructor).update(
+          {
+            allExecuted: workflow.allExecuted,
+          },
+          {
             where: {
               key: workflow.key,
             },
             transaction,
-        });
+          },
+        );
 
         execution.workflow = workflow;
 

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

@@ -213,7 +213,9 @@ export default class Processor {
     const { instructions } = this.options.plugin;
     const instruction = instructions.get(node.type);
     if (typeof instruction.resume !== 'function') {
-      return Promise.reject(new Error('`resume` should be implemented'));
+      return Promise.reject(
+        new Error(`"resume" method should be implemented for [${node.type}] instruction of node (#${node.id})`),
+      );
     }
 
     return this.exec(instruction.resume.bind(instruction), node, job);

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

@@ -413,6 +413,82 @@ describe('workflow > Processor', () => {
       expect(jobs.length).toEqual(5);
     });
 
+    it('condition contains loop (target as 0)', async () => {
+      const n1 = await workflow.createNode({
+        type: 'condition',
+      });
+
+      const n2 = await workflow.createNode({
+        type: 'loop',
+        branchIndex: BRANCH_INDEX.ON_TRUE,
+        upstreamId: n1.id,
+        config: {
+          target: 0,
+        },
+      });
+
+      const n3 = await workflow.createNode({
+        type: 'echo',
+        branchIndex: 0,
+        upstreamId: n2.id,
+      });
+
+      const n4 = await workflow.createNode({
+        type: 'echo',
+        upstreamId: n1.id,
+      });
+
+      await n1.setDownstream(n4);
+
+      const post = await PostRepo.create({ values: { title: 't1' } });
+
+      await sleep(500);
+
+      const [e1] = await workflow.getExecutions();
+      expect(e1.status).toEqual(EXECUTION_STATUS.RESOLVED);
+      const jobs = await e1.getJobs({ order: [['id', 'ASC']] });
+      expect(jobs.length).toBe(3);
+      expect(jobs[0].status).toBe(JOB_STATUS.RESOLVED);
+    });
+
+    it('condition contains loop (target as 2)', async () => {
+      const n1 = await workflow.createNode({
+        type: 'condition',
+      });
+
+      const n2 = await workflow.createNode({
+        type: 'loop',
+        branchIndex: BRANCH_INDEX.ON_TRUE,
+        upstreamId: n1.id,
+        config: {
+          target: 2,
+        },
+      });
+
+      const n3 = await workflow.createNode({
+        type: 'echo',
+        branchIndex: 0,
+        upstreamId: n2.id,
+      });
+
+      const n4 = await workflow.createNode({
+        type: 'echo',
+        upstreamId: n1.id,
+      });
+
+      await n1.setDownstream(n4);
+
+      const post = await PostRepo.create({ values: { title: 't1' } });
+
+      await sleep(500);
+
+      const [e1] = await workflow.getExecutions();
+      expect(e1.status).toEqual(EXECUTION_STATUS.RESOLVED);
+      const jobs = await e1.getJobs({ order: [['id', 'ASC']] });
+      expect(jobs.length).toBe(5);
+      expect(jobs[0].status).toBe(JOB_STATUS.RESOLVED);
+    });
+
     it('parallel branches contains condition', async () => {
       const n1 = await workflow.createNode({
         type: 'parallel',

packages/plugins/@nocobase/plugin-workflow/src/server/__tests__/instructions/request.test.ts

@@ -1,5 +1,9 @@
-import { Application, Gateway } from '@nocobase/server';
+import jwt from 'jsonwebtoken';
+
+import { Gateway } from '@nocobase/server';
 import Database from '@nocobase/database';
+import { MockServer } from '@nocobase/test';
+
 import { getApp, sleep } from '..';
 import { RequestConfig } from '../../instructions/request';
 import { EXECUTION_STATUS, JOB_STATUS } from '../../constants';
@@ -11,14 +15,20 @@ const URL_400 = `http://localhost:${PORT}/api/400`;
 const URL_TIMEOUT = `http://localhost:${PORT}/api/timeout`;
 
 describe('workflow > instructions > request', () => {
-  let app: Application;
+  let app: MockServer;
   let db: Database;
   let PostRepo;
   let WorkflowModel;
   let workflow;
 
   beforeEach(async () => {
-    app = await getApp({ manual: true });
+    app = await getApp({
+      resourcer: {
+        prefix: '/api',
+      },
+      plugins: ['users', 'auth'],
+      manual: true,
+    });
 
     app.use(async (ctx, next) => {
       if (ctx.path === '/api/400') {
@@ -29,13 +39,14 @@ describe('workflow > instructions > request', () => {
         return ctx.throw(new Error('timeout'));
       }
       if (ctx.path === '/api/data') {
+        await sleep(100);
         ctx.withoutDataWrapping = true;
         ctx.body = {
           meta: { title: ctx.query.title },
           data: { title: ctx.request.body['title'] },
         };
       }
-      next();
+      await next();
     });
 
     Gateway.getInstance().start({
@@ -62,8 +73,8 @@ describe('workflow > instructions > request', () => {
 
   afterEach(() => app.destroy());
 
-  describe('request', () => {
+  describe('request static app routes', () => {
-    it('request', async () => {
+    it('get data', async () => {
       await workflow.createNode({
         type: 'request',
         config: {
@@ -80,6 +91,7 @@ describe('workflow > instructions > request', () => {
       expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED);
       const [job] = await execution.getJobs();
       expect(job.status).toEqual(JOB_STATUS.RESOLVED);
+      expect(job.result).toEqual({ meta: {}, data: {} });
     });
 
     it('request - timeout', async () => {
@@ -249,4 +261,41 @@ describe('workflow > instructions > request', () => {
       expect(jobs[0].result).toBe(2);
     });
   });
+
+  describe('request db resource', () => {
+    it('request db resource', async () => {
+      const user = await db.getRepository('users').create({});
+
+      const token = jwt.sign(
+        {
+          userId: typeof user.id,
+        },
+        process.env.APP_KEY,
+        {
+          expiresIn: '1d',
+        },
+      );
+
+      const n1 = await workflow.createNode({
+        type: 'request',
+        config: {
+          url: `http://localhost:${PORT}/api/categories`,
+          method: 'POST',
+          headers: [{ name: 'Authorization', value: `Bearer ${token}` }],
+        } as RequestConfig,
+      });
+
+      await PostRepo.create({ values: { title: 't1' } });
+
+      await sleep(500);
+
+      const category = await db.getRepository('categories').findOne({});
+
+      const [execution] = await workflow.getExecutions();
+      expect(execution.status).toBe(EXECUTION_STATUS.RESOLVED);
+      const [job] = await execution.getJobs();
+      expect(job.status).toBe(JOB_STATUS.RESOLVED);
+      expect(job.result.data).toMatchObject({});
+    });
+  });
 });

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

@@ -32,7 +32,7 @@ interface ScheduleMode {
   on?(this: ScheduleTrigger, workflow: WorkflowModel): void;
   off?(this: ScheduleTrigger, workflow: WorkflowModel): void;
   shouldCache(this: ScheduleTrigger, workflow: WorkflowModel, now: Date): Promise<boolean> | boolean;
-  trigger(this: ScheduleTrigger, workflow: WorkflowModel, now: Date): any;
+  trigger(this: ScheduleTrigger, workflow: WorkflowModel, now: Date): Promise<number> | number;
 }
 
 const ScheduleModes = new Map<number, ScheduleMode>();
@@ -80,29 +80,31 @@ ScheduleModes.set(SCHEDULE_MODE.CONSTANT, {
     // NOTE: align to second start
     const startTime = parseDateWithoutMs(startsOn);
     if (!startTime || startTime > timestamp) {
-      return;
+      return 0;
     }
 
     if (repeat) {
       if (typeof repeat === 'number') {
         if (Math.round(timestamp - startTime) % repeat) {
-          return;
+          return 0;
         }
       }
 
       if (endsOn) {
         const endTime = parseDateWithoutMs(endsOn);
         if (!endTime || endTime < timestamp) {
-          return;
+          return 0;
         }
       }
     } else {
       if (startTime !== timestamp) {
-        return;
+        return 0;
       }
     }
 
-    return this.plugin.trigger(workflow, { date: now });
+    this.plugin.trigger(workflow, { date: now });
+
+    return 1;
   },
 });
 
@@ -233,14 +235,14 @@ ScheduleModes.set(SCHEDULE_MODE.COLLECTION_FIELD, {
     // when repeat is number, means repeat after startsOn
     // (now - startsOn) % repeat <= cacheCycle
     if (repeat) {
-      const tsFn = DialectTimestampFnMap[db.options.dialect!];
+      const tsFn = DialectTimestampFnMap[db.options.dialect];
       if (typeof repeat === 'number' && repeat > this.cacheCycle && tsFn) {
-        conditions.push(
+        const modExp = fn(
-          where(
+          'MOD',
-            fn('MOD', literal(`${Math.round(timestamp / 1000)} - ${tsFn(startsOn.field)}`), Math.round(repeat / 1000)),
+          literal(`${Math.round(timestamp / 1000)} - ${tsFn(startsOn.field)}`),
-            { [Op.lt]: Math.round(this.cacheCycle / 1000) },
+          Math.round(repeat / 1000),
-          ),
         );
+        conditions.push(where(modExp, { [Op.lt]: Math.round(this.cacheCycle / 1000) }));
         // conditions.push(literal(`mod(${timestamp} - ${tsFn(startsOn.field)} * 1000, ${repeat}) < ${this.cacheCycle}`));
       }
 
@@ -283,7 +285,7 @@ ScheduleModes.set(SCHEDULE_MODE.COLLECTION_FIELD, {
 
     const startTimestamp = getOnTimestampWithOffset(startsOn, now);
     if (!startTimestamp) {
-      return false;
+      return 0;
     }
 
     const conditions: any[] = [
@@ -302,26 +304,26 @@ ScheduleModes.set(SCHEDULE_MODE.COLLECTION_FIELD, {
         },
       });
 
-      const tsFn = DialectTimestampFnMap[this.plugin.app.db.options.dialect!];
+      const tsFn = DialectTimestampFnMap[this.plugin.app.db.options.dialect];
       if (typeof repeat === 'number' && tsFn) {
-        conditions.push(
+        const modExp = fn(
-          where(
+          'MOD',
-            fn('MOD', literal(`${Math.round(timestamp / 1000)} - ${tsFn(startsOn.field)}`), Math.round(repeat / 1000)),
+          literal(`${Math.round(timestamp / 1000)} - ${tsFn(startsOn.field)}`),
-            { [Op.eq]: 0 },
+          Math.round(repeat / 1000),
-          ),
         );
+        conditions.push(where(modExp, { [Op.eq]: 0 }));
         // conditions.push(literal(`MOD(CAST(${timestamp} AS BIGINT) - CAST((FLOOR(${tsFn(startsOn.field)}) AS BIGINT) * 1000), ${repeat}) = 0`));
       }
 
       if (endsOn) {
         const endTimestamp = getOnTimestampWithOffset(endsOn, now);
         if (!endTimestamp) {
-          return false;
+          return 0;
         }
 
         if (typeof endsOn === 'string') {
           if (endTimestamp <= timestamp) {
-            return false;
+            return 0;
           }
         } else {
           conditions.push({
@@ -342,10 +344,15 @@ ScheduleModes.set(SCHEDULE_MODE.COLLECTION_FIELD, {
 
     const repo = this.plugin.app.db.getRepository(collection);
     const instances = await repo.find({
-      filter: {
+      where: {
-        $and: conditions,
+        [Op.and]: conditions,
       },
       appends,
+      ...(workflow.config.limit
+        ? {
+            limit: Math.max(workflow.config.limit - workflow.allExecuted, 0),
+          }
+        : {}),
     });
 
     instances.forEach((item) => {
@@ -354,6 +361,8 @@ ScheduleModes.set(SCHEDULE_MODE.COLLECTION_FIELD, {
         data: item.toJSON(),
       });
     });
+
+    return instances.length;
   },
 });
 
@@ -424,7 +433,7 @@ export default class ScheduleTrigger extends Trigger {
   }
 
   init() {
-    if (this.plugin.app.getPlugin('multi-app-share-collection').enabled && this.plugin.app.name !== 'main') {
+    if (this.plugin.app.getPlugin('multi-app-share-collection')?.enabled && this.plugin.app.name !== 'main') {
       return;
     }
 
@@ -465,7 +474,6 @@ export default class ScheduleTrigger extends Trigger {
   async onTick(now) {
     // NOTE: trigger workflows in sequence when sqlite due to only one transaction
     const isSqlite = this.plugin.app.db.options.dialect === 'sqlite';
-
     return Array.from(this.cache.values()).reduce(
       (prev, workflow) => {
         if (!this.shouldTrigger(workflow, now)) {
@@ -482,19 +490,9 @@ export default class ScheduleTrigger extends Trigger {
   }
 
   async reload() {
-    const WorkflowModel = this.plugin.app.db.getCollection('workflows').model;
+    const WorkflowRepo = this.plugin.app.db.getRepository('workflows');
-    const workflows = await WorkflowModel.findAll({
+    const workflows = await WorkflowRepo.find({
-      where: { enabled: true, type: 'schedule' },
+      filter: { enabled: true, type: 'schedule' },
-      include: [
-        {
-          association: 'executions',
-          attributes: ['id', 'createdAt'],
-          separate: true,
-          limit: 1,
-          order: [['createdAt', 'DESC']],
-        },
-      ],
-      group: ['id'],
     });
 
     // NOTE: clear cached jobs in last cycle