feat: implement build trigger system (#1880)

* feat: implement build trigger system

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

* feat: refactor build trigger system and fix error

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

---------

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

Author: AllureCurtain

ceres/src/build_trigger/changes_calculator.rs

@@ -0,0 +1,116 @@
+use std::{
+    path::{Path, PathBuf},
+    sync::Arc,
+};
+
+use common::errors::MegaError;
+use git_internal::hash::ObjectHash;
+use jupiter::storage::Storage;
+
+use crate::{
+    api_service::{cache::GitObjectCache, mono_api_service::MonoApiService},
+    build_trigger::{SerializableBuildInfo, SerializableStatus, TriggerContext},
+    model::change_list::{ClDiffFile, ClFilesRes},
+};
+
+pub struct ChangesCalculator {
+    storage: Storage,
+    git_object_cache: Arc<GitObjectCache>,
+}
+
+impl ChangesCalculator {
+    pub fn new(storage: Storage, git_object_cache: Arc<GitObjectCache>) -> Self {
+        Self {
+            storage,
+            git_object_cache,
+        }
+    }
+
+    pub async fn get_builds_for_commit(
+        &self,
+        context: &TriggerContext,
+    ) -> Result<Vec<SerializableBuildInfo>, MegaError> {
+        let old_files = self.get_commit_blobs(&context.from_hash).await?;
+        let new_files = self.get_commit_blobs(&context.commit_hash).await?;
+        let diff_files = self.cl_files_list(old_files, new_files).await?;
+
+        let changes = self.build_changes(&context.repo_path, diff_files)?;
+
+        Ok(vec![SerializableBuildInfo { changes }])
+    }
+
+    fn build_changes(
+        &self,
+        cl_path: &str,
+        cl_diff_files: Vec<ClDiffFile>,
+    ) -> Result<Vec<SerializableStatus>, MegaError> {
+        let cl_base = PathBuf::from(cl_path);
+        let path_str = cl_base.to_str().ok_or_else(|| {
+            MegaError::Other(format!("CL base path is not valid UTF-8: {:?}", cl_base))
+        })?;
+
+        let changes = cl_diff_files
+            .into_iter()
+            .map(|m| {
+                let mut item: ClFilesRes = m.into();
+                item.path = cl_base.join(item.path).to_string_lossy().to_string();
+                item
+            })
+            .collect::<Vec<_>>();
+
+        let counter_changes = changes
+            .iter()
+            .filter(|&s| PathBuf::from(&s.path).starts_with(&cl_base))
+            .map(|s| {
+                let rel = Path::new(&s.path)
+                    .strip_prefix(path_str)
+                    .map_err(|_| {
+                        MegaError::Other(format!("Invalid project-relative path: {}", s.path))
+                    })?
+                    .to_string_lossy()
+                    .replace('\\', "/")
+                    .trim_start_matches('/')
+                    .to_string();
+
+                let status = if s.action == "new" {
+                    SerializableStatus::Added(rel)
+                } else if s.action == "deleted" {
+                    SerializableStatus::Removed(rel)
+                } else if s.action == "modified" {
+                    SerializableStatus::Modified(rel)
+                } else {
+                    return Err(MegaError::Other(format!(
+                        "Unsupported change action: {}",
+                        s.action
+                    )));
+                };
+                Ok(status)
+            })
+            .collect::<Result<Vec<_>, MegaError>>()?;
+
+        Ok(counter_changes)
+    }
+
+    async fn get_commit_blobs(
+        &self,
+        commit_hash: &str,
+    ) -> Result<Vec<(PathBuf, ObjectHash)>, MegaError> {
+        let api_service = MonoApiService {
+            storage: self.storage.clone(),
+            git_object_cache: self.git_object_cache.clone(),
+        };
+        api_service.get_commit_blobs(commit_hash).await
+    }
+
+    async fn cl_files_list(
+        &self,
+        old_files: Vec<(PathBuf, ObjectHash)>,
+        new_files: Vec<(PathBuf, ObjectHash)>,
+    ) -> Result<Vec<ClDiffFile>, MegaError> {
+        let api_service = MonoApiService {
+            storage: self.storage.clone(),
+            git_object_cache: self.git_object_cache.clone(),
+        };
+        api_service.cl_files_list(old_files, new_files).await
+    }
+}

ceres/src/build_trigger/dispatcher.rs

@@ -0,0 +1,107 @@
+use std::sync::Arc;
+
+use bellatrix::{Bellatrix, orion_client::OrionBuildRequest};
+use common::errors::MegaError;
+use jupiter::storage::Storage;
+
+use crate::build_trigger::{BuildTrigger, BuildTriggerPayload, SerializableBuildInfo};
+
+/// Handles dispatching build triggers to the build execution layer (Bellatrix/Orion).
+pub struct BuildDispatcher {
+    storage: Storage,
+    bellatrix: Arc<Bellatrix>,
+}
+
+impl BuildDispatcher {
+    pub fn new(storage: Storage, bellatrix: Arc<Bellatrix>) -> Self {
+        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))
+        })?;
+
+        let db_record = self
+            .storage
+            .build_trigger_storage()
+            .insert(
+                trigger.trigger_type.to_string(),
+                trigger.trigger_source.to_string(),
+                trigger_payload,
+            )
+            .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),
+        };
+
+        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
+                    );
+                }
+            }
+        });
+
+        Ok(db_record.id)
+    }
+}

ceres/src/build_trigger/git_push_handler.rs

