update mq export;perf: redis cache (#4465)

* perf: redis cache

* update mq export

* perf: website sync error tip

* add error worker

Author: c121914yu

packages/global/core/dataset/constants.ts

@@ -51,7 +51,8 @@ export const DatasetTypeMap = {
 export enum DatasetStatusEnum {
   active = 'active',
   syncing = 'syncing',
-  waiting = 'waiting'
+  waiting = 'waiting',
+  error = 'error'
 }
 export const DatasetStatusMap = {
   [DatasetStatusEnum.active]: {
@@ -62,6 +63,9 @@ export const DatasetStatusMap = {
   },
   [DatasetStatusEnum.waiting]: {
     label: i18nT('common:core.dataset.status.waiting')
+  },
+  [DatasetStatusEnum.error]: {
+    label: i18nT('dataset:status_error')
   }
 };
 

packages/global/core/dataset/type.d.ts

@@ -209,6 +209,7 @@ export type DatasetListItemType = {
 
 export type DatasetItemType = Omit<DatasetSchemaType, 'vectorModel' | 'agentModel' | 'vlmModel'> & {
   status: `${DatasetStatusEnum}`;
+  errorMsg?: string;
   vectorModel: EmbeddingModelItemType;
   agentModel: LLMModelItemType;
   vlmModel?: LLMModelItemType;

packages/service/common/bullmq/index.ts

@@ -69,6 +69,11 @@ export function getWorker<DataType, ReturnType = void>(
   newWorker.on('error', (error) => {
     addLog.error(`MQ Worker [${name}]: ${error.message}`, error);
   });
+  newWorker.on('failed', (jobId, error) => {
+    addLog.error(`MQ Worker [${name}]: ${error.message}`, error);
+  });
   workers.set(name, newWorker);
   return newWorker;
 }
+
+export * from 'bullmq';

packages/service/common/vectorStore/controller.ts

@@ -7,6 +7,7 @@ import { EmbeddingModelItemType } from '@fastgpt/global/core/ai/model.d';
 import { MILVUS_ADDRESS, PG_ADDRESS, OCEANBASE_ADDRESS } from './constants';
 import { MilvusCtrl } from './milvus/class';
 import { setRedisCache, getRedisCache, delRedisCache, CacheKeyEnum } from '../redis/cache';
+import { throttle } from 'lodash';
 
 const getVectorObj = () => {
   if (PG_ADDRESS) return new PgVectorCtrl();
@@ -15,7 +16,12 @@ const getVectorObj = () => {
 
   return new PgVectorCtrl();
 };
+
 const getChcheKey = (teamId: string) => `${CacheKeyEnum.team_vector_count}:${teamId}`;
+const onDelCache = throttle((teamId: string) => delRedisCache(getChcheKey(teamId)), 30000, {
+  leading: true,
+  trailing: true
+});
 
 const Vector = getVectorObj();
 
@@ -59,7 +65,7 @@ export const insertDatasetDataVector = async ({
     vector: vectors[0]
   });
 
-  delRedisCache(getChcheKey(props.teamId));
+  onDelCache(props.teamId);
 
   return {
     tokens,
@@ -69,6 +75,6 @@ export const insertDatasetDataVector = async ({
 
 export const deleteDatasetDataVector = async (props: DelDatasetVectorCtrlProps) => {
   const result = await Vector.delete(props);
-  delRedisCache(getChcheKey(props.teamId));
+  onDelCache(props.teamId);
   return result;
 };

packages/service/core/dataset/websiteSync/index.ts

@@ -34,23 +34,44 @@ export const addWebsiteSyncJob = (data: WebsiteSyncJobData) => {
 export const getWebsiteSyncDatasetStatus = async (datasetId: string) => {
   const jobId = await websiteSyncQueue.getDeduplicationJobId(datasetId);
   if (!jobId) {
-    return DatasetStatusEnum.active;
+    return {
+      status: DatasetStatusEnum.active,
+      errorMsg: undefined
+    };
   }
   const job = await websiteSyncQueue.getJob(jobId);
   if (!job) {
-    return DatasetStatusEnum.active;
+    return {
+      status: DatasetStatusEnum.active,
+      errorMsg: undefined
+    };
   }
 
   const jobState = await job.getState();
 
+  if (jobState === 'failed' || jobState === 'unknown') {
+    return {
+      status: DatasetStatusEnum.error,
+      errorMsg: job.failedReason
+    };
+  }
   if (['waiting-children', 'waiting'].includes(jobState)) {
-    return DatasetStatusEnum.waiting;
+    return {
+      status: DatasetStatusEnum.waiting,
+      errorMsg: undefined
+    };
   }
   if (jobState === 'active') {
-    return DatasetStatusEnum.syncing;
+    return {
+      status: DatasetStatusEnum.syncing,
+      errorMsg: undefined
+    };
   }
 
-  return DatasetStatusEnum.active;
+  return {
+    status: DatasetStatusEnum.active,
+    errorMsg: undefined
+  };
 };
 
 // Scheduler setting

packages/web/i18n/en/dataset.json

@@ -124,6 +124,7 @@
   "split_sign_question": "question mark",
   "split_sign_semicolon": "semicolon",
   "start_sync_website_tip": "Confirm to start synchronizing data? \nThe old data will be deleted and retrieved again, please confirm!",
+  "status_error": "Running exception",
   "sync_collection_failed": "Synchronization collection error, please check whether the source file can be accessed normally",
   "sync_schedule": "Timing synchronization",
   "sync_schedule_tip": "Only existing collections will be synchronized. \nIncludes linked collections and all collections in the API knowledge base. \nThe system will poll for updates every day, and the specific update time cannot be determined.",

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

@@ -125,6 +125,7 @@
   "split_sign_question": "问号",
   "split_sign_semicolon": "分号",
   "start_sync_website_tip": "确认开始同步数据？将会删除旧数据后重新获取，请确认！",
+  "status_error": "运行异常",
   "sync_collection_failed": "同步集合错误，请检查是否能正常访问源文件",
   "sync_schedule": "定时同步",
   "sync_schedule_tip": "仅会同步已存在的集合。包括链接集合以及 API 知识库里所有集合。系统会每天进行轮询更新，无法确定具体的更新时间。",

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

@@ -124,6 +124,7 @@
   "split_sign_question": "問號",
   "split_sign_semicolon": "分號",
   "start_sync_website_tip": "確認開始同步資料？\n將會刪除舊資料後重新獲取，請確認！",
+  "status_error": "運行異常",
   "sync_collection_failed": "同步集合錯誤，請檢查是否能正常存取來源文件",
   "sync_schedule": "定時同步",
   "sync_schedule_tip": "只會同步已存在的集合。\n包括連結集合以及 API 知識庫裡所有集合。\n系統會每天進行輪詢更新，無法確定特定的更新時間。",

projects/app/next.config.js

@@ -83,6 +83,7 @@ const nextConfig = {
     serverComponentsExternalPackages: [
       'mongoose',
       'pg',
+      'bullmq',
       '@zilliz/milvus2-sdk-node',
       "tiktoken",
     ],

projects/app/src/pageComponents/dataset/detail/CollectionCard/Header.tsx

@@ -33,6 +33,7 @@ import HeaderTagPopOver from './HeaderTagPopOver';
 import MyBox from '@fastgpt/web/components/common/MyBox';
 import Icon from '@fastgpt/web/components/common/Icon';
 import MyTag from '@fastgpt/web/components/common/Tag/index';
+import QuestionTip from '@fastgpt/web/components/common/MyTooltip/QuestionTip';
 
 const FileSourceSelector = dynamic(() => import('../Import/components/FileSourceSelector'));
 
@@ -321,6 +322,14 @@ const Header = ({}: {}) => {
                       {t('common:core.dataset.status.waiting')}
                     </MyTag>
                   )}
+                  {datasetDetail.status === DatasetStatusEnum.error && (
+                    <MyTag colorSchema="red" showDot px={3} h={'36px'}>
+                      <HStack spacing={1}>
+                        <Box>{t('dataset:status_error')}</Box>
+                        <QuestionTip color={'red.500'} label={datasetDetail.errorMsg} />
+                      </HStack>
+                    </MyTag>
+                  )}
                 </>
               ) : (
                 <Button