更新依赖项，优化 WebSocket 处理，添加文件流管道功能，改进用户认证逻辑

src/modules/ws-proxy/index.ts

@@ -1,136 +1,29 @@
-import { WebSocketServer } from 'ws';
-import { nanoid } from 'nanoid';
 import { WsProxyManager } from './manager.ts';
-import { getLoginUser } from '@/modules/auth.ts';
+import { getLoginUserByToken } from '@/modules/auth.ts';
 import { logger } from '../logger.ts';
 export const wsProxyManager = new WsProxyManager();
-
-export const upgrade = (request: any, socket: any, head: any) => {
-  const req = request as any;
-  const url = new URL(req.url, 'http://localhost');
-  const id = url.searchParams.get('id');
-  if (url.pathname === '/ws/proxy') {
-    console.log('upgrade', request.url, id);
-    wss.handleUpgrade(req, socket, head, (ws) => {
-      // 这里手动触发 connection 事件
-      console.log('emitting connection event');
-      // @ts-ignore
-      wss.emit('connection', ws, req);
-    });
-    return true;
-  }
-  return false;
-};
-export const wss = new WebSocketServer({
-  noServer: true,
-  path: '/ws/proxy',
-});
-
-wss.on('connection', async (ws, req) => {
-  console.log('connected', req.url);
-  const url = new URL(req.url, 'http://localhost');
-  const _id = url?.searchParams?.get('id');
-  const id = _id || nanoid();
-  const loginUser = await getLoginUser(req);
-  if (!loginUser) {
-    console.log('未登录，断开连接');
-    ws.send(JSON.stringify({ code: 401, message: '未登录' }));
-    setTimeout(() => {
-      ws.close();
-    }, 1000);
-    return;
-  }
-  const user = loginUser.tokenUser.username;
-  const userApp = user + '-' + id;
-  console.log('注册 ws 连接', userApp);
-  wsProxyManager.register(userApp, { user, ws });
-  ws.send(
-    JSON.stringify({
-      type: 'connected',
-      user: user,
-      id,
-    }),
-  );
-  ws.on('message', async (event: Buffer) => {
-    const eventData = event.toString();
-    if (!eventData) {
-      return;
-    }
-    const data = JSON.parse(eventData);
-    logger.debug('message', data);
-  });
-  ws.on('close', () => {
-    logger.debug('ws closed');
-    wsProxyManager.unregister(userApp);
-  });
-});
-
-export class WssApp {
-  wss: WebSocketServer;
-  bunWSS = websocket;
-  constructor() {
-    this.wss = wss;
-  }
-  upgrade(request: any, socket: any, head: any) {
-    // return upgrade(request, socket, head);
-    return bunUpgrade(request);
-  }
-}
-
-export const bunUpgrade = (request: Request) => {
-  const url = new URL(request.url, 'http://localhost');
-  const isUpgrade = url.pathname === '/ws/proxy';
-  if (isUpgrade) {
-    console.log('upgrade', request.url);
-
-    // 使用 Bun 原生 WebSocket
-    new Response(null, {
-      status: 101,
-      headers: {
-        'Upgrade': 'websocket',
-      },
-    });
-    return true;
-  }
-
-  return false;
-};
-// Bun WebSocket 处理器
-export const websocket = {
-  async open(ws: any) {
-    console.log('WebSocket opened');
-    const { url, token } = ws.data;
-
-    const urlObj = new URL(url, 'http://localhost');
-    const _id = urlObj.searchParams.get('id');
-    const id = _id || nanoid();
-
-    // 创建一个模拟的 request 对象用于认证
-    const mockReq: any = {
-      url: url,
-      headers: {
-        authorization: token ? `Bearer ${token}` : undefined,
-      },
-    };
-
-    const loginUser = await getLoginUser(mockReq);
-
-    if (!loginUser) {
-      console.log('未登录,断开连接');
+import { WebScoketListenerFun } from '@kevisual/router/src/server/server-type.ts'
+export const wssFun: WebScoketListenerFun = async (req, res) => {
+  // do nothing, just to enable ws upgrade event
+  const { id, ws, token, data, emitter } = req;
+  logger.debug('ws proxy connected, id=', id, ' token=', token, ' data=', data);
+  // console.log('req', req)
+  const { type } = data || {};
+  if (type === 'registryClient') {
+    const loginUser = await getLoginUserByToken(token);
+    if (!loginUser?.tokenUser) {
+      logger.debug('未登录，断开连接');
       ws.send(JSON.stringify({ code: 401, message: '未登录' }));
-      ws.close();
+      setTimeout(() => {
+        ws.close(401, 'Unauthorized');
+      }, 1000);
       return;
     }
-
-    const user = loginUser.tokenUser.username;
+    const user = loginUser?.tokenUser?.username;
     const userApp = user + '-' + id;
-    console.log('注册 ws 连接', userApp);
-
-    ws.data.userApp = userApp;
-    ws.data.user = user;
-
+    logger.debug('注册 ws 连接', userApp);
+    // @ts-ignore
     wsProxyManager.register(userApp, { user, ws });
-
     ws.send(
       JSON.stringify({
         type: 'connected',
@@ -138,26 +31,22 @@ export const websocket = {
         id,
       }),
     );
-  },
-
-  async message(ws: any, message: string) {
-    try {
-      const data = JSON.parse(message);
-      logger.debug('message', data);
-    } catch (error) {
-      logger.error('Failed to parse message', error);
-    }
-  },
-
-  close(ws: any) {
-    const { userApp } = ws.data;
-    logger.debug('ws closed', userApp);
-    if (userApp) {
+    emitter.once('close--' + id, () => {
+      logger.debug('ws emitter closed');
       wsProxyManager.unregister(userApp);
-    }
-  },
-
-  error(ws: any, error: Error) {
-    console.error('WebSocket error:', error);
-  },
-};
+    });
+    // @ts-ignore
+    ws.data.userApp = userApp;
+    return;
+  }
+  // @ts-ignore
+  const userApp = ws.data.userApp;
+  logger.debug('message', data, ' userApp=', userApp);
+  const wsMessage = wsProxyManager.get(userApp);
+  if (wsMessage) {
+    wsMessage.sendResponse(data);
+  } else {
+    // @ts-ignore
+    logger.debug('账号应用未注册，无法处理消息。未授权？', ws.data);
+  }
+}