fix(plugin-workflow): fix validations (#9111)

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

@@ -21,7 +21,7 @@ export type LoopInstructionConfig = {
   target: any;
   condition?:
     | {
-        checkpoint?: number;
+        checkpoint?: 0 | 1;
         continueOnFalse?: boolean;
         calculation?: any;
         expression?: string;
@@ -55,6 +55,17 @@ function calculateCondition(node: FlowNodeModel, processor: Processor) {
 
 export default class extends Instruction {
   configSchema = Joi.object({
+    target: Joi.alternatives().try(Joi.number(), Joi.array(), Joi.string()),
+    condition: Joi.alternatives()
+      .try(
+        Joi.object({
+          checkpoint: Joi.number().valid(0, 1).default(0),
+          continueOnFalse: Joi.boolean(),
+          calculation: Joi.object(),
+        }),
+        Joi.boolean().valid(false),
+      )
+      .default(false),
     exit: Joi.number().valid(EXIT.RETURN, EXIT.BREAK, EXIT.CONTINUE),
   });
 

packages/plugins/@nocobase/plugin-workflow-loop/src/server/__tests__/env-limit.test.ts

@@ -0,0 +1,161 @@
+/**
+ * This file is part of the NocoBase (R) project.
+ * Copyright (c) 2020-2024 NocoBase Co., Ltd.
+ * Authors: NocoBase Team.
+ *
+ * This project is dual-licensed under AGPL-3.0 and NocoBase Commercial License.
+ * For more information, please refer to: https://www.nocobase.com/agreement.
+ */
+
+import Database from '@nocobase/database';
+import { Application } from '@nocobase/server';
+import { EXECUTION_STATUS, JOB_STATUS } from '@nocobase/plugin-workflow';
+import { getApp } from '@nocobase/plugin-workflow-test';
+import Plugin from '..';
+
+describe('workflow > instruction > loop > process.env.WORKFLOW_LOOP_LIMIT', () => {
+  let app: Application;
+  let db: Database;
+  let PostRepo;
+  let WorkflowModel;
+  let workflow;
+  let plugin;
+
+  afterEach(() => app.destroy());
+
+  describe('limit = 1', () => {
+    let original: string | undefined;
+
+    beforeEach(async () => {
+      original = process.env.WORKFLOW_LOOP_LIMIT;
+      process.env.WORKFLOW_LOOP_LIMIT = '1';
+
+      app = await getApp({
+        plugins: [Plugin],
+      });
+      plugin = app.pm.get('workflow');
+
+      db = app.db;
+      WorkflowModel = db.getCollection('workflows').model;
+      PostRepo = db.getCollection('posts').repository;
+
+      workflow = await WorkflowModel.create({
+        enabled: true,
+        sync: true,
+        type: 'collection',
+        config: {
+          mode: 1,
+          collection: 'posts',
+        },
+      });
+    });
+
+    afterEach(() => {
+      process.env.WORKFLOW_LOOP_LIMIT = original;
+    });
+
+    it('limit exceeded should stop loop', async () => {
+      const n1 = await workflow.createNode({
+        type: 'loop',
+        config: {
+          target: 10,
+        },
+      });
+
+      const n2 = await workflow.createNode({
+        type: 'echo',
+        upstreamId: n1.id,
+        branchIndex: 0,
+      });
+
+      await PostRepo.create({ values: { title: 't1' } });
+
+      const [execution] = await workflow.getExecutions();
+      expect(execution.status).toBe(EXECUTION_STATUS.ERROR);
+      const jobs = await execution.getJobs({ order: [['id', 'ASC']] });
+      expect(jobs.length).toBe(2);
+      expect(jobs[0].status).toBe(JOB_STATUS.ERROR);
+      expect(jobs[0].result).toEqual({ looped: 1, done: 1, exceeded: true });
+    });
+
+    it('limit not exceeded should be resolved', async () => {
+      const n1 = await workflow.createNode({
+        type: 'loop',
+        config: {
+          target: 10,
+        },
+      });
+
+      const n2 = await workflow.createNode({
+        type: 'echo',
+        upstreamId: n1.id,
+        branchIndex: 0,
+      });
+
+      await PostRepo.create({ values: { title: 't1' } });
+
+      const [execution] = await workflow.getExecutions();
+      expect(execution.status).toBe(EXECUTION_STATUS.ERROR);
+      const jobs = await execution.getJobs({ order: [['id', 'ASC']] });
+      expect(jobs.length).toBe(2);
+      expect(jobs[0].status).toBe(JOB_STATUS.ERROR);
+      expect(jobs[0].result).toEqual({ looped: 1, done: 1, exceeded: true });
+    });
+  });
+
+  describe('limit = 0', () => {
+    let original: string | undefined;
+
+    beforeEach(async () => {
+      original = process.env.WORKFLOW_LOOP_LIMIT;
+      process.env.WORKFLOW_LOOP_LIMIT = '0';
+
+      app = await getApp({
+        plugins: [Plugin],
+      });
+      plugin = app.pm.get('workflow');
+
+      db = app.db;
+      WorkflowModel = db.getCollection('workflows').model;
+      PostRepo = db.getCollection('posts').repository;
+
+      workflow = await WorkflowModel.create({
+        enabled: true,
+        sync: true,
+        type: 'collection',
+        config: {
+          mode: 1,
+          collection: 'posts',
+        },
+      });
+    });
+
+    afterEach(() => {
+      process.env.WORKFLOW_LOOP_LIMIT = original;
+    });
+
+    it('zero disables limit', async () => {
+      const n1 = await workflow.createNode({
+        type: 'loop',
+        config: {
+          target: 10,
+        },
+      });
+
+      const n2 = await workflow.createNode({
+        type: 'echo',
+        upstreamId: n1.id,
+        branchIndex: 0,
+      });
+
+      await PostRepo.create({ values: { title: 't1' } });
+
+      const [execution] = await workflow.getExecutions();
+      expect(execution.status).toBe(EXECUTION_STATUS.RESOLVED);
+      const jobs = await execution.getJobs({ order: [['id', 'ASC']] });
+      expect(jobs.length).toBe(11);
+      expect(jobs[0].status).toBe(JOB_STATUS.RESOLVED);
+      expect(jobs[0].result).toEqual({ looped: 10, done: 10 });
+    });
+  });
+});

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

@@ -1091,185 +1091,4 @@ describe('workflow > instructions > loop', () => {
       expect(jobs[0].status).toBe(JOB_STATUS.RESOLVED);
     });
   });
-
-  describe('validation', () => {
-    let agent;
-    let validationWorkflow;
-
-    beforeEach(async () => {
-      agent = (app as any).agent();
-      validationWorkflow = await WorkflowModel.create({
-        enabled: true,
-        type: 'asyncTrigger',
-      });
-    });
-
-    it('should reject when exit is invalid', async () => {
-      const { status } = await agent.resource('workflows.nodes', validationWorkflow.id).create({
-        values: { type: 'loop', config: { exit: 99 } },
-      });
-      expect(status).toBe(400);
-    });
-
-    it('should accept with valid exit value', async () => {
-      const { status } = await agent.resource('workflows.nodes', validationWorkflow.id).create({
-        values: { type: 'loop', config: { exit: EXIT.RETURN } },
-      });
-      expect(status).toBe(200);
-    });
-
-    it('should accept with empty config', async () => {
-      const { status } = await agent.resource('workflows.nodes', validationWorkflow.id).create({
-        values: { type: 'loop', config: {} },
-      });
-      expect(status).toBe(200);
-    });
-  });
-});
-
-describe('process.env.WORKFLOW_LOOP_LIMIT', () => {
-  let app: Application;
-  let db: Database;
-  let PostRepo;
-  let WorkflowModel;
-  let workflow;
-  let plugin;
-
-  afterEach(() => app.destroy());
-
-  describe('limit = 1', () => {
-    let original: string | undefined;
-
-    beforeEach(async () => {
-      original = process.env.WORKFLOW_LOOP_LIMIT;
-      process.env.WORKFLOW_LOOP_LIMIT = '1';
-
-      app = await getApp({
-        plugins: [Plugin],
-      });
-      plugin = app.pm.get('workflow');
-
-      db = app.db;
-      WorkflowModel = db.getCollection('workflows').model;
-      PostRepo = db.getCollection('posts').repository;
-
-      workflow = await WorkflowModel.create({
-        enabled: true,
-        sync: true,
-        type: 'collection',
-        config: {
-          mode: 1,
-          collection: 'posts',
-        },
-      });
-    });
-
-    afterEach(() => {
-      process.env.WORKFLOW_LOOP_LIMIT = original;
-    });
-
-    it('limit exceeded should stop loop', async () => {
-      const n1 = await workflow.createNode({
-        type: 'loop',
-        config: {
-          target: 10,
-        },
-      });
-
-      const n2 = await workflow.createNode({
-        type: 'echo',
-        upstreamId: n1.id,
-        branchIndex: 0,
-      });
-
-      await PostRepo.create({ values: { title: 't1' } });
-
-      const [execution] = await workflow.getExecutions();
-      expect(execution.status).toBe(EXECUTION_STATUS.ERROR);
-      const jobs = await execution.getJobs({ order: [['id', 'ASC']] });
-      expect(jobs.length).toBe(2);
-      expect(jobs[0].status).toBe(JOB_STATUS.ERROR);
-      expect(jobs[0].result).toEqual({ looped: 1, done: 1, exceeded: true });
-    });
-
-    it('limit not exceeded should be resolved', async () => {
-      const n1 = await workflow.createNode({
-        type: 'loop',
-        config: {
-          target: 10,
-        },
-      });
-
-      const n2 = await workflow.createNode({
-        type: 'echo',
-        upstreamId: n1.id,
-        branchIndex: 0,
-      });
-
-      await PostRepo.create({ values: { title: 't1' } });
-
-      const [execution] = await workflow.getExecutions();
-      expect(execution.status).toBe(EXECUTION_STATUS.ERROR);
-      const jobs = await execution.getJobs({ order: [['id', 'ASC']] });
-      expect(jobs.length).toBe(2);
-      expect(jobs[0].status).toBe(JOB_STATUS.ERROR);
-      expect(jobs[0].result).toEqual({ looped: 1, done: 1, exceeded: true });
-    });
-  });
-
-  describe('limit = 0', () => {
-    let original: string | undefined;
-
-    beforeEach(async () => {
-      original = process.env.WORKFLOW_LOOP_LIMIT;
-      process.env.WORKFLOW_LOOP_LIMIT = '0';
-
-      app = await getApp({
-        plugins: [Plugin],
-      });
-      plugin = app.pm.get('workflow');
-
-      db = app.db;
-      WorkflowModel = db.getCollection('workflows').model;
-      PostRepo = db.getCollection('posts').repository;
-
-      workflow = await WorkflowModel.create({
-        enabled: true,
-        sync: true,
-        type: 'collection',
-        config: {
-          mode: 1,
-          collection: 'posts',
-        },
-      });
-    });
-
-    afterEach(() => {
-      process.env.WORKFLOW_LOOP_LIMIT = original;
-    });
-
-    it('zero disables limit', async () => {
-      const n1 = await workflow.createNode({
-        type: 'loop',
-        config: {
-          target: 10,
-        },
-      });
-
-      const n2 = await workflow.createNode({
-        type: 'echo',
-        upstreamId: n1.id,
-        branchIndex: 0,
-      });
-
-      await PostRepo.create({ values: { title: 't1' } });
-
-      const [execution] = await workflow.getExecutions();
-      expect(execution.status).toBe(EXECUTION_STATUS.RESOLVED);
-      const jobs = await execution.getJobs({ order: [['id', 'ASC']] });
-      expect(jobs.length).toBe(11);
-      expect(jobs[0].status).toBe(JOB_STATUS.RESOLVED);
-      expect(jobs[0].result).toEqual({ looped: 10, done: 10 });
-    });
-  });
 });

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

@@ -0,0 +1,60 @@
+/**
+ * This file is part of the NocoBase (R) project.
+ * Copyright (c) 2020-2024 NocoBase Co., Ltd.
+ * Authors: NocoBase Team.
+ *
+ * This project is dual-licensed under AGPL-3.0 and NocoBase Commercial License.
+ * For more information, please refer to: https://www.nocobase.com/agreement.
+ */
+
+import { MockServer, ExtendedAgent } from '@nocobase/test';
+import { getApp } from '@nocobase/plugin-workflow-test';
+import { WorkflowModel } from '@nocobase/plugin-workflow';
+import { EXIT } from '../../constants';
+import Plugin from '..';
+
+describe('workflow > instruction > loop > validation', () => {
+  let app: MockServer;
+  let agent: ExtendedAgent;
+  let validationWorkflow: WorkflowModel;
+
+  beforeEach(async () => {
+    app = await getApp({
+      plugins: [Plugin],
+    });
+
+    const db = app.db;
+    const WorkflowRepo = db.getCollection('workflows').repository;
+    agent = (app as MockServer).agent();
+    validationWorkflow = await WorkflowRepo.create({
+      values: {
+        enabled: true,
+        sync: true,
+        type: 'asyncTrigger',
+      },
+    });
+  });
+
+  afterEach(() => app.destroy());
+
+  it('should reject when exit is invalid', async () => {
+    const { status } = await agent.resource('workflows.nodes', validationWorkflow.id).create({
+      values: { type: 'loop', config: { exit: 99 } },
+    });
+    expect(status).toBe(400);
+  });
+
+  it('should accept with valid exit value', async () => {
+    const { status } = await agent.resource('workflows.nodes', validationWorkflow.id).create({
+      values: { type: 'loop', config: { exit: EXIT.RETURN } },
+    });
+    expect(status).toBe(200);
+  });
+
+  it('should accept with empty config', async () => {
+    const { status } = await agent.resource('workflows.nodes', validationWorkflow.id).create({
+      values: { type: 'loop', config: {} },
+    });
+    expect(status).toBe(200);
+  });
+});

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

@@ -33,16 +33,16 @@ export default class MailerInstruction extends Instruction {
         user: Joi.string(),
         pass: Joi.string(),
       }),
