Merge branch 'main' into next

packages/plugins/@nocobase/plugin-workflow-sql/src/client/SQLInstruction.tsx

@@ -51,6 +51,12 @@ export default class extends Instruction {
         `,
       },
     },
+    withMeta: {
+      type: 'boolean',
+      'x-decorator': 'FormItem',
+      'x-component': 'Checkbox',
+      'x-content': `{{t("Include meta information of this query in result", { ns: "${NAMESPACE}" })}}`,
+    },
   };
   scope = {
     sqlDescription() {

packages/plugins/@nocobase/plugin-workflow-sql/src/client/__e2e__/dataCURD.test.ts

@@ -119,7 +119,7 @@ test.describe('select data', () => {
     const jobs = getWorkflowNodeExecutionsObj[0].jobs;
     const sqlNodeJob = jobs.find((job) => job.nodeId.toString() === sqlNodeId);
     const sqlNodeJobResult = sqlNodeJob.result;
-    const nodeResultRecordOrgname = sqlNodeJobResult[0][0].orgname;
+    const nodeResultRecordOrgname = sqlNodeJobResult[0].orgname;
     expect(nodeResultRecordOrgname).toBe('公司名称(单行文本)1');
     // 4、后置处理：删除工作流
     await apiDeleteWorkflow(workflowId);
@@ -215,7 +215,7 @@ test.describe('select data', () => {
     const jobs = getWorkflowNodeExecutionsObj[0].jobs;
     const sqlNodeJob = jobs.find((job) => job.nodeId.toString() === sqlNodeId);
     const sqlNodeJobResult = sqlNodeJob.result;
-    const nodeResultRecordLength = sqlNodeJobResult[0].length;
+    const nodeResultRecordLength = sqlNodeJobResult.length;
     expect(nodeResultRecordLength).toBe(8);
     // 4、后置处理：删除工作流
     await apiDeleteWorkflow(workflowId);
@@ -317,7 +317,7 @@ test.describe('select data', () => {
     const jobs = getWorkflowNodeExecutionsObj[0].jobs;
     const sqlNodeJob = jobs.find((job) => job.nodeId.toString() === sqlNodeId);
     const sqlNodeJobResult = sqlNodeJob.result;
-    const nodeResultRecordOrgname = sqlNodeJobResult[0][0].orgname;
+    const nodeResultRecordOrgname = sqlNodeJobResult[0].orgname;
     expect(nodeResultRecordOrgname).toBe('公司名称(单行文本)1');
     // 4、后置处理：删除工作流
     await apiDeleteWorkflow(workflowId);
@@ -419,7 +419,7 @@ test.describe('select data', () => {
     const jobs = getWorkflowNodeExecutionsObj[0].jobs;
     const sqlNodeJob = jobs.find((job) => job.nodeId.toString() === sqlNodeId);
     const sqlNodeJobResult = sqlNodeJob.result;
-    const nodeResultRecordLength = sqlNodeJobResult[0].length;
+    const nodeResultRecordLength = sqlNodeJobResult.length;
     expect(nodeResultRecordLength).toBe(7);
     // 4、后置处理：删除工作流
     await apiDeleteWorkflow(workflowId);
@@ -524,7 +524,7 @@ test.describe('select data', () => {
     const jobs = getWorkflowNodeExecutionsObj[0].jobs;
     const sqlNodeJob = jobs.find((job) => job.nodeId.toString() === sqlNodeId);
     const sqlNodeJobResult = sqlNodeJob.result;
-    const nodeResultRecordOrgname = sqlNodeJobResult[0][0].orgname;
+    const nodeResultRecordOrgname = sqlNodeJobResult[0].orgname;
     expect(nodeResultRecordOrgname).toBe(triggerNodeCollectionRecordOne);
     // 4、后置处理：删除工作流
     await apiDeleteWorkflow(workflowId);
@@ -629,7 +629,7 @@ test.describe('select data', () => {
     const jobs = getWorkflowNodeExecutionsObj[0].jobs;
     const sqlNodeJob = jobs.find((job) => job.nodeId.toString() === sqlNodeId);
     const sqlNodeJobResult = sqlNodeJob.result;
-    const nodeResultRecordLength = sqlNodeJobResult[0].length;
+    const nodeResultRecordLength = sqlNodeJobResult.length;
     expect(nodeResultRecordLength).toBe(8);
     // 4、后置处理：删除工作流
     await apiDeleteWorkflow(workflowId);
@@ -713,10 +713,10 @@ test.describe('insert data', () => {
     const jobs = getWorkflowNodeExecutionsObj[0].jobs;
     const sqlNodeJob = jobs.find((job) => job.nodeId.toString() === sqlNodeId);
     const sqlNodeJobResult = sqlNodeJob.result;
-    const nodeResultRecordOrgname = sqlNodeJobResult[0][0].orgname;
+    const nodeResultRecordOrgname = sqlNodeJobResult[0].orgname;
     expect(nodeResultRecordOrgname).toBe('公司名称(单行文本)1');
 
-    const insertRecordId = sqlNodeJobResult[0][0].id;
+    const insertRecordId = sqlNodeJobResult[0].id;
     const getRecords = await apiGetRecord(SQLNodeCollectionName, insertRecordId);
     const getRecordsObj = JSON.parse(JSON.stringify(getRecords));
     expect(getRecordsObj.orgname).toBe('公司名称(单行文本)1');
@@ -801,7 +801,7 @@ test.describe('insert data', () => {
     const jobs = getWorkflowNodeExecutionsObj[0].jobs;
     const sqlNodeJob = jobs.find((job) => job.nodeId.toString() === sqlNodeId);
     const sqlNodeJobResult = sqlNodeJob.result;
-    const nodeResultRecordLength = sqlNodeJobResult[0].length;
+    const nodeResultRecordLength = sqlNodeJobResult.length;
     expect(nodeResultRecordLength).toBe(3);
     // 4、后置处理：删除工作流
     await apiDeleteWorkflow(workflowId);
@@ -903,7 +903,7 @@ test.describe('insert data', () => {
     const jobs = getWorkflowNodeExecutionsObj[0].jobs;
     const sqlNodeJob = jobs.find((job) => job.nodeId.toString() === sqlNodeId);
     const sqlNodeJobResult = sqlNodeJob.result;
-    const nodeResultRecordOrgname = sqlNodeJobResult[0][0].orgname;
+    const nodeResultRecordOrgname = sqlNodeJobResult[0].orgname;
     expect(nodeResultRecordOrgname).toBe('公司名称(单行文本)1');
     // 4、后置处理：删除工作流
     await apiDeleteWorkflow(workflowId);
@@ -1001,7 +1001,7 @@ test.describe('update data', () => {
     const jobs = getWorkflowNodeExecutionsObj[0].jobs;
     const sqlNodeJob = jobs.find((job) => job.nodeId.toString() === sqlNodeId);
     const sqlNodeJobResult = sqlNodeJob.result;
-    const nodeResultRecordLength = sqlNodeJobResult[0].length;
+    const nodeResultRecordLength = sqlNodeJobResult.length;
     expect(nodeResultRecordLength).toBe(8);
 
     // 4、后置处理：删除工作流
@@ -1107,7 +1107,7 @@ test.describe('update data', () => {
     const jobs = getWorkflowNodeExecutionsObj[0].jobs;
     const sqlNodeJob = jobs.find((job) => job.nodeId.toString() === sqlNodeId);
     const sqlNodeJobResult = sqlNodeJob.result;
-    const nodeResultRecordOrgname = sqlNodeJobResult[0][0].orgname;
+    const nodeResultRecordOrgname = sqlNodeJobResult[0].orgname;
     expect(nodeResultRecordOrgname).toBe('orgname');
 
     // 4、后置处理：删除工作流
@@ -1206,7 +1206,7 @@ test.describe('delete data', () => {
     const jobs = getWorkflowNodeExecutionsObj[0].jobs;
     const sqlNodeJob = jobs.find((job) => job.nodeId.toString() === sqlNodeId);
     const sqlNodeJobResult = sqlNodeJob.result;
-    const nodeResultRecordOrgname = sqlNodeJobResult[0][0].orgname;
+    const nodeResultRecordOrgname = sqlNodeJobResult[0].orgname;
     expect(nodeResultRecordOrgname).toBe('公司名称(单行文本)1');
 
     // 4、后置处理：删除工作流
@@ -1312,7 +1312,7 @@ test.describe('delete data', () => {
     const jobs = getWorkflowNodeExecutionsObj[0].jobs;
     const sqlNodeJob = jobs.find((job) => job.nodeId.toString() === sqlNodeId);
     const sqlNodeJobResult = sqlNodeJob.result;
-    const nodeResultRecordOrgname = sqlNodeJobResult[0][0].orgname;
+    const nodeResultRecordOrgname = sqlNodeJobResult[0].orgname;
     expect(nodeResultRecordOrgname).toBe('公司名称(单行文本)1');
 
     // 4、后置处理：删除工作流

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

@@ -2,5 +2,6 @@
   "SQL action": "SQL 操作",
   "Execute a SQL statement in database.": "在数据库中执行一个 SQL 语句",
   "Select a data source to execute SQL.": "选择一个数据源来执行 SQL",
-  "SQL query result could be used through <1>JSON query node</1> (Commercial plugin).": "SQL 执行的结果可在 <1>JSON 解析节点</1> 中使用（商业插件）。"
+  "SQL query result could be used through <1>JSON query node</1> (Commercial plugin).": "SQL 执行的结果可在 <1>JSON 解析节点</1> 中使用（商业插件）。",
+  "Include meta information of this query in result": "在结果中包含此查询的元信息"
 }

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

@@ -11,10 +11,9 @@ import { Processor, Instruction, JOB_STATUS, FlowNodeModel } from '@nocobase/plu
 
 export default class extends Instruction {
   async run(node: FlowNodeModel, input, processor: Processor) {
+    const dataSourceName = node.config.dataSource || 'main';
     // @ts-ignore
-    const { db } = this.workflow.app.dataSourceManager.dataSources.get(
+    const { db } = this.workflow.app.dataSourceManager.dataSources.get(dataSourceName).collectionManager;
-      node.config.dataSource || 'main',
-    ).collectionManager;
     if (!db) {
       throw new Error(`type of data source "${node.config.dataSource}" is not database`);
     }
@@ -26,14 +25,14 @@ export default class extends Instruction {
     }
 
     // @ts-ignore
-    const result = await db.sequelize.query(sql, {
+    const [result, meta] = await db.sequelize.query(sql, {
-      transaction: processor.transaction,
+      transaction: this.workflow.useDataSourceTransaction(dataSourceName, processor.transaction),
       // plain: true,
       // model: db.getCollection(node.config.collection).model
     });
 
     return {
-      result,
+      result: node.config.withMeta ? [result, meta] : result,
       status: JOB_STATUS.RESOLVED,
     };
   }

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

@@ -117,8 +117,6 @@ describe('workflow > instructions > sql', () => {
       const [execution] = await workflow.getExecutions();
       const [sqlJob] = await execution.getJobs({ order: [['id', 'ASC']] });
       expect(sqlJob.status).toBe(JOB_STATUS.RESOLVED);
-      // expect(queryJob.status).toBe(JOB_STATUS.RESOLVED);
-      // expect(queryJob.result.read).toBe(post.id);
     });
 
     it('update', async () => {
@@ -208,20 +206,56 @@ describe('workflow > instructions > sql', () => {
         },
       });
 
+      const CategoryCollection = db.getCollection('categories');
+
       const n1 = await w2.createNode({
         type: 'sql',
         config: {
-          sql: `select count(id) from ${PostCollection.quotedTableName()}`,
+          sql: `select count(id) as count from ${CategoryCollection.quotedTableName()}`,
         },
       });
 
-      const CategoryRepo = db.getCollection('categories').repository;
+      const category = await CategoryCollection.repository.create({ values: { title: 't1' } });
-      const category = await CategoryRepo.create({ values: { title: 't1' } });
 
       const [execution] = await w2.getExecutions();
       expect(execution.status).toBe(EXECUTION_STATUS.RESOLVED);
       const [job] = await execution.getJobs();
       expect(job.status).toBe(JOB_STATUS.RESOLVED);
+      expect(job.result.length).toBe(1);
+      expect(job.result[0].count).toBe(1);
+    });
+  });
+
+  describe('legacy', () => {
+    it('withMeta', async () => {
+      const w2 = await WorkflowModel.create({
+        enabled: true,
+        sync: true,
+        type: 'collection',
+        config: {
+          mode: 1,
+          collection: 'categories',
+        },
+      });
+
+      const CategoryCollection = db.getCollection('categories');
+
+      const n1 = await w2.createNode({
+        type: 'sql',
+        config: {
+          sql: `select count(id) as count from ${CategoryCollection.quotedTableName()}`,
+          withMeta: true,
+        },
+      });
+
+      const category = await CategoryCollection.repository.create({ values: { title: 't1' } });
+
+      const [execution] = await w2.getExecutions();
+      const [sqlJob] = await execution.getJobs({ order: [['id', 'ASC']] });
+      expect(sqlJob.status).toBe(JOB_STATUS.RESOLVED);
+      expect(sqlJob.result.length).toBe(2);
+      expect(sqlJob.result[0].length).toBe(1);
+      expect(sqlJob.result[0][0].count).toBe(1);
     });
   });
 
@@ -249,10 +283,9 @@ describe('workflow > instructions > sql', () => {
       const [execution] = await workflow.getExecutions();
       expect(execution.status).toBe(EXECUTION_STATUS.RESOLVED);
       const [job] = await execution.getJobs();
-      expect(job.result.length).toBe(2);
+      expect(job.result.length).toBe(1);
-      expect(job.result[0].length).toBe(1);
       // @ts-ignore
-      expect(job.result[0][0].id).toBe(post.id);
+      expect(job.result[0].id).toBe(post.id);
     });
   });
 });

packages/plugins/@nocobase/plugin-workflow-sql/src/server/migrations/20240902174913-add-raw-result-mode.ts

@@ -0,0 +1,40 @@
+/**
+ * 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 { Migration } from '@nocobase/server';
+
+export default class extends Migration {
+  appVersion = '<1.3.16-beta';
+  async up() {
+    const { db } = this.context;
+
+    const NodeRepo = db.getRepository('flow_nodes');
+    await db.sequelize.transaction(async (transaction) => {
+      const nodes = await NodeRepo.find({
+        filter: {
+          type: 'sql',
+        },
+        transaction,
+      });
+
+      await nodes.reduce(
+        (promise, node) =>
+          promise.then(() => {
+            node.set('config', { ...node.config, withMeta: true });
+            node.changed('config', true);
+            return node.save({
+              silent: true,
+              transaction,
+            });
+          }),
+        Promise.resolve(),
+      );
+    });
+  }
+}