From 2da34dfc2c70a8dcd20d65b4f8f37a40d8b0d86f Mon Sep 17 00:00:00 2001 From: Pine Date: Thu, 3 Sep 2026 10:11:56 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20MQTT=20=E4=B8=80=E6=AC=A1=E6=80=A7?= =?UTF-8?q?=E5=AF=86=E7=A0=81=E9=80=82=E9=85=8D=20-=20=E7=A6=81=E7=94=A8?= =?UTF-8?q?=E8=87=AA=E5=8A=A8=E9=87=8D=E8=BF=9E=EF=BC=8C=E6=96=AD=E5=BC=80?= =?UTF-8?q?=E5=90=8E=E9=87=8D=E6=96=B0=E8=8E=B7=E5=8F=96=E5=87=AD=E8=AF=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. reconnectPeriod: 0 禁用 mqtt.js 自动重连（旧密码已失效） 2. close 事件：延迟后重新调用 connect() 获取新凭证 3. error 事件：强制断开并重新获取凭证 4. 确保每次重连都使用新的一次性密码 --- console/src/services/chatSocket.ts | 242 +++++++++++++++++++++++++++++ 1 file changed, 242 insertions(+) create mode 100644 console/src/services/chatSocket.ts diff --git a/console/src/services/chatSocket.ts b/console/src/services/chatSocket.ts new file mode 100644 index 0000000..8d78cbe --- /dev/null +++ b/console/src/services/chatSocket.ts @@ -0,0 +1,242 @@ +/** + * chatSocket.ts — IM 实时通道（PineSound 方案：桌面端直连 MQTT broker）。 + * + * 连接流程： + * 1. GET /api/im/mqtt/credentials 获取 broker WS 地址、username、password、client_id + * 2. mqtt.connect(broker_ws_url, { username, password, clientId, will }) + * 3. 订阅： + * - {prefix}/+/+/out/{userId} 所有频道发给自己的消息 + * - {prefix}/notify/{userId} 系统通知（好友申请/工单/@提及） + * - {prefix}/presence/{userId} 自己的上下线确认 + * 4. 发送消息：publish 到 {prefix}/{type}/{conversation_id}/in + * + * 对外接口与旧 WebSocket 版保持一致（on/send/connect/disconnect/status）， + * communicationStore 和沟通页面无需改动。 + */ +import mqtt, { type MqttClient } from "mqtt"; +import { getApiToken, getApiUrl } from "../api/config"; +import { listConversations } from "../api/modules/communication"; + +type WsHandler = (data: Record) => void; + +interface MqttCredentials { + broker_ws_url: string; + username: string; + password: string; + client_id: string; + topic_prefix: string; + presence_topic: string; +} + +const RECONNECT_DELAY_MS = 3000; + +class ChatSocket { + private client: MqttClient | null = null; + private creds: MqttCredentials | null = null; + private userId = ""; + private convTypeMap = new Map(); // conversation_id → type + private stopped = false; + private handlers = new Map>(); + status: "idle" | "connecting" | "open" | "closed" = "idle"; + onStatusChange: ((s: ChatSocket["status"]) => void) | null = null; + + /** 获取 MQTT 凭证（经 server-core /im/mqtt/credentials 转发）。 */ + private async fetchCredentials(): Promise { + try { + const resp = await fetch(getApiUrl("/im/mqtt/credentials"), { + headers: { Authorization: `Bearer ${getApiToken()}` }, + }); + if (!resp.ok) return null; + return (await resp.json()) as MqttCredentials; + } catch { + return null; + } + } + + /** 加载会话列表，建立 conversation_id → type 映射（发送消息时需要）。 */ + private async loadConvTypes(): Promise { + try { + const resp = await listConversations(); + for (const c of resp.items || []) { + this.convTypeMap.set(c.conversation_id, c.type); + } + } catch { + /* ignore */ + } + } + + connect(): void { + if (this.stopped) return; + this.setStatus("connecting"); + void this.fetchCredentials().then((creds) => { + if (!creds) { + // EMQX 未配置或凭证接口不可用，延迟重试 + setTimeout(() => this.connect(), RECONNECT_DELAY_MS * 2); + return; + } + this.creds = creds; + this.userId = creds.username; + void this.loadConvTypes().then(() => this.open()); + }); + } + + private open(): void { + if (this.stopped || !this.creds) return; + const { broker_ws_url, username, password, client_id, topic_prefix, presence_topic } = this.creds; + + try { + const client = mqtt.connect(broker_ws_url, { + username, + password, + clientId: client_id, + clean: true, + reconnectPeriod: 0, // 禁用自动重连：密码是一次性的，重连必须重新获取凭证 + connectTimeout: 10000, + // Will Message：断线时自动发布离线 + will: { + topic: presence_topic, + payload: JSON.stringify({ status: "offline", user_id: this.userId }), + qos: 1, + retain: false, + }, + }); + this.client = client; + + client.on("connect", () => { + this.setStatus("open"); + // 订阅所有发给自己的消息 + 系统通知 + const msgTopic = `${topic_prefix}/+/+/out/${this.userId}`; + const notifyTopic = `${topic_prefix}/notify/${this.userId}`; + client.subscribe([msgTopic, notifyTopic], { qos: 1 }); + // 上线通知 + client.publish(presence_topic, JSON.stringify({ status: "online", user_id: this.userId }), { qos: 1 }); + this.emit("ready", { user_id: this.userId }); + }); + + client.on("message", (topic, payload) => { + try { + const data = JSON.parse(payload.toString()) as { + type?: string; + data?: Record; + }; + // 频道消息：topic 形如 {prefix}/{type}/{conv_id}/out/{user_id} + if (topic.includes("/out/")) { + if (data.type === "message" && data.data) { + this.emit("message", data.data); + } else if (data.type) { + this.emit(data.type, data.data ?? {}); + } + return; + } + // 系统通知：{prefix}/notify/{user_id}，payload 直接是事件 + if (topic.includes("/notify/")) { + if (data.type) { + this.emit(data.type, data.data ?? {}); + } + return; + } + } catch { + /* ignore malformed */ + } + }); + + client.on("reconnect", () => this.setStatus("connecting")); + client.on("close", () => { + this.setStatus("closed"); + // 断开后重新获取凭证（一次性密码），延迟重连 + if (!this.stopped) { + setTimeout(() => this.connect(), RECONNECT_DELAY_MS); + } + }); + client.on("error", (err) => { + console.warn("[chatSocket] MQTT error:", err.message); + // 认证失败或其他错误时，重新获取凭证并重连 + if (!this.stopped) { + this.client?.end(true); + this.client = null; + setTimeout(() => this.connect(), RECONNECT_DELAY_MS); + } + }); + } catch (err) { + console.warn("[chatSocket] MQTT connect failed:", err); + setTimeout(() => this.connect(), RECONNECT_DELAY_MS); + } + } + + /** 发送消息（PineSound 方案：publish 到频道 server_listen_topic）。 */ + send(payload: Record): boolean { + if (!this.client || !this.creds || this.status !== "open") return false; + + const action = payload.action as string | undefined; + if (action === "send") { + const convId = String(payload.conversation_id || ""); + const type = this.convTypeMap.get(convId) || "group"; + const topic = `${this.creds.topic_prefix}/${type}/${convId}/in`; + const body = { + sender_id: this.userId, + content: payload.content, + msg_type: payload.msg_type || "text", + metadata: payload.metadata || {}, + client_msg_id: payload.client_msg_id || `c_${Date.now()}`, + timestamp: Date.now(), + }; + this.client.publish(topic, JSON.stringify(body), { qos: 1 }); + return true; + } + + if (action === "read") { + // read 走 REST（已在 store 中处理），MQTT 不处理 + return true; + } + + if (action === "ping") { + // MQTT 有原生 keepalive，无需应用层 ping + return true; + } + + return false; + } + + /** 外部告知新会话的类型（创建会话后调用，使 send 能正确路由）。 */ + registerConversation(convId: string, type: string): void { + this.convTypeMap.set(convId, type); + } + + /** 订阅事件：message / ready / friend_request / friend_accepted / mention / cs_ticket / system */ + on(event: string, handler: WsHandler): () => void { + if (!this.handlers.has(event)) this.handlers.set(event, new Set()); + this.handlers.get(event)!.add(handler); + return () => this.handlers.get(event)?.delete(handler); + } + + private emit(event: string, data: Record): void { + this.handlers.get(event)?.forEach((h) => { + try { + h(data); + } catch { + /* handler error ignored */ + } + }); + } + + private setStatus(s: ChatSocket["status"]): void { + this.status = s; + this.onStatusChange?.(s); + } + + disconnect(): void { + this.stopped = true; + if (this.creds) { + this.client?.publish( + this.creds.presence_topic, + JSON.stringify({ status: "offline", user_id: this.userId }), + { qos: 1 }, + ); + } + this.client?.end(true); + this.client = null; + this.setStatus("closed"); + } +} + +export const chatSocket = new ChatSocket();