perf: app cron job (#3360)

docSite/content/zh-cn/docs/development/upgrading/4815.md

@@ -28,6 +28,19 @@ weight: 809
 - Sandbox 镜像，可以不更新
 
 
+## 运行初始化脚本
+
+从任意终端，发起 1 个 HTTP 请求。其中 {{rootkey}} 替换成环境变量里的 `rootkey`；{{host}} 替换成**FastGPT 域名**。
+
+```bash
+curl --location --request POST 'https://{{host}}/admin/initv4815' \
+--header 'rootkey: {{rootkey}}' \
+--header 'Content-Type: application/json'
+```
+
+会重置应用定时执行的字段，把 null 去掉，减少索引大小。
+
+
 ## 完整更新内容
 
 1. 新增 - API 知识库, 见 [API 知识库介绍](/docs/guide/knowledge_base/api_dataset/)，外部文件库会被弃用。
@@ -42,9 +55,10 @@ weight: 809
 10. 优化 - 字符串变量替换，未赋值的变量会转成 undefined，而不是保留原来 id 串。
 11. 优化 - 全局变量默认值在 API 生效，并且自定义变量支持默认值。
 12. 优化 - 增加 HTTP Body 的 JSON 解析，正则将 undefined 转 null，减少 Body 解析错误。
-13. 修复 - 分享链接点赞鉴权问题。
+13. 优化 - 定时执行增加运行日志，增加重试，减少报错概率。
-14. 修复 - 对话页面切换自动执行应用时，会误触发非自动执行应用。
+14. 修复 - 分享链接点赞鉴权问题。
-15. 修复 - 语言播放鉴权问题。
+15. 修复 - 对话页面切换自动执行应用时，会误触发非自动执行应用。
-16. 修复 - 插件应用知识库引用上限始终为 3000
+16. 修复 - 语言播放鉴权问题。
-17. 修复 - 工作流编辑记录存储上限，去掉本地存储，增加异常离开时，强制自动保存。
+17. 修复 - 插件应用知识库引用上限始终为 3000
-18. 修复 - 工作流特殊变量替换问题。（$开头的字符串无法替换）
+18. 修复 - 工作流编辑记录存储上限，去掉本地存储，增加异常离开时，强制自动保存。
+19. 修复 - 工作流特殊变量替换问题。（$开头的字符串无法替换）

packages/global/core/chat/constants.ts

@@ -33,6 +33,7 @@ export enum ChatSourceEnum {
   online = 'online',
   share = 'share',
   api = 'api',
+  cronJob = 'cronJob',
   team = 'team',
   feishu = 'feishu',
   official_account = 'official_account',
@@ -52,6 +53,9 @@ export const ChatSourceMap = {
   [ChatSourceEnum.api]: {
     name: i18nT('common:core.chat.logs.api')
   },
+  [ChatSourceEnum.cronJob]: {
+    name: i18nT('chat:source_cronJob')
+  },
   [ChatSourceEnum.team]: {
     name: i18nT('common:core.chat.logs.team')
   },

packages/service/core/app/schema.ts

@@ -121,6 +121,13 @@ const AppSchema = new Schema({
 
 AppSchema.index({ teamId: 1, updateTime: -1 });
 AppSchema.index({ teamId: 1, type: 1 });
-AppSchema.index({ scheduledTriggerConfig: 1, scheduledTriggerNextTime: -1 });
+AppSchema.index(
+  { scheduledTriggerConfig: 1, scheduledTriggerNextTime: -1 },
+  {
+    partialFilterExpression: {
+      scheduledTriggerConfig: { $exists: true }
+    }
+  }
+);
 
 export const MongoApp = getMongoModel<AppType>(AppCollectionName, AppSchema);

packages/web/i18n/en/chat.json

@@ -43,6 +43,7 @@
   "select_file": "Upload File",
   "select_file_img": "Upload file / image",
   "select_img": "Upload Image",
+  "source_cronJob": "Scheduled execution",
   "stream_output": "Stream Output",
   "unsupported_file_type": "Unsupported file types",
   "upload": "Upload",

packages/web/i18n/zh-CN/chat.json

@@ -43,6 +43,7 @@
   "select_file": "上传文件",
   "select_file_img": "上传文件/图片",
   "select_img": "上传图片",
+  "source_cronJob": "定时执行",
   "stream_output": "流输出",
   "unsupported_file_type": "不支持的文件类型",
   "upload": "上传",

packages/web/i18n/zh-Hant/chat.json

@@ -43,6 +43,7 @@
   "select_file": "上傳檔案",
   "select_file_img": "上傳檔案 / 圖片",
   "select_img": "上傳圖片",
+  "source_cronJob": "定時執行",
   "stream_output": "串流輸出",
   "unsupported_file_type": "不支援的檔案類型",
   "upload": "上傳",

projects/app/src/pages/api/admin/initv4815.ts

@@ -0,0 +1,27 @@
+import type { NextApiRequest, NextApiResponse } from 'next';
+import { authCert } from '@fastgpt/service/support/permission/auth/common';
+import { NextAPI } from '@/service/middleware/entry';
+import { MongoApp } from '@fastgpt/service/core/app/schema';
+
+/* 初始化发布的版本 */
+async function handler(req: NextApiRequest, res: NextApiResponse) {
+  await authCert({ req, authRoot: true });
+
+  // scheduledTriggerConfig为 null 的，都转成 unExist
+  return MongoApp.updateMany(
+    {
+      $or: [
+        { scheduledTriggerConfig: { $eq: null } },
+        { 'scheduledTriggerConfig.cronString': { $eq: '' } }
+      ]
+    },
+    {
+      $unset: {
+        scheduledTriggerConfig: '',
+        scheduledTriggerNextTime: ''
+      }
+    }
+  );
+}
+
+export default NextAPI(handler);

projects/app/src/pages/api/core/app/version/publish.ts

@@ -10,6 +10,7 @@ import { PostPublishAppProps } from '@/global/core/app/api';
 import { WritePermissionVal } from '@fastgpt/global/support/permission/constant';
 import { ApiRequestProps } from '@fastgpt/service/type/next';
 import { AppTypeEnum } from '@fastgpt/global/core/app/constants';
+import { getScheduleTriggerApp } from '@/service/core/app/utils';
 
 async function handler(
   req: ApiRequestProps<PostPublishAppProps>,
@@ -52,12 +53,17 @@ async function handler(
         updateTime: new Date(),
         version: 'v2',
         // 只有发布才会更新定时器
-        ...(isPublish && {
+        ...(isPublish &&
-          scheduledTriggerConfig: chatConfig?.scheduledTriggerConfig,
+          (chatConfig?.scheduledTriggerConfig?.cronString
-          scheduledTriggerNextTime: chatConfig?.scheduledTriggerConfig?.cronString
+            ? {
-            ? getNextTimeByCronStringAndTimezone(chatConfig.scheduledTriggerConfig)
+                $set: {
-            : null
+                  scheduledTriggerConfig: chatConfig.scheduledTriggerConfig,
-        }),
+                  scheduledTriggerNextTime: getNextTimeByCronStringAndTimezone(
+                    chatConfig.scheduledTriggerConfig
+                  )
+                }
+              }
+            : { $unset: { scheduledTriggerConfig: '', scheduledTriggerNextTime: '' } })),
         'pluginData.nodeVersion': _id
       },
       {
@@ -66,6 +72,8 @@ async function handler(
     );
   });
 
+  await getScheduleTriggerApp();
+
   return {};
 }
 

projects/app/src/service/core/app/utils.ts

@@ -1,10 +1,13 @@
 import { getUserChatInfoAndAuthTeamPoints } from '@/service/support/permission/auth/team';
 import { pushChatUsage } from '@/service/support/wallet/usage/push';
-import { defaultApp } from '@/web/core/app/constants';
 import { getNextTimeByCronStringAndTimezone } from '@fastgpt/global/common/string/time';
 import { getNanoid } from '@fastgpt/global/common/string/tools';
 import { delay, retryFn } from '@fastgpt/global/common/system/utils';
-import { ChatItemValueTypeEnum } from '@fastgpt/global/core/chat/constants';
+import {
+  ChatItemValueTypeEnum,
+  ChatRoleEnum,
+  ChatSourceEnum
+} from '@fastgpt/global/core/chat/constants';
 import {
   getWorkflowEntryNodeIds,
   initWorkflowEdgeStatus,
@@ -15,12 +18,16 @@ import { addLog } from '@fastgpt/service/common/system/log';
 import { MongoApp } from '@fastgpt/service/core/app/schema';
 import { WORKFLOW_MAX_RUN_TIMES } from '@fastgpt/service/core/workflow/constants';
 import { dispatchWorkFlow } from '@fastgpt/service/core/workflow/dispatch';
+import { DispatchNodeResponseKeyEnum } from '@fastgpt/global/core/workflow/runtime/constants';
+import { UserChatItemValueItemType } from '@fastgpt/global/core/chat/type';
+import { saveChat } from '@fastgpt/service/core/chat/saveChat';
+import { getAppLatestVersion } from '@fastgpt/service/core/app/version/controller';
 
 export const getScheduleTriggerApp = async () => {
   // 1. Find all the app
   const apps = await retryFn(() => {
     return MongoApp.find({
-      scheduledTriggerConfig: { $ne: null },
+      scheduledTriggerConfig: { $exists: true },
       scheduledTriggerNextTime: { $lte: new Date() }
     });
   });
@@ -34,11 +41,22 @@ export const getScheduleTriggerApp = async () => {
         await delay(Math.floor(Math.random() * 60 * 1000));
         const { user } = await getUserChatInfoAndAuthTeamPoints(app.tmbId);
 
-        await retryFn(async () => {
+        // Get app latest version
-          if (!app.scheduledTriggerConfig) return;
+        const { nodes, edges, chatConfig } = await getAppLatestVersion(app._id, app);
 
-          const { flowUsages } = await dispatchWorkFlow({
+        const chatId = getNanoid();
-            chatId: getNanoid(),
+        const userQuery: UserChatItemValueItemType[] = [
+          {
+            type: ChatItemValueTypeEnum.text,
+            text: {
+              content: app.scheduledTriggerConfig?.defaultPrompt
+            }
+          }
+        ];
+
+        const { flowUsages, assistantResponses, flowResponses } = await retryFn(() => {
+          return dispatchWorkFlow({
+            chatId,
             user,
             mode: 'chat',
             runningAppInfo: {
@@ -47,33 +65,48 @@ export const getScheduleTriggerApp = async () => {
               tmbId: String(app.tmbId)
             },
             uid: String(app.tmbId),
-            runtimeNodes: storeNodes2RuntimeNodes(
+            runtimeNodes: storeNodes2RuntimeNodes(nodes, getWorkflowEntryNodeIds(nodes)),
-              app.modules,
+            runtimeEdges: initWorkflowEdgeStatus(edges),
-              getWorkflowEntryNodeIds(app.modules)
-            ),
-            runtimeEdges: initWorkflowEdgeStatus(app.edges),
             variables: {},
-            query: [
+            query: userQuery,
-              {
+            chatConfig,
-                type: ChatItemValueTypeEnum.text,
-                text: {
-                  content: app.scheduledTriggerConfig?.defaultPrompt
-                }
-              }
-            ],
-            chatConfig: defaultApp.chatConfig,
             histories: [],
             stream: false,
             maxRunTimes: WORKFLOW_MAX_RUN_TIMES
           });
-          pushChatUsage({
+        });
-            appName: app.name,
+
-            appId: app._id,
+        // Save chat
-            teamId: String(app.teamId),
+        await saveChat({
-            tmbId: String(app.tmbId),
+          chatId,
-            source: UsageSourceEnum.cronJob,
+          appId: app._id,
-            flowUsages
+          teamId: String(app.teamId),
-          });
+          tmbId: String(app.tmbId),
+          nodes,
+          appChatConfig: chatConfig,
+          variables: {},
+          isUpdateUseTime: false, // owner update use time
+          newTitle: 'Cron Job',
+          source: ChatSourceEnum.cronJob,
+          content: [
+            {
+              obj: ChatRoleEnum.Human,
+              value: userQuery
+            },
+            {
+              obj: ChatRoleEnum.AI,
+              value: assistantResponses,
+              [DispatchNodeResponseKeyEnum.nodeResponse]: flowResponses
+            }
+          ]
+        });
+        pushChatUsage({
+          appName: app.name,
+          appId: app._id,
+          teamId: String(app.teamId),
+          tmbId: String(app.tmbId),
+          source: UsageSourceEnum.cronJob,
+          flowUsages
         });
 
         // update next time