@@ -0,0 +1,66 @@
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use chrono::Utc;
+use common::errors::MegaError;
+use jupiter::storage::Storage;
+
+use super::changes_calculator::ChangesCalculator;
+use crate::{
+    api_service::cache::GitObjectCache,
+    build_trigger::{
+        BuildTrigger, BuildTriggerPayload, BuildTriggerType, GitPushPayload, TriggerContext,
+        TriggerHandler,
+    },
+};
+
+/// Handler for Git push triggers.
+pub struct GitPushHandler {
+    changes_calculator: ChangesCalculator,
+}
+
+impl GitPushHandler {
+    pub fn new(storage: Storage, git_object_cache: Arc<GitObjectCache>) -> Self {
+        Self {
+            changes_calculator: ChangesCalculator::new(storage, git_object_cache),
+        }
+    }
+}
+
+#[async_trait]
+impl TriggerHandler for GitPushHandler {
+    async fn handle(&self, context: &TriggerContext) -> Result<BuildTrigger, MegaError> {
+        let builds = self
+            .changes_calculator
+            .get_builds_for_commit(context)
+            .await?;
+
+        let cl_link = context.cl_link.clone().unwrap_or_else(|| {
+            format!(
+                "push-{}-{}",
+                Utc::now().timestamp_millis(),
+                &context.commit_hash[..8.min(context.commit_hash.len())]
+            )
+        });
+
+        Ok(BuildTrigger {
+            trigger_type: context.trigger_type,
+            trigger_source: context.trigger_source,
+            trigger_time: Utc::now(),
+            payload: BuildTriggerPayload::GitPush(GitPushPayload {
+                repo: context.repo_path.clone(),
+                from_hash: context.from_hash.clone(),
+                commit_hash: context.commit_hash.clone(),
+                cl_link,
+                cl_id: context.cl_id,
+                builds: serde_json::to_value(&builds)
+                    .map_err(|e| MegaError::Other(format!("Failed to serialize builds: {}", e)))?,
+                triggered_by: context.triggered_by.clone(),
+            }),
+        })
+    }
+
+    fn trigger_type(&self) -> BuildTriggerType {
+        BuildTriggerType::GitPush
+    }
+}

ceres/src/build_trigger/manual_handler.rs

@@ -0,0 +1,107 @@
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use chrono::Utc;
+use common::errors::MegaError;
+use jupiter::storage::Storage;
+
+use super::changes_calculator::ChangesCalculator;
+use crate::{
+    api_service::cache::GitObjectCache,
+    build_trigger::{
+        BuildTrigger, BuildTriggerPayload, BuildTriggerType, ManualPayload, TriggerContext,
+        TriggerHandler,
+    },
+};
+
+/// Handler for manual build triggers.
+pub struct ManualHandler {
+    storage: Storage,
+    changes_calculator: ChangesCalculator,
+}
+
+impl ManualHandler {
+    pub fn new(storage: Storage, git_object_cache: Arc<GitObjectCache>) -> Self {
+        Self {
+            storage: storage.clone(),
+            changes_calculator: ChangesCalculator::new(storage, git_object_cache),
+        }
+    }
+
+    /// Get the parent commit hash for a given commit.
+    /// Returns the first parent if available, otherwise returns the same commit hash.
+    async fn get_parent_commit(&self, commit_hash: &str) -> Result<String, MegaError> {
+        let commit = self
+            .storage
+            .mono_storage()
+            .get_commit_by_hash(commit_hash)
+            .await?
+            .ok_or_else(|| {
+                MegaError::Other(format!("[code:404] Commit not found: {}", commit_hash))
+            })?;
+
+        // Parse parents_id JSON array
+        let parent_ids: Vec<String> =
+            serde_json::from_value(commit.parents_id.clone()).unwrap_or_default();
+
+        // Use first parent if available, otherwise return same hash (initial commit case)
+        Ok(parent_ids
+            .first()
+            .cloned()
+            .unwrap_or_else(|| commit_hash.to_string()))
+    }
+}
+
+#[async_trait]
+impl TriggerHandler for ManualHandler {
+    async fn handle(&self, context: &TriggerContext) -> Result<BuildTrigger, MegaError> {
+        let from_hash = if context.from_hash == context.commit_hash {
+            self.get_parent_commit(&context.commit_hash).await?
+        } else {
+            context.from_hash.clone()
+        };
+
+        let adjusted_context = TriggerContext {
+            from_hash: from_hash.clone(),
+            ..context.clone()
+        };
+
+        let builds = self
+            .changes_calculator
+            .get_builds_for_commit(&adjusted_context)
+            .await?;
+
+        let cl_link = context.cl_link.clone().unwrap_or_else(|| {
+            format!(
+                "manual-{}-{}",
+                Utc::now().timestamp_millis(),
+                &context.commit_hash[..8.min(context.commit_hash.len())]
+            )
+        });
+
+        Ok(BuildTrigger {
+            trigger_type: BuildTriggerType::Manual,
+            trigger_source: context.trigger_source,
+            trigger_time: Utc::now(),
+            payload: BuildTriggerPayload::Manual(ManualPayload {
+                repo: context.repo_path.clone(),
+                commit_hash: context.commit_hash.clone(),
+                triggered_by: context
+                    .triggered_by
+                    .clone()
+                    .unwrap_or_else(|| "unknown".to_string()),
+                builds: serde_json::to_value(&builds)
+                    .map_err(|e| MegaError::Other(format!("Failed to serialize builds: {}", e)))?,
+                params: context.params.clone(),
+                cl_link: cl_link.clone(),
+                cl_id: context.cl_id,
+                ref_name: context.ref_name.clone(),
+                ref_type: context.ref_type.clone(),
+            }),
+        })
+    }
+
+    fn trigger_type(&self) -> BuildTriggerType {
+        BuildTriggerType::Manual
+    }
+}