fix: MQTT 一次性密码适配 - 禁用自动重连，断开后重新获取凭证

1. reconnectPeriod: 0 禁用 mqtt.js 自动重连（旧密码已失效） 2. close 事件：延迟后重新调用 connect() 获取新凭证 3. error 事件：强制断开并重新获取凭证 4. 确保每次重连都使用新的一次性密码
2026-09-03 10:11:56 +08:00
parent 1403d5b83a
commit 2da34dfc2c
1 changed files with 242 additions and 0 deletions
@@ -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<string, unknown>) => 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<string, string>(); // conversation_id → type
+  private stopped = false;
+  private handlers = new Map<string, Set<WsHandler>>();
+  status: "idle" | "connecting" | "open" | "closed" = "idle";
+  onStatusChange: ((s: ChatSocket["status"]) => void) | null = null;
+
+  /** 获取 MQTT 凭证（经 server-core /im/mqtt/credentials 转发）。 */
+  private async fetchCredentials(): Promise<MqttCredentials | null> {
+    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<void> {
+    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<string, unknown>;
+          };
+          // 频道消息：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<string, unknown>): 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<string, unknown>): 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();