fix(mqtt): persist topics and support advanced options

apps/desktop/src/components/connection/ConnectionDialog.vue

@@ -710,14 +710,18 @@ const mqttClientId = ref("");
 const mqttProtocolVersion = ref<"v3" | "v4" | "v5">("v5");
 const mqttTransportMode = ref<"tcp" | "websocket">("tcp");
 const mqttWsPath = ref("/mqtt");
-const mqttAuthKind = ref<"none" | "password">("none");
+const mqttAuthKind = ref<"none" | "password" | "certificate">("none");
 const mqttUsername = ref("");
 const mqttPassword = ref("");
+const mqttCaCertPath = ref("");
+const mqttClientCertPath = ref("");
+const mqttClientKeyPath = ref("");
 const mqttTls = ref(false);
 const mqttTlsSkipVerify = ref(false);
 const mqttKeepAliveSecs = ref(60);
 const mqttConnectTimeoutSecs = ref(30);
 const mqttMaxPacketSizeBytes = ref(16 * 1024 * 1024);
+const mqttSavedTopics = ref<MqttConnectionConfig["savedTopics"]>([]);
 const nacosPrimaryAddressPlaceholder = computed(() => {
   return "http://127.0.0.1:8848/nacos";
 });
@@ -1231,15 +1235,29 @@ function resetMqttFields(config?: Partial<MqttConnectionConfig>) {
   mqttKeepAliveSecs.value = Math.max(1, config?.keepAliveSecs || 60);
   mqttConnectTimeoutSecs.value = Math.max(1, config?.connectTimeoutSecs || 30);
   mqttMaxPacketSizeBytes.value = Math.min(268435455, Math.max(1024, config?.maxPacketSizeBytes || 16 * 1024 * 1024));
+  mqttSavedTopics.value = config?.savedTopics ?? [];
   const auth = config?.auth;
   if (auth && auth.kind === "password") {
     mqttAuthKind.value = "password";
     mqttUsername.value = auth.username || "";
     mqttPassword.value = auth.password || "";
+    mqttCaCertPath.value = "";
+    mqttClientCertPath.value = "";
+    mqttClientKeyPath.value = "";
+  } else if (auth && auth.kind === "certificate") {
+    mqttAuthKind.value = "certificate";
+    mqttUsername.value = "";
+    mqttPassword.value = "";
+    mqttCaCertPath.value = auth.caCertPath || "";
+    mqttClientCertPath.value = auth.clientCertPath || "";
+    mqttClientKeyPath.value = auth.clientKeyPath || "";
   } else {
     mqttAuthKind.value = "none";
     mqttUsername.value = "";
     mqttPassword.value = "";
+    mqttCaCertPath.value = "";
+    mqttClientCertPath.value = "";
+    mqttClientKeyPath.value = "";
   }
 }
 
@@ -1252,7 +1270,12 @@ function hydrateMqttFields(value: unknown) {
 }
 
 function buildMqttExternalConfig(): MqttConnectionConfig {
-  const auth: MqttConnectionConfig["auth"] = mqttAuthKind.value === "password" ? { kind: "password", username: mqttUsername.value, password: mqttPassword.value } : { kind: "none" };
+  const auth: MqttConnectionConfig["auth"] =
+    mqttAuthKind.value === "password"
+      ? { kind: "password", username: mqttUsername.value, password: mqttPassword.value }
+      : mqttAuthKind.value === "certificate"
+        ? { kind: "certificate", caCertPath: mqttCaCertPath.value || undefined, clientCertPath: mqttClientCertPath.value || undefined, clientKeyPath: mqttClientKeyPath.value || undefined }
+        : { kind: "none" };
 
   return {
     host: mqttHost.value.trim(),
@@ -1266,6 +1289,7 @@ function buildMqttExternalConfig(): MqttConnectionConfig {
     keepAliveSecs: Math.max(1, mqttKeepAliveSecs.value),
     connectTimeoutSecs: Math.max(1, mqttConnectTimeoutSecs.value),
     maxPacketSizeBytes: Math.min(268435455, Math.max(1024, mqttMaxPacketSizeBytes.value)),
+    savedTopics: mqttSavedTopics.value,
     wsPath: mqttTransportMode.value === "websocket" ? mqttWsPath.value || "/mqtt" : undefined,
   };
 }
@@ -6050,6 +6074,7 @@ function openExternalUrl(url: string) {
                     <div class="col-span-3 flex gap-2">
                       <Button size="sm" :variant="mqttAuthKind === 'none' ? 'default' : 'outline'" @click="mqttAuthKind = 'none'">{{ t("connection.mqAuthNone") }}</Button>
                       <Button size="sm" :variant="mqttAuthKind === 'password' ? 'default' : 'outline'" @click="mqttAuthKind = 'password'">{{ t("connection.mqAuthBasic") }}</Button>
+                      <Button size="sm" :variant="mqttAuthKind === 'certificate' ? 'default' : 'outline'" @click="mqttAuthKind = 'certificate'">{{ t("connection.mqttAuthCertificate") }}</Button>
                     </div>
                   </div>
                   <template v-if="mqttAuthKind === 'password'">
@@ -6062,6 +6087,20 @@ function openExternalUrl(url: string) {
                       <Input v-model="mqttPassword" type="password" class="col-span-3" :placeholder="t('connection.mqttPasswordPlaceholder')" />
                     </div>
                   </template>
+                  <template v-else-if="mqttAuthKind === 'certificate'">
+                    <div class="grid grid-cols-4 items-center gap-4">
+                      <Label :class="connectionLabelClass">{{ t("connection.mqttCaCertPath") }}</Label>
+                      <Input v-model="mqttCaCertPath" class="col-span-3" placeholder="/path/to/ca.pem" />
+                    </div>
+                    <div class="grid grid-cols-4 items-center gap-4">
+                      <Label :class="connectionLabelClass">{{ t("connection.mqttClientCertPath") }}</Label>
+                      <Input v-model="mqttClientCertPath" class="col-span-3" placeholder="/path/to/client.crt" />
+                    </div>
+                    <div class="grid grid-cols-4 items-center gap-4">
+                      <Label :class="connectionLabelClass">{{ t("connection.mqttClientKeyPath") }}</Label>
+                      <Input v-model="mqttClientKeyPath" class="col-span-3" placeholder="/path/to/client.key" />
+                    </div>
+                  </template>
                   <div class="grid grid-cols-4 items-center gap-4">
                     <Label :class="connectionLabelClass">{{ t("connection.mqttTls") }}</Label>
                     <div class="col-span-3 flex items-center gap-2">

apps/desktop/src/components/mqtt/MqttAdminConsole.vue

@@ -20,6 +20,7 @@ const brokerInfo = ref<MqttBrokerInfo | null>(null);
 const topicTree = ref<MqttTopicNode | null>(null);
 const messages = ref<MqttMessage[]>([]);
 const subscribedTopics = ref<[string, string][]>([]);
+const noLocalSubscribe = ref(false);
 const selectedTopic = ref<string>(props.initialTopic ?? "");
 const loading = ref(true);
 const error = ref<string | null>(null);
@@ -52,9 +53,9 @@ async function refreshData() {
   }
 }
 
-async function handleSubscribe(topic: string) {
+async function handleSubscribe(topic: string, noLocal = noLocalSubscribe.value) {
   try {
-    await mqttSubscribe(props.connectionId, topic, "atmostonce");
+    await mqttSubscribe(props.connectionId, topic, "atmostonce", noLocal);
     await refreshData();
   } catch (e) {
     error.value = String(e);
@@ -194,6 +195,7 @@ onUnmounted(() => {
             "
           >
             <input class="flex-1 h-7 px-2 text-xs rounded border bg-transparent" :placeholder="t('connection.mqttSubscribePlaceholder')" />
+            <label class="flex items-center gap-1 text-[10px] whitespace-nowrap"><input v-model="noLocalSubscribe" type="checkbox" />{{ t("connection.mqttNoLocal") }}</label>
             <Button size="sm" variant="ghost" class="h-7 text-xs" type="submit">{{ t("connection.mqttSubscribe") }}</Button>
           </form>
         </div>
@@ -216,13 +218,13 @@ onUnmounted(() => {
               </select>
             </div>
           </div>
-          <div class="flex-1 overflow-auto">
+          <div class="flex-1 overflow-auto flex flex-col">
             <div v-if="messages.length === 0" class="text-xs text-muted-foreground p-4 text-center">{{ t("connection.mqttNoMessages") }}</div>
             <div
               v-for="(msg, i) in messages"
               :key="i"
-              class="px-3 py-2 border-b last:border-b-0 cursor-pointer text-xs"
-              :class="msg.direction === 'sent' ? 'bg-emerald-50/60 dark:bg-emerald-950/20 hover:bg-emerald-100/60 dark:hover:bg-emerald-950/40' : 'hover:bg-muted/50'"
+              class="px-3 py-2 border-b last:border-b-0 cursor-pointer text-xs self-start w-full"
+              :class="msg.direction === 'sent' ? 'bg-emerald-50/60 dark:bg-emerald-950/20 hover:bg-emerald-100/60 dark:hover:bg-emerald-950/40 ml-auto max-w-[85%] rounded-l-md' : 'hover:bg-muted/50 mr-auto max-w-[85%] rounded-r-md'"
               @click="handleTopicClick(msg.topic)"
             >
               <div class="flex items-center gap-2 mb-0.5">

apps/desktop/src/components/mqtt/MqttPublishDialog.vue

@@ -32,7 +32,7 @@ const qosHints = computed(() => [t("connection.mqttQosAtMostOnce"), t("connectio
 const payloadPlaceholder = computed(() => {
   switch (encoding.value) {
     case "json":
-      return t("connection.mqttPayloadPlaceholderJson");
+      return t("connection.mqttPayloadPlaceholderJson", { example: '{"key": "value"}' });
     case "base64":
       return t("connection.mqttPayloadPlaceholderBase64");
     case "hex":

apps/desktop/src/i18n/locales/en.ts

@@ -526,6 +526,10 @@ export default {
     mqttUsername: "Username",
     mqttUsernamePlaceholder: "Optional",
     mqttPassword: "Password",
+    mqttAuthCertificate: "Client certificate",
+    mqttCaCertPath: "CA certificate path",
+    mqttClientCertPath: "Client certificate path",
+    mqttClientKeyPath: "Client private key path",
     mqttPasswordPlaceholder: "Optional",
     mqttTls: "TLS",
     mqttTlsSkipVerify: "Skip certificate verification",
@@ -542,6 +546,7 @@ export default {
     mqttMessages: "Messages",
     mqttPublish: "Publish Message",
     mqttSubscribe: "Subscribe",
+    mqttNoLocal: "No Local",
     mqttUnsubscribe: "Unsubscribe",
     mqttNoMessages: "No messages",
     mqttNoTopics: "No subscribed topics",
@@ -571,7 +576,7 @@ export default {
     mqttPayloadEncoding: "Payload encoding",
     mqttPayload: "Payload",
     mqttPayloadPlaceholderPlaintext: "Enter message content...",
-    mqttPayloadPlaceholderJson: 'Enter JSON, for example {"key": "value"}',
+    mqttPayloadPlaceholderJson: "Enter JSON, for example {example}",
     mqttPayloadPlaceholderBase64: "Enter Base64-encoded data",
     mqttPayloadPlaceholderHex: "Enter hexadecimal data, for example 48 65 6C 6C 6F",
     mqttPayloadPlaceholderStructured: "Enter a JSON value; it will be encoded as {encoding}",

apps/desktop/src/i18n/locales/es.ts

@@ -792,7 +792,7 @@ export default withEnglishFallback({
     mqttPayloadEncoding: "Codificación del Payload",
     mqttPayload: "Payload",
     mqttPayloadPlaceholderPlaintext: "Introduce el contenido del mensaje...",
-    mqttPayloadPlaceholderJson: 'Introduce JSON, por ejemplo {"key": "value"}',
+    mqttPayloadPlaceholderJson: "Introduce JSON, por ejemplo {example}",
     mqttPayloadPlaceholderBase64: "Introduce datos codificados en Base64",
     mqttPayloadPlaceholderHex: "Introduce datos hexadecimales, por ejemplo 48 65 6C 6C 6F",
     mqttPayloadPlaceholderStructured: "Introduce un valor JSON; se codificará como {encoding}",
@@ -812,6 +812,11 @@ export default withEnglishFallback({
     zookeeperAuthMethod: "Método de autenticación",
     zookeeperAuthDigest: "Digest (predeterminado)",
     zookeeperAuthSaslDigest: "SASL DIGEST-MD5",
+    mqttAuthCertificate: "Certificado de cliente",
+    mqttCaCertPath: "Ruta del certificado CA",
+    mqttClientCertPath: "Ruta del certificado de cliente",
+    mqttClientKeyPath: "Ruta de la clave privada del cliente",
+    mqttNoLocal: "Prohibir reenvío local",
   },
   editor: {
     duckdbDraining: "La consulta anterior de DuckDB aún se está deteniendo. Vuelva a intentarlo en breve.",

apps/desktop/src/i18n/locales/it.ts

@@ -790,7 +790,7 @@ export default withEnglishFallback({
     mqttPayloadEncoding: "Codifica del Payload",
     mqttPayload: "Payload",
     mqttPayloadPlaceholderPlaintext: "Inserisci il contenuto del messaggio...",
-    mqttPayloadPlaceholderJson: 'Inserisci JSON, ad esempio {"key": "value"}',
+    mqttPayloadPlaceholderJson: "Inserisci JSON, ad esempio {example}",
     mqttPayloadPlaceholderBase64: "Inserisci dati codificati in Base64",
     mqttPayloadPlaceholderHex: "Inserisci dati esadecimali, ad esempio 48 65 6C 6C 6F",
     mqttPayloadPlaceholderStructured: "Inserisci un valore JSON; verrà codificato come {encoding}",
@@ -810,6 +810,11 @@ export default withEnglishFallback({
     zookeeperAuthMethod: "Metodo di autenticazione",
     zookeeperAuthDigest: "Digest (predefinito)",
     zookeeperAuthSaslDigest: "SASL DIGEST-MD5",
+    mqttAuthCertificate: "Certificato client",
+    mqttCaCertPath: "Percorso del certificato CA",
+    mqttClientCertPath: "Percorso del certificato client",
+    mqttClientKeyPath: "Percorso della chiave privata client",
+    mqttNoLocal: "Disabilita inoltro locale",
   },
   editor: {
     duckdbDraining: "La query DuckDB precedente è ancora in fase di arresto. Riprova a breve.",

apps/desktop/src/i18n/locales/ja.ts

@@ -790,7 +790,7 @@ export default withEnglishFallback({
     mqttPayloadEncoding: "ペイロードのエンコード",
     mqttPayload: "ペイロード",
     mqttPayloadPlaceholderPlaintext: "メッセージ内容を入力...",
-    mqttPayloadPlaceholderJson: 'JSON を入力（例: {"key": "value"}）',
+    mqttPayloadPlaceholderJson: "JSON を入力（例: {example}）",
     mqttPayloadPlaceholderBase64: "Base64 エンコード済みデータを入力",
     mqttPayloadPlaceholderHex: "16 進数データを入力（例: 48 65 6C 6C 6F）",
     mqttPayloadPlaceholderStructured: "JSON 値を入力してください。{encoding} としてエンコードされます",
@@ -831,6 +831,11 @@ export default withEnglishFallback({
     zookeeperAuthMethod: "認証方式",
     zookeeperAuthDigest: "Digest（デフォルト）",
     zookeeperAuthSaslDigest: "SASL DIGEST-MD5",
+    mqttAuthCertificate: "クライアント証明書",
+    mqttCaCertPath: "CA証明書パス",
+    mqttClientCertPath: "クライアント証明書パス",
+    mqttClientKeyPath: "クライアント秘密鍵パス",
+    mqttNoLocal: "ローカル転送禁止",
   },
   editor: {
     duckdbDraining: "前回の DuckDB クエリはまだ停止処理中です。しばらくしてから再試行してください。",

apps/desktop/src/i18n/locales/pt-BR.ts

@@ -791,7 +791,7 @@ export default withEnglishFallback({
     mqttPayloadEncoding: "Codificação do Payload",
     mqttPayload: "Payload",
     mqttPayloadPlaceholderPlaintext: "Digite o conteúdo da mensagem...",
-    mqttPayloadPlaceholderJson: 'Digite JSON, por exemplo {"key": "value"}',
+    mqttPayloadPlaceholderJson: "Digite JSON, por exemplo {example}",
     mqttPayloadPlaceholderBase64: "Digite dados codificados em Base64",
     mqttPayloadPlaceholderHex: "Digite dados hexadecimais, por exemplo 48 65 6C 6C 6F",
     mqttPayloadPlaceholderStructured: "Digite um valor JSON; ele será codificado como {encoding}",
@@ -811,6 +811,11 @@ export default withEnglishFallback({
     zookeeperAuthMethod: "Método de autenticação",
     zookeeperAuthDigest: "Digest (padrão)",
     zookeeperAuthSaslDigest: "SASL DIGEST-MD5",
+    mqttAuthCertificate: "Certificado do cliente",
+    mqttCaCertPath: "Caminho do certificado CA",
+    mqttClientCertPath: "Caminho do certificado do cliente",
+    mqttClientKeyPath: "Caminho da chave privada do cliente",
+    mqttNoLocal: "Proibir encaminhamento local",
   },
   editor: {
     duckdbDraining: "A consulta anterior do DuckDB ainda está sendo interrompida. Tente novamente em breve.",

apps/desktop/src/i18n/locales/zh-CN.ts

@@ -527,6 +527,10 @@ export default withEnglishFallback({
     mqttUsername: "用户名",
     mqttUsernamePlaceholder: "选填",
     mqttPassword: "密码",
+    mqttAuthCertificate: "客户端证书",
+    mqttCaCertPath: "CA 证书路径",
+    mqttClientCertPath: "客户端证书路径",
+    mqttClientKeyPath: "客户端私钥路径",
     mqttPasswordPlaceholder: "选填",
     mqttTls: "TLS",
     mqttTlsSkipVerify: "跳过证书校验",
@@ -543,6 +547,7 @@ export default withEnglishFallback({
     mqttMessages: "消息",
     mqttPublish: "发布消息",
     mqttSubscribe: "订阅",
+    mqttNoLocal: "禁止本地转发",
     mqttUnsubscribe: "取消订阅",
     mqttNoMessages: "暂无消息",
     mqttNoTopics: "暂无已订阅的 Topic",
@@ -572,7 +577,7 @@ export default withEnglishFallback({
     mqttPayloadEncoding: "Payload 编码",
     mqttPayload: "消息内容 (Payload)",
     mqttPayloadPlaceholderPlaintext: "输入消息内容...",
-    mqttPayloadPlaceholderJson: '输入 JSON，例如 {"key": "value"}',
+    mqttPayloadPlaceholderJson: "输入 JSON，例如 {example}",
     mqttPayloadPlaceholderBase64: "输入 Base64 编码的数据",
     mqttPayloadPlaceholderHex: "输入十六进制数据，例如 48 65 6C 6C 6F",
     mqttPayloadPlaceholderStructured: "输入 JSON 值，发送时将自动编码为 {encoding}",

apps/desktop/src/i18n/locales/zh-TW.ts

@@ -790,7 +790,7 @@ export default withEnglishFallback({
     mqttPayloadEncoding: "Payload 編碼",
     mqttPayload: "訊息內容 (Payload)",
     mqttPayloadPlaceholderPlaintext: "輸入訊息內容...",
-    mqttPayloadPlaceholderJson: '輸入 JSON，例如 {"key": "value"}',
+    mqttPayloadPlaceholderJson: "輸入 JSON，例如 {example}",
     mqttPayloadPlaceholderBase64: "輸入 Base64 編碼的資料",
     mqttPayloadPlaceholderHex: "輸入十六進位資料，例如 48 65 6C 6C 6F",
     mqttPayloadPlaceholderStructured: "輸入 JSON 值，傳送時將自動編碼為 {encoding}",
@@ -810,6 +810,11 @@ export default withEnglishFallback({
     zookeeperAuthMethod: "認證方式",
     zookeeperAuthDigest: "Digest（預設）",
     zookeeperAuthSaslDigest: "SASL DIGEST-MD5",
+    mqttAuthCertificate: "用戶端憑證",
+    mqttCaCertPath: "CA 憑證路徑",
+    mqttClientCertPath: "用戶端憑證路徑",
+    mqttClientKeyPath: "用戶端私鑰路徑",
+    mqttNoLocal: "禁止本機轉送",
   },
   editor: {
     duckdbDraining: "上一筆 DuckDB 查詢仍在停止，請稍後重試。",

apps/desktop/src/lib/backend/mqtt-http.ts

@@ -9,7 +9,7 @@ export async function mqttGetBrokerInfo(_connectionId: string) {
   mqttWebNotAvailable("mqttGetBrokerInfo");
 }
 
-export async function mqttSubscribe(_connectionId: string, _topic: string, _qos?: string | null) {
+export async function mqttSubscribe(_connectionId: string, _topic: string, _qos?: string | null, _noLocal?: boolean) {
   mqttWebNotAvailable("mqttSubscribe");
 }
 

apps/desktop/src/lib/backend/mqtt-tauri.ts

@@ -4,8 +4,8 @@ export async function mqttGetBrokerInfo(connectionId: string) {
   return invoke("mqtt_get_broker_info", { connectionId });
 }
 
-export async function mqttSubscribe(connectionId: string, topic: string, qos?: string | null) {
-  return invoke("mqtt_subscribe", { connectionId, topic, qos: qos ?? null });
+export async function mqttSubscribe(connectionId: string, topic: string, qos?: string | null, noLocal?: boolean) {
+  return invoke("mqtt_subscribe", { connectionId, topic, qos: qos ?? null, noLocal: noLocal ?? false });
 }
 
 export async function mqttUnsubscribe(connectionId: string, topic: string) {

apps/desktop/src/types/mqtt.ts

@@ -26,6 +26,13 @@ export interface MqttConnectionConfig {
   connectTimeoutSecs: number;
   maxPacketSizeBytes: number;
   wsPath?: string;
+  savedTopics?: MqttSavedTopic[];
+}
+
+export interface MqttSavedTopic {
+  topic: string;
+  qos: MqttQoS;
+  noLocal?: boolean;
 }
 
 export interface MqttBrokerInfo {

crates/dbx-core/src/mqtt/client.rs

@@ -1,6 +1,8 @@
 //! MQTT 客户端封装（基于 rumqttc），管理 MQTT broker 的长连接和消息收发。
 
 use std::collections::{HashMap, HashSet, VecDeque};
+use std::fs::File;
+use std::io::BufReader;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Weak};
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
@@ -12,8 +14,8 @@ use rumqttc::tokio_rustls::rustls::{
     pki_types::{CertificateDer, ServerName, UnixTime},
 };
 use rumqttc::v5::{
-    mqttbytes::QoS as QoSV5, AsyncClient as AsyncClientV5, Event as EventV5, Incoming as IncomingV5,
-    MqttOptions as MqttOptionsV5,
+    mqttbytes::{v5::Filter as MqttV5Filter, QoS as QoSV5},
+    AsyncClient as AsyncClientV5, Event as EventV5, Incoming as IncomingV5, MqttOptions as MqttOptionsV5,
 };
 use rumqttc::{
     mqttbytes::Protocol, valid_filter, valid_topic, AsyncClient as AsyncClientV4, Event as EventV4,
@@ -25,7 +27,7 @@ use tokio::task::JoinHandle;
 
 use super::types::{
     MqttAuth, MqttBrokerInfo, MqttConnectionConfig, MqttMessage, MqttMessageDirection, MqttPublishRequest, MqttQoS,
-    MqttTlsVerificationMode, MqttTopicNode, MqttTransport,
+    MqttSavedTopic, MqttTlsVerificationMode, MqttTopicNode, MqttTransport,
 };
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -56,6 +58,7 @@ struct PendingSubscribe {
     sequence: u64,
     topic: String,
     qos: MqttQoS,
+    no_local: bool,
     add_to_desired: bool,
     completion: Option<SubscriptionCompletion>,
 }
@@ -308,6 +311,7 @@ pub struct MqttClient {
     granted_subscriptions: RwLock<Vec<(String, MqttQoS)>>,
     /// 需要在无会话重连后恢复的 topic 集合
     desired_subscriptions: RwLock<Vec<(String, MqttQoS)>>,
+    no_local_topics: RwLock<HashSet<String>>,
     /// 当前连接是否已收到有效 CONNACK
     connected: AtomicBool,
     /// 订阅请求与 packet id 的关联状态
@@ -337,7 +341,7 @@ impl MqttClient {
         if !(1024..=268_435_455).contains(&config.max_packet_size_bytes) {
             return Err("MQTT 最大报文大小必须在 1024 到 268435455 字节之间".to_string());
         }
-        let plan = build_connect_plan(&config);
+        let plan = build_connect_plan(&config)?;
         match plan.backend {
             MqttBackendKind::V4 => Self::connect_v4(config, plan).await,
             MqttBackendKind::V5 => Self::connect_v5(config, plan).await,
@@ -388,12 +392,16 @@ impl MqttClient {
         backend: MqttBackendClient,
         shutdown_notify: Arc<Notify>,
     ) -> Arc<Self> {
+        let desired_subscriptions = config.saved_topics.iter().map(|saved| (saved.topic.clone(), saved.qos)).collect();
+        let no_local_topics =
+            config.saved_topics.iter().filter(|saved| saved.no_local).map(|saved| saved.topic.clone()).collect();
         Arc::new(Self {
             backend,
             config,
             subscriptions: RwLock::new(Vec::new()),
             granted_subscriptions: RwLock::new(Vec::new()),
-            desired_subscriptions: RwLock::new(Vec::new()),
+            desired_subscriptions: RwLock::new(desired_subscriptions),
+            no_local_topics: RwLock::new(no_local_topics),
             connected: AtomicBool::new(false),
             subscription_requests: Mutex::new(SubscriptionRequestTracker::default()),
             subscription_send_lock: Mutex::new(()),
@@ -775,7 +783,8 @@ impl MqttClient {
             if has_pending {
                 continue;
             }
-            if let Err(error) = self.queue_subscribe_request(topic.clone(), qos, false, None).await {
+            let no_local = self.no_local_topics.read().await.contains(&topic);
+            if let Err(error) = self.queue_subscribe_request(topic.clone(), qos, no_local, false, None).await {
                 log::warn!("MQTT 重连后恢复订阅 {topic} 失败: {error}");
             }
         }
@@ -833,6 +842,11 @@ impl MqttClient {
                 upsert_subscription(&self.granted_subscriptions, &pending.topic, granted_qos).await;
                 self.seen_retained.write().await.clear_filter(&pending.topic);
                 if pending.add_to_desired {
+                    if pending.no_local {
+                        self.no_local_topics.write().await.insert(pending.topic.clone());
+                    } else {
+                        self.no_local_topics.write().await.remove(&pending.topic);
+                    }
                     upsert_subscription(&self.desired_subscriptions, &pending.topic, pending.qos).await;
                 }
                 Ok(())
@@ -859,6 +873,7 @@ impl MqttClient {
             remove_subscription(&self.subscriptions, &pending.topic).await;
             remove_subscription(&self.granted_subscriptions, &pending.topic).await;
             remove_subscription(&self.desired_subscriptions, &pending.topic).await;
+            self.no_local_topics.write().await.remove(&pending.topic);
             self.seen_retained.write().await.clear_filter(&pending.topic);
         } else if let Err(error) = &result {
             log::warn!("{error}");
@@ -873,6 +888,7 @@ impl MqttClient {
         &self,
         topic: String,
         qos: MqttQoS,
+        no_local: bool,
         add_to_desired: bool,
         completion: Option<SubscriptionCompletion>,
     ) -> Result<(), String> {
@@ -887,13 +903,14 @@ impl MqttClient {
                 sequence,
                 topic: topic.clone(),
                 qos,
+                no_local,
                 add_to_desired,
                 completion,
             });
             sequence
         };
 
-        if let Err(error) = self.backend.subscribe(&topic, mqtt_qos(qos)).await {
+        if let Err(error) = self.backend.subscribe(&topic, mqtt_qos(qos), no_local).await {
             self.subscription_requests.lock().await.queued_subscribes.retain(|pending| pending.sequence != sequence);
             return Err(error);
         }
@@ -941,15 +958,16 @@ impl MqttClient {
     }
 
     /// 订阅 topic
-    pub async fn subscribe(&self, topic: &str, qos: MqttQoS) -> Result<(), String> {
+    pub async fn subscribe(&self, topic: &str, qos: MqttQoS, no_local: bool) -> Result<(), String> {
         validate_topic_filter(topic)?;
-        if self
+        let has_desired = self
             .desired_subscriptions
             .read()
             .await
             .iter()
-            .any(|(current, current_qos)| current == topic && *current_qos == qos)
-        {
+            .any(|(current, current_qos)| current == topic && *current_qos == qos);
+        let no_local_matches = self.no_local_topics.read().await.contains(topic);
+        if has_desired && no_local_matches == no_local {
             return Ok(());
         }
         if self.subscription_requests.lock().await.has_pending_topic(topic) {
@@ -957,7 +975,7 @@ impl MqttClient {
         }
 
         let (completion, result) = oneshot::channel();
-        self.queue_subscribe_request(topic.to_string(), qos, true, Some(completion)).await?;
+        self.queue_subscribe_request(topic.to_string(), qos, no_local, true, Some(completion)).await?;
         result.await.map_err(|_| "订阅确认通道已关闭，请检查 MQTT 连接状态".to_string())?
     }
 
@@ -1031,6 +1049,19 @@ impl MqttClient {
         self.subscriptions.read().await.clone()
     }
 
+    /// 获取持久化意图中的订阅 Topic，用于保存连接配置。
+    pub async fn desired_topics(&self) -> Vec<(String, MqttQoS)> {
+        self.desired_subscriptions.read().await.clone()
+    }
+
+    pub async fn desired_topic_configs(&self) -> Vec<MqttSavedTopic> {
+        let desired = self.desired_subscriptions.read().await.clone();
+        let no_local = self.no_local_topics.read().await;
+        desired
+            .into_iter()
+            .map(|(topic, qos)| MqttSavedTopic { no_local: no_local.contains(&topic), topic, qos })
+            .collect()
+    }
     /// 获取消息缓冲区中的消息
     pub async fn get_messages(&self, topic_filter: Option<&str>, limit: usize) -> Vec<MqttMessage> {
         let buffer = self.message_buffer.read().await;
@@ -1117,13 +1148,18 @@ impl Drop for MqttClient {
 }
 
 impl MqttBackendClient {
-    async fn subscribe(&self, topic: &str, qos: QoSV4) -> Result<(), String> {
+    async fn subscribe(&self, topic: &str, qos: QoSV4, no_local: bool) -> Result<(), String> {
         match self {
             MqttBackendClient::V4(client) => {
+                if no_local {
+                    return Err("MQTT 3.x 不支持 No Local 订阅选项".to_string());
+                }
                 client.subscribe(topic, qos).await.map_err(|e| format!("订阅 topic 失败: {e}"))
             }
             MqttBackendClient::V5(client) => {
-                client.subscribe(topic, qos_v4_to_v5(qos)).await.map_err(|e| format!("订阅 topic 失败: {e}"))
+                let mut filter = MqttV5Filter::new(topic.to_string(), qos_v4_to_v5(qos));
+                filter.nolocal = no_local;
+                client.subscribe_many([filter]).await.map_err(|e| format!("订阅 topic 失败: {e}"))
             }
         }
     }
@@ -1159,7 +1195,7 @@ impl MqttBackendClient {
     }
 }
 
-fn build_connect_plan(config: &MqttConnectionConfig) -> MqttConnectPlan {
+fn build_connect_plan(config: &MqttConnectionConfig) -> Result<MqttConnectPlan, String> {
     let backend = match config.protocol_version {
         super::types::MqttProtocolVersion::V3 | super::types::MqttProtocolVersion::V4 => MqttBackendKind::V4,
         super::types::MqttProtocolVersion::V5 => MqttBackendKind::V5,
@@ -1170,13 +1206,27 @@ fn build_connect_plan(config: &MqttConnectionConfig) -> MqttConnectPlan {
         super::types::MqttProtocolVersion::V5 => None,
     };
     let tls_verification_mode = config.tls_verification_mode();
-    let transport = build_transport(config.transport, tls_verification_mode);
+    let transport = build_transport(config.transport, tls_verification_mode, &config.auth)?;
 
-    MqttConnectPlan { backend, broker_addr: config.broker_addr_for_transport(), transport, protocol }
+    Ok(MqttConnectPlan { backend, broker_addr: config.broker_addr_for_transport(), transport, protocol })
 }
 
-fn build_transport(transport: MqttTransport, tls_verification_mode: Option<MqttTlsVerificationMode>) -> Transport {
-    match (transport, tls_verification_mode) {
+fn build_transport(
+    transport: MqttTransport,
+    tls_verification_mode: Option<MqttTlsVerificationMode>,
+    auth: &MqttAuth,
+) -> Result<Transport, String> {
+    if let Some(mode) = tls_verification_mode {
+        if matches!(auth, MqttAuth::Certificate { .. }) {
+            let tls = certificate_tls_configuration(auth, mode)?;
+            return Ok(match transport {
+                MqttTransport::Tcp => Transport::tls_with_config(tls),
+                MqttTransport::WebSocket => Transport::wss_with_config(tls),
+            });
+        }
+    }
+
+    Ok(match (transport, tls_verification_mode) {
         (MqttTransport::Tcp, None) => Transport::Tcp,
         (MqttTransport::Tcp, Some(MqttTlsVerificationMode::VerifyServerCert)) => Transport::tls_with_default_config(),
         (MqttTransport::Tcp, Some(MqttTlsVerificationMode::SkipServerCertVerification)) => {
@@ -1189,9 +1239,57 @@ fn build_transport(transport: MqttTransport, tls_verification_mode: Option<MqttT
         (MqttTransport::WebSocket, Some(MqttTlsVerificationMode::SkipServerCertVerification)) => {
             Transport::wss_with_config(insecure_tls_configuration())
         }
-    }
+    })
 }
 
+fn certificate_tls_configuration(auth: &MqttAuth, mode: MqttTlsVerificationMode) -> Result<TlsConfiguration, String> {
+    let MqttAuth::Certificate { ca_cert_path, client_cert_path, client_key_path } = auth else {
+        return Err("MQTT 证书认证配置无效".to_string());
+    };
+    let client_cert_path = client_cert_path.as_deref().ok_or("MQTT 证书认证缺少客户端证书路径")?;
+    let client_key_path = client_key_path.as_deref().ok_or("MQTT 证书认证缺少客户端私钥路径")?;
+
+    let mut root_cert_store = rustls::RootCertStore::empty();
+    if let Some(ca_cert_path) = ca_cert_path.as_deref().filter(|path| !path.trim().is_empty()) {
+        let file = File::open(ca_cert_path).map_err(|e| format!("读取 MQTT CA 证书失败: {e}"))?;
+        let certs = rustls_pemfile::certs(&mut BufReader::new(file))
+            .collect::<Result<Vec<_>, _>>()
+            .map_err(|e| format!("解析 MQTT CA 证书失败: {e}"))?;
+        let (added, _) = root_cert_store.add_parsable_certificates(certs);
+        if added == 0 {
+            return Err("MQTT CA 证书文件不包含有效证书".to_string());
+        }
+    } else {
+        root_cert_store.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
+    }
+
+    let builder = match mode {
+        MqttTlsVerificationMode::VerifyServerCert => {
+            rustls::ClientConfig::builder().with_root_certificates(root_cert_store)
+        }
+        MqttTlsVerificationMode::SkipServerCertVerification => {
+            let provider = Arc::new(rustls::crypto::ring::default_provider());
+            rustls::ClientConfig::builder()
+                .dangerous()
+                .with_custom_certificate_verifier(Arc::new(NoCertificateVerification { provider }))
+        }
+    };
+    let cert_file = File::open(client_cert_path).map_err(|e| format!("读取 MQTT 客户端证书失败: {e}"))?;
+    let certs = rustls_pemfile::certs(&mut BufReader::new(cert_file))
+        .collect::<Result<Vec<_>, _>>()
+        .map_err(|e| format!("解析 MQTT 客户端证书失败: {e}"))?;
+    if certs.is_empty() {
+        return Err("MQTT 客户端证书文件不包含有效证书".to_string());
+    }
+    let key_file = File::open(client_key_path).map_err(|e| format!("读取 MQTT 客户端私钥失败: {e}"))?;
+    let mut key_reader = BufReader::new(key_file);
+    let key = rustls_pemfile::private_key(&mut key_reader)
+        .map_err(|e| format!("解析 MQTT 客户端私钥失败: {e}"))?
+        .ok_or("MQTT 客户端私钥文件不包含有效私钥")?;
+    let config =
+        builder.with_client_auth_cert(certs, key).map_err(|e| format!("构建 MQTT 客户端 TLS 配置失败: {e}"))?;
+    Ok(TlsConfiguration::from(config))
+}
 fn insecure_tls_configuration() -> TlsConfiguration {
     let provider = Arc::new(rustls::crypto::ring::default_provider());
     let config = rustls::ClientConfig::builder()
@@ -1503,7 +1601,7 @@ mod tests {
     }
 
     #[test]
-    fn connect_plan_uses_ws_path_and_insecure_tls_transport() {
+    fn connect_plan_uses_ws_path_and_insecure_tls_transport() -> Result<(), Box<dyn std::error::Error>> {
         let config = MqttConnectionConfig {
             host: "broker.example.com".to_string(),
             port: 8084,
@@ -1514,13 +1612,15 @@ mod tests {
             ..Default::default()
         };
 
-        let plan = build_connect_plan(&config);
+        let plan = build_connect_plan(&config)?;
         assert_eq!(plan.broker_addr, "wss://broker.example.com:8084/custom/mqtt");
         assert_eq!(plan.backend, MqttBackendKind::V5);
         match plan.transport {
             Transport::Wss(TlsConfiguration::Rustls(_)) => {}
             _ => panic!("tlsSkipVerify 应使用显式 Rustls 自定义校验器"),
         }
+
+        Ok(())
     }
 
     #[tokio::test]
@@ -1572,7 +1672,7 @@ mod tests {
         let config = mqtt_config(port, crate::mqtt::types::MqttProtocolVersion::V4);
         let client = MqttClient::connect(config).await.unwrap();
 
-        client.subscribe("device/reconnect/#", MqttQoS::AtMostOnce).await.unwrap();
+        client.subscribe("device/reconnect/#", MqttQoS::AtMostOnce, false).await.unwrap();
         assert!(client.broker_info().await.connected);
         assert_eq!(client.list_topics().await.len(), 1);
 
@@ -1597,7 +1697,9 @@ mod tests {
 
         let subscribing_client = Arc::clone(&client);
         let subscribe =
-            tokio::spawn(async move { subscribing_client.subscribe("device/ack/#", MqttQoS::AtLeastOnce).await });
+            tokio::spawn(
+                async move { subscribing_client.subscribe("device/ack/#", MqttQoS::AtLeastOnce, false).await },
+            );
         subscribe_seen.await.unwrap();
         assert!(client.list_topics().await.is_empty());
         allow_suback.send(()).unwrap();
@@ -1687,7 +1789,7 @@ mod tests {
         let config = mqtt_config(port, crate::mqtt::types::MqttProtocolVersion::V4);
         let client = MqttClient::connect(config).await.unwrap();
 
-        client.subscribe("device/downgraded/#", MqttQoS::ExactlyOnce).await.unwrap();
+        client.subscribe("device/downgraded/#", MqttQoS::ExactlyOnce, false).await.unwrap();
         assert_eq!(client.list_topics().await, vec![("device/downgraded/#".to_string(), MqttQoS::AtMostOnce)]);
 
         client.disconnect().await;
@@ -1700,14 +1802,14 @@ mod tests {
         let config = mqtt_config(port, crate::mqtt::types::MqttProtocolVersion::V4);
         let client = MqttClient::connect(config).await.unwrap();
 
-        client.subscribe("retained/#", MqttQoS::AtMostOnce).await.unwrap();
+        client.subscribe("retained/#", MqttQoS::AtMostOnce, false).await.unwrap();
         wait_for_message_count(&client, 2).await;
         let initial = client.get_messages(None, 10).await;
         assert_eq!(initial.iter().filter(|message| message.payload_text.as_deref() == Some("v1")).count(), 1);
         assert!(initial.iter().any(|message| message.payload_text.as_deref() == Some("v2")));
 
         client.unsubscribe("retained/#").await.unwrap();
-        client.subscribe("retained/#", MqttQoS::AtMostOnce).await.unwrap();
+        client.subscribe("retained/#", MqttQoS::AtMostOnce, false).await.unwrap();
         wait_for_message_count(&client, 3).await;
         assert_eq!(
             client
@@ -1777,7 +1879,7 @@ mod tests {
         let config = mqtt_config(port, crate::mqtt::types::MqttProtocolVersion::V4);
         let client = MqttClient::connect(config).await.unwrap();
 
-        let error = client.subscribe("device/rejected/#", MqttQoS::AtMostOnce).await.unwrap_err();
+        let error = client.subscribe("device/rejected/#", MqttQoS::AtMostOnce, false).await.unwrap_err();
         assert!(error.contains("拒绝"), "unexpected error: {error}");
         assert!(client.list_topics().await.is_empty());
 

crates/dbx-core/src/mqtt/service.rs

@@ -31,8 +31,8 @@ pub async fn get_broker_info(client: &Arc<MqttClient>) -> Result<MqttBrokerInfo,
 }
 
 /// 订阅 topic
-pub async fn subscribe(client: &Arc<MqttClient>, topic: &str, qos: MqttQoS) -> Result<(), String> {
-    client.subscribe(topic, qos).await
+pub async fn subscribe(client: &Arc<MqttClient>, topic: &str, qos: MqttQoS, no_local: bool) -> Result<(), String> {
+    client.subscribe(topic, qos, no_local).await
 }
 
 /// 取消订阅 topic

crates/dbx-core/src/mqtt/types.rs

@@ -32,7 +32,14 @@ pub enum MqttAuth {
     #[serde(rename = "password")]
     Password { username: String, password: String },
     #[serde(rename = "certificate")]
-    Certificate { ca_cert_path: Option<String>, client_cert_path: Option<String>, client_key_path: Option<String> },
+    Certificate {
+        #[serde(rename = "caCertPath")]
+        ca_cert_path: Option<String>,
+        #[serde(rename = "clientCertPath")]
+        client_cert_path: Option<String>,
+        #[serde(rename = "clientKeyPath")]
+        client_key_path: Option<String>,
+    },
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -88,6 +95,9 @@ pub struct MqttConnectionConfig {
     /// WebSocket 路径（仅 WebSocket 传输时使用）
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub ws_path: Option<String>,
+    /// 已保存的订阅 Topic，跨重连和应用重启恢复。
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub saved_topics: Vec<MqttSavedTopic>,
 }
 
 fn default_keep_alive() -> u64 {
@@ -117,6 +127,7 @@ impl Default for MqttConnectionConfig {
             connect_timeout_secs: default_connect_timeout(),
             max_packet_size_bytes: default_max_packet_size(),
             ws_path: None,
+            saved_topics: Vec::new(),
         }
     }
 }
@@ -133,6 +144,9 @@ impl MqttConnectionConfig {
         if parsed.client_id.trim().is_empty() {
             return Err("MQTT Client ID 不能为空".to_string());
         }
+        if matches!(parsed.auth, MqttAuth::Certificate { .. }) && !parsed.tls {
+            return Err("MQTT 证书认证必须启用 TLS".to_string());
+        }
         if !(1024..=268_435_455).contains(&parsed.max_packet_size_bytes) {
             return Err("MQTT 最大报文大小必须在 1024 到 268435455 字节之间".to_string());
         }
@@ -170,6 +184,17 @@ impl MqttConnectionConfig {
     }
 }
 
+/// 已保存的 MQTT 订阅配置。
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MqttSavedTopic {
+    pub topic: String,
+    #[serde(default)]
+    pub qos: MqttQoS,
+    #[serde(default)]
+    pub no_local: bool,
+}
+
 /// MQTT 消息服务质量
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
 #[serde(rename_all = "lowercase")]

docs/mqtt-issues-fix-plan.md

@@ -0,0 +1,58 @@
+# MQTT Issues 修复计划
+
+关联 issue：
+
+- [#5471](https://github.com/t8y2/dbx/issues/5471)：JSON 发布面板消失、重新打开后 Topic 丢失（P0）
+- [#5490](https://github.com/t8y2/dbx/issues/5490)：断线后 Topic 丢失、缺少 SSL 证书登录（P0）
+- [#5450](https://github.com/t8y2/dbx/issues/5450)：Payload 切换为 JSON 后发送框消失（P0）
+- [#5448](https://github.com/t8y2/dbx/issues/5448)：禁止 MQTT 本地转发、消息分区展示（P1）
+- [#5456](https://github.com/t8y2/dbx/issues/5456)：保存 Topic 配置（P1）
+
+## 现状与根因
+
+1. JSON placeholder 中包含原始 `{...}`，切换 JSON 时会被 Vue I18n 当作插值表达式编译，导致发布组件渲染异常。
+2. MQTT Topic 目前只保存在 `MqttClient` 的进程内存中；同一客户端的网络重连可以恢复，但重建客户端或重启 DBX 后会丢失。
+3. `MqttAuth` 已声明 `certificate` 类型，但连接 UI 只提供无认证和账号密码，Rust TLS 构建也只应用账号密码。
+4. 消息已经携带 `sent/received` 方向，但订阅请求没有 MQTT 5 `No Local` 选项，因此发布到自己订阅的 Topic 时会看到 broker 回送副本。
+
+## 实施顺序
+
+### 1. P0：JSON 发布面板
+
+- 将 JSON 示例改为安全参数插值，避免语言包直接出现未转义花括号。
+- 保持 Payload textarea 和发布按钮在编码切换后继续挂载。
+- 增加切换所有 Payload 编码的组件回归测试，并覆盖非法 JSON。
+
+### 2. P0/P1：Topic 持久化
+
+- 在 MQTT external config 中新增带默认值的订阅配置数组，保持旧连接配置兼容。
+- 订阅成功后保存 Topic、QoS 和订阅选项；取消订阅成功后删除配置。
+- 客户端创建和首次 CONNACK 后恢复保存的订阅；保留现有网络重连恢复逻辑。
+- 使用按连接 ID 的合并更新，避免覆盖用户同时修改的其他连接字段。
+
+### 3. P0：TLS 与证书登录
+
+- 增加 CA、客户端证书、客户端私钥的选择和校验。
+- Rust 端构建系统 CA、自定义 CA、mTLS 和跳过服务端验证的 Rustls 配置。
+- 只保存证书路径，不把私钥内容写入连接配置。
+
+### 4. P1：No Local 与消息分流
+
+- MQTT 5 使用 `Filter.nolocal`；MQTT 3.x 明确提示协议不支持。
+- 订阅配置保存 `noLocal` 等 MQTT 5 选项。
+- 消息列表按方向左右展示，并提供全部/接收/发送过滤。
+
+## 验收标准
+
+- 选择 JSON 后发布面板不消失，合法 JSON 可发布，非法 JSON 有错误提示。
+- 断线重连、关闭并重新打开 DBX 后，已保存 Topic、QoS 和选项自动恢复。
+- TLS、CA、自签 CA、客户端证书认证均可连接，错误证书给出明确错误。
+- MQTT 5 开启 No Local 后不出现自己的 broker 回送副本；关闭后发送和接收消息仍能按方向区分。
+
+## 验证命令
+
+```text
+pnpm typecheck
+pnpm test
+cargo test -p dbx-core --features mq-admin mqtt
+```

packages/app-tests/mqttPayloadCodec.test.ts

@@ -54,3 +54,8 @@ test("MQTT JSON publishing sends canonical bytes instead of the original text",
   assert.match(source, /payloadText:\s*encoding\.value === "plaintext" \? payloadText\.value : null/);
   assert.doesNotMatch(source, /encoding\.value === "plaintext" \|\| encoding\.value === "json"/);
 });
+test("MQTT JSON placeholder uses a named interpolation for literal braces", () => {
+  const source = readFileSync("apps/desktop/src/components/mqtt/MqttPublishDialog.vue", "utf8");
+  assert.match(source, /mqttPayloadPlaceholderJson", \{ example:/);
+  assert.doesNotMatch(source, /mqttPayloadPlaceholderJson"\);/);
+});

src-tauri/src/commands/mqtt_cmd.rs

@@ -1,3 +1,4 @@
+use serde_json::json;
 use std::sync::Arc;
 use tauri::State;
 
@@ -19,6 +20,25 @@ async fn get_mqtt_client(
     }
 }
 
+async fn persist_mqtt_topics(
+    state: &AppState,
+    connection_id: &str,
+    client: &dbx_core::mqtt::client::MqttClient,
+) -> Result<(), String> {
+    let saved_topics = client.desired_topic_configs().await;
+    let Some(mut config) = state.configs.read().await.get(connection_id).cloned() else {
+        return Ok(());
+    };
+    let mut external_config = config.external_config.take().unwrap_or_else(|| json!({}));
+    let Some(external_object) = external_config.as_object_mut() else {
+        return Err("MQTT external_config 必须是 JSON 对象".to_string());
+    };
+    external_object.insert("savedTopics".to_string(), serde_json::to_value(saved_topics).map_err(|e| e.to_string())?);
+    config.external_config = Some(external_config);
+    state.storage.save_connections(std::slice::from_ref(&config)).await?;
+    state.configs.write().await.insert(connection_id.to_string(), config);
+    Ok(())
+}
 /// 获取 broker 基本信息
 #[tauri::command]
 pub async fn mqtt_get_broker_info(
@@ -36,9 +56,11 @@ pub async fn mqtt_subscribe(
     connection_id: String,
     topic: String,
     qos: Option<MqttQoS>,
+    no_local: Option<bool>,
 ) -> Result<(), String> {
     let client = get_mqtt_client(&state, &connection_id).await?;
-    service::subscribe(&client, &topic, qos.unwrap_or_default()).await
+    service::subscribe(&client, &topic, qos.unwrap_or_default(), no_local.unwrap_or(false)).await?;
+    persist_mqtt_topics(state.inner(), &connection_id, &client).await
 }
 
 /// 取消订阅 topic
@@ -49,7 +71,8 @@ pub async fn mqtt_unsubscribe(
     topic: String,
 ) -> Result<(), String> {
     let client = get_mqtt_client(&state, &connection_id).await?;
-    service::unsubscribe(&client, &topic).await
+    service::unsubscribe(&client, &topic).await?;
+    persist_mqtt_topics(state.inner(), &connection_id, &client).await
 }
 
 /// 发布消息