fix(mqtt): preserve saved connections when persisting MQTT topics

crates/dbx-core/src/storage.rs

@@ -2438,6 +2438,47 @@ impl Storage {
         .await
     }
 
+    /// Update only the persisted MQTT saved-topic metadata for one connection.
+    ///
+    /// Unlike `save_connections`, this is an in-place update and must not replace
+    /// the saved connection list or touch separately stored connection secrets.
+    pub async fn save_connection_mqtt_saved_topics(
+        &self,
+        connection_id: &str,
+        saved_topics: serde_json::Value,
+    ) -> Result<(), String> {
+        let connection_id = connection_id.to_string();
+        self.with_conn(move |conn| {
+            let json = conn
+                .query_row("SELECT config_json FROM connections WHERE id = ?1", [&connection_id], |row| {
+                    row.get::<_, String>(0)
+                })
+                .optional()
+                .map_err(|error| error.to_string())?
+                .ok_or_else(|| format!("Connection config not found: {connection_id}"))?;
+            let mut config: ConnectionConfig = serde_json::from_str(&json).map_err(|error| error.to_string())?;
+            let mut external_config = config.external_config.take().unwrap_or_else(|| serde_json::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(), saved_topics);
+            config.external_config = Some(external_config);
+            let updated_json = serde_json::to_string(&config).map_err(|error| error.to_string())?;
+            conn.execute("UPDATE connections SET config_json = ?1 WHERE id = ?2", params![updated_json, connection_id])
+                .map(
+                    |updated| {
+                        if updated == 0 {
+                            Err(format!("Connection config not found: {connection_id}"))
+                        } else {
+                            Ok(())
+                        }
+                    },
+                )
+                .map_err(|error| error.to_string())?
+        })
+        .await
+    }
+
     /// Update only the persisted driver identity for an existing connection.
     /// This is used after runtime driver fallback and deliberately leaves all
     /// other connection metadata and separately stored secrets untouched.
@@ -4424,6 +4465,46 @@ mod tests {
         let _ = std::fs::remove_file(path);
     }
 
+    #[tokio::test]
+    async fn save_connection_mqtt_saved_topics_updates_only_target_and_preserves_secrets() {
+        let path = temp_db_path("mqtt-saved-topics");
+        let storage = Storage::open(&path).await.unwrap();
+        let target = mq_connection("target", "target-secret");
+        let untouched = mq_connection("untouched", "untouched-secret");
+        storage.save_connections(&[target.clone(), untouched.clone()]).await.unwrap();
+
+        let saved_topics = serde_json::json!([{
+            "topic": "sensors/temperature",
+            "qos": "atleastonce",
+            "noLocal": false,
+        }]);
+        storage.save_connection_mqtt_saved_topics(&target.id, saved_topics.clone()).await.unwrap();
+
+        let loaded = storage.load_connections().await.unwrap();
+        assert_eq!(loaded.len(), 2);
+        let updated = loaded.iter().find(|config| config.id == target.id).unwrap();
+        assert_eq!(updated.external_config.as_ref().unwrap()["savedTopics"], saved_topics);
+        assert_eq!(mq_token(updated), Some("target-secret"));
+        assert_eq!(loaded.iter().find(|config| config.id == untouched.id), Some(&untouched));
+        assert_eq!(mq_token(loaded.iter().find(|config| config.id == untouched.id).unwrap()), Some("untouched-secret"));
+
+        assert!(storage.save_connection_mqtt_saved_topics("missing", serde_json::json!([])).await.is_err());
+        let _ = std::fs::remove_file(path);
+    }
+
+    #[tokio::test]
+    async fn save_connection_mqtt_saved_topics_rejects_non_object_external_config() {
+        let path = temp_db_path("mqtt-saved-topics-invalid");
+        let storage = Storage::open(&path).await.unwrap();
+        let mut target = mq_connection("target", "target-secret");
+        target.external_config = Some(serde_json::json!("invalid"));
+        storage.save_connections(std::slice::from_ref(&target)).await.unwrap();
+
+        let error = storage.save_connection_mqtt_saved_topics(&target.id, serde_json::json!([])).await.unwrap_err();
+        assert!(error.contains("external_config"));
+        assert_eq!(storage.load_connections().await.unwrap(), vec![target]);
+        let _ = std::fs::remove_file(path);
+    }
     #[tokio::test]
     async fn save_connection_driver_profile_updates_only_the_target_metadata() {
         let path = temp_db_path("connection-driver-profile");

src-tauri/src/commands/mqtt_cmd.rs

@@ -26,17 +26,19 @@ async fn persist_mqtt_topics(
     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 {
+    if !state.configs.read().await.contains_key(connection_id) {
         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);
+    }
+    let saved_topics = serde_json::to_value(saved_topics).map_err(|e| e.to_string())?;
+    state.storage.save_connection_mqtt_saved_topics(connection_id, saved_topics.clone()).await?;
+    if let Some(config) = state.configs.write().await.get_mut(connection_id) {
+        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(), saved_topics);
+        config.external_config = Some(external_config);
+    }
     Ok(())
 }
 /// 获取 broker 基本信息