-      from: Joi.string(),
-      to: Joi.array().items(Joi.string()),
-      cc: Joi.array().items(Joi.string()),
-      bcc: Joi.array().items(Joi.string()),
-      subject: Joi.string(),
-      contentType: Joi.string().valid('html', 'text').default('html'),
-      html: Joi.string(),
-      text: Joi.string(),
-      ignoreFail: Joi.boolean().default(false),
     }),
+    from: Joi.string(),
+    to: Joi.array().items(Joi.string()),
+    cc: Joi.array().items(Joi.string()),
+    bcc: Joi.array().items(Joi.string()),
+    subject: Joi.string(),
+    contentType: Joi.string().valid('html', 'text').default('html'),
+    html: Joi.string(),
+    text: Joi.string(),
+    ignoreFail: Joi.boolean().default(false),
   });
 
   private static transporterMap = new Map<string, Transporter>();

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

@@ -80,7 +80,7 @@ describe('workflow > instructions > mailer > validation', () => {
     const { status } = await agent.resource('workflows.nodes', workflow.id).create({
       values: {
         type: 'mailer',
-        config: { provider: { host: 'smtp.example.com', from: '{{$variable.from}}' } },
+        config: { provider: { host: 'smtp.example.com' }, from: '{{$variable.from}}' },
       },
     });
     expect(status).toBe(200);
@@ -90,7 +90,7 @@ describe('workflow > instructions > mailer > validation', () => {
     const { status } = await agent.resource('workflows.nodes', workflow.id).create({
       values: {
         type: 'mailer',
-        config: { provider: { host: 'smtp.example.com', to: ['{{$context.data.email}}'] } },
+        config: { provider: { host: 'smtp.example.com' }, to: ['{{$context.data.email}}'] },
       },
     });
     expect(status).toBe(200);
@@ -103,9 +103,9 @@ describe('workflow > instructions > mailer > validation', () => {
         config: {
           provider: {
             host: 'smtp.example.com',
-            cc: ['{{$context.data.cc}}'],
-            bcc: ['{{$context.data.bcc}}'],
           },
+          cc: ['{{$context.data.cc}}'],
+          bcc: ['{{$context.data.bcc}}'],
         },
       },
     });

packages/plugins/@nocobase/plugin-workflow-notification/package.json

@@ -14,6 +14,7 @@
   "homepage.zh-CN": "https://docs-cn.nocobase.com/handbook/workflow-smtp-mailer",
   "devDependencies": {
     "antd": "5.x",
+    "joi": "^17.13.3",
     "react": "18.x"
   },
   "peerDependencies": {

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

@@ -7,11 +7,17 @@
  * For more information, please refer to: https://www.nocobase.com/agreement.
  */
 
+import Joi from 'joi';
 import NotificationsServerPlugin from '@nocobase/plugin-notification-manager';
 
 import { Processor, Instruction, JOB_STATUS, FlowNodeModel } from '@nocobase/plugin-workflow';
 
 export default class extends Instruction {
+  configSchema = Joi.object({
+    channelName: Joi.string(),
+    ignoreFail: Joi.boolean().default(false),
+  });
+
   async run(node: FlowNodeModel, prevJob, processor: Processor) {
     const { ignoreFail, ...config } = node.config;
     const options = processor.getParsedValue(config, node.id);

packages/plugins/@nocobase/plugin-workflow-response-message/package.json

@@ -11,6 +11,9 @@
   "homepage": "https://docs.nocobase.com/handbook/workflow-response-message",
   "homepage.ru-RU": "https://docs-ru.nocobase.com/handbook/workflow-response-message",
   "homepage.zh-CN": "https://docs-cn.nocobase.com/handbook/workflow-response-message",
+  "devDependencies": {
+    "joi": "^17.13.3"
+  },
   "peerDependencies": {
     "@nocobase/client": "2.x",
     "@nocobase/plugin-workflow": "2.x",

packages/plugins/@nocobase/plugin-workflow-response-message/src/server/ResponseMessageInstruction.ts

@@ -17,12 +17,17 @@
  */
 
 import { Instruction, Processor, JOB_STATUS, FlowNodeModel } from '@nocobase/plugin-workflow';
+import Joi from 'joi';
 
 interface Config {
   message?: string;
 }
 
 export default class extends Instruction {
+  configSchema = Joi.object({
+    message: Joi.string(),
+  });
+
   async run(node: FlowNodeModel, prevJob, processor: Processor) {
     const { httpContext } = processor.options;
 

packages/plugins/@nocobase/plugin-workflow-sql/package.json

@@ -14,6 +14,7 @@
   "homepage.zh-CN": "https://docs-cn.nocobase.com/handbook/workflow-sql",
   "devDependencies": {
     "antd": "5.x",
+    "joi": "^17.13.3",
     "react": "18.x",
     "react-i18next": "^11.15.1"
   },

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

@@ -9,6 +9,7 @@
 
 import { SequelizeCollectionManager } from '@nocobase/data-source-manager';
 import { Processor, Instruction, JOB_STATUS, FlowNodeModel } from '@nocobase/plugin-workflow';
+import Joi from 'joi';
 
 export type SQLInstructionConfig = {
   dataSource?: string;
@@ -19,6 +20,19 @@ export type SQLInstructionConfig = {
 };
 
 export default class extends Instruction {
+  configSchema = Joi.object({
+    dataSource: Joi.string(),
+    sql: Joi.string(),
+    withMeta: Joi.boolean().default(false),
+    unsafeInjection: Joi.boolean().default(false),
+    variables: Joi.array().items(
+      Joi.object({
+        name: Joi.string().required(),
+        value: Joi.any(),
+      }),
+    ),
+  });
+
   async run(node: FlowNodeModel, input, processor: Processor) {
     const dataSourceName = node.config.dataSource || 'main';
     const { collectionManager } = this.workflow.app.dataSourceManager.dataSources.get(dataSourceName);
@@ -26,7 +40,7 @@ export default class extends Instruction {
       throw new Error(`type of data source "${node.config.dataSource}" is not database`);
     }
 
-    const { unsafeInjection = false, variables: variablesConfig = [] } = node.config;
+    const { unsafeInjection = false, variables = [] } = node.config;
 
     let sql = '';
     let replacements = null;
@@ -34,10 +48,11 @@ export default class extends Instruction {
       sql = processor.getParsedValue(node.config.sql || '', node.id).trim();
     } else {
       sql = (node.config.sql || '').trim();
+      const parameters = processor.getParsedValue(variables, node.id);
       replacements = {};
-      for (const { name, value } of variablesConfig) {
+      for (const { name, value } of parameters) {
         if (name) {
-          replacements[name] = processor.getParsedValue(value, node.id);
+          replacements[name] = value;
         }
       }
     }
@@ -67,7 +82,7 @@ export default class extends Instruction {
     sql: sqlConfig,
     withMeta,
     unsafeInjection = false,
-    variables: variablesConfig = [],
+    variables = [],
   }: SQLInstructionConfig = {}) {
     if (!sqlConfig) {
       return {
@@ -90,7 +105,7 @@ export default class extends Instruction {
       } else {
         sql = sqlConfig.trim();
         replacements = {};
-        for (const { name, value } of variablesConfig) {
+        for (const { name, value } of variables) {
           if (name) {
             replacements[name] = value;
           }