WebSocket

Server Runtime 的实时通信层：SocketServer / RoomGateway / GameGateway / GameEventPublisher / BroadcastHub。

概述

WebSocket 层基于 Socket.IO 实现，承担：

客户端连接生命周期管理（含身份认证与重连）
房间相关事件路由（room:join / room:leave / room:ready / room:start）
游戏内事件路由（game:action / game:sync / game:leave）
引擎事件到 Socket 事件的转发（脱敏后广播）
BroadcastPort 接口实现

SocketServer

Socket.IO 服务端封装 + BroadcastPort 实现。

SocketServerDeps

interface SocketServerDeps {
  playerSessionService: PlayerSessionService
  roomGateway: RoomGateway
  gameGateway: GameGateway
}

SocketServerOptions

interface SocketServerOptions {
  path?: string
  cors?: {
    origin: string | string[]
    methods?: string[]
    credentials?: boolean
  }
}

类

class SocketServer implements BroadcastPort {
  constructor(
    httpServer: HttpServer,
    deps: SocketServerDeps,
    options?: SocketServerOptions
  )
  readonly io: Server
  async close(): Promise<void>
  // BroadcastPort 实现（见下）
}

行为

创建 Socket.IO Server 并挂载到 HTTP server。
io.use(authMiddleware)：身份认证中间件，从 socket.handshake.auth.playerId 或 X-Player-Id header 取 playerId；缺失则拒绝连接（HTTP 路径有 playerIdentityMiddleware 兜底；Socket 必须显式提供）。
io.on('connection', handleConnection)：处理新连接。
维护 playerSockets: Map<playerId, Socket> 以支持 O(1) 的 sendToPlayer 路由。
该 Map 与 PlayerSessionService 的 PlayerSession 状态并行——SocketServer 关心路由，PlayerSessionService 关心完整状态。

handleConnection 流程

从 socket.data.playerId 取 playerId。
playerSockets.set(playerId, socket)。
playerSessionService.connect(playerId, socket.id) 建立 / 恢复连接。
重连恢复：若 session.roomId 存在则 socket.join('room:${roomId}')；若 session.gameId 存在则 socket.join('game:${gameId}')。
roomGateway.registerHandlers(socket) + gameGateway.registerHandlers(socket)。
注册 socket.on('disconnect', ...)。

handleDisconnect 流程

仅当当前注册的 socket 仍是本 socket 时才清除 playerSockets（防止新连接覆盖后被误删）。
playerSessionService.disconnect(socket.id) 标记 PlayerSession 离线。

BroadcastPort 实现

方法	行为
`broadcastToGame(gameId, event, payload)`	`io.to('game:${gameId}').emit(event, payload)`
`broadcastToRoom(roomId, event, payload)`	`io.to('room:${roomId}').emit(event, payload)`
`sendToPlayer(playerId, event, payload)`	取 `playerSockets.get(playerId)`，存在则 `socket.emit(event, payload)`
`isPlayerConnected(playerId)`	`playerSockets.get(playerId)?.connected ?? false`
`joinPlayerToRoom(playerId, room)`	让玩家 socket 加入 Socket.IO 房间（用于 `room:start` 时把玩家加入 `game:${gameId}`）
`leavePlayerFromRoom(playerId, room)`	让玩家 socket 离开房间

close()

async close(): Promise<void>

优雅关闭：断开所有 socket。

RoomGateway

房间相关 Socket 事件网关。

RoomGatewayDeps

interface RoomGatewayDeps {
  roomService: RoomService
  playerSessionService: PlayerSessionService
  broadcaster: BroadcastPort
}

RoomAck

interface RoomAck<T = unknown> {
  ok: boolean
  code?: string
  message?: string
  data?: T
}

类

class RoomGateway {
  constructor(deps: RoomGatewayDeps)
  registerHandlers(socket: Socket): void
}

处理事件

事件	payload	说明
`room:join`	`{ roomId }`	加入房间
`room:leave`	`{ roomId }`	离开房间
`room:ready`	`{ roomId, ready? }`	标记玩家就绪状态（Phase 6 简化：仅广播通知）
`room:start`	`{ roomId }`	房主开始游戏

room:join 行为

校验 roomId 存在。
roomService.joinRoom(roomId, playerId)。
playerSessionService.bindRoom(playerId, roomId)。
socket.join('room:${roomId}')。
broadcaster.broadcastToRoom(roomId, 'room:playerJoined', { roomId, playerId, playerIds, currentPlayers, maxPlayers })。
ack({ ok: true, data: { roomId, gameType, ownerId, playerIds, maxPlayers, minPlayers, status, gameSessionId } })。

room:leave 行为

校验 roomId 存在。
roomService.leaveRoom(roomId, playerId)。
playerSessionService.unbindRoom(playerId)。
socket.leave('room:${roomId}')。
broadcaster.broadcastToRoom(roomId, 'room:playerLeft', { roomId, playerId, playerIds, currentPlayers })。
ack({ ok: true, data: { roomId, playerIds, status } })。

room:ready 行为

Phase 6 简化：Room 实体不维护 per-player ready 状态，仅向房间其他成员广播 room:playerReady 通知。开始游戏的硬性条件由 RoomService.startRoom 校验（minPlayers 满足即可）。

room:start 行为

校验 roomId 存在。
取出 room，校验 room.ownerId === playerId，否则 ack({ ok: false, code: 'PLAYER_NOT_ALLOWED', ... })。
roomService.startRoom(roomId) → GameService.startGame（内部已广播 game:started 到 room 与初始 game:state）。
把每位玩家的 socket 加入 game:${gameId}：broadcaster.joinPlayerToRoom(pid, 'game:${gameId}')。
绑定 PlayerSession.gameId：playerSessionService.bindGame(pid, gameId)。
ack({ ok: true, data: { roomId, gameId, gameType, playerIds, status } })。

初始事件丢失

GameService.startGame 期间 engine.start() 派发的初始事件（GAME_STARTED / 初始发牌等）会被 publisher 广播到 game:${gameId} 房间，但此时玩家 socket 尚未加入该房间——这些初始事件会丢失。客户端通过 game:started + game:state 重建初始视图。后续由 Action 触发的事件都会正确送达（socket 已 join）。

GameGateway

游戏内 Socket 事件网关。

GameGatewayDeps

interface GameGatewayDeps {
  gameService: GameService
  playerSessionService: PlayerSessionService
  broadcaster: BroadcastPort
}

GameAck

interface GameAck<T = unknown> {
  ok: boolean
  code?: string
  message?: string
  data?: T
}

类

class GameGateway {
  constructor(deps: GameGatewayDeps)
  registerHandlers(socket: Socket): void
}

处理事件

事件	payload	说明
`game:action`	`ClientActionRequest`	执行玩家 Action
`game:sync`	`{ gameId }`	拉取当前 `PublicGameState`
`game:leave`	`{ gameId }`	主动离开对局

game:action 行为

校验 payload.gameId / payload.actionId / payload.action.type 存在。
playerId 来自 socket.data.playerId——客户端 payload 中即使带 playerId 也会被丢弃。
gameService.executeAction(gameId, playerId, actionId, { type, payload })。
ack({ ok: result.success, data: result })——失败时 ok: false，但 data 仍含完整 ActionResult（含 error.code）。

game:sync 行为

校验 gameId 存在。
gameService.sync(gameId, playerId)。
ack({ ok: true, data: sync })。

客户端重连后调用 game:sync 拿到完整状态以重建本地视图，避免仅依赖可能丢失的增量 game:event。

game:leave 行为

校验 gameId 存在。
playerSessionService.unbindGame(playerId) 清空 PlayerSession.gameId。
socket.leave('game:${gameId}')。
broadcaster.broadcastToGame(gameId, 'game:playerLeft', { gameId, playerId })。
ack({ ok: true, data: { gameId, playerId } })。

与断线不同：主动离开会清空 PlayerSession.gameId，并让 socket 离开 game 房间。GameSession 本身不会被销毁（其他玩家仍在进行），仅是本玩家不再接收事件。

GameEventPublisher

把引擎 EventBus 上的事件转发为 Socket 事件。

类

class GameEventPublisher {
  constructor(
    gameId: string,
    gameType: string,
    eventSerializer: GameEventSerializer,
    broadcaster: BroadcastPort,
    logger: Logger
  )
  attach(engine: GameEngine): () => void
}

attach()

attach(engine: GameEngine): () => void

订阅引擎所有事件（通过 engine.subscribeAll）；返回取消订阅函数。

行为

引擎在 createGame / start / dispatch / 链式 dispatch 期间同步派发事件，本方法将这些事件实时转发给 socket 客户端。
每个事件经 eventSerializer.serialize(event, gameType) 脱敏：
- 返回 null 表示抑制广播（如斗地主 DDZ_CARDS_DEALT 的底牌信息）。
- 否则填充 gameId 后通过 broadcaster.broadcastToGame(gameId, 'game:event', serialized) 广播到 game:${gameId} 房间。
序列化失败不阻断 dispatch 流程；记录错误后继续。

一会话一 publisher

GameEventPublisher 设计为独立组件：它不参与游戏规则、Action 验证或 GameState 修改。一个 GameSession 对应一个 GameEventPublisher 实例，在 GameService.startGame 中创建并 attach(session.engine)，在 removeGame 中调用返回的 unsubscribe() 清理。

BroadcastHub

实现 BroadcastPort 接口的延迟绑定转发器。

类

class BroadcastHub implements BroadcastPort {
  setTarget(target: BroadcastPort): void
  broadcastToGame(gameId: string, event: string, payload: unknown): void
  broadcastToRoom(roomId: string, event: string, payload: unknown): void
  sendToPlayer(playerId: string, event: string, payload: unknown): void
  isPlayerConnected(playerId: string): boolean
  joinPlayerToRoom(playerId: string, room: string): void
  leavePlayerFromRoom(playerId: string, room: string): void
}

设计目的：打破循环依赖

GameService 需要 BroadcastPort（构造期）
SocketServer 实现 BroadcastPort（但需要 GameService 来处理 socket 事件）

解决：先构造 BroadcastHub 作为占位 BroadcastPort 注入 GameService；待 SocketServer 构造完毕后调用 broadcastHub.setTarget(socketServer) 把转发目标指向它。

行为

setTarget(target) 之前：所有方法变为 no-op（isPlayerConnected 返回 false）。
setTarget(target) 之后：所有方法转发到 target。
测试时可不调用 setTarget，或注入自定义 target。

事件类型

SocketGameEvent

interface SocketGameEvent {
  gameId: string
  stateVersion: number
  type: string
  payload: unknown
  playerId?: string
}

game:event 事件体（已脱敏）。gameId 由 GameEventPublisher 在转发时填充。

SocketConnectionEvent

interface SocketConnectionEvent {
  gameId?: string
  roomId?: string
  playerId: string
  connected: boolean
}

player:connection 事件体（不带 socketId，仅 playerId + connected）。由 PlayerSessionService.broadcastConnection 在 connect / disconnect 时发出。

BroadcastPort

interface BroadcastPort {
  broadcastToGame(gameId: string, event: string, payload: unknown): void
  broadcastToRoom(roomId: string, event: string, payload: unknown): void
  sendToPlayer(playerId: string, event: string, payload: unknown): void
  isPlayerConnected(playerId: string): boolean
  joinPlayerToRoom(playerId: string, room: string): void
  leavePlayerFromRoom(playerId: string, room: string): void
}

Application 层通过此接口向 Socket 层推送事件，避免直接依赖 Socket.IO。实现位于 SocketServer；测试时可注入 mock 实现。

Socket 事件清单

客户端 → 服务器

事件	payload	ack
`room:join`	`{ roomId }`	`RoomAck<{ roomId, gameType, ownerId, playerIds, maxPlayers, minPlayers, status, gameSessionId }>`
`room:leave`	`{ roomId }`	`RoomAck<{ roomId, playerIds, status }>`
`room:ready`	`{ roomId, ready? }`	`RoomAck<{ roomId, playerId, ready }>`
`room:start`	`{ roomId }`	`RoomAck<{ roomId, gameId, gameType, playerIds, status }>`
`game:action`	`ClientActionRequest`	`GameAck<ActionResult>`
`game:sync`	`{ gameId }`	`GameAck<GameSyncResponse>`
`game:leave`	`{ gameId }`	`GameAck<{ gameId, playerId }>`

服务器 → 客户端

事件	payload	触发场景
`game:started`	`{ gameId, roomId, gameType, playerIds, stateVersion }`	`GameService.startGame` 时广播到 room
`game:state`	`{ gameId, stateVersion, state }`	每次 Action 成功后向每位玩家推送 per-player view
`game:event`	`SocketGameEvent`	引擎事件经脱敏后实时广播到 `game:${gameId}`
`game:finished`	`{ gameId, roomId, stateVersion, winnerIds }`	`GameService.finishGame` 时广播
`game:playerLeft`	`{ gameId, playerId }`	玩家主动 `game:leave`
`game:error`	`{ scope?, event?, code, message }`	Gateway 捕获异常时通知调用方
`room:playerJoined`	`{ roomId, playerId, playerIds, currentPlayers, maxPlayers }`	玩家加入房间
`room:playerLeft`	`{ roomId, playerId, playerIds, currentPlayers }`	玩家离开房间
`room:playerReady`	`{ roomId, playerId, ready }`	玩家就绪状态切换
`player:connection`	`SocketConnectionEvent`	玩家 connect / disconnect

示例

客户端连接与加入房间

// 客户端（伪代码）
import { io } from 'socket.io-client'

const socket = io('http://localhost:3000', {
  auth: { playerId: 'p1' }   // 或通过 X-Player-Id header
})

// 加入房间
socket.emit('room:join', { roomId: 'room_xxx' }, (ack) => {
  console.log(ack.ok, ack.data?.playerIds)
})

// 房主开始游戏
socket.emit('room:start', { roomId: 'room_xxx' }, (ack) => {
  console.log(ack.ok, ack.data?.gameId)
})

// 监听服务器事件
socket.on('game:started', (payload) => {
  console.log('game started:', payload.gameId)
})
socket.on('game:state', (payload) => {
  console.log('my state version:', payload.stateVersion)
  console.log('my hand:', payload.state.myHand)
})
socket.on('game:event', (event) => {
  console.log('event:', event.type, event.payload)
})

执行玩家 Action

socket.emit(
  'game:action',
  {
    gameId: 'game_xxx',
    actionId: 'client_abc123',       // 客户端生成的幂等键
    action: {
      type: 'PLAY_CARD',
      payload: { cardIds: ['<cardId>'] }
    }
  },
  (ack) => {
    if (!ack.ok) {
      console.log('action rejected:', ack.data?.error)
    } else {
      console.log('action ok, stateVersion:', ack.data?.stateVersion)
    }
  }
)

服务器端自定义 BroadcastPort（测试用）

import { createServer } from 'decklet/server'

const runtime = createServer()

// BroadcastHub 已被 setTarget 到 SocketServer；
// 测试时可注入自定义 target
runtime.broadcastHub.setTarget({
  broadcastToGame(gameId, event, payload) {
    console.log('[mock] broadcastToGame', gameId, event, payload)
  },
  broadcastToRoom(roomId, event, payload) {
    console.log('[mock] broadcastToRoom', roomId, event, payload)
  },
  sendToPlayer(playerId, event, payload) {
    console.log('[mock] sendToPlayer', playerId, event, payload)
  },
  isPlayerConnected() { return true },
  joinPlayerToRoom() {},
  leavePlayerFromRoom() {}
})

注意事项

playerId 必须从 socket.data.playerId 注入（authMiddleware 已校验），严禁从客户端 payload 接收 playerId。
game:action 的 ack 在 ok: false 时仍返回 data: ActionResult（含 error.code / error.message），便于客户端区分业务级失败与系统级异常。
game:start 期间初始事件丢失是已知设计——客户端通过 game:started + game:state 重建初始视图。
SocketServer.playerSockets 与 PlayerSessionService 的 PlayerSession 状态并行：SocketServer 关心路由（playerId → socket），PlayerSessionService 关心完整状态（含 roomId / gameId / connected / lastConnectedAt）。
GameEventPublisher 序列化失败不阻断 dispatch 流程——记录错误后继续，避免影响其他玩家。
BroadcastHub 在 setTarget 之前所有方法为 no-op，这避免了构造期的循环依赖问题。

WebSocket ​

概述 ​

SocketServer ​

SocketServerDeps ​

SocketServerOptions ​

类 ​

行为 ​

handleConnection 流程 ​

handleDisconnect 流程 ​

BroadcastPort 实现 ​

close() ​

RoomGateway ​

RoomGatewayDeps ​

RoomAck ​

类 ​

处理事件 ​

room:join 行为 ​

room:leave 行为 ​

room:ready 行为 ​

room:start 行为 ​

GameGateway ​

GameGatewayDeps ​

GameAck ​

类 ​

处理事件 ​

game:action 行为 ​

game:sync 行为 ​

game:leave 行为 ​

GameEventPublisher ​

类 ​

attach() ​

行为 ​

BroadcastHub ​

类 ​

设计目的：打破循环依赖 ​

行为 ​

事件类型 ​

SocketGameEvent ​

SocketConnectionEvent ​

BroadcastPort ​

Socket 事件清单 ​

客户端 → 服务器 ​

服务器 → 客户端 ​

示例 ​

客户端连接与加入房间 ​

执行玩家 Action ​

服务器端自定义 BroadcastPort（测试用） ​

注意事项 ​

WebSocket

概述

SocketServer

SocketServerDeps

SocketServerOptions

类

行为

handleConnection 流程

handleDisconnect 流程

BroadcastPort 实现

close()

RoomGateway

RoomGatewayDeps

RoomAck

类

处理事件

room:join 行为

room:leave 行为

room:ready 行为

room:start 行为

GameGateway

GameGatewayDeps

GameAck

类

处理事件

game:action 行为

game:sync 行为

game:leave 行为

GameEventPublisher

类

attach()

行为

BroadcastHub

类

设计目的：打破循环依赖

行为

事件类型

SocketGameEvent

SocketConnectionEvent

BroadcastPort

Socket 事件清单

客户端 → 服务器

服务器 → 客户端

示例

客户端连接与加入房间

执行玩家 Action

服务器端自定义 BroadcastPort（测试用）

注意事项