/** * 聊天 WebSocket 处理器 * * 事件: * - onConnected(userId) * - onTyping(isTyping) * - onSentence(index, text, total, done) ← AI 逐句推送 * - onVoiceReady(index, audioUrl) * - onError(errorMessage) */ import { getApiConfig } from "@/core/net/config/api_config"; import { ExceptionHandler } from "@/core/errors"; import { AppEnvUtil } from "@/utils/app-env"; import { Logger } from "@/utils/logger"; import { CommercialActionSchema, ChatActionSchema, type ChatAction, type CommercialAction, } from "@/data/schemas/chat"; import { createClientMessageId } from "@/lib/chat/client_message_id"; const log = new Logger("ChatWebSocket"); const RECONNECT_DELAY_MS = 3000; export interface SentencePayload { index: number; text: string; total: number; done: boolean; } export interface PaywallStatusPayload { locked: boolean; reason: string | null; } export class ChatWebSocket { private ws: WebSocket | null = null; private reconnectTimer: ReturnType | null = null; private connectionUrl: string | null = null; private disposed = false; onConnected: ((userId: string) => void) | null = null; onTyping: ((isTyping: boolean) => void) | null = null; onSentence: ((payload: SentencePayload) => void) | null = null; onVoiceReady: ((index: number, audioUrl: string) => void) | null = null; onImage: ((url: string) => void) | null = null; onPaywallStatus: ((payload: PaywallStatusPayload) => void) | null = null; onCommercialAction: ((action: CommercialAction) => void) | null = null; onChatAction: ((action: ChatAction) => void) | null = null; onError: ((errorMessage: string) => void) | null = null; constructor( private readonly serverUrl: string, private readonly token: string, ) {} get isConnected(): boolean { return this.ws?.readyState === WebSocket.OPEN; } connect(): void { if (this.disposed) return; if (this.ws && this.ws.readyState !== WebSocket.CLOSED) return; try { const url = buildChatWebSocketUrl(this.serverUrl, this.token); this.connectionUrl = url; logWebSocketDebug(`↔ WS CONNECT ${redactWebSocketUrl(url)}`); this.ws = new WebSocket(url); this.ws.onopen = () => { logWebSocketDebug(`↔ WS OPEN ${redactWebSocketUrl(url)}`); // 连接成功;userId 由后端在第一条消息中下发 }; this.ws.onmessage = (e) => { logWebSocketFrame("← WS MESSAGE", url, e.data); this.handleMessage(e.data); }; this.ws.onerror = () => { logWebSocketError(`✕ WS ERROR ${redactWebSocketUrl(url)}`); this.onError?.("WebSocket error"); }; this.ws.onclose = (event) => { logWebSocketDebug(`↔ WS CLOSE ${redactWebSocketUrl(url)}`, { code: event.code, reason: event.reason, wasClean: event.wasClean, reconnectInMs: this.disposed ? null : RECONNECT_DELAY_MS, }); if (!this.disposed) { this.reconnectTimer = setTimeout( () => this.connect(), RECONNECT_DELAY_MS, ); } }; } catch (e) { logWebSocketError("✕ WS CONNECT FAILED", e); this.onError?.( ExceptionHandler.message(e, "WebSocket connection failed."), ); } } sendMessage( content: string, clientMessageId: string = createClientMessageId(), ): boolean { const payload = { type: "message", content, clientMessageId }; if (!this.isConnected) { logWebSocketWarn("→ WS SEND SKIPPED: socket is not open", payload); return false; } logWebSocketFrame( "→ WS SEND", this.connectionUrl ?? this.serverUrl, payload, ); this.ws?.send(JSON.stringify(payload)); return true; } disconnect(): void { this.disposed = true; if (this.reconnectTimer) { clearTimeout(this.reconnectTimer); this.reconnectTimer = null; } this.ws?.close(); this.ws = null; this.connectionUrl = null; } private handleMessage(data: unknown): void { if (typeof data !== "string") { logWebSocketWarn("← WS MESSAGE IGNORED: non-string payload", data); return; } let payload: { type?: string; userId?: string; isTyping?: boolean; index?: number; text?: string; total?: number; done?: boolean; audioUrl?: string; error?: string; data?: { image?: { type?: string | null; url?: string | null; }; lockDetail?: Partial; actionId?: string; type?: string; copy?: string; ctaLabel?: string; target?: string; ruleId?: string; kind?: string; orderId?: string | null; }; }; try { payload = JSON.parse(data); } catch (e) { logWebSocketError("← WS MESSAGE PARSE FAILED", e); return; } switch (payload.type) { case "connected": if (payload.userId) this.onConnected?.(payload.userId); break; case "typing": this.onTyping?.(payload.isTyping ?? false); break; case "sentence": this.onSentence?.({ index: payload.index ?? 0, text: payload.text ?? "", total: payload.total ?? 0, done: payload.done ?? false, }); break; case "voice_ready": this.onVoiceReady?.(payload.index ?? 0, payload.audioUrl ?? ""); break; case "image": { const url = payload.data?.image?.url; if (url) this.onImage?.(url); break; } case "paywall_status": { const lockDetail = payload.data?.lockDetail; this.onPaywallStatus?.({ locked: lockDetail?.locked ?? false, reason: lockDetail?.reason ?? null, }); break; } case "commercial_action": { const action = CommercialActionSchema.safeParse(payload.data); if (action.success) this.onCommercialAction?.(action.data); break; } case "chat_action": { const action = ChatActionSchema.safeParse(payload.data); if (action.success) this.onChatAction?.(action.data); break; } case "error": this.onError?.(payload.error ?? "Unknown error"); break; } } } function buildChatWebSocketUrl(serverUrl: string, token: string): string { if (AppEnvUtil.isDevelopment()) return serverUrl; return `${serverUrl}?token=${encodeURIComponent(token)}`; } function redactWebSocketUrl(url: string): string { try { const parsed = new URL(url); if (parsed.searchParams.has("token")) { parsed.searchParams.set("token", ""); } return parsed.toString(); } catch { return url.replace(/([?&]token=)[^&]+/i, "$1"); } } function logWebSocketFrame( label: string, url: string, payload: unknown, ): void { if (AppEnvUtil.isProduction()) return; const body = formatWebSocketLogValue(payload); const bodyLabel = label.startsWith("→") ? "request body" : "response body"; log.debug( `${label} ${redactWebSocketUrl(url)}${body ? `\n${bodyLabel}:\n${body}` : ""}`, ); } function logWebSocketDebug(message: string, payload?: unknown): void { if (AppEnvUtil.isProduction()) return; const body = formatWebSocketLogValue(payload); log.debug(`${message}${body ? `\nbody:\n${body}` : ""}`); } function logWebSocketWarn(message: string, payload?: unknown): void { if (AppEnvUtil.isProduction()) return; const body = formatWebSocketLogValue(payload); log.warn(`${message}${body ? `\nbody:\n${body}` : ""}`); } function logWebSocketError(message: string, payload?: unknown): void { if (AppEnvUtil.isProduction()) return; const body = formatWebSocketLogValue(payload); log.error(`${message}${body ? `\nbody:\n${body}` : ""}`); } function formatWebSocketLogValue(payload: unknown): string { if (payload instanceof Error) { return Logger.formatValue({ message: payload.message, stack: payload.stack, }); } return Logger.formatValue(payload); } /** 默认创建实例:从 env 读取 WS URL。 */ export function createChatWebSocket(token: string): ChatWebSocket { const base = getApiConfig().wsUrl; return new ChatWebSocket(base, token); }