Merge branch 'web3infra-foundation:main' into main

Author: genedna

Cargo.toml

@@ -78,6 +78,7 @@ smallvec = "1.15.1"
 bytes = "1.11"
 chrono = { version = "0.4.43", features = ["serde"] }
 hex = "0.4.3"
+sha2 = "0.10.9"
 
 idgenerator = "2.0.0"
 config = "0.15.19"

api-model/src/buck2/types.rs

@@ -39,3 +39,75 @@ impl ProjectRelativePath {
         opt.map(|s| Self(s.to_owned()))
     }
 }
+
+/// Supported read modes for log APIs.
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema, PartialEq, Eq, Default)]
+#[serde(rename_all = "lowercase")]
+pub enum LogReadMode {
+    #[default]
+    Full,
+    Segment,
+}
+
+/// Log stream event emitted by Orion worker/build processing.
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
+pub struct LogEvent {
+    pub task_id: String,
+    pub repo_name: String,
+    pub build_id: String,
+    pub line: String,
+    pub is_end: bool,
+}
+
+/// Log segment read result.
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
+pub struct LogSegment {
+    pub build_id: String,
+    pub offset: u64,
+    pub len: usize,
+    pub data: String,
+    pub next_offset: u64,
+    pub file_size: u64,
+    /// Whether we reached end of file
+    pub eof: bool,
+}
+
+/// Query parameters for target log APIs.
+#[derive(Debug, Clone, Deserialize, ToSchema)]
+pub struct TargetLogQuery {
+    #[serde(default)]
+    pub r#type: LogReadMode,
+    pub offset: Option<usize>,
+    pub limit: Option<usize>,
+}
+
+/// Query parameters for task history log APIs.
+#[derive(Debug, Clone, Deserialize, ToSchema)]
+pub struct TaskHistoryQuery {
+    pub task_id: String,
+    pub build_id: String,
+    pub repo: String,
+    pub start: Option<usize>,
+    pub end: Option<usize>,
+}
+
+/// Log lines response for history reads.
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
+pub struct LogLinesResponse {
+    pub data: Vec<String>,
+    pub len: usize,
+}
+
+/// Log lines response for target reads.
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
+pub struct TargetLogLinesResponse {
+    pub data: Vec<String>,
+    pub len: usize,
+    pub build_id: String,
+}
+
+/// Error response for log-related APIs.
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
+pub struct LogErrorResponse {
+    pub message: String,
+}

api-model/src/lib.rs

@@ -1,4 +1,3 @@
 pub mod buck2;
 pub mod common;
 pub mod git;
-pub mod orion;

api-model/src/orion/log.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
+    }
+}

api-model/src/orion/mod.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)
+    }
+}