diff --git a/ceres/src/pack/handler.rs b/ceres/src/pack/handler.rs index 37b24b429..9825260f0 100644 --- a/ceres/src/pack/handler.rs +++ b/ceres/src/pack/handler.rs @@ -1,10 +1,9 @@ use std::{ collections::HashSet, - io::Cursor, pin::Pin, sync::{ atomic::{AtomicUsize, Ordering}, - mpsc::{self, Receiver}, + mpsc::Receiver, }, }; @@ -42,27 +41,6 @@ pub trait PackHandler: Send + Sync { (head_hash, refs) } - async fn unpack(&self, pack_config: &PackConfig, pack_file: Bytes) -> Result<(), GitError> { - // #[cfg(debug_assertions)] - // { - // let datetime = chrono::Utc::now().naive_utc(); - // let path = format!("{}.pack", datetime); - // let mut output = std::fs::File::create(path).unwrap(); - // output.write_all(&pack_file).unwrap(); - // } - let (sender, receiver) = mpsc::channel(); - let p = Pack::new( - None, - Some(1024 * 1024 * 1024 * pack_config.pack_decode_mem_size), - Some(pack_config.pack_decode_cache_path.clone()), - pack_config.clean_cache_after_decode, - ); - - p.decode_async(Cursor::new(pack_file), sender); //Pack moved here - - self.save_entry(receiver).await - } - async fn unpack_stream( &self, pack_config: &PackConfig, @@ -81,7 +59,7 @@ pub trait PackHandler: Send + Sync { Ok(receiver) } - async fn save_entry(&self, rx: Receiver) -> Result<(), GitError>; + async fn handle_receiver(&self, rx: Receiver) -> Result<(), GitError>; /// Asynchronously retrieves the full pack data for the specified repository path. /// This function collects commits and nodes from the storage and packs them into diff --git a/ceres/src/pack/import_repo.rs b/ceres/src/pack/import_repo.rs index 6127f7a1d..756d87c5e 100644 --- a/ceres/src/pack/import_repo.rs +++ b/ceres/src/pack/import_repo.rs @@ -49,7 +49,7 @@ impl PackHandler for ImportRepo { self.find_head_hash(refs) } - async fn save_entry(&self, receiver: Receiver) -> Result<(), GitError> { + async fn handle_receiver(&self, receiver: Receiver) -> Result<(), GitError> { let storage = self.context.services.git_db_storage.clone(); let mut entry_list = vec![]; let mut join_tasks = vec![]; diff --git a/ceres/src/pack/monorepo.rs b/ceres/src/pack/monorepo.rs index 9ebbe7ad8..c51766532 100644 --- a/ceres/src/pack/monorepo.rs +++ b/ceres/src/pack/monorepo.rs @@ -10,6 +10,9 @@ use std::{ }; use async_trait::async_trait; +use futures::future::join_all; +use tokio::sync::mpsc; +use tokio_stream::wrappers::ReceiverStream; use callisto::raw_blob; use common::{errors::MegaError, utils::MEGA_BRANCH_NAME}; @@ -23,8 +26,6 @@ use mercury::{ pack::entry::Entry, }, }; -use tokio::sync::mpsc; -use tokio_stream::wrappers::ReceiverStream; use venus::{ import_repo::import_refs::{RefCommand, Refs}, monorepo::mr::MergeRequest, @@ -114,12 +115,12 @@ impl PackHandler for MonoRepo { self.find_head_hash(refs) } - async fn save_entry(&self, receiver: Receiver) -> Result<(), GitError> { + async fn handle_receiver(&self, receiver: Receiver) -> Result<(), GitError> { let storage = self.context.services.mega_storage.clone(); let (mut mr, mr_exist) = self.get_mr().await; - let mut commit_size = 0; + let mut unpack_res = Ok(()); if mr_exist { if mr.from_hash == self.from_hash.clone().unwrap() { let to_hash = self.to_hash.clone().unwrap(); @@ -130,7 +131,20 @@ impl PackHandler for MonoRepo { .add_mr_comment(mr.id, 0, Some(comment)) .await .unwrap(); - commit_size = self.save_entry(receiver).await; + unpack_res = self.save_entry(receiver).await; + if unpack_res.is_err() { + mr.close(); + storage + .add_mr_comment( + mr.id, + 0, + Some("Mega closed MR due to multi commit detected".to_string()), + ) + .await + .unwrap(); + } + } else { + tracing::info!("repeat commit with mr: {}, do nothing", mr.id); } } else { mr.close(); @@ -141,23 +155,12 @@ impl PackHandler for MonoRepo { } storage.update_mr(mr.clone()).await.unwrap(); } else { - commit_size = self.save_entry(receiver).await; - - storage.save_mr(mr.clone()).await.unwrap(); + unpack_res = self.save_entry(receiver).await; + if unpack_res.is_ok() { + storage.save_mr(mr.clone()).await.unwrap(); + } }; - - if commit_size > 1 { - mr.close(); - storage - .add_mr_comment( - mr.id, - 0, - Some("Mega closed MR due to multi commit detected".to_string()), - ) - .await - .unwrap(); - } - Ok(()) + unpack_res } // monorepo full pack should follow the shallow clone command 'git clone --depth=1' @@ -366,22 +369,36 @@ impl MonoRepo { ) } - async fn save_entry(&self, receiver: Receiver) -> i32 { + async fn save_entry(&self, receiver: Receiver) -> Result<(), GitError> { let storage = self.context.services.mega_storage.clone(); let mut entry_list = Vec::new(); - - let mut commit_size = 0; + let mut join_tasks = vec![]; + let mut commit_id = String::new(); for entry in receiver { - if entry.obj_type == ObjectType::Commit { - commit_size += 1; + if commit_id.is_empty() { + if entry.obj_type == ObjectType::Commit { + commit_id = entry.hash.to_plain_str(); + } + } else { + if entry.obj_type == ObjectType::Commit { + return Err(GitError::CustomError( + "only single commit support in each commit".to_string(), + )); + } + if entry_list.len() >= 1000 { + let stg_clone = storage.clone(); + let commit_id = commit_id.clone(); + let handle = tokio::spawn(async move { + stg_clone.save_entry(&commit_id, entry_list).await.unwrap(); + }); + join_tasks.push(handle); + entry_list = vec![]; + } } entry_list.push(entry); - if entry_list.len() >= 1000 { - storage.save_entry(entry_list).await.unwrap(); - entry_list = Vec::new(); - } } - storage.save_entry(entry_list).await.unwrap(); - commit_size + join_all(join_tasks).await; + storage.save_entry(&commit_id, entry_list).await.unwrap(); + Ok(()) } } diff --git a/ceres/src/protocol/smart.rs b/ceres/src/protocol/smart.rs index fd180845b..c71094c5c 100644 --- a/ceres/src/protocol/smart.rs +++ b/ceres/src/protocol/smart.rs @@ -217,13 +217,13 @@ impl SmartProtocol { .await .unwrap(); + // can not block main thread here. let ph_clone = pack_handler.clone(); - let unpack_success = tokio::task::spawn_blocking(move || { + let unpack_result = tokio::task::spawn_blocking(move || { let handle = tokio::runtime::Handle::current(); - handle.block_on(async { ph_clone.save_entry(receiver).await }) + handle.block_on(async { ph_clone.handle_receiver(receiver).await }) }) - .await - .is_ok(); + .await.unwrap(); // write "unpack ok\n to report" add_pkt_line_string(&mut report_status, "unpack ok\n".to_owned()); @@ -240,81 +240,17 @@ impl SmartProtocol { // a.The reference can have changed since the reference discovery phase was originally sent, meaning someone pushed in the meantime. // b.The reference being pushed could be a non-fast-forward reference and the update hooks or configuration could be set to not allow that, etc. // c.Also, some references can be updated while others can be rejected. - if unpack_success { - if !default_exist { - command.default_branch = true; - default_exist = true; + match unpack_result { + Ok(_) => { + if !default_exist { + command.default_branch = true; + default_exist = true; + } + pack_handler.update_refs(&command).await.unwrap(); } - pack_handler.update_refs(&command).await.unwrap(); - } else { - command.failed(String::from("unpack failed")); - } - } - add_pkt_line_string(&mut report_status, command.get_status()); - } - report_status.put(&PKT_LINE_END_MARKER[..]); - let length = report_status.len(); - let mut buf = self.build_side_band_format(report_status, length); - buf.put(&PKT_LINE_END_MARKER[..]); - Ok(buf.into()) - } - - // preserve for ssh server - pub async fn git_receive_pack(&mut self, mut body_bytes: Bytes) -> Result { - if body_bytes.len() < 1_000 { - tracing::debug!("bytes from client: {:?}", body_bytes); - } else { - tracing::debug!("{} bytes from client", body_bytes.len()) - } - while !body_bytes.starts_with(&[b'P', b'A', b'C', b'K']) && !body_bytes.is_empty() { - let (bytes_take, mut pkt_line) = read_pkt_line(&mut body_bytes); - if bytes_take != 0 { - let command = self.parse_ref_command(&mut pkt_line); - self.parse_capabilities(&String::from_utf8(pkt_line.to_vec()).unwrap()); - tracing::debug!( - "parse ref_command: {:?}, with caps:{:?}", - command, - self.capabilities - ); - self.command_list.push(command); - } - } - // handles situation when client send b"0000" - if body_bytes.is_empty() { - return Ok(body_bytes); - } - // After receiving the pack data from the sender, the receiver sends a report - let mut report_status = BytesMut::new(); - let pack_handler = self.pack_handler().await; - //1. unpack progress - let unpack_success = pack_handler - .unpack(&self.context.config.pack, body_bytes) - .await - .is_ok(); - - // write "unpack ok\n to report" - add_pkt_line_string(&mut report_status, "unpack ok\n".to_owned()); - - let mut default_exist = pack_handler.check_default_branch().await; - - //2. update each refs and build report - for mut command in self.command_list.clone() { - if command.ref_type == RefType::Tag { - // just update if refs type is tag - pack_handler.update_refs(&command).await.unwrap(); - } else { - // Updates can be unsuccessful for a number of reasons. - // a.The reference can have changed since the reference discovery phase was originally sent, meaning someone pushed in the meantime. - // b.The reference being pushed could be a non-fast-forward reference and the update hooks or configuration could be set to not allow that, etc. - // c.Also, some references can be updated while others can be rejected. - if unpack_success { - if !default_exist { - command.default_branch = true; - default_exist = true; + Err(ref err) => { + command.failed(err.to_string()); } - pack_handler.update_refs(&command).await.unwrap(); - } else { - command.failed(String::from("unpack failed")); } } add_pkt_line_string(&mut report_status, command.get_status()); diff --git a/gateway/src/api_service/import_service.rs b/gateway/src/api_service/import_service.rs index 0ac0ba159..8937d072c 100644 --- a/gateway/src/api_service/import_service.rs +++ b/gateway/src/api_service/import_service.rs @@ -21,8 +21,8 @@ pub struct ImportRepoService { #[async_trait] impl ApiHandler for ImportRepoService { - async fn get_blob_as_string(&self, _object_id: &str) -> Result { - unreachable!() + async fn get_blob_as_string(&self, _path: PathBuf, _filename: &str) -> Result { + unimplemented!() } async fn get_latest_commit(&self, path: PathBuf) -> Result { diff --git a/gateway/src/api_service/mod.rs b/gateway/src/api_service/mod.rs index 3c74011e3..e2379b637 100644 --- a/gateway/src/api_service/mod.rs +++ b/gateway/src/api_service/mod.rs @@ -16,7 +16,7 @@ const SIGNATURE_END: &str = "-----END PGP SIGNATURE-----"; #[async_trait] pub trait ApiHandler: Send + Sync { - async fn get_blob_as_string(&self, object_id: &str) -> Result; + async fn get_blob_as_string(&self, path: PathBuf, filename: &str) -> Result; async fn get_latest_commit(&self, path: PathBuf) -> Result; @@ -52,7 +52,7 @@ pub trait ApiHandler: Send + Sync { let truncated_text = filtered_text.chars().take(50).collect::(); truncated_text.to_owned() } else { - "".to_owned() + content } } } diff --git a/gateway/src/api_service/mono_service.rs b/gateway/src/api_service/mono_service.rs index 5709d56b7..7e0e27336 100644 --- a/gateway/src/api_service/mono_service.rs +++ b/gateway/src/api_service/mono_service.rs @@ -1,4 +1,4 @@ -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::path::{Component, Path, PathBuf}; use std::str::FromStr; use std::sync::Arc; @@ -6,7 +6,7 @@ use std::sync::Arc; use axum::async_trait; use callisto::db_enums::ConvType; -use callisto::{mega_blob, mega_tree, raw_blob}; +use callisto::{mega_blob, mega_commit, mega_tree, raw_blob}; use common::errors::MegaError; use jupiter::storage::batch_save_model; use jupiter::storage::mega_storage::MegaStorage; @@ -16,7 +16,7 @@ use mercury::internal::object::blob::Blob; use mercury::internal::object::commit::Commit; use mercury::internal::object::tree::{Tree, TreeItem, TreeItemMode}; use venus::monorepo::converter; -use venus::monorepo::mr::{MergeOperation, MergeResult}; +use venus::monorepo::mr::MergeOperation; use crate::api_service::{ApiHandler, SIGNATURE_END}; use crate::model::create_file::CreateFileInfo; @@ -31,11 +31,23 @@ pub struct MonorepoService { #[async_trait] impl ApiHandler for MonorepoService { - async fn get_blob_as_string(&self, object_id: &str) -> Result { - let plain_text = match self.storage.get_raw_blob_by_hash(object_id).await { - Ok(Some(model)) => String::from_utf8(model.data.unwrap()).unwrap(), - _ => String::new(), - }; + async fn get_blob_as_string( + &self, + path: PathBuf, + filename: &str, + ) -> Result { + let (_, tree) = self.search_tree_by_path(&path).await.unwrap(); + let mut plain_text = String::new(); + if let Some(item) = tree.tree_items.into_iter().find(|x| x.name == filename) { + plain_text = match self + .storage + .get_raw_blob_by_hash(&item.id.to_plain_str()) + .await + { + Ok(Some(model)) => String::from_utf8(model.data.unwrap()).unwrap(), + _ => String::new(), + }; + } Ok(BlobObjects { plain_text }) } @@ -84,14 +96,15 @@ impl ApiHandler for MonorepoService { async fn get_tree_commit_info(&self, path: PathBuf) -> Result { match self.search_tree_by_path(&path).await { Ok((_, tree)) => { - let mut commit_map = HashMap::new(); - let mut tree_to_commit = HashMap::new(); + // let mut commit_map = HashMap::new(); + let mut item_to_commit = HashMap::new(); let trees = self .storage .get_trees_by_hashes( tree.tree_items .iter() + .filter(|x| x.mode == TreeItemMode::Tree) .map(|x| x.id.to_plain_str()) .collect(), ) @@ -99,33 +112,53 @@ impl ApiHandler for MonorepoService { .unwrap(); for tree in trees { - let commit_id = tree.commit_id; - tree_to_commit.insert(tree.tree_id, commit_id.clone()); - - let commit = if commit_map.contains_key(&commit_id) { - commit_map.get(&commit_id).cloned() - } else { - self.storage.get_commit_by_hash(&commit_id).await.unwrap() - }; - if let Some(commit) = commit { - commit_map.insert(commit.commit_id.clone(), commit); - } + item_to_commit.insert(tree.tree_id, tree.commit_id); + } + + let blobs = self + .storage + .get_mega_blobs_by_hashes( + tree.tree_items + .iter() + .filter(|x| x.mode == TreeItemMode::Blob) + .map(|x| x.id.to_plain_str()) + .collect(), + ) + .await + .unwrap(); + + for blob in blobs { + item_to_commit.insert(blob.blob_id, blob.commit_id); } let mut items = Vec::new(); - for item in tree.tree_items { - let mut info: TreeCommitItem = item.clone().into(); - let commit: Commit = commit_map - .get(tree_to_commit.get(&item.id.to_plain_str()).unwrap()) - .unwrap() - .clone() - .into(); + let commit_ids: HashSet = item_to_commit.values().cloned().collect(); - info.oid = commit.id.to_plain_str(); - info.message = - self.remove_useless_str(commit.message.clone(), SIGNATURE_END.to_owned()); - info.date = commit.committer.timestamp.to_string(); + let commits = self + .storage + .get_commits_by_hashes(&commit_ids.into_iter().collect()) + .await + .unwrap(); + let commit_map: HashMap = commits + .into_iter() + .map(|x| (x.commit_id.clone(), x)) + .collect(); + for item in tree.tree_items { + let mut info: TreeCommitItem = item.clone().into(); + if let Some(commit_id) = item_to_commit.get(&item.id.to_plain_str()) { + if let Some(model) = commit_map.get(commit_id) { + let commit: Commit = model.clone().into(); + info.oid = commit.id.to_plain_str(); + info.message = self.remove_useless_str( + commit.message.clone(), + SIGNATURE_END.to_owned(), + ); + info.date = commit.committer.timestamp.to_string(); + } else { + tracing::error!("failed fecth commit: {}", commit_id) + } + } items.push(info); } Ok(TreeCommitInfo { @@ -148,15 +181,26 @@ impl MonorepoService { pub async fn create_monorepo_file(&self, file_info: CreateFileInfo) -> Result<(), GitError> { let path = PathBuf::from(file_info.path); + let mut save_trees = vec![]; + + let (update_trees, search_tree) = self.search_tree_by_path(&path).await.unwrap(); + let mut t_items = search_tree.tree_items; let new_item = if file_info.is_directory { - let blob = converter::generate_git_keep(); + if t_items + .iter() + .any(|x| x.mode == TreeItemMode::Tree && x.name == file_info.name) + { + return Err(GitError::CustomError("Duplicate name".to_string())); + } + let blob = converter::generate_git_keep_with_timestamp(); let tree_item = TreeItem { mode: TreeItemMode::Blob, id: blob.id, name: String::from(".gitkeep"), }; let child_tree = Tree::from_tree_items(vec![tree_item]).unwrap(); + save_trees.push(child_tree.clone()); TreeItem { mode: TreeItemMode::Tree, id: child_tree.id, @@ -164,7 +208,7 @@ impl MonorepoService { } } else { let blob = Blob::from_content(&file_info.content.unwrap()); - let mega_blob: mega_blob::Model = blob.clone().into(); + let mega_blob: mega_blob::Model = (&blob).into(); let mega_blob: mega_blob::ActiveModel = mega_blob.into(); let raw_blob: raw_blob::Model = blob.clone().into(); let raw_blob: raw_blob::ActiveModel = raw_blob.into(); @@ -180,40 +224,34 @@ impl MonorepoService { name: file_info.name.clone(), } }; - - let (tree_vec, search_tree) = self.search_tree_by_path(&path).await.unwrap(); - - let mut t_items = search_tree.tree_items; - // todo: need check if file exist? t_items.push(new_item); - let new_tree = Tree::from_tree_items(t_items).unwrap(); + let p_tree = Tree::from_tree_items(t_items).unwrap(); let refs = self.storage.get_ref("/").await.unwrap().unwrap(); let commit = Commit::from_tree_id( - new_tree.id, + p_tree.id, vec![SHA1::from_str(&refs.ref_commit_hash).unwrap()], &format!("create file {} commit", file_info.name), ); - let tree_model: mega_tree::Model = new_tree.into(); - let tree_model: mega_tree::ActiveModel = tree_model.into(); - - batch_save_model(self.storage.get_connection(), vec![tree_model]) - .await - .unwrap(); - - self.update_parent_tree(path, tree_vec, commit) + let commit_id = self + .update_parent_tree(path, update_trees, commit) .await .unwrap(); + save_trees.push(p_tree); + for save_t in save_trees { + let mut tree_model: mega_tree::Model = save_t.into(); + tree_model.commit_id.clone_from(&commit_id); + let tree_model: mega_tree::ActiveModel = tree_model.into(); + batch_save_model(self.storage.get_connection(), vec![tree_model]) + .await + .unwrap(); + } Ok(()) } - pub async fn merge_mr(&self, op: MergeOperation) -> Result { - let mut res = MergeResult { - result: true, - err_message: "".to_owned(), - }; + pub async fn merge_mr(&self, op: MergeOperation) -> Result<(), MegaError> { if let Some(mut mr) = self.storage.get_open_mr_by_id(op.mr_id).await.unwrap() { let refs = self.storage.get_ref(&mr.path).await.unwrap().unwrap(); @@ -237,8 +275,7 @@ impl MonorepoService { .unwrap(); if mr.path != "/" { let path = PathBuf::from(mr.path.clone()); - - // beacuse only need parent tree so we skip current directory + // beacuse only parent tree is needed so we skip current directory let (tree_vec, _) = self .search_tree_by_path(path.parent().unwrap()) .await @@ -248,17 +285,15 @@ impl MonorepoService { .unwrap(); // remove refs start with path self.storage.remove_refs(&mr.path).await.unwrap(); - // todo: self.clean_dangling_commits().await; + // TODO: self.clean_dangling_commits().await; } } else { - res.result = false; - "ref hash conflict".clone_into(&mut res.err_message); + return Err(MegaError::with_message("ref hash conflict")); } } else { - res.result = false; - "Invalid mr id".clone_into(&mut res.err_message); + return Err(MegaError::with_message("Invalid mr id")); } - Ok(res) + Ok(()) } /// Searches for a tree and affected parent by path. @@ -326,7 +361,7 @@ impl MonorepoService { mut path: PathBuf, mut tree_vec: Vec, commit: Commit, - ) -> Result<(), GitError> { + ) -> Result { let mut save_trees = Vec::new(); let mut p_commit_id = String::new(); @@ -380,7 +415,7 @@ impl MonorepoService { batch_save_model(self.storage.get_connection(), save_trees) .await .unwrap(); - Ok(()) + Ok(p_commit_id) } } diff --git a/gateway/src/api_service/router.rs b/gateway/src/api_service/router.rs index b18845e06..b79943726 100644 --- a/gateway/src/api_service/router.rs +++ b/gateway/src/api_service/router.rs @@ -1,4 +1,4 @@ -use std::{collections::HashMap, path::PathBuf}; +use std::path::PathBuf; use axum::{ extract::{Query, State}, @@ -11,12 +11,15 @@ use axum::{ use jupiter::context::Context; use venus::{ import_repo::repo::Repo, - monorepo::mr::{MergeOperation, MergeResult}, + monorepo::mr::{CommonResult, MergeOperation}, }; use crate::{ api_service::import_service::ImportRepoService, - model::{create_file::CreateFileInfo, query::CodePreviewQuery}, + model::{ + create_file::CreateFileInfo, + query::{BlobContentQuery, CodePreviewQuery}, + }, }; use crate::{ api_service::mono_service::MonorepoService, @@ -26,8 +29,6 @@ use crate::{api_service::ApiHandler, model::objects::TreeBriefInfo}; #[derive(Clone)] pub struct ApiServiceState { - // pub mega_storage: Arc, - // pub git_db_storage: Arc, pub context: Context, } @@ -41,25 +42,24 @@ impl ApiServiceState { pub async fn api_handler(&self, path: PathBuf) -> Box { let import_dir = self.context.config.monorepo.import_dir.clone(); if path.starts_with(&import_dir) && path != import_dir { - let repo: Repo = self + if let Some(model) = self .context .services .git_db_storage .find_git_repo(path.to_str().unwrap()) .await .unwrap() - .unwrap() - .into(); - - Box::new(ImportRepoService { - storage: self.context.services.git_db_storage.clone(), - repo, - }) - } else { - Box::new(MonorepoService { - storage: self.context.services.mega_storage.clone(), - }) + { + let repo: Repo = model.into(); + return Box::new(ImportRepoService { + storage: self.context.services.git_db_storage.clone(), + repo, + }); + } } + Box::new(MonorepoService { + storage: self.context.services.mega_storage.clone(), + }) } } @@ -80,13 +80,13 @@ pub fn routers() -> Router { } async fn get_blob_object( - Query(query): Query>, + Query(query): Query, state: State, ) -> Result, (StatusCode, String)> { - let object_id = query.get("object_id").unwrap(); let res = state - .monorepo() - .get_blob_as_string(object_id) + .api_handler(query.path.clone().into()) + .await + .get_blob_as_string(query.path.into(), &query.name) .await .unwrap(); Ok(Json(res)) @@ -123,20 +123,29 @@ async fn init(state: State) { async fn create_file( state: State, Json(json): Json, -) -> Result, (StatusCode, String)> { - state +) -> Result, (StatusCode, String)> { + let res = state .monorepo() .create_monorepo_file(json.clone()) - .await - .unwrap(); - Ok(Json(json)) + .await; + let res = if res.is_err() { + CommonResult::failed(&res.err().unwrap().to_string()) + } else { + CommonResult::succrss() + }; + Ok(Json(res)) } async fn merge( state: State, Json(json): Json, -) -> Result, (StatusCode, String)> { - let res = state.monorepo().merge_mr(json.clone()).await.unwrap(); +) -> Result, (StatusCode, String)> { + let res = state.monorepo().merge_mr(json.clone()).await; + let res = if res.is_err() { + CommonResult::failed(&res.err().unwrap().to_string()) + } else { + CommonResult::succrss() + }; Ok(Json(res)) } diff --git a/gateway/src/git_protocol/ssh.rs b/gateway/src/git_protocol/ssh.rs index 64ddc4590..8110d2008 100644 --- a/gateway/src/git_protocol/ssh.rs +++ b/gateway/src/git_protocol/ssh.rs @@ -6,7 +6,7 @@ use std::sync::{Arc, Mutex}; use async_trait::async_trait; use bytes::{Bytes, BytesMut}; use chrono::{DateTime, Duration, Utc}; -use futures::StreamExt; +use futures::{stream, StreamExt}; use russh::server::{self, Auth, Msg, Response, Session}; use russh::{Channel, ChannelId}; use russh_keys::key; @@ -226,8 +226,12 @@ impl SshServer { async fn handle_receive_pack(&mut self, channel: ChannelId, session: &mut Session) { let smart_protocol = self.smart_protocol.as_mut().unwrap(); + let data = self.data_combined.clone(); + let stream = stream::once(async move { + Ok(Bytes::from(data)) + }); let buf = smart_protocol - .git_receive_pack(Bytes::from(self.data_combined.to_vec())) + .git_receive_pack_stream(Box::pin(stream)) .await .unwrap(); tracing::info!("report status: {:?}", buf); diff --git a/gateway/src/model/query.rs b/gateway/src/model/query.rs index 51a720af8..e5c552504 100644 --- a/gateway/src/model/query.rs +++ b/gateway/src/model/query.rs @@ -7,6 +7,7 @@ pub struct DirectoryQuery { #[serde(default = "default_path")] pub repo_path: String, } + #[allow(dead_code)] #[derive(Debug, Deserialize)] pub struct CodePreviewQuery { @@ -16,6 +17,13 @@ pub struct CodePreviewQuery { pub path: String, } +#[derive(Debug, Deserialize)] +pub struct BlobContentQuery { + #[serde(default)] + pub name: String, + #[serde(default = "default_path")] + pub path: String, +} fn default_path() -> String { "/".to_string() diff --git a/jupiter/src/storage/mega_storage.rs b/jupiter/src/storage/mega_storage.rs index a2d7c2364..a49b54ad0 100644 --- a/jupiter/src/storage/mega_storage.rs +++ b/jupiter/src/storage/mega_storage.rs @@ -1,6 +1,6 @@ -use std::sync::Arc; +use std::sync::{Arc, Mutex}; -use common::config::StorageConfig; +use rayon::iter::{IntoParallelRefIterator, ParallelIterator}; use sea_orm::ActiveValue::NotSet; use sea_orm::{ ActiveModelTrait, ColumnTrait, DatabaseConnection, EntityTrait, IntoActiveModel, QueryFilter, @@ -9,8 +9,9 @@ use sea_orm::{ use callisto::db_enums::{ConvType, MergeStatus}; use callisto::{ - mega_commit, mega_mr, mega_mr_comment, mega_mr_conv, mega_refs, mega_tree, raw_blob, + mega_blob, mega_commit, mega_mr, mega_mr_comment, mega_mr_conv, mega_refs, mega_tree, raw_blob, }; +use common::config::StorageConfig; use common::errors::MegaError; use common::utils::generate_id; use mercury::internal::object::MegaObjectModel; @@ -189,42 +190,54 @@ impl MegaStorage { Ok(()) } - pub async fn save_entry(&self, entry_list: Vec) -> Result<(), MegaError> { - let mut commits = Vec::new(); - let mut trees = Vec::new(); - let mut blobs = Vec::new(); - let mut raw_blobs = Vec::new(); - let mut tags = Vec::new(); + pub async fn save_entry( + &self, + commit_id: &str, + entry_list: Vec, + ) -> Result<(), MegaError> { + let (commits, trees, blobs, raw_blobs, tags) = ( + Mutex::new(Vec::new()), + Mutex::new(Vec::new()), + Mutex::new(Vec::new()), + Mutex::new(Vec::new()), + Mutex::new(Vec::new()), + ); - for entry in entry_list { + entry_list.par_iter().for_each(|entry| { let raw_obj = entry.process_entry(); let model = raw_obj.convert_to_mega_model(); match model { - MegaObjectModel::Commit(commit) => commits.push(commit.into_active_model()), - MegaObjectModel::Tree(tree) => { - trees.push(tree.clone().into_active_model()); + MegaObjectModel::Commit(commit) => { + commits.lock().unwrap().push(commit.into_active_model()) } - MegaObjectModel::Blob(blob, raw) => { - blobs.push(blob.clone().into_active_model()); - raw_blobs.push(raw.into_active_model()); + MegaObjectModel::Tree(mut tree) => { + commit_id.clone_into(&mut tree.commit_id); + trees.lock().unwrap().push(tree.into_active_model()); } - MegaObjectModel::Tag(tag) => tags.push(tag.into_active_model()), + MegaObjectModel::Blob(mut blob, raw) => { + commit_id.clone_into(&mut blob.commit_id); + blobs.lock().unwrap().push(blob.clone().into_active_model()); + raw_blobs.lock().unwrap().push(raw.into_active_model()); + } + MegaObjectModel::Tag(tag) => tags.lock().unwrap().push(tag.into_active_model()), } - } + }); - batch_save_model(self.get_connection(), commits) + batch_save_model(self.get_connection(), commits.into_inner().unwrap()) .await .unwrap(); - batch_save_model(self.get_connection(), trees) + batch_save_model(self.get_connection(), trees.into_inner().unwrap()) .await .unwrap(); - batch_save_model(self.get_connection(), blobs) + batch_save_model(self.get_connection(), blobs.into_inner().unwrap()) .await .unwrap(); - batch_save_model(self.get_connection(), raw_blobs) + batch_save_model(self.get_connection(), raw_blobs.into_inner().unwrap()) + .await + .unwrap(); + batch_save_model(self.get_connection(), tags.into_inner().unwrap()) .await .unwrap(); - batch_save_model(self.get_connection(), tags).await.unwrap(); Ok(()) } @@ -313,6 +326,17 @@ impl MegaStorage { .unwrap()) } + pub async fn get_mega_blobs_by_hashes( + &self, + hashes: Vec, + ) -> Result, MegaError> { + Ok(mega_blob::Entity::find() + .filter(mega_blob::Column::BlobId.is_in(hashes)) + .all(self.get_connection()) + .await + .unwrap()) + } + pub async fn get_raw_blobs_by_hashes( &self, hashes: Vec, diff --git a/mega/config.toml b/mega/config.toml index d13aa606f..c993e83cc 100644 --- a/mega/config.toml +++ b/mega/config.toml @@ -51,7 +51,7 @@ obs_endpoint = "https://obs.cn-east-3.myhuaweicloud.com" [monorepo] -## Only import directory support multi-branch commit and tag, repo under regular directory only support main branch only +## Only import directory support multi-branch commit and tag, monorepo only support main branch ## Mega treats files under this directory as import repo and other directories as monorepo import_dir = "/third-part" diff --git a/mercury/Cargo.toml b/mercury/Cargo.toml index e43520a75..3b93802a8 100644 --- a/mercury/Cargo.toml +++ b/mercury/Cargo.toml @@ -25,7 +25,7 @@ uuid = { version = "1.7.0", features = ["v4"] } sha1_smol = "1.0.0" threadpool = "1.8.1" num_cpus.workspace = true -dashmap = "6.0.0" +dashmap = "5.5.3" tokio.workspace = true lru-mem = "0.3.0" bincode = "1.3.3" diff --git a/mercury/src/errors.rs b/mercury/src/errors.rs index cddab1d37..647ea1355 100644 --- a/mercury/src/errors.rs +++ b/mercury/src/errors.rs @@ -83,6 +83,9 @@ pub enum GitError { #[error("Network Error: {0}")] NetworkError(String), + + #[error("{0}")] + CustomError(String), } impl From for GitError { diff --git a/mercury/src/internal/model/blob.rs b/mercury/src/internal/model/blob.rs index 64c3468ba..97aea30d7 100644 --- a/mercury/src/internal/model/blob.rs +++ b/mercury/src/internal/model/blob.rs @@ -5,8 +5,8 @@ use common::utils::generate_id; use crate::{hash::SHA1, internal::object::blob::Blob}; -impl From for mega_blob::Model { - fn from(value: Blob) -> Self { +impl From<&Blob> for mega_blob::Model { + fn from(value: &Blob) -> Self { mega_blob::Model { id: generate_id(), blob_id: value.id.to_plain_str(), diff --git a/mercury/src/internal/object/mod.rs b/mercury/src/internal/object/mod.rs index 8a042344c..01c9d7b71 100644 --- a/mercury/src/internal/object/mod.rs +++ b/mercury/src/internal/object/mod.rs @@ -74,45 +74,19 @@ pub enum MegaObjectModel { impl GitObject { pub fn convert_to_mega_model(self) -> MegaObjectModel { match self { - GitObject::Commit(commit) => { - let mega_commit: mega_commit::Model = commit.into(); - MegaObjectModel::Commit(mega_commit) - } - GitObject::Tree(tree) => { - let mega_tree: mega_tree::Model = tree.into(); - MegaObjectModel::Tree(mega_tree) - } - GitObject::Blob(blob) => { - let mega_blob: mega_blob::Model = blob.clone().into(); - let raw_blob: raw_blob::Model = blob.into(); - MegaObjectModel::Blob(mega_blob, raw_blob) - } - GitObject::Tag(tag) => { - let mega_tag: mega_tag::Model = tag.into(); - MegaObjectModel::Tag(mega_tag) - } + GitObject::Commit(commit) => MegaObjectModel::Commit(commit.into()), + GitObject::Tree(tree) => MegaObjectModel::Tree(tree.into()), + GitObject::Blob(blob) => MegaObjectModel::Blob((&blob).into(), blob.into()), + GitObject::Tag(tag) => MegaObjectModel::Tag(tag.into()), } } pub fn convert_to_git_model(self) -> GitObjectModel { match self { - GitObject::Commit(commit) => { - let commit: git_commit::Model = commit.into(); - GitObjectModel::Commit(commit) - } - GitObject::Tree(tree) => { - let tree: git_tree::Model = tree.into(); - GitObjectModel::Tree(tree) - } - GitObject::Blob(blob) => { - let git_blob: git_blob::Model = (&blob).into(); - let raw_blob: raw_blob::Model = blob.into(); - GitObjectModel::Blob(git_blob, raw_blob) - } - GitObject::Tag(tag) => { - let tag: git_tag::Model = tag.into(); - GitObjectModel::Tag(tag) - } + GitObject::Commit(commit) => GitObjectModel::Commit(commit.into()), + GitObject::Tree(tree) => GitObjectModel::Tree(tree.into()), + GitObject::Blob(blob) => GitObjectModel::Blob((&blob).into(), blob.into()), + GitObject::Tag(tag) => GitObjectModel::Tag(tag.into()), } } } diff --git a/moon/src/components/TopNavbar.js b/moon/src/components/TopNavbar.js index e49ac51b1..6879f4311 100644 --- a/moon/src/components/TopNavbar.js +++ b/moon/src/components/TopNavbar.js @@ -8,14 +8,14 @@ const TopNavbar = () => { {/* level first */}
- logo + logo
- - + +
{ setShowTree(true); - router.push(`/?object_id=${file.id}`); + router.push({ + query: { ...router.query, name: `${file.name}` } + }); }; const handleDirectoryClick = async (directory) => { - router.push(`/?repo_path=/projects/freighter&object_id=${directory.id}`); + const c_path = router.query.path || ''; + const newPath = `${c_path}/${directory.name}`; + + // router.push(`/?path=${directory.path}`); + router.push({ + pathname: router.pathname, + query: { ...router.query, path: newPath }, + }); + setShowTree(true); try { - const response = await fetch(`/api/tree?repo_path=/projects/freighter&object_id=${encodeURIComponent(directory.id)}`); + const response = await fetch(`/api/tree?path=${encodeURIComponent(directory.path)}`); if (!response.ok) { throw new Error('Failed to fetch tree data'); @@ -162,10 +172,12 @@ const HomePage = ({ rootDirectory, directory, readmeContent, fileContent, TreeDa const onExpand = async (keys, { expanded, node }) => { // push new url and query to router console.log("OnExpanded!"); - router.push({ query: { repo_path: "/projects/freighter", object_id: node.key } }); + console.log("keys", keys); + console.log("node", node.path); + // router.push({ query: { repo_path: "/projects/freighter", object_id: node.key } }); var responseData = ''; try { - const response = await fetch(`/api/tree?repo_path=/projects/freighter&object_id=${encodeURIComponent(node.key)}`); + const response = await fetch(`/api/tree?path=${encodeURIComponent(node.path)}`); if (!response.ok) { throw new Error('Failed to fetch tree data'); @@ -196,15 +208,15 @@ const HomePage = ({ rootDirectory, directory, readmeContent, fileContent, TreeDa if (index === 0) { console.log("clicked root path"); setShowTree(false); - router.push(`/?repo_path=/projects/freighter`); + router.push(`/?path=/`); } else { setCurrentPath(currentPath.slice(0, index + 1)); - router.push(`/?repo_path=/projects/freighter&object_id=${key}`); + router.push(`/?path=${key}`); // reRender the tree for back to clicked dir var responseData = ''; try { - const response = await fetch(`/api/tree?repo_path=/projects/freighter&object_id=${encodeURIComponent(key)}`); + const response = await fetch(`/api/tree?path=${encodeURIComponent(key)}`); if (!response.ok) { throw new Error('Failed to fetch tree data'); @@ -265,10 +277,10 @@ const HomePage = ({ rootDirectory, directory, readmeContent, fileContent, TreeDa Name - commit + Message - commitData + Date @@ -298,8 +310,8 @@ const HomePage = ({ rootDirectory, directory, readmeContent, fileContent, TreeDa handleDirectoryClick(project)}>{project.name} )} - {project.commit_msg} - {project.commit_date} + {project.message} + {project.date} ))} @@ -356,13 +368,13 @@ const HomePage = ({ rootDirectory, directory, readmeContent, fileContent, TreeDa export async function getServerSideProps(context) { const MEGA_URL = 'http://localhost:8000'; // get the parameters form context - const { repo_path, object_id } = context.query; - const rootDirectory = (await axios.get(`${MEGA_URL}/api/v1/tree?repo_path=/projects/freighter`)).data; + const { path, name } = context.query; + const rootDirectory = (await axios.get(`${MEGA_URL}/api/v1/tree?path=/`)).data; - // obtain the current directory, the root directory only has the 'path' parameter without the 'id' parameter. Both parameters exist for non-root directories - const response = repo_path && object_id - ? await axios.get(`${MEGA_URL}/api/v1/tree?repo_path=/projects/freighter&object_id=${encodeURIComponent(object_id)}`) - : await axios.get(`${MEGA_URL}/api/v1/tree?repo_path=/projects/freighter`); + // obtain the current directory + const response = path + ? await axios.get(`${MEGA_URL}/api/v1/tree-commit-info?path=${encodeURIComponent(path)}`) + : await axios.get(`${MEGA_URL}/api/v1/tree-commit-info?path=/`); const directory = response.data; var readmeContent = ''; @@ -370,10 +382,10 @@ export async function getServerSideProps(context) { var TreeData = ''; // get the file content - if (object_id) { + if (name) { try { - const fileResponse = await axios.get(`${MEGA_URL}/api/v1/blob?object_id=${object_id}`, { withCredentials: true }); - fileContent = fileResponse.data.row_data; + const fileResponse = await axios.get(`${MEGA_URL}/api/v1/blob?path=${path}&name=${name}`, { withCredentials: true }); + fileContent = fileResponse.data.plain_text; } catch (error) { console.error("Error fetching file content:", error); } @@ -383,8 +395,8 @@ export async function getServerSideProps(context) { for (const project of directory.items || []) { if (project.name === 'README.md' && project.content_type === 'file') { try { - const response = await axios.get(`${MEGA_URL}/api/v1/blob?object_id=${project.id}`, { withCredentials: true }); - readmeContent = response.data.row_data; + const response = await axios.get(`${MEGA_URL}/api/v1/blob?path=${path}&name=${project.name}`, { withCredentials: true }); + readmeContent = response.data.plain_text; break; } catch (error) { console.error("Error fetching README content:", error); diff --git a/venus/src/monorepo/converter.rs b/venus/src/monorepo/converter.rs index 806a12944..d669044e6 100644 --- a/venus/src/monorepo/converter.rs +++ b/venus/src/monorepo/converter.rs @@ -13,6 +13,14 @@ pub fn generate_git_keep() -> Blob { Blob::from_content(&git_keep_content) } +pub fn generate_git_keep_with_timestamp() -> Blob { + let git_keep_content = format!( + "This file was used to maintain the git tree, generate at:{}", + chrono::Utc::now().naive_utc() + ); + Blob::from_content(&git_keep_content) +} + pub fn init_trees(git_keep: &Blob) -> (HashMap, Tree) { let tree_item = TreeItem { mode: TreeItemMode::Blob, @@ -83,7 +91,7 @@ impl MegaModelConverter { self.traverse_for_update(child_tree); } else { let blob = self.blob_maps.get(&item.id).unwrap(); - let mut mega_blob: mega_blob::Model = blob.to_owned().into(); + let mut mega_blob: mega_blob::Model = blob.into(); mega_blob.commit_id = self.commit.id.to_plain_str(); self.mega_blobs .borrow_mut() diff --git a/venus/src/monorepo/mr.rs b/venus/src/monorepo/mr.rs index 27af72f6d..65b36a263 100644 --- a/venus/src/monorepo/mr.rs +++ b/venus/src/monorepo/mr.rs @@ -78,7 +78,22 @@ pub struct MergeOperation { #[derive(PartialEq, Eq, Debug, Clone, Default, Serialize, Deserialize)] -pub struct MergeResult { +pub struct CommonResult { pub result: bool, pub err_message: String, } + +impl CommonResult { + pub fn succrss() -> Self { + CommonResult { + result: true, + err_message: "".to_owned(), + } + } + pub fn failed(err_message: &str) -> Self { + CommonResult { + result: false, + err_message: err_message.to_string(), + } + } +}