wechat publish (#6607)

* wechat publish * update test * doc
2026-05-11 01:02:01 +08:00 · 2026-03-23 11:57:05 +08:00
parent c84c45398a
commit c37b3aa0e8
56 changed files with 2303 additions and 139 deletions
@@ -0,0 +1,119 @@
+import { authFrequencyLimit } from '../../../common/system/frequencyLimit/utils';
+import type {
+  AuthOutLinkInitProps,
+  AuthOutLinkLimitProps,
+  AuthOutLinkResponse
+} from '@fastgpt/global/support/outLink/api';
+import { axios } from '../../../common/api/axios';
+import { OutLinkErrEnum } from '@fastgpt/global/common/error/code/outLink';
+import type { OutLinkSchema } from '@fastgpt/global/support/outLink/type';
+import { addMinutes } from 'date-fns';
+import { S3_KEY_PATH_INVALID_CHARS } from '../../../common/s3/constants';
+import { UserError } from '@fastgpt/global/common/error/utils';
+
+export type TokenAuthResponseType = {
+  success: boolean;
+  msg?: string;
+  message?: string;
+  data?: AuthOutLinkResponse;
+};
+
+export const authOutLinkInit = async ({
+  tokenUrl,
+  outLinkUid
+}: AuthOutLinkInitProps): Promise<AuthOutLinkResponse> => {
+  if (!tokenUrl) return { uid: outLinkUid };
+
+  const { data } = await axios<TokenAuthResponseType>({
+    baseURL: tokenUrl,
+    url: '/shareAuth/init',
+    method: 'POST',
+    data: {
+      token: outLinkUid
+    }
+  });
+  if (data?.success !== true) {
+    return Promise.reject(data?.message || data?.msg || OutLinkErrEnum.unAuthUser);
+  }
+
+  const uid = data?.data?.uid;
+  if (
+    !uid ||
+    typeof uid !== 'string' ||
+    Buffer.byteLength(uid) > 255 ||
+    S3_KEY_PATH_INVALID_CHARS.test(uid)
+  ) {
+    return Promise.reject(new UserError('Invalid UID'));
+  }
+
+  return { uid };
+};
+
+const authIpLimit = async ({ ip, outLink }: { ip: string; outLink: OutLinkSchema }) => {
+  if (!outLink.limit || !outLink.limit.QPM) {
+    return;
+  }
+
+  try {
+    await authFrequencyLimit({
+      eventId: `${outLink._id}-${ip}`,
+      maxAmount: outLink.limit.QPM,
+      expiredTime: addMinutes(new Date(), 1)
+    });
+  } catch (error) {
+    return Promise.reject(new UserError(`每分钟仅能请求 ${outLink.limit.QPM} 次~`));
+  }
+};
+
+export async function authOutLinkLimit({
+  outLink,
+  ip,
+  outLinkUid,
+  question
+}: AuthOutLinkLimitProps): Promise<AuthOutLinkResponse> {
+  if (!outLink.limit) {
+    return { uid: outLinkUid };
+  }
+
+  //   expiredTime already to string
+  if (outLink.limit.expiredTime && new Date(outLink.limit.expiredTime).getTime() < Date.now()) {
+    return Promise.reject(new UserError('分享链接已过期'));
+  }
+
+  if (
+    outLink.limit.maxUsagePoints &&
+    outLink.limit.maxUsagePoints > -1 &&
+    outLink.usagePoints > outLink.limit.maxUsagePoints
+  ) {
+    return Promise.reject(new UserError('链接超出使用限制'));
+  }
+
+  // ip limit
+  if (ip) {
+    await authIpLimit({ ip, outLink });
+  }
+
+  // url auth. send request
+  if (!outLink.limit.hookUrl) {
+    return { uid: outLinkUid };
+  }
+  try {
+    const { data } = await axios<TokenAuthResponseType>({
+      baseURL: outLink.limit.hookUrl,
+      url: '/shareAuth/start',
+      method: 'POST',
+      data: {
+        token: outLinkUid,
+        question
+      }
+    });
+
+    if (data?.success !== true) {
+      return Promise.reject(new UserError(data?.message || data?.msg || '身份校验失败'));
+    }
+
+    return { uid: data?.data?.uid || outLinkUid };
+  } catch (error) {
+    return Promise.reject(new UserError('身份校验失败'));
+  }
+}
@@ -0,0 +1,327 @@
+import { ChatRoleEnum } from '@fastgpt/global/core/chat/constants';
+import type { UserChatItemValueItemType } from '@fastgpt/global/core/chat/type';
+import { DispatchNodeResponseKeyEnum } from '@fastgpt/global/core/workflow/runtime/constants';
+import {
+  getWorkflowEntryNodeIds,
+  getMaxHistoryLimitFromNodes,
+  storeEdges2RuntimeEdges,
+  storeNodes2RuntimeNodes
+} from '@fastgpt/global/core/workflow/runtime/utils';
+import type { OutlinkAppType, OutLinkSchema } from '@fastgpt/global/support/outLink/type';
+import { getAppLatestVersion } from '../../../core/app/version/controller';
+import { MongoApp } from '../../../core/app/schema';
+import { getChatItems } from '../../../core/chat/controller';
+import { pushChatRecords } from '../../../core/chat/saveChat';
+import { dispatchWorkFlow } from '../../../core/workflow/dispatch';
+import { getUserChatInfo } from '../../../support/user/team/utils';
+import { getRunningUserInfoByTmbId } from '../../../support/user/team/utils';
+import { SseResponseEventEnum } from '@fastgpt/global/core/workflow/runtime/constants';
+import type { NextApiResponse } from 'next';
+import { authOutLinkLimit } from './auth';
+import { addOutLinkUsage } from '../../../support/outLink/tools';
+import { getLogger, LogCategories } from '../../../common/logger';
+import { appendRedisCache } from '../../../common/redis/cache';
+import { getErrText } from '@fastgpt/global/common/error/utils';
+import { getUsageSourceByPublishChannel } from '@fastgpt/global/support/wallet/usage/tools';
+import { getChatSourceByPublishChannel } from '@fastgpt/global/core/chat/utils';
+import { WORKFLOW_MAX_RUN_TIMES } from '../../../core/workflow/constants';
+import { mongoSessionRun } from '../../../common/mongo/sessionRun';
+import { MongoChat } from '../../../core/chat/chatSchema';
+import { getNanoid } from '@fastgpt/global/common/string/tools';
+import { MongoChatItem } from '../../../core/chat/chatItemSchema';
+
+const logger = getLogger(LogCategories.MODULE.OUTLINK);
+
+// 新开历史记录, 把原来 chatId 替换
+const RESET_CHAT_INPUT: Record<string, boolean> = {
+  Reset: true,
+  '/reset': true
+};
+const RESET_CHAT_REPLY = '对话已重置。\n\n The chat records have been reset.';
+export const resetChat = ({ appId, chatId }: { appId: string; chatId: string }) => {
+  const newChatId = getNanoid(26);
+  return mongoSessionRun(async (session) => {
+    await MongoChat.updateOne(
+      {
+        appId,
+        chatId
+      },
+      {
+        $set: {
+          chatId: newChatId
+        }
+      },
+      { session }
+    );
+    await MongoChatItem.updateMany(
+      {
+        appId,
+        chatId
+      },
+      {
+        $set: {
+          chatId: newChatId
+        }
+      },
+      { session }
+    );
+  });
+};
+
+export type outLinkInvokeChatProps<T extends OutlinkAppType> = {
+  outLinkConfig: OutLinkSchema<T>;
+  chatId: string; // specific chat
+  query: UserChatItemValueItemType[];
+  res?: NextApiResponse;
+  messageId: string;
+  chatUserId: string;
+  replyCallback?: (replyContent: string) => Promise<any>;
+  streamId?: string;
+};
+
+const DEFAULT_REPLY = 'This is default reply';
+
+export const STREAM_END_FLAG = '[DONE]';
+export const STREAM_CACHE_KEY_PREFIX = 'streamResponse:';
+
+export async function outlinkInvokeChat<T extends OutlinkAppType>({
+  outLinkConfig,
+  chatId,
+  query,
+  res,
+  messageId,
+  chatUserId,
+  replyCallback,
+  streamId
+}: outLinkInvokeChatProps<T>) {
+  const streamResKey = `${STREAM_CACHE_KEY_PREFIX}${streamId}`;
+
+  try {
+    // Get app workflow config
+    const [app, { nodes, chatConfig, edges }, { timezone, externalProvider }] = await Promise.all([
+      MongoApp.findById(outLinkConfig.appId).lean(),
+      getAppLatestVersion(outLinkConfig.appId),
+      getUserChatInfo(outLinkConfig.tmbId)
+    ]);
+
+    if (!nodes || !chatConfig || !app) {
+      return Promise.reject('Invalid chat');
+    }
+
+    // Check whether the chatId is valid
+    const userQuestion = query.find((item) => item.text)?.text?.content || '';
+    if (RESET_CHAT_INPUT[userQuestion]) {
+      await resetChat({ appId: outLinkConfig.appId, chatId });
+      await replyCallback?.(RESET_CHAT_REPLY);
+      if (streamId) {
+        await appendRedisCache(streamResKey, RESET_CHAT_REPLY, 60);
+        await appendRedisCache(streamResKey, STREAM_END_FLAG, 60);
+      }
+      return;
+    }
+
+    // Load chat histories and global variables in parallel
+    const [{ histories }, chatDetail] = await Promise.all([
+      getChatItems({
+        appId: outLinkConfig.appId,
+        chatId,
+        offset: 0,
+        limit: getMaxHistoryLimitFromNodes(nodes),
+        field: `obj value`
+      }),
+      MongoChat.findOne({ appId: outLinkConfig.appId, chatId }, 'source variableList variables')
+    ]);
+
+    // dedupe
+    if (histories.find((item) => item.dataId === messageId)) {
+      return; // dupelicated messaage, do noting
+    }
+
+    await authOutLinkLimit({
+      outLinkUid: chatUserId,
+      outLink: outLinkConfig as any, // HACK, we do not need to provide app: T
+      question: userQuestion,
+      ip: chatId
+    });
+
+    const workflowStreamResponse = streamId
+      ? async ({
+          write,
+          event,
+          data
+        }: {
+          write?: (text: string) => void;
+          event: SseResponseEventEnum;
+          data: Record<string, any>;
+        }) => {
+          if (event === SseResponseEventEnum.answer || event === SseResponseEventEnum.fastAnswer) {
+            try {
+              const text = data.choices?.[0]?.delta?.content;
+              if (text) {
+                await appendRedisCache(streamResKey, text, 60);
+              }
+            } catch (error) {
+              logger.error('Outlink real-time streaming failed', {
+                streamId,
+                messageId,
+                error
+              });
+            }
+          }
+        }
+      : undefined;
+
+    // Append first
+    await appendRedisCache(streamResKey, '', 120);
+
+    // Merge global variables from database
+    let variables = {};
+    if (chatDetail?.variables) {
+      variables = {
+        ...chatDetail.variables
+      };
+    }
+
+    const {
+      assistantResponses,
+      newVariables,
+      flowResponses,
+      flowUsages,
+      durationSeconds,
+      system_memories
+    } = await dispatchWorkFlow({
+      apiVersion: 'v2',
+      res,
+      mode: 'chat',
+      usageSource: getUsageSourceByPublishChannel(outLinkConfig.type),
+      runningAppInfo: {
+        id: String(app._id),
+        name: app.name,
+        teamId: app.teamId,
+        tmbId: app.tmbId
+      },
+      runningUserInfo: await getRunningUserInfoByTmbId(app.tmbId),
+      uid: chatUserId || outLinkConfig.tmbId,
+      chatId,
+      variables,
+      histories,
+      query: query,
+      chatConfig,
+      stream: streamId ? true : false,
+      workflowStreamResponse,
+      runtimeEdges: storeEdges2RuntimeEdges(edges),
+      runtimeNodes: storeNodes2RuntimeNodes(nodes, getWorkflowEntryNodeIds(nodes)),
+      maxRunTimes: WORKFLOW_MAX_RUN_TIMES,
+      retainDatasetCite: false
+    });
+
+    // Format results
+    let responseContent = assistantResponses
+      .map((response) => {
+        return response.text?.content;
+      })
+      .filter(Boolean)
+      .join('\n')
+      .trim();
+    if (responseContent.length === 0) {
+      responseContent = DEFAULT_REPLY;
+    }
+
+    const replyResult = await (async () => {
+      try {
+        if (streamId) {
+          // if streamId is provided, do not reply
+          return;
+        }
+        const result = await replyCallback?.(responseContent);
+
+        if (result.errcode !== 0) {
+          logger.error('Outlink official account reply failed', {
+            errmsg: result.errmsg
+          });
+          return {
+            success: false,
+            errmsg: result.errmsg
+          };
+        }
+
+        logger.debug('Outlink official account reply success', {
+          responseContent,
+          result
+        });
+        return {
+          success: true,
+          data: result
+        };
+      } catch (error) {
+        logger.error('Outlink reply callback failed', { error });
+        return {
+          success: false,
+          errmsg: getErrText(error)
+        };
+      }
+    })();
+
+    // Save and reply
+    await pushChatRecords({
+      chatId,
+      appId: app._id,
+      teamId: outLinkConfig.teamId,
+      tmbId: outLinkConfig.tmbId,
+      outLinkUid: chatUserId,
+      nodes,
+      appChatConfig: chatConfig,
+      variables: newVariables,
+      newTitle: String(userQuestion || '').slice(0, 8),
+      shareId: outLinkConfig.shareId,
+      source: getChatSourceByPublishChannel(outLinkConfig.type),
+      sourceName: outLinkConfig.name,
+      userContent: {
+        dataId: messageId,
+        obj: ChatRoleEnum.Human,
+        value: query
+      },
+      aiContent: {
+        obj: ChatRoleEnum.AI,
+        value: assistantResponses,
+        [DispatchNodeResponseKeyEnum.nodeResponse]: flowResponses,
+        memories: system_memories
+      },
+      metadata: {},
+      durationSeconds,
+      errorMsg: replyResult?.success ? undefined : replyResult?.errmsg
+    });
+
+    const totalPoints = flowUsages.reduce((sum, item) => sum + (item.totalPoints || 0), 0);
+    addOutLinkUsage({
+      shareId: outLinkConfig.shareId,
+      totalPoints: totalPoints
+    });
+
+    if (streamId) {
+      await appendRedisCache(streamResKey, STREAM_END_FLAG, 60);
+    }
+  } catch (error) {
+    logger.error('Outlink invoke chat failed', {
+      shareId: outLinkConfig.shareId,
+      chatId,
+      messageId,
+      streamId,
+      error
+    });
+
+    try {
+      await appendRedisCache(streamResKey, STREAM_END_FLAG, 60);
+
+      await replyCallback?.(`App run error: ${getErrText(error)}`);
+    } catch (error) {
+      logger.error('Outlink invoke chat fallback reply failed', {
+        shareId: outLinkConfig.shareId,
+        chatId,
+        messageId,
+        streamId,
+        error
+      });
+    }
+  }
+}