From e8885f77983daf7218a895ffc664feec7aaf650a Mon Sep 17 00:00:00 2001 From: jinmo <9446650@qq.com> Date: Fri, 7 Aug 2026 13:43:09 +0800 Subject: [PATCH] fix(mqtt): preserve saved connections when persisting MQTT topics --- crates/dbx-core/src/storage.rs | 81 ++++++++++++++++++++++++++++++ src-tauri/src/commands/mqtt_cmd.rs | 22 ++++---- 2 files changed, 93 insertions(+), 10 deletions(-) diff --git a/crates/dbx-core/src/storage.rs b/crates/dbx-core/src/storage.rs index 27a6bdbe8..b129033f7 100644 --- a/crates/dbx-core/src/storage.rs +++ b/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"); diff --git a/src-tauri/src/commands/mqtt_cmd.rs b/src-tauri/src/commands/mqtt_cmd.rs index 96eea74cd..2a2c557b5 100644 --- a/src-tauri/src/commands/mqtt_cmd.rs +++ b/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 基本信息