feat(nacos): support DataId mapping in config sync

Closes #6439

crates/dbx-core/src/nacos/batch.rs

@@ -1,3 +1,4 @@
+use std::collections::{HashMap, HashSet};
 use std::path::Path;
 use std::sync::Arc;
 
@@ -158,17 +159,20 @@ async fn transfer_configs(
     request: &NacosConfigTransferRequest,
 ) -> Result<Vec<NacosConfigUpsert>, String> {
     let target_group = request.target_group.as_deref().map(str::trim).filter(|group| !group.is_empty());
+    let mut data_id_mappings = validated_data_id_mappings(request)?;
     let source = resolve_selector(source_admin, &request.source).await?;
     let configs = source
         .into_iter()
         .map(|config| {
+            let source_key = (config.group.clone(), config.data_id.clone());
             let content = config
                 .content
                 .ok_or_else(|| format!("Nacos configuration {}/{} has no content", config.group, config.data_id))?;
             let group = target_group.map(str::to_string).unwrap_or(config.group);
+            let data_id = data_id_mappings.remove(&source_key).unwrap_or(config.data_id);
             Ok(NacosConfigUpsert {
                 namespace: Some(request.target_namespace.clone()),
-                data_id: config.data_id,
+                data_id,
                 group,
                 content,
                 config_type: config.config_type,
@@ -178,7 +182,12 @@ async fn transfer_configs(
             })
         })
         .collect::<Result<Vec<_>, String>>()?;
-    let mut target_keys = std::collections::HashSet::new();
+    if let Some(((group, data_id), _)) = data_id_mappings.into_iter().next() {
+        return Err(format!(
+            "Nacos configuration {group}/{data_id} no longer exists in the selected source configurations"
+        ));
+    }
+    let mut target_keys = HashSet::new();
     for config in &configs {
         let namespace = config.namespace.as_deref().unwrap_or_default();
         let key = (namespace, config.group.as_str(), config.data_id.as_str());
@@ -192,6 +201,47 @@ async fn transfer_configs(
     Ok(configs)
 }
 
+fn validated_data_id_mappings(
+    request: &NacosConfigTransferRequest,
+) -> Result<HashMap<(String, String), String>, String> {
+    if request.data_id_mappings.is_empty() {
+        return Ok(HashMap::new());
+    }
+    if !matches!(request.source.scope, NacosConfigSelectionScope::Selected) {
+        return Err("Nacos Data ID mappings are only supported for selected configurations".to_string());
+    }
+
+    let selected_keys: HashSet<_> =
+        request.source.keys.iter().map(|key| (key.group.as_str(), key.data_id.as_str())).collect();
+    let mut mappings = HashMap::with_capacity(request.data_id_mappings.len());
+    for mapping in &request.data_id_mappings {
+        let source_key = (mapping.source_group.as_str(), mapping.source_data_id.as_str());
+        if !selected_keys.contains(&source_key) {
+            return Err(format!(
+                "Nacos configuration {}/{} is not selected for Data ID mapping",
+                mapping.source_group, mapping.source_data_id
+            ));
+        }
+        let target_data_id = mapping.target_data_id.trim();
+        if target_data_id.is_empty() {
+            return Err(format!(
+                "Nacos configuration {}/{} target Data ID cannot be blank",
+                mapping.source_group, mapping.source_data_id
+            ));
+        }
+        if mappings
+            .insert((mapping.source_group.clone(), mapping.source_data_id.clone()), target_data_id.to_string())
+            .is_some()
+        {
+            return Err(format!(
+                "Nacos configuration {}/{} has multiple target Data ID mappings",
+                mapping.source_group, mapping.source_data_id
+            ));
+        }
+    }
+    Ok(mappings)
+}
+
 async fn resolve_selector(
     admin: Arc<dyn NacosAdmin>,
     selector: &NacosConfigSelector,
@@ -842,15 +892,23 @@ mod tests {
             },
             target_namespace: "target-ns".to_string(),
             target_group: target_group.map(str::to_string),
+            data_id_mappings: Vec::new(),
             conflict_policy: NacosConflictPolicy::default(),
         }
     }
 
-    fn source_admin_with_config() -> Arc<MockAdmin> {
-        let admin = Arc::new(MockAdmin::default());
+    fn data_id_mapping(source_group: &str, source_data_id: &str, target_data_id: &str) -> NacosConfigDataIdMapping {
+        NacosConfigDataIdMapping {
+            source_group: source_group.to_string(),
+            source_data_id: source_data_id.to_string(),
+            target_data_id: target_data_id.to_string(),
+        }
+    }
+
+    fn insert_source_config(admin: &MockAdmin, data_id: &str, group: &str, content: &str) {
         admin.insert(NacosConfigItem {
-            data_id: "app".to_string(),
-            group: "SOURCE_GROUP".to_string(),
+            data_id: data_id.to_string(),
+            group: group.to_string(),
             namespace: "source-ns".to_string(),
             app_name: None,
             desc: None,
@@ -858,8 +916,13 @@ mod tests {
             config_type: Some("text".to_string()),
             md5: None,
             encrypted_data_key: None,
-            content: Some("value".to_string()),
+            content: Some(content.to_string()),
         });
+    }
+
+    fn source_admin_with_config() -> Arc<MockAdmin> {
+        let admin = Arc::new(MockAdmin::default());
+        insert_source_config(&admin, "app", "SOURCE_GROUP", "value");
         admin
     }
 
@@ -916,4 +979,132 @@ mod tests {
         assert_eq!(configs.len(), 1);
         assert_eq!(configs[0].group, "SOURCE_GROUP");
     }
+
+    #[tokio::test]
+    async fn transfer_with_data_id_mapping_uses_the_trimmed_target_data_id() {
+        let source_admin = source_admin_with_config();
+        let mut request = transfer_request(None);
+        request.data_id_mappings = vec![data_id_mapping("SOURCE_GROUP", "app", "  renamed.yaml  ")];
+
+        let configs = transfer_configs(source_admin, &request).await.unwrap();
+
+        assert_eq!(configs.len(), 1);
+        assert_eq!(configs[0].data_id, "renamed.yaml");
+    }
+
+    #[tokio::test]
+    async fn transfer_with_multiple_selected_configs_maps_only_the_edited_data_id() {
+        let source_admin = source_admin_with_config();
+        insert_source_config(&source_admin, "worker", "OTHER_GROUP", "worker value");
+        let mut request = transfer_request(None);
+        request.source.keys.push(NacosConfigKey {
+            namespace: Some("source-ns".to_string()),
+            data_id: "worker".to_string(),
+            group: "OTHER_GROUP".to_string(),
+        });
+        request.data_id_mappings = vec![data_id_mapping("OTHER_GROUP", "worker", "worker-prod")];
+
+        let configs = transfer_configs(source_admin, &request).await.unwrap();
+
+        assert_eq!(configs.len(), 2);
+        assert!(configs.iter().any(|config| config.group == "SOURCE_GROUP" && config.data_id == "app"));
+        assert!(configs.iter().any(|config| config.group == "OTHER_GROUP" && config.data_id == "worker-prod"));
+    }
+
+    #[tokio::test]
+    async fn transfer_rejects_blank_data_id_mapping() {
+        let source_admin = source_admin_with_config();
+        let mut request = transfer_request(None);
+        request.data_id_mappings = vec![data_id_mapping("SOURCE_GROUP", "app", "   ")];
+
+        let error = transfer_configs(source_admin, &request).await.unwrap_err();
+
+        assert!(error.contains("target Data ID cannot be blank"), "unexpected error: {error}");
+    }
+
+    #[tokio::test]
+    async fn transfer_rejects_duplicate_mappings_for_the_same_source_key() {
+        let source_admin = source_admin_with_config();
+        let mut request = transfer_request(None);
+        request.data_id_mappings =
+            vec![data_id_mapping("SOURCE_GROUP", "app", "first"), data_id_mapping("SOURCE_GROUP", "app", "second")];
+
+        let error = transfer_configs(source_admin, &request).await.unwrap_err();
+
+        assert!(error.contains("multiple target Data ID mappings"), "unexpected error: {error}");
+    }
+
+    #[tokio::test]
+    async fn transfer_rejects_data_id_mapping_outside_selected_scope() {
+        let source_admin = source_admin_with_config();
+        let mut request = transfer_request(None);
+        request.source.scope = NacosConfigSelectionScope::Filtered;
+        request.data_id_mappings = vec![data_id_mapping("SOURCE_GROUP", "app", "renamed")];
+
+        let error = transfer_configs(source_admin, &request).await.unwrap_err();
+
+        assert!(error.contains("only supported for selected"), "unexpected error: {error}");
+    }
+
+    #[tokio::test]
+    async fn transfer_rejects_data_id_mapping_for_an_unselected_source_key() {
+        let source_admin = source_admin_with_config();
+        let mut request = transfer_request(None);
+        request.data_id_mappings = vec![data_id_mapping("SOURCE_GROUP", "not-selected", "renamed")];
+
+        let error = transfer_configs(source_admin, &request).await.unwrap_err();
+
+        assert!(error.contains("SOURCE_GROUP/not-selected is not selected"), "unexpected error: {error}");
+    }
+
+    #[tokio::test]
+    async fn transfer_rejects_a_stale_selected_source_mapping() {
+        let source_admin = source_admin_with_config();
+        let mut request = transfer_request(None);
+        request.source.keys[0].data_id = "deleted".to_string();
+        request.data_id_mappings = vec![data_id_mapping("SOURCE_GROUP", "deleted", "renamed")];
+
+        let error = transfer_configs(source_admin, &request).await.unwrap_err();
+
+        assert_eq!(error, "not found");
+    }
+
+    #[tokio::test]
+    async fn transfer_rejects_target_key_collisions_after_data_id_mapping() {
+        let source_admin = source_admin_with_config();
+        insert_source_config(&source_admin, "worker", "SOURCE_GROUP", "worker value");
+        let mut request = transfer_request(Some("TARGET_GROUP"));
+        request.source.keys.push(NacosConfigKey {
+            namespace: Some("source-ns".to_string()),
+            data_id: "worker".to_string(),
+            group: "SOURCE_GROUP".to_string(),
+        });
+        request.data_id_mappings = vec![data_id_mapping("SOURCE_GROUP", "worker", "app")];
+
+        let error = transfer_configs(source_admin, &request).await.unwrap_err();
+
+        assert!(error.contains("target-ns/TARGET_GROUP/app"), "unexpected error: {error}");
+    }
+
+    #[tokio::test]
+    async fn preview_and_apply_share_the_same_mapped_data_id_plan() {
+        let source_admin = source_admin_with_config();
+        let target_admin = Arc::new(MockAdmin::default());
+        let mut request = transfer_request(None);
+        request.data_id_mappings = vec![data_id_mapping("SOURCE_GROUP", "app", "app-prod")];
+
+        let preview = preview_transfer(source_admin.clone(), target_admin.clone(), &request).await.unwrap();
+        let report = apply_transfer(source_admin, target_admin.clone(), &request, &preview.plan_hash).await.unwrap();
+
+        assert_eq!((preview.created, report.created), (1, 1));
+        let published = target_admin
+            .get_config(NacosConfigKey {
+                namespace: Some("target-ns".to_string()),
+                data_id: "app-prod".to_string(),
+                group: "SOURCE_GROUP".to_string(),
+            })
+            .await
+            .unwrap();
+        assert_eq!(published.content.as_deref(), Some("value"));
+    }
 }

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

@@ -816,6 +816,14 @@ pub struct NacosBatchReport {
     pub items: Vec<NacosBatchItemResult>,
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "camelCase")]
+pub struct NacosConfigDataIdMapping {
+    pub source_group: String,
+    pub source_data_id: String,
+    pub target_data_id: String,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 #[serde(rename_all = "camelCase")]
 pub struct NacosConfigTransferRequest {
@@ -827,6 +835,8 @@ pub struct NacosConfigTransferRequest {
     #[serde(default)]
     pub target_group: Option<String>,
     #[serde(default)]
+    pub data_id_mappings: Vec<NacosConfigDataIdMapping>,
+    #[serde(default)]
     pub conflict_policy: NacosConflictPolicy,
 }