feat: implement task_id for build triggers (#1902)

Signed-off-by: allure <1550220889@qq.com>

Author: AllureCurtain

ceres/src/build_trigger/dispatcher.rs

@@ -1,6 +1,9 @@
 use std::sync::Arc;
 
-use bellatrix::{Bellatrix, orion_client::OrionBuildRequest};
+use bellatrix::{
+    Bellatrix,
+    orion_client::{BuildInfo, OrionBuildRequest},
+};
 use common::errors::MegaError;
 use jupiter::storage::Storage;
 
@@ -17,91 +20,77 @@ impl BuildDispatcher {
         Self { storage, bellatrix }
     }
 
-    /// Dispatch a build trigger.
-    ///
-    /// This method:
-    /// 1. Persists the trigger to the database
-    /// 2. Sends the build request to Bellatrix/Orion asynchronously
-    ///
-    /// Returns the ID of the created trigger record.
     pub async fn dispatch(&self, trigger: BuildTrigger) -> Result<i64, MegaError> {
         let trigger_payload = serde_json::to_value(&trigger.payload).map_err(|e| {
             tracing::error!("Failed to serialize payload: {}", e);
             MegaError::Other(format!("Failed to serialize payload: {}", e))
         })?;
 
+        // Determine task_id based on whether build system is enabled
+        let task_id: Option<uuid::Uuid> = if self.bellatrix.enable_build() {
+            // Extract data from payload using pattern matching
+            let (cl_link, repo, builds_json, cl_id) = match &trigger.payload {
+                BuildTriggerPayload::GitPush(p) => (&p.cl_link, &p.repo, &p.builds, p.cl_id),
+                BuildTriggerPayload::Manual(p) => (&p.cl_link, &p.repo, &p.builds, p.cl_id),
+                BuildTriggerPayload::Retry(p) => (&p.cl_link, &p.repo, &p.builds, p.cl_id),
+                BuildTriggerPayload::Webhook(p) => (&p.cl_link, &p.repo, &p.builds, p.cl_id),
+                BuildTriggerPayload::Schedule(p) => (&p.cl_link, &p.repo, &p.builds, p.cl_id),
+                BuildTriggerPayload::WebEdit(p) => (&p.cl_link, &p.repo, &p.builds, p.cl_id),
+            };
+
+            let builds: Vec<SerializableBuildInfo> = serde_json::from_value(builds_json.clone())
+                .map_err(|e| {
+                    tracing::error!("Failed to deserialize builds from payload: {}", e);
+                    MegaError::Other(format!("Failed to deserialize builds from payload: {}", e))
+                })?;
+
+            let bellatrix_builds: Vec<BuildInfo> = builds
+                .into_iter()
+                .map(|info| BuildInfo {
+                    changes: info.changes.into_iter().map(|s| s.into()).collect(),
+                })
+                .collect();
+
+            let req = OrionBuildRequest {
+                cl_link: cl_link.to_string(),
+                mount_path: repo.to_string(),
+                cl: cl_id.unwrap_or(0),
+                builds: bellatrix_builds,
+            };
+
+            let task_id_str = self.bellatrix.on_post_receive(req).await.map_err(|e| {
+                tracing::error!("Failed to dispatch build to Bellatrix: {}", e);
+                MegaError::Other(format!("Failed to dispatch build to Bellatrix: {}", e))
+            })?;
+
+            let task_uuid = uuid::Uuid::parse_str(&task_id_str).map_err(|e| {
+                tracing::error!("Invalid task_id format '{}': {}", task_id_str, e);
+                MegaError::Other(format!("Invalid task_id format '{}': {}", task_id_str, e))
+            })?;
+
+            Some(task_uuid)
+        } else {
+            tracing::info!("BuildDispatcher: Build system disabled, skipping Orion call");
+            None
+        };
+
+        // Insert trigger record with task_id (complete record in one operation)
         let db_record = self
             .storage
             .build_trigger_storage()
             .insert(
                 trigger.trigger_type.to_string(),
                 trigger.trigger_source.to_string(),
                 trigger_payload,
+                task_id,
             )
             .await?;
 
-        // Persist to database
-        tracing::info!("BuildDispatcher: Persisted trigger ID: {}", db_record.id);
-
-        if !self.bellatrix.enable_build() {
-            tracing::info!("BuildDispatcher: Completed (build system disabled)");
-            return Ok(db_record.id);
-        }
-
-        // Extract data from payload using pattern matching
-        let (cl_link, repo, builds_json, cl_id) = match &trigger.payload {
-            BuildTriggerPayload::GitPush(p) => (&p.cl_link, &p.repo, &p.builds, p.cl_id),
-            BuildTriggerPayload::Manual(p) => (&p.cl_link, &p.repo, &p.builds, p.cl_id),
-            BuildTriggerPayload::Retry(p) => (&p.cl_link, &p.repo, &p.builds, p.cl_id),
-            BuildTriggerPayload::Webhook(p) => (&p.cl_link, &p.repo, &p.builds, p.cl_id),
-            BuildTriggerPayload::Schedule(p) => (&p.cl_link, &p.repo, &p.builds, p.cl_id),
-            BuildTriggerPayload::WebEdit(p) => (&p.cl_link, &p.repo, &p.builds, p.cl_id),
-        };
-
-        let builds: Vec<SerializableBuildInfo> = serde_json::from_value(builds_json.clone())
-            .map_err(|e| {
-                tracing::error!("Failed to deserialize builds from payload: {}", e);
-                MegaError::Other(format!("Failed to deserialize builds from payload: {}", e))
-            })?;
-
-        let bellatrix_builds: Vec<bellatrix::orion_client::BuildInfo> = builds
-            .into_iter()
-            .enumerate()
-            .map(|(idx, info)| {
-                tracing::debug!("  Build [{}]: {} change(s)", idx + 1, info.changes.len());
-                bellatrix::orion_client::BuildInfo {
-                    changes: info.changes.into_iter().map(|s| s.into()).collect(),
-                }
-            })
-            .collect();
-
-        let req = OrionBuildRequest {
-            cl_link: cl_link.to_string(),
-            mount_path: repo.to_string(),
-            cl: cl_id.unwrap_or(0),
-            builds: bellatrix_builds,
-        };
-
-        // Dispatch asynchronously
-        let bellatrix = self.bellatrix.clone();
-        let trigger_id = db_record.id;
-        tokio::spawn(async move {
-            match bellatrix.on_post_receive(req).await {
-                Ok(_) => {
-                    tracing::info!(
-                        "BuildDispatcher: Build request sent to Bellatrix (Trigger ID: {})",
-                        trigger_id
-                    );
-                }
-                Err(err) => {
-                    tracing::error!(
-                        "BuildDispatcher: Failed to dispatch build (Trigger ID: {}): {}",
-                        trigger_id,
-                        err
-                    );
-                }
-            }
-        });
+        tracing::info!(
+            "BuildDispatcher: Trigger persisted (ID: {}, Task ID: {:?})",
+            db_record.id,
+            task_id
+        );
 
         Ok(db_record.id)
     }

ceres/src/pack/monorepo.rs

@@ -1054,13 +1054,16 @@ impl MonoRepo {
                 triggered_by: self.username.clone(),
             };
 
-            let _trigger_id = BuildTriggerService::handle_git_push_event(
+            if let Err(e) = BuildTriggerService::handle_git_push_event(
                 self.storage.clone(),
                 self.git_object_cache.clone(),
                 self.bellatrix.clone(),
                 event,
             )
-            .await?;
+            .await
+            {
+                tracing::error!("Failed to trigger CI pipeline: {}", e);
+            }
         }
 
         let check_reg = CheckerRegistry::new(self.storage.clone().into(), self.username());

jupiter/src/storage/build_trigger_storage.rs

@@ -14,12 +14,14 @@ pub struct BuildTriggerStorage {
 }
 
 impl BuildTriggerStorage {
-    /// Insert a new build trigger
+    /// Insert a new build trigger with optional task_id.
+    /// When task_id is provided, inserts a complete record in one operation.
     pub async fn insert(
         &self,
         trigger_type: String,
         trigger_source: String,
         trigger_payload: serde_json::Value,
+        task_id: Option<uuid::Uuid>,
     ) -> Result<build_triggers::Model, MegaError> {
         let now = chrono::Utc::now().naive_utc();
 
@@ -29,7 +31,7 @@ impl BuildTriggerStorage {
             trigger_source: ActiveValue::Set(trigger_source),
             trigger_payload: ActiveValue::Set(trigger_payload),
             trigger_time: ActiveValue::Set(now),
-            task_id: ActiveValue::NotSet,
+            task_id: ActiveValue::Set(task_id),
             updated_at: ActiveValue::Set(now),
         };
 
@@ -169,7 +171,7 @@ mod tests {
 
         let inserted = storage
             .build_trigger_storage()
-            .insert("git_push".to_string(), "user".to_string(), payload)
+            .insert("git_push".to_string(), "user".to_string(), payload, None)
             .await
             .unwrap();
 
@@ -185,6 +187,42 @@ mod tests {
         assert_eq!(retrieved.trigger_source, "user");
     }
 
+    #[tokio::test]
+    async fn test_insert_with_task_id() {
+        let temp_dir = tempdir().unwrap();
+        let storage = test_storage(temp_dir.path()).await;
+
+        let payload = serde_json::json!({
+            "type": "manual",
+            "repo": "/test",
+            "commit_hash": "abc123",
+            "cl_link": "test_link",
+            "builds": []
+        });
+
+        let task_id = uuid::Uuid::new_v4();
+
+        let inserted = storage
+            .build_trigger_storage()
+            .insert(
+                "manual".to_string(),
+                "user".to_string(),
+                payload,
+                Some(task_id),
+            )
+            .await
+            .unwrap();
+
+        let retrieved = storage
+            .build_trigger_storage()
+            .get_by_id(inserted.id)
+            .await
+            .unwrap()
+            .unwrap();
+
+        assert_eq!(retrieved.task_id, Some(task_id));
+    }
+
     #[tokio::test]
     async fn test_get_recent() {
         let temp_dir = tempdir().unwrap();
@@ -201,7 +239,12 @@ mod tests {
         for _ in 0..3 {
             storage
                 .build_trigger_storage()
-                .insert("manual".to_string(), "user".to_string(), payload.clone())
+                .insert(
+                    "manual".to_string(),
+                    "user".to_string(),
+                    payload.clone(),
+                    None,
+                )
                 .await
                 .unwrap();
         }
@@ -229,7 +272,7 @@ mod tests {
 
             storage
                 .build_trigger_storage()
-                .insert("manual".to_string(), "user".to_string(), payload)
+                .insert("manual".to_string(), "user".to_string(), payload, None)
                 .await
                 .unwrap();
         }

orion-server/bellatrix/src/lib.rs

@@ -23,8 +23,8 @@ impl Bellatrix {
         self.build_config.enable_build
     }
 
-    pub async fn on_post_receive(&self, req: OrionBuildRequest) -> anyhow::Result<()> {
-        self.orion.trigger_build(req).await?;
-        Ok(())
+    pub async fn on_post_receive(&self, req: OrionBuildRequest) -> anyhow::Result<String> {
+        let task_id = self.orion.trigger_build(req).await?;
+        Ok(task_id)
     }
 }

orion-server/bellatrix/src/orion_client/mod.rs

@@ -48,6 +48,12 @@ pub struct OrionBuildRequest {
     pub builds: Vec<BuildInfo>,
 }
 
+/// Response from Orion task handler containing the assigned task ID.
+#[derive(Deserialize, Debug)]
+pub struct TaskResponse {
+    pub task_id: String,
+}
+
 #[derive(Clone)]
 pub(crate) struct OrionClient {
     base_url: String,
@@ -62,12 +68,15 @@ impl OrionClient {
         }
     }
 
-    pub async fn trigger_build(&self, req: OrionBuildRequest) -> anyhow::Result<()> {
+    /// Trigger a build on Orion and return the assigned task ID.
+    pub async fn trigger_build(&self, req: OrionBuildRequest) -> anyhow::Result<String> {
         let url = format!("{}/task", self.base_url);
         tracing::info!("Try to trigger build with params:{:?}", req);
         let res = self.client.post(&url).json(&req).send().await?;
         if res.status().is_success() {
-            Ok(())
+            let task_response: TaskResponse = res.json().await?;
+            tracing::info!("Received task_id from Orion: {}", task_response.task_id);
+            Ok(task_response.task_id)
         } else {
             tracing::error!("Failed to trigger build: {}", res.status());
             Err(anyhow::anyhow!("Failed to trigger build: {}", res.status()))