Files
whale-town-end/src/business/zulip/zulip.service.ts
moyin e282c9dd16 service:完善Zulip服务的连接管理和错误处理
- 增强WebSocket连接状态监控
- 优化错误处理和重连机制
- 完善服务层的日志记录
- 提升连接稳定性和可靠性

支持远程WebSocket连接的服务层改进
2026-01-05 11:14:22 +08:00

802 lines
22 KiB
TypeScript
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
/**
* Zulip集成主服务
*
* 功能描述:
* - 作为Zulip集成系统的主要协调服务
* - 整合各个子服务,提供统一的业务接口
* - 处理游戏客户端与Zulip之间的核心业务逻辑
*
* 主要方法:
* - handlePlayerLogin(): 处理玩家登录和Zulip客户端初始化
* - handlePlayerLogout(): 处理玩家登出和资源清理
* - sendChatMessage(): 处理游戏聊天消息发送到Zulip
* - processZulipMessage(): 处理从Zulip接收的消息
*
* 使用场景:
* - WebSocket网关调用处理消息路由
* - 会话管理和状态维护
* - 消息格式转换和过滤
*
* @author angjustinl
* @version 1.0.0
* @since 2025-12-25
*/
import { Injectable, Logger, Inject } from '@nestjs/common';
import { randomUUID } from 'crypto';
import { SessionManagerService } from './services/session_manager.service';
import { MessageFilterService } from './services/message_filter.service';
import { ZulipEventProcessorService } from './services/zulip_event_processor.service';
import {
IZulipClientPoolService,
IZulipConfigService,
} from '../../core/zulip/interfaces/zulip-core.interfaces';
/**
* 玩家登录请求接口
*/
export interface PlayerLoginRequest {
token: string;
socketId: string;
}
/**
* 聊天消息请求接口
*/
export interface ChatMessageRequest {
socketId: string;
content: string;
scope: string;
}
/**
* 位置更新请求接口
*/
export interface PositionUpdateRequest {
socketId: string;
x: number;
y: number;
mapId: string;
}
/**
* 登录响应接口
*/
export interface LoginResponse {
success: boolean;
sessionId?: string;
userId?: string;
username?: string;
currentMap?: string;
error?: string;
}
/**
* 聊天消息响应接口
*/
export interface ChatMessageResponse {
success: boolean;
messageId?: number | string;
error?: string;
}
/**
* Zulip集成主服务类
*
* 职责:
* - 作为Zulip集成系统的主要协调服务
* - 整合各个子服务,提供统一的业务接口
* - 处理游戏客户端与Zulip之间的核心业务逻辑
* - 管理玩家会话和消息路由
*
* 主要方法:
* - handlePlayerLogin(): 处理玩家登录和Zulip客户端初始化
* - handlePlayerLogout(): 处理玩家登出和资源清理
* - sendChatMessage(): 处理游戏聊天消息发送到Zulip
* - updatePlayerPosition(): 更新玩家位置信息
*
* 使用场景:
* - WebSocket网关调用处理消息路由
* - 会话管理和状态维护
* - 消息格式转换和过滤
* - 游戏与Zulip的双向通信桥梁
*/
@Injectable()
export class ZulipService {
private readonly logger = new Logger(ZulipService.name);
private readonly DEFAULT_MAP = 'whale_port';
constructor(
@Inject('ZULIP_CLIENT_POOL_SERVICE')
private readonly zulipClientPool: IZulipClientPoolService,
private readonly sessionManager: SessionManagerService,
private readonly messageFilter: MessageFilterService,
private readonly eventProcessor: ZulipEventProcessorService,
@Inject('ZULIP_CONFIG_SERVICE')
private readonly configManager: IZulipConfigService,
) {
this.logger.log('ZulipService初始化完成');
// 启动事件处理
this.initializeEventProcessing();
}
/**
* 处理玩家登录
*
* 功能描述:
* 验证游戏Token创建Zulip客户端建立会话映射关系
*
* 业务逻辑:
* 1. 验证游戏Token的有效性
* 2. 获取用户的Zulip API Key
* 3. 创建用户专用的Zulip客户端实例
* 4. 注册Zulip事件队列
* 5. 建立Socket_ID与Zulip_Queue_ID的映射关系
* 6. 返回登录成功确认
*
* @param request 玩家登录请求数据
* @returns Promise<LoginResponse>
*
* @throws UnauthorizedException 当Token验证失败时
* @throws InternalServerErrorException 当系统操作失败时
*/
async handlePlayerLogin(request: PlayerLoginRequest): Promise<LoginResponse> {
const startTime = Date.now();
this.logger.log('开始处理玩家登录', {
operation: 'handlePlayerLogin',
socketId: request.socketId,
timestamp: new Date().toISOString(),
});
try {
// 1. 验证请求参数
if (!request.token || !request.token.trim()) {
this.logger.warn('登录失败Token为空', {
operation: 'handlePlayerLogin',
socketId: request.socketId,
});
return {
success: false,
error: 'Token不能为空',
};
}
if (!request.socketId || !request.socketId.trim()) {
this.logger.warn('登录失败socketId为空', {
operation: 'handlePlayerLogin',
});
return {
success: false,
error: 'socketId不能为空',
};
}
// 2. 验证游戏Token并获取用户信息
// TODO: 实际项目中应该调用认证服务验证Token
// 这里暂时使用模拟数据
const userInfo = await this.validateGameToken(request.token);
if (!userInfo) {
this.logger.warn('登录失败Token验证失败', {
operation: 'handlePlayerLogin',
socketId: request.socketId,
});
return {
success: false,
error: 'Token验证失败',
};
}
// 3. 生成会话ID
const sessionId = randomUUID();
// 调试日志:检查用户信息
this.logger.log('用户信息检查', {
operation: 'handlePlayerLogin',
userId: userInfo.userId,
hasZulipApiKey: !!userInfo.zulipApiKey,
zulipApiKeyLength: userInfo.zulipApiKey?.length || 0,
zulipEmail: userInfo.zulipEmail,
email: userInfo.email,
});
// 4. 创建Zulip客户端如果有API Key
let zulipQueueId = `queue_${sessionId}`;
if (userInfo.zulipApiKey) {
try {
const zulipConfig = this.configManager.getZulipConfig();
const clientInstance = await this.zulipClientPool.createUserClient(userInfo.userId, {
username: userInfo.zulipEmail || userInfo.email,
apiKey: userInfo.zulipApiKey,
realm: zulipConfig.zulipServerUrl,
});
if (clientInstance.queueId) {
zulipQueueId = clientInstance.queueId;
}
this.logger.log('Zulip客户端创建成功', {
operation: 'handlePlayerLogin',
userId: userInfo.userId,
queueId: zulipQueueId,
});
} catch (zulipError) {
const err = zulipError as Error;
this.logger.warn('Zulip客户端创建失败使用本地模式', {
operation: 'handlePlayerLogin',
userId: userInfo.userId,
error: err.message,
});
// Zulip客户端创建失败不影响登录使用本地模式
}
}
// 5. 创建游戏会话
const session = await this.sessionManager.createSession(
request.socketId,
userInfo.userId,
zulipQueueId,
userInfo.username,
this.DEFAULT_MAP,
{ x: 400, y: 300 },
);
const duration = Date.now() - startTime;
this.logger.log('玩家登录处理完成', {
operation: 'handlePlayerLogin',
socketId: request.socketId,
sessionId,
userId: userInfo.userId,
username: userInfo.username,
currentMap: session.currentMap,
duration,
timestamp: new Date().toISOString(),
});
return {
success: true,
sessionId,
userId: userInfo.userId,
username: userInfo.username,
currentMap: session.currentMap,
};
} catch (error) {
const err = error as Error;
const duration = Date.now() - startTime;
this.logger.error('玩家登录处理失败', {
operation: 'handlePlayerLogin',
socketId: request.socketId,
error: err.message,
duration,
timestamp: new Date().toISOString(),
}, err.stack);
return {
success: false,
error: '登录失败,请稍后重试',
};
}
}
/**
* 验证游戏Token
*
* 功能描述:
* 验证游戏Token的有效性返回用户信息
*
* @param token 游戏Token
* @returns Promise<UserInfo | null> 用户信息验证失败返回null
* @private
*/
private async validateGameToken(token: string): Promise<{
userId: string;
username: string;
email: string;
zulipEmail?: string;
zulipApiKey?: string;
} | null> {
// TODO: 实际项目中应该调用认证服务验证Token (登录godot所获取的JWT token)
// 这里暂时使用模拟数据进行开发测试
this.logger.debug('验证游戏Token', {
operation: 'validateGameToken',
tokenLength: token.length,
});
// 模拟Token验证
// 实际实现应该:
// 1. 调用LoginService验证Token
// 2. 从数据库获取用户的Zulip API Key
// 3. 返回完整的用户信息
if (token.startsWith('invalid')) {
return null;
}
// 从Token中提取用户ID模拟
const userId = `user_${token.substring(0, 8)}`;
// 为测试用户提供真实的 Zulip API Key
let zulipApiKey = undefined;
let zulipEmail = undefined;
// 检查是否是配置了真实 Zulip API Key 的测试用户
const hasTestApiKey = token.includes('lCPWCPf');
const hasUserApiKey = token.includes('W2KhXaQx');
const hasOldApiKey = token.includes('MZ1jEMQo');
const isRealUserToken = token === 'real_user_token_with_zulip_key_123';
this.logger.log('Token检查', {
operation: 'validateGameToken',
userId,
tokenPrefix: token.substring(0, 20),
hasUserApiKey,
hasOldApiKey,
isRealUserToken,
});
if (isRealUserToken || hasUserApiKey || hasTestApiKey || hasOldApiKey) {
// 使用用户的真实 API Key
// 注意这个API Key对应的Zulip用户邮箱是 user8@zulip.xinghangee.icu
zulipApiKey = 'lCPWCPfGh7WUHxwN56GF8oYXOpqNfGF8';
zulipEmail = 'angjustinl@mail.angforever.top';
this.logger.log('配置真实Zulip API Key', {
operation: 'validateGameToken',
userId,
zulipEmail,
hasApiKey: true,
});
}
return {
userId,
username: `Player_${userId.substring(5, 10)}`,
email: `${userId}@example.com`,
// 实际项目中从数据库获取
zulipEmail,
zulipApiKey,
};
}
/**
* 处理玩家登出
*
* 功能描述:
* 清理玩家会话注销Zulip事件队列释放相关资源
*
* 业务逻辑:
* 1. 获取会话信息
* 2. 注销Zulip事件队列
* 3. 清理Zulip客户端实例
* 4. 删除会话映射关系
* 5. 记录登出日志
*
* @param socketId WebSocket连接ID
* @returns Promise<void>
*/
async handlePlayerLogout(socketId: string): Promise<void> {
const startTime = Date.now();
this.logger.log('开始处理玩家登出', {
operation: 'handlePlayerLogout',
socketId,
timestamp: new Date().toISOString(),
});
try {
// 1. 获取会话信息
const session = await this.sessionManager.getSession(socketId);
if (!session) {
this.logger.log('会话不存在,跳过登出处理', {
operation: 'handlePlayerLogout',
socketId,
});
return;
}
// 2. 清理Zulip客户端资源
if (session.userId) {
try {
await this.zulipClientPool.destroyUserClient(session.userId);
this.logger.log('Zulip客户端清理完成', {
operation: 'handlePlayerLogout',
userId: session.userId,
});
} catch (zulipError) {
const err = zulipError as Error;
this.logger.warn('Zulip客户端清理失败', {
operation: 'handlePlayerLogout',
userId: session.userId,
error: err.message,
});
// 继续执行会话清理
}
}
// 3. 删除会话映射
await this.sessionManager.destroySession(socketId);
const duration = Date.now() - startTime;
this.logger.log('玩家登出处理完成', {
operation: 'handlePlayerLogout',
socketId,
userId: session.userId,
duration,
timestamp: new Date().toISOString(),
});
} catch (error) {
const err = error as Error;
const duration = Date.now() - startTime;
this.logger.error('玩家登出处理失败', {
operation: 'handlePlayerLogout',
socketId,
error: err.message,
duration,
timestamp: new Date().toISOString(),
}, err.stack);
// 登出失败不抛出异常,确保连接能够正常断开
}
}
/**
* 处理聊天消息发送
*
* 功能描述:
* 处理游戏客户端发送的聊天消息转发到对应的Zulip Stream/Topic
*
* 业务逻辑:
* 1. 获取玩家当前位置和会话信息
* 2. 根据位置确定目标Stream和Topic
* 3. 进行消息内容过滤和频率检查
* 4. 使用玩家的Zulip客户端发送消息
* 5. 返回发送结果确认
*
* @param request 聊天消息请求数据
* @returns Promise<ChatMessageResponse>
*/
async sendChatMessage(request: ChatMessageRequest): Promise<ChatMessageResponse> {
const startTime = Date.now();
this.logger.log('开始处理聊天消息发送', {
operation: 'sendChatMessage',
socketId: request.socketId,
contentLength: request.content.length,
scope: request.scope,
timestamp: new Date().toISOString(),
});
try {
// 1. 获取会话信息
const session = await this.sessionManager.getSession(request.socketId);
if (!session) {
this.logger.warn('发送消息失败:会话不存在', {
operation: 'sendChatMessage',
socketId: request.socketId,
});
return {
success: false,
error: '会话不存在,请重新登录',
};
}
// 2. 上下文注入根据位置确定目标Stream
const context = await this.sessionManager.injectContext(request.socketId);
const targetStream = context.stream;
const targetTopic = context.topic || 'General';
// 3. 消息验证(内容过滤、频率限制、权限验证)
const validationResult = await this.messageFilter.validateMessage(
session.userId,
request.content,
targetStream,
session.currentMap,
);
if (!validationResult.allowed) {
this.logger.warn('消息验证失败', {
operation: 'sendChatMessage',
socketId: request.socketId,
userId: session.userId,
reason: validationResult.reason,
});
return {
success: false,
error: validationResult.reason || '消息发送失败',
};
}
// 使用过滤后的内容(如果有)
const messageContent = validationResult.filteredContent || request.content;
// 4. 发送消息到Zulip
const sendResult = await this.zulipClientPool.sendMessage(
session.userId,
targetStream,
targetTopic,
messageContent,
);
if (!sendResult.success) {
// Zulip发送失败记录日志但不影响本地消息显示
this.logger.warn('Zulip消息发送失败使用本地模式', {
operation: 'sendChatMessage',
socketId: request.socketId,
userId: session.userId,
error: sendResult.error,
});
// 即使Zulip发送失败也返回成功本地模式
// 实际项目中可以根据需求决定是否返回失败
}
const duration = Date.now() - startTime;
this.logger.log('聊天消息发送完成', {
operation: 'sendChatMessage',
socketId: request.socketId,
userId: session.userId,
targetStream,
targetTopic,
zulipSuccess: sendResult.success,
messageId: sendResult.messageId,
duration,
timestamp: new Date().toISOString(),
});
return {
success: true,
messageId: sendResult.messageId,
};
} catch (error) {
const err = error as Error;
const duration = Date.now() - startTime;
this.logger.error('聊天消息发送失败', {
operation: 'sendChatMessage',
socketId: request.socketId,
error: err.message,
duration,
timestamp: new Date().toISOString(),
}, err.stack);
return {
success: false,
error: '消息发送失败,请稍后重试',
};
}
}
/**
* 更新玩家位置
*
* 功能描述:
* 更新玩家在游戏世界中的位置信息,用于消息路由和上下文注入
*
* @param request 位置更新请求数据
* @returns Promise<boolean> 是否更新成功
*/
async updatePlayerPosition(request: PositionUpdateRequest): Promise<boolean> {
this.logger.debug('更新玩家位置', {
operation: 'updatePlayerPosition',
socketId: request.socketId,
mapId: request.mapId,
position: { x: request.x, y: request.y },
timestamp: new Date().toISOString(),
});
try {
// 验证参数
if (!request.socketId || !request.socketId.trim()) {
this.logger.warn('更新位置失败socketId为空', {
operation: 'updatePlayerPosition',
});
return false;
}
if (!request.mapId || !request.mapId.trim()) {
this.logger.warn('更新位置失败mapId为空', {
operation: 'updatePlayerPosition',
socketId: request.socketId,
});
return false;
}
// 调用SessionManager更新位置信息
const result = await this.sessionManager.updatePlayerPosition(
request.socketId,
request.mapId,
request.x,
request.y,
);
if (result) {
this.logger.debug('玩家位置更新成功', {
operation: 'updatePlayerPosition',
socketId: request.socketId,
mapId: request.mapId,
});
}
return result;
} catch (error) {
const err = error as Error;
this.logger.error('更新玩家位置失败', {
operation: 'updatePlayerPosition',
socketId: request.socketId,
error: err.message,
timestamp: new Date().toISOString(),
}, err.stack);
return false;
}
}
/**
* 处理从Zulip接收的消息
*
* 功能描述:
* 处理Zulip事件队列推送的消息转换格式后发送给相关的游戏客户端
*
* @param zulipMessage Zulip消息对象
* @returns Promise<{targetSockets: string[], message: any}>
*/
async processZulipMessage(zulipMessage: any): Promise<{
targetSockets: string[];
message: {
t: string;
from: string;
txt: string;
bubble: boolean;
};
}> {
this.logger.debug('处理Zulip消息', {
operation: 'processZulipMessage',
messageId: zulipMessage.id,
stream: zulipMessage.stream_id,
sender: zulipMessage.sender_email,
timestamp: new Date().toISOString(),
});
try {
// 1. 根据Stream确定目标地图
const streamName = zulipMessage.display_recipient || zulipMessage.stream_name;
const mapId = this.configManager.getMapIdByStream(streamName);
if (!mapId) {
this.logger.debug('未找到Stream对应的地图', {
operation: 'processZulipMessage',
streamName,
});
return {
targetSockets: [],
message: {
t: 'chat_render',
from: zulipMessage.sender_full_name || 'Unknown',
txt: zulipMessage.content || '',
bubble: true,
},
};
}
// 2. 获取目标地图中的所有玩家Socket
const targetSockets = await this.sessionManager.getSocketsInMap(mapId);
// 3. 转换消息格式为游戏协议
const gameMessage = {
t: 'chat_render' as const,
from: zulipMessage.sender_full_name || 'Unknown',
txt: zulipMessage.content || '',
bubble: true,
};
this.logger.log('Zulip消息处理完成', {
operation: 'processZulipMessage',
messageId: zulipMessage.id,
mapId,
targetCount: targetSockets.length,
});
return {
targetSockets,
message: gameMessage,
};
} catch (error) {
const err = error as Error;
this.logger.error('处理Zulip消息失败', {
operation: 'processZulipMessage',
messageId: zulipMessage.id,
error: err.message,
timestamp: new Date().toISOString(),
}, err.stack);
return {
targetSockets: [],
message: {
t: 'chat_render',
from: 'System',
txt: '',
bubble: false,
},
};
}
}
/**
* 获取会话信息
*
* 功能描述:
* 根据socketId获取会话信息
*
* @param socketId WebSocket连接ID
* @returns Promise<GameSession | null>
*/
async getSession(socketId: string) {
return this.sessionManager.getSession(socketId);
}
/**
* 获取地图中的所有Socket
*
* 功能描述:
* 获取指定地图中所有在线玩家的Socket ID列表
*
* @param mapId 地图ID
* @returns Promise<string[]>
*/
async getSocketsInMap(mapId: string): Promise<string[]> {
return this.sessionManager.getSocketsInMap(mapId);
}
/**
* 获取事件处理器实例
*
* 功能描述:
* 返回ZulipEventProcessorService实例用于设置消息分发器
*
* @returns ZulipEventProcessorService 事件处理器实例
*/
getEventProcessor(): ZulipEventProcessorService {
return this.eventProcessor;
}
/**
* 初始化事件处理
*
* 功能描述:
* 启动Zulip事件处理循环用于接收和处理从Zulip服务器返回的消息
*
* @private
*/
private async initializeEventProcessing(): Promise<void> {
try {
this.logger.log('开始初始化Zulip事件处理');
// 启动事件处理循环
await this.eventProcessor.startEventProcessing();
this.logger.log('Zulip事件处理初始化完成');
} catch (error) {
const err = error as Error;
this.logger.error('初始化Zulip事件处理失败', {
operation: 'initializeEventProcessing',
error: err.message,
}, err.stack);
}
}
}