Merge branch 'codex/websocket' into codex/zongmeng-v1

src/react-app/routes/game/settings/components/AccountSettingsTab.tsx

@@ -425,6 +425,7 @@ export function AccountSettingsTab() {
       <SettingsSection
         title="注销账号"
         description="此操作无法撤销，请确认已了解注销后的影响。"
+        className="border border-crimson/35 bg-crimson/8 px-4 pb-4"
       >
         {deletionExpanded ? (
           <div className="grid gap-4">

src/server/routes/api/realtime.router.test.ts

@@ -1,16 +1,32 @@
+import { isAllowedRealtimeOrigin } from '@server/lib/http/realtimeOrigin';
 import { Hono } from 'hono';
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
-import { isAllowedRealtimeOrigin } from '@server/lib/http/realtimeOrigin';
 
 const {
   recordRealtimeConnectionCloseMock,
+  recordRealtimeConnectionHeartbeatMock,
   recordRealtimeConnectionOpenMock,
+  subscribePlayerStateEventsMock,
+  subscribeWorldChatMessagesMock,
+  upgradeRequestMock,
   upgradeWebSocketMock,
-} = vi.hoisted(() => ({
-  recordRealtimeConnectionCloseMock: vi.fn(),
-  recordRealtimeConnectionOpenMock: vi.fn(),
-  upgradeWebSocketMock: vi.fn(),
-}));
+} = vi.hoisted(() => {
+  const upgradeRequestMock = vi.fn();
+  return {
+    recordRealtimeConnectionCloseMock: vi.fn(),
+    recordRealtimeConnectionHeartbeatMock: vi.fn(),
+    recordRealtimeConnectionOpenMock: vi.fn(),
+    subscribePlayerStateEventsMock: vi.fn(),
+    subscribeWorldChatMessagesMock: vi.fn(),
+    upgradeRequestMock,
+    upgradeWebSocketMock: vi.fn((createEvents: (context: any) => any) => {
+      return async (context: any) => {
+        const handlers = await createEvents(context);
+        return upgradeRequestMock(context, handlers);
+      };
+    }),
+  };
+});
 
 vi.mock('@server/lib/hono/middleware', () => ({
   requireActiveCultivatorRef:
@@ -32,15 +48,16 @@ vi.mock('hono/bun', () => ({
 
 vi.mock('@server/lib/services/onlinePresenceService', () => ({
   recordRealtimeConnectionClose: recordRealtimeConnectionCloseMock,
+  recordRealtimeConnectionHeartbeat: recordRealtimeConnectionHeartbeatMock,
   recordRealtimeConnectionOpen: recordRealtimeConnectionOpenMock,
 }));
 
 vi.mock('@server/lib/services/playerStateBroadcaster', () => ({
-  subscribePlayerStateEvents: () => vi.fn(),
+  subscribePlayerStateEvents: subscribePlayerStateEventsMock,
 }));
 
 vi.mock('@server/lib/services/worldChatBroadcaster', () => ({
-  subscribeWorldChatMessages: () => vi.fn(),
+  subscribeWorldChatMessages: subscribeWorldChatMessagesMock,
 }));
 
 import realtimeRouter from './realtime.router';
@@ -48,12 +65,15 @@ import realtimeRouter from './realtime.router';
 const originalOrigins = process.env.PUBLIC_WEB_ORIGINS;
 
 afterEach(() => {
+  vi.useRealTimers();
   process.env.PUBLIC_WEB_ORIGINS = originalOrigins;
 });
 
 beforeEach(() => {
   vi.clearAllMocks();
-  upgradeWebSocketMock.mockImplementation((_context, handlers) => {
+  subscribePlayerStateEventsMock.mockImplementation(() => vi.fn());
+  subscribeWorldChatMessagesMock.mockImplementation(() => vi.fn());
+  upgradeRequestMock.mockImplementation((_context, handlers) => {
     const ws = {
       readyState: 1,
       send: vi.fn(),
@@ -88,14 +108,11 @@ describe('realtime origin guard', () => {
     process.env.PUBLIC_WEB_ORIGINS = 'https://app.example.com';
     const app = new Hono().route('/api/realtime', realtimeRouter);
 
-    const response = await app.request(
-      '/api/realtime?channels=player-state',
-      {
-        headers: {
-          Origin: 'https://app.example.com',
-        },
+    const response = await app.request('/api/realtime?channels=player-state', {
+      headers: {
+        Origin: 'https://app.example.com',
       },
-    );
+    });
 
     expect(response.status).toBe(200);
     expect(recordRealtimeConnectionOpenMock).toHaveBeenCalledWith(
@@ -105,4 +122,187 @@ describe('realtime origin guard', () => {
       'cultivator-1',
     );
   });
+
+  it('keeps the connection alive when client activity uses a new Hono WSContext', async () => {
+    vi.useFakeTimers();
+    process.env.PUBLIC_WEB_ORIGINS = 'https://app.example.com';
+    let handlers: any;
+    upgradeRequestMock.mockImplementation((_context, nextHandlers) => {
+      handlers = nextHandlers;
+      return new Response(null, { status: 200 });
+    });
+    const app = new Hono().route('/api/realtime', realtimeRouter);
+
+    const response = await app.request('/api/realtime?channels=player-state', {
+      headers: {
+        Origin: 'https://app.example.com',
+      },
+    });
+    expect(response.status).toBe(200);
+
+    const send = vi.fn();
+    const close = vi.fn();
+    const openContext = { readyState: 1, send, close };
+    const messageContext = { readyState: 1, send, close };
+    handlers.onOpen?.({}, openContext);
+
+    await vi.advanceTimersByTimeAsync(25_000);
+    expect(send).toHaveBeenCalledWith(expect.stringContaining('"type":"ping"'));
+
+    handlers.onMessage?.(
+      { data: JSON.stringify({ type: 'client.activity' }) },
+      messageContext,
+    );
+    await vi.advanceTimersByTimeAsync(50_000);
+
+    expect(close).not.toHaveBeenCalled();
+    handlers.onClose?.();
+  });
+
+  it('uses elapsed time when a heartbeat interval callback is delayed', async () => {
+    process.env.PUBLIC_WEB_ORIGINS = 'https://app.example.com';
+    let handlers: any;
+    upgradeRequestMock.mockImplementation((_context, nextHandlers) => {
+      handlers = nextHandlers;
+      return new Response(null, { status: 200 });
+    });
+    let heartbeatTick: (() => void) | undefined;
+    vi.spyOn(globalThis, 'setInterval').mockImplementation((callback) => {
+      heartbeatTick = callback as () => void;
+      return {} as ReturnType<typeof setInterval>;
+    });
+    vi.spyOn(globalThis, 'clearInterval').mockImplementation(() => {});
+    const now = vi.spyOn(Date, 'now').mockReturnValue(1_000);
+    const app = new Hono().route('/api/realtime', realtimeRouter);
+
+    const response = await app.request('/api/realtime?channels=player-state', {
+      headers: {
+        Origin: 'https://app.example.com',
+      },
+    });
+    expect(response.status).toBe(200);
+
+    const send = vi.fn();
+    const close = vi.fn();
+    handlers.onOpen?.({}, { readyState: 1, send, close });
+    now.mockReturnValue(51_001);
+    heartbeatTick?.();
+
+    expect(close).toHaveBeenCalledWith(4_000, 'heartbeat timeout');
+  });
+
+  it('isolates heartbeat activity between websocket upgrade closures', async () => {
+    vi.useFakeTimers();
+    process.env.PUBLIC_WEB_ORIGINS = 'https://app.example.com';
+    const handlers: any[] = [];
+    upgradeRequestMock.mockImplementation((_context, nextHandlers) => {
+      handlers.push(nextHandlers);
+      return new Response(null, { status: 200 });
+    });
+    const app = new Hono().route('/api/realtime', realtimeRouter);
+    const request = () =>
+      app.request('/api/realtime?channels=player-state', {
+        headers: { Origin: 'https://app.example.com' },
+      });
+
+    await request();
+    await request();
+    expect(handlers).toHaveLength(2);
+
+    const firstSocket = { readyState: 1, send: vi.fn(), close: vi.fn() };
+    const secondSocket = { readyState: 1, send: vi.fn(), close: vi.fn() };
+    handlers[0].onOpen?.({}, firstSocket);
+    handlers[1].onOpen?.({}, secondSocket);
+
+    await vi.advanceTimersByTimeAsync(25_000);
+    handlers[0].onMessage?.(
+      { data: JSON.stringify({ type: 'pong' }) },
+      firstSocket,
+    );
+    await vi.advanceTimersByTimeAsync(50_000);
+
+    expect(firstSocket.close).not.toHaveBeenCalled();
+    expect(secondSocket.close).toHaveBeenCalledWith(4_000, 'heartbeat timeout');
+    handlers[0].onClose?.();
+  });
+
+  it('reserves connection capacity before websocket open', async () => {
+    process.env.PUBLIC_WEB_ORIGINS = 'https://app.example.com';
+    const handlers: any[] = [];
+    upgradeRequestMock.mockImplementation((_context, nextHandlers) => {
+      handlers.push(nextHandlers);
+      return new Response(null, { status: 200 });
+    });
+    const app = new Hono().route('/api/realtime', realtimeRouter);
+    const request = () =>
+      app.request('/api/realtime?channels=player-state', {
+        headers: { Origin: 'https://app.example.com' },
+      });
+
+    const responses = await Promise.all([
+      request(),
+      request(),
+      request(),
+      request(),
+    ]);
+
+    expect(responses.map((response) => response.status)).toEqual([
+      200, 200, 200, 429,
+    ]);
+    expect(handlers).toHaveLength(3);
+    for (const websocketHandlers of handlers) {
+      const socket = { readyState: 1, send: vi.fn(), close: vi.fn() };
+      websocketHandlers.onOpen?.({}, socket);
+      websocketHandlers.onClose?.();
+    }
+  });
+
+  it('finishes cleanup when an unsubscribe callback throws', async () => {
+    process.env.PUBLIC_WEB_ORIGINS = 'https://app.example.com';
+    const throwingUnsubscribe = vi.fn(() => {
+      throw new Error('unsubscribe failed');
+    });
+    const safeUnsubscribe = vi.fn();
+    subscribePlayerStateEventsMock.mockReturnValue(throwingUnsubscribe);
+    subscribeWorldChatMessagesMock.mockReturnValue(safeUnsubscribe);
+    vi.spyOn(console, 'warn').mockImplementation(() => {});
+    let handlers: any;
+    upgradeRequestMock.mockImplementation((_context, nextHandlers) => {
+      handlers = nextHandlers;
+      return new Response(null, { status: 200 });
+    });
+    const app = new Hono().route('/api/realtime', realtimeRouter);
+    await app.request('/api/realtime?channels=player-state%2Cworld-chat', {
+      headers: { Origin: 'https://app.example.com' },
+    });
+    handlers.onOpen?.({}, { readyState: 1, send: vi.fn(), close: vi.fn() });
+
+    expect(() => handlers.onClose?.()).not.toThrow();
+    expect(throwingUnsubscribe).toHaveBeenCalledOnce();
+    expect(safeUnsubscribe).toHaveBeenCalledOnce();
+    expect(recordRealtimeConnectionCloseMock).toHaveBeenCalledOnce();
+  });
+
+  it('cleans up when sending to the websocket throws', async () => {
+    process.env.PUBLIC_WEB_ORIGINS = 'https://app.example.com';
+    let handlers: any;
+    upgradeRequestMock.mockImplementation((_context, nextHandlers) => {
+      handlers = nextHandlers;
+      return new Response(null, { status: 200 });
+    });
+    const app = new Hono().route('/api/realtime', realtimeRouter);
+    await app.request('/api/realtime?channels=player-state', {
+      headers: { Origin: 'https://app.example.com' },
+    });
+    const socket = {
+      readyState: 1,
+      send: vi.fn(() => {
+        throw new Error('socket ended');
+      }),
+      close: vi.fn(),
+    };
+
+    expect(() => handlers.onOpen?.({}, socket)).not.toThrow();
+    expect(recordRealtimeConnectionCloseMock).toHaveBeenCalledOnce();
+  });
 });

src/server/routes/api/realtime.router.ts

@@ -13,13 +13,16 @@ import {
   type RealtimeChannel,
   type RealtimeServerEvent,
 } from '@shared/contracts/realtime';
+import { Hono, type Context, type MiddlewareHandler } from 'hono';
 import { upgradeWebSocket } from 'hono/bun';
 import type { WSContext } from 'hono/ws';
-import { Hono, type Context } from 'hono';
 
 const router = new Hono<AppEnv>();
 const HEARTBEAT_INTERVAL_MS = 25_000;
-const MAX_MISSED_PONGS = 2;
+const MAX_MISSED_HEARTBEATS = 2;
+const HEARTBEAT_TIMEOUT_MS = HEARTBEAT_INTERVAL_MS * MAX_MISSED_HEARTBEATS;
+const HEARTBEAT_TIMEOUT_CLOSE_CODE = 4_000;
+const HEARTBEAT_TIMEOUT_CLOSE_REASON = 'heartbeat timeout';
 const MAX_MESSAGE_BYTES = 512;
 const MAX_CONNECTIONS_PER_USER = 3;
 const MAX_CONNECTIONS_PER_CULTIVATOR = 3;
@@ -29,6 +32,21 @@ const userConnectionCounts = new Map<string, number>();
 const cultivatorConnectionCounts = new Map<string, number>();
 const ipConnectionCounts = new Map<string, number>();
 
+type ConnectionIdentity = {
+  userId: string;
+  cultivatorId: string;
+  ip: string;
+};
+
+type ConnectionReservation = ConnectionIdentity & {
+  release: () => void;
+};
+
+const pendingConnectionReservations = new WeakMap<
+  Context<AppEnv>,
+  ConnectionReservation
+>();
+
 function incrementCount(map: Map<string, number>, key: string) {
   map.set(key, (map.get(key) ?? 0) + 1);
 }
@@ -64,38 +82,51 @@ function parseChannels(raw: string | undefined): RealtimeChannel[] {
       allowed.has(item as RealtimeChannel),
     );
 
-  return channels.length > 0 ? Array.from(new Set(channels)) : [...REALTIME_CHANNELS];
+  return channels.length > 0
+    ? Array.from(new Set(channels))
+    : [...REALTIME_CHANNELS];
 }
 
-function canAcceptConnection(args: {
-  userId: string;
-  cultivatorId: string;
-  ip: string;
-}) {
-  return (
-    (userConnectionCounts.get(args.userId) ?? 0) < MAX_CONNECTIONS_PER_USER &&
-    (cultivatorConnectionCounts.get(args.cultivatorId) ?? 0) <
-      MAX_CONNECTIONS_PER_CULTIVATOR &&
-    (ipConnectionCounts.get(args.ip) ?? 0) < MAX_CONNECTIONS_PER_IP
-  );
-}
-
-function sendEnvelope(ws: WSContext, event: RealtimeServerEvent) {
-  if (ws.readyState !== 1) {
-    return;
+function reserveConnection(
+  identity: ConnectionIdentity,
+): ConnectionReservation | null {
+  if (
+    (userConnectionCounts.get(identity.userId) ?? 0) >=
+      MAX_CONNECTIONS_PER_USER ||
+    (cultivatorConnectionCounts.get(identity.cultivatorId) ?? 0) >=
+      MAX_CONNECTIONS_PER_CULTIVATOR ||
+    (ipConnectionCounts.get(identity.ip) ?? 0) >= MAX_CONNECTIONS_PER_IP
+  ) {
+    return null;
   }
 
-  ws.send(JSON.stringify(event));
+  incrementCount(userConnectionCounts, identity.userId);
+  incrementCount(cultivatorConnectionCounts, identity.cultivatorId);
+  incrementCount(ipConnectionCounts, identity.ip);
+  let released = false;
+
+  return {
+    ...identity,
+    release() {
+      if (released) {
+        return;
+      }
+      released = true;
+      decrementCount(userConnectionCounts, identity.userId);
+      decrementCount(cultivatorConnectionCounts, identity.cultivatorId);
+      decrementCount(ipConnectionCounts, identity.ip);
+    },
+  };
 }
 
-router.get('/', async (c, next) => {
+const requireRealtimeOrigin = (async (c, next) => {
   if (!isAllowedRealtimeOrigin(c.req.header('origin'))) {
     return c.json({ success: false, error: 'Forbidden origin' }, 403);
   }
   await next();
-});
+}) satisfies MiddlewareHandler<AppEnv>;
 
-router.get('/', requireActiveCultivatorRef(), async (c) => {
+const reserveRealtimeConnection = (async (c, next) => {
   const user = c.get('user');
   if (!user) {
     return c.json({ success: false, error: '未授权访问' }, 401);
@@ -106,115 +137,163 @@ router.get('/', requireActiveCultivatorRef(), async (c) => {
     return c.json({ success: false, error: '当前没有活跃角色' }, 404);
   }
 
-  const channels = parseChannels(c.req.query('channels'));
-  const ip = getClientIp(c);
-  if (
-    !canAcceptConnection({
-      userId: user.id,
-      cultivatorId: ref.cultivatorId,
-      ip,
-    })
-  ) {
+  const reservation = reserveConnection({
+    userId: user.id,
+    cultivatorId: ref.cultivatorId,
+    ip: getClientIp(c),
+  });
+  if (!reservation) {
     return c.json({ success: false, error: '实时连接过多' }, 429);
   }
 
-  let cleanup: () => void = () => {};
+  pendingConnectionReservations.set(c, reservation);
+  try {
+    await next();
+  } catch (error) {
+    pendingConnectionReservations.delete(c);
+    reservation.release();
+    throw error;
+  }
 
-  return upgradeWebSocket(c, {
-    onOpen(_event, ws) {
-      incrementCount(userConnectionCounts, user.id);
-      incrementCount(cultivatorConnectionCounts, ref.cultivatorId);
-      incrementCount(ipConnectionCounts, ip);
-      recordRealtimeConnectionOpen(ref.cultivatorId);
+  if (pendingConnectionReservations.get(c) === reservation) {
+    pendingConnectionReservations.delete(c);
+    reservation.release();
+  }
+}) satisfies MiddlewareHandler<AppEnv>;
 
-      const unsubscribers: Array<() => void> = [];
-      if (channels.includes('player-state')) {
-        unsubscribers.push(
-          subscribePlayerStateEvents(ref.cultivatorId, (events) => {
-            sendEnvelope(ws, {
-              type: 'player-state.events',
-              payload: { events },
-            });
-          }),
-        );
-      }
-      if (channels.includes('world-chat')) {
-        unsubscribers.push(
-          subscribeWorldChatMessages((message) => {
-            sendEnvelope(ws, {
-              type: 'world-chat.message',
-              payload: message,
-            });
-          }),
-        );
-      }
+router.get(
+  '/',
+  requireRealtimeOrigin,
+  requireActiveCultivatorRef(),
+  reserveRealtimeConnection,
+  upgradeWebSocket((rawContext) => {
+    const c = rawContext as Context<AppEnv>;
+    const reservation = pendingConnectionReservations.get(c);
+    if (!reservation) {
+      throw new Error('Realtime connection reservation is missing');
+    }
+    pendingConnectionReservations.delete(c);
 
-      let missedPongs = 0;
-      let closed = false;
-      let heartbeat: ReturnType<typeof setInterval> | null = setInterval(() => {
-        missedPongs += 1;
-        if (missedPongs > MAX_MISSED_PONGS) {
-          ws.close();
-          cleanup();
-          return;
-        }
-        recordRealtimeConnectionHeartbeat(ref.cultivatorId);
-        sendEnvelope(ws, {
-          type: 'ping',
-          payload: { serverTime: new Date().toISOString() },
-        });
-      }, HEARTBEAT_INTERVAL_MS);
-      cleanup = () => {
-        if (closed) {
-          return;
-        }
-        closed = true;
-        for (const unsubscribe of unsubscribers) {
-          unsubscribe();
-        }
-        decrementCount(userConnectionCounts, user.id);
-        decrementCount(cultivatorConnectionCounts, ref.cultivatorId);
-        decrementCount(ipConnectionCounts, ip);
-        recordRealtimeConnectionClose(ref.cultivatorId);
-        if (heartbeat) {
-          clearInterval(heartbeat);
-          heartbeat = null;
-        }
-      };
-
-      sendEnvelope(ws, {
-        type: 'ready',
-        payload: { cultivatorId: ref.cultivatorId, channels },
-      });
-      Object.assign(ws, {
-        __daoyouPong: () => {
-          missedPongs = 0;
-        },
-      });
-    },
-    onMessage(event, ws) {
-      const raw = event.data;
-      if (typeof raw !== 'string' || raw.length > MAX_MESSAGE_BYTES) {
-        ws.close();
+    const channels = parseChannels(c.req.query('channels'));
+    const { cultivatorId } = reservation;
+    let lastClientActivityAt = Date.now();
+    let heartbeat: ReturnType<typeof setInterval> | undefined;
+    let closed = false;
+    const unsubscribers: Array<() => void> = [];
+    const cleanup = () => {
+      if (closed) {
         return;
       }
-
-      try {
-        const parsed = JSON.parse(raw) as { type?: unknown };
-        if (parsed.type === 'pong') {
-          (ws as unknown as { __daoyouPong?: () => void }).__daoyouPong?.();
-        }
-      } catch {
-        // Ignore invalid client frames; realtime v1 is server-push only.
+      closed = true;
+      if (heartbeat) {
+        clearInterval(heartbeat);
+      }
+      for (const unsubscribe of unsubscribers) {
+        try {
+          unsubscribe();
+        } catch (error) {
+          console.warn('[realtime] unsubscribe failed', {
+            cultivatorId,
+            error,
+          });
+        }
+      }
+      reservation.release();
+      recordRealtimeConnectionClose(cultivatorId);
+    };
+    const closeConnection = (ws: WSContext, code: number, reason: string) => {
+      if (closed) {
+        return;
+      }
+      try {
+        ws.close(code, reason);
+      } catch (error) {
+        console.warn('[realtime] websocket close failed', {
+          cultivatorId,
+          error,
+        });
       }
-    },
-    onClose() {
       cleanup();
-    },
-    onError() {
-      cleanup();
-    },
-  });
-});
+    };
+    const sendEnvelope = (ws: WSContext, event: RealtimeServerEvent) => {
+      if (closed) {
+        return false;
+      }
+      try {
+        ws.send(JSON.stringify(event));
+        return true;
+      } catch (error) {
+        console.warn('[realtime] websocket send failed', {
+          cultivatorId,
+          error,
+        });
+        closeConnection(ws, 1_011, 'websocket send failed');
+        return false;
+      }
+    };
+
+    return {
+      onOpen(_event, ws) {
+        lastClientActivityAt = Date.now();
+        recordRealtimeConnectionOpen(cultivatorId);
+
+        if (channels.includes('player-state')) {
+          unsubscribers.push(
+            subscribePlayerStateEvents(cultivatorId, (events) => {
+              sendEnvelope(ws, {
+                type: 'player-state.events',
+                payload: { events },
+              });
+            }),
+          );
+        }
+        if (channels.includes('world-chat')) {
+          unsubscribers.push(
+            subscribeWorldChatMessages((message) => {
+              sendEnvelope(ws, {
+                type: 'world-chat.message',
+                payload: message,
+              });
+            }),
+          );
+        }
+
+        heartbeat = setInterval(() => {
+          const timeSinceLastClientActivity = Date.now() - lastClientActivityAt;
+          if (timeSinceLastClientActivity > HEARTBEAT_TIMEOUT_MS) {
+            closeConnection(
+              ws,
+              HEARTBEAT_TIMEOUT_CLOSE_CODE,
+              HEARTBEAT_TIMEOUT_CLOSE_REASON,
+            );
+            return;
+          }
+          recordRealtimeConnectionHeartbeat(cultivatorId);
+          sendEnvelope(ws, {
+            type: 'ping',
+            payload: { serverTime: new Date().toISOString() },
+          });
+        }, HEARTBEAT_INTERVAL_MS);
+        sendEnvelope(ws, {
+          type: 'ready',
+          payload: { cultivatorId, channels },
+        });
+      },
+      onMessage(event, ws) {
+        const raw = event.data;
+        if (typeof raw !== 'string' || raw.length > MAX_MESSAGE_BYTES) {
+          closeConnection(ws, 1_003, 'invalid client message');
+          return;
+        }
+
+        // Any bounded client frame proves that this connection is still alive.
+        lastClientActivityAt = Date.now();
+      },
+      onClose() {
+        cleanup();
+      },
+    };
+  }),
+);
 
 export default router;