From 076dd99f95a5c05489b30fb7cf890195809e5076 Mon Sep 17 00:00:00 2001 From: wujian0327 <353981613@qq.com> Date: Tue, 8 Oct 2024 17:10:05 +0800 Subject: [PATCH 1/2] enalbe mega to cache public repo --- common/src/model.rs | 3 + gateway/src/api/ztm_router.rs | 3 +- gateway/src/https_server.rs | 25 +++- gemini/src/http/cache_repo.rs | 222 ++++++++++++++++++++++++++++++++++ gemini/src/http/handler.rs | 24 ++-- gemini/src/http/mod.rs | 1 + gemini/src/lib.rs | 2 - gemini/src/util.rs | 15 +++ 8 files changed, 273 insertions(+), 22 deletions(-) create mode 100644 gemini/src/http/cache_repo.rs diff --git a/common/src/model.rs b/common/src/model.rs index 21778b572..8ed9e57f5 100644 --- a/common/src/model.rs +++ b/common/src/model.rs @@ -14,6 +14,9 @@ pub struct ZtmOptions { #[arg(long)] pub bootstrap_node: Option, + + #[arg(long, default_value_t = false)] + pub cache_repo: bool, } #[derive(Deserialize, Debug)] diff --git a/gateway/src/api/ztm_router.rs b/gateway/src/api/ztm_router.rs index cd423f011..f8fa655bf 100644 --- a/gateway/src/api/ztm_router.rs +++ b/gateway/src/api/ztm_router.rs @@ -50,6 +50,7 @@ async fn repo_provide( state.inner.context.clone(), path, alias, + get_peerid(), ) .await { @@ -110,7 +111,7 @@ async fn alias_to_path( return Err((StatusCode::BAD_REQUEST, String::from("Alias not provide\n"))); } }; - let res = context + let res: Option = context .services .ztm_storage .get_path_from_alias(alias) diff --git a/gateway/src/https_server.rs b/gateway/src/https_server.rs index a3316c676..ab990c5e6 100644 --- a/gateway/src/https_server.rs +++ b/gateway/src/https_server.rs @@ -8,6 +8,7 @@ use axum::{http, Router}; use axum_server::tls_rustls::RustlsConfig; use clap::Args; +use gemini::http::cache_repo::cache_public_repository; use tower::ServiceBuilder; use tower_http::cors::{Any, CorsLayer}; use tower_http::decompression::RequestDecompressionLayer; @@ -187,7 +188,7 @@ pub fn check_run_with_ztm(config: Config, ztm: ZtmOptions, http_port: u16) { Some(bootstrap_node) => { tracing::info!( "The bootstrap node is {}, prepare to join ztm network", - bootstrap_node + bootstrap_node.clone() ); let (peer_id, _) = vault::init(); let ztm_agent: LocalZTMAgent = LocalZTMAgent { @@ -195,9 +196,29 @@ pub fn check_run_with_ztm(config: Config, ztm: ZtmOptions, http_port: u16) { }; ztm_agent.clone().start_ztm_agent(); thread::sleep(time::Duration::from_secs(3)); + + let bootstrap_node_clone = bootstrap_node.clone(); + let config_clone = config.clone(); + let ztm_agent_clone = ztm_agent.clone(); tokio::spawn(async move { - run_ztm_client(bootstrap_node, config, peer_id, ztm_agent, http_port).await + run_ztm_client( + bootstrap_node_clone, + config_clone, + peer_id, + ztm_agent_clone, + http_port, + ) + .await }); + + if ztm.cache_repo { + thread::sleep(time::Duration::from_secs(3)); + tokio::spawn(async move { + let context = Context::new(config.clone()).await; + context.services.mono_storage.init_monorepo().await; + cache_public_repository(bootstrap_node, context, ztm_agent).await + }); + } } None => { tracing::info!("The bootstrap node is not set, prepare to start mega server locally"); diff --git a/gemini/src/http/cache_repo.rs b/gemini/src/http/cache_repo.rs new file mode 100644 index 000000000..7ba6da33c --- /dev/null +++ b/gemini/src/http/cache_repo.rs @@ -0,0 +1,222 @@ +use std::{fs, path::PathBuf, time::Duration}; + +use callisto::ztm_path_mapping; +use common::utils::generate_id; +use jupiter::context::Context; +use tokio::process::Command; +use vault::get_peerid; + +use crate::{ + util::{get_git_model_by_path, handle_response}, + ztm::agent::LocalZTMAgent, + RepoInfo, +}; + +use super::handler::{repo_folk_alias, repo_provide}; + +pub async fn cache_public_repository( + bootstrap_node: String, + context: Context, + agent: LocalZTMAgent, +) { + let mut interval = tokio::time::interval(Duration::from_secs(60 * 5)); + loop { + interval.tick().await; + cache_public_repository_handler(bootstrap_node.clone(), context.clone(), agent.clone()) + .await; + } +} + +async fn cache_public_repository_handler( + bootstrap_node: String, + context: Context, + agent: LocalZTMAgent, +) { + tracing::info!("Start caching public repositories"); + // get public repo by bootstrap_node + let url = format!("{bootstrap_node}/api/v1/repo_list"); + let request_result = reqwest::get(url.clone()).await; + let response_text = match handle_response(request_result).await { + Ok(s) => s, + Err(s) => { + tracing::error!("GET {url} failed,{s}"); + return; + } + }; + let repo_list: Vec = match serde_json::from_slice(response_text.as_bytes()) { + Ok(p) => p, + Err(e) => { + tracing::error!("{}", e); + return; + } + }; + for repo in repo_list { + if !repo.peer_online && repo.origin != get_peerid() { + continue; + } + let bootstrap_node = bootstrap_node.clone(); + let context = context.clone(); + let agent = agent.clone(); + let handle = tokio::spawn(async { + clone_and_share_repo(bootstrap_node, context, agent, repo).await; + }); + handle.await.unwrap(); + } +} + +async fn clone_and_share_repo( + bootstrap_node: String, + context: Context, + agent: LocalZTMAgent, + repo: RepoInfo, +) { + let alias = repo.name; + let repo_path = format!("/third-part/{}", alias.clone()); + + if get_git_model_by_path(context.clone(), repo_path) + .await + .is_some() + { + //exist + //TODO update if origin commit change + tracing::info!("Repository {} exists", alias); + return; + } + + let res: Option = context + .services + .ztm_storage + .get_path_from_alias(&alias) + .await + .unwrap(); + if res.is_none() { + //clone repo from other peer + let clone_url = repo_folk_alias(agent.agent_port, repo.identifier.to_string()).await; + let clone_url = match clone_url { + Ok(d) => d, + Err(_) => return, + }; + tracing::info!("Clone {} with local port: {}", repo.identifier, clone_url); + match clone_repository(clone_url, alias.clone()).await { + Ok(_) => { + tracing::info!("Clone {} to local successfully", repo.identifier); + let _ = share_repository(alias, context, bootstrap_node, repo.origin).await; + } + Err(e) => { + tracing::error!("Clone {} to local failed:{}", repo.identifier, e); + } + } + } +} + +async fn clone_repository(repo_url: String, name: String) -> Result<(), String> { + let base_dir = + PathBuf::from(std::env::var("MEGA_BASE_DIR").unwrap_or_else(|_| "/tmp/.mega".to_string())); + let target_dir = base_dir.join("tmp").join(name.clone()); + + if target_dir.exists() { + fs::remove_dir_all(&target_dir).unwrap(); + } + + tracing::info!( + "Exec: git clone {} {}", + repo_url, + target_dir.to_str().unwrap() + ); + + let mut cmd = Command::new("git") + .arg("clone") + .arg(repo_url) + .arg(target_dir.to_str().unwrap()) + .spawn() + .map_err(|e| format!("Failed to execute process: {}", e))?; + + let status = cmd + .wait() + .await + .map_err(|e| format!("Failed to execute process: {}", e))?; + + tracing::info!("Git clone with result: {}", status); + + change_remote_url(target_dir.clone(), name).await?; + push_to_new_remote(target_dir).await?; + Ok(()) +} + +async fn change_remote_url(repo_path: PathBuf, name: String) -> Result<(), String> { + tracing::info!("Exec: git remote remove origin"); + let _output = Command::new("git") + .arg("remote") + .arg("remove") + .arg("origin") + .current_dir(repo_path.clone()) + .output() + .await + .map_err(|e| format!("Failed to execute process: {}", e))?; + + let new_path = &format!("http://localhost:8000/third-part/{}", name); + tracing::info!("Exec: git remote add origin {}", new_path); + let _output = Command::new("git") + .arg("remote") + .arg("add") + .arg("origin") + .arg(new_path) + .current_dir(repo_path.clone()) + .output() + .await + .map_err(|e| format!("Failed to execute process: {}", e))?; + + Ok(()) +} + +async fn push_to_new_remote(repo_path: PathBuf) -> Result<(), String> { + tracing::info!("Exec: git push origin master"); + let mut cmd = Command::new("git") + .arg("push") + .arg("origin") + .arg("master") + .current_dir(repo_path.clone()) + .spawn() + .map_err(|e| format!("Failed to execute process: {}", e))?; + + let status = cmd + .wait() + .await + .map_err(|e| format!("Failed to execute process: {}", e))?; + + tracing::info!("Git push with result: {}", status); + + Ok(()) +} + +async fn share_repository( + name: String, + context: Context, + bootstrap_node: String, + origin: String, +) -> Result<(), String> { + let repo_path = format!("/third-part/{}", name); + let model: ztm_path_mapping::Model = ztm_path_mapping::Model { + id: generate_id(), + alias: name.clone(), + repo_path: repo_path.to_string(), + created_at: chrono::Utc::now().naive_utc(), + updated_at: chrono::Utc::now().naive_utc(), + }; + context + .services + .ztm_storage + .save_alias_mapping(model.clone()) + .await + .map_err(|e| format!("{}", e))?; + let res = repo_provide(bootstrap_node, context, repo_path.clone(), name, origin).await; + match res { + Ok(_) => { + tracing::info!("Share repo {} successfully", repo_path); + } + Err(e) => { + tracing::error!(e); + } + } + Ok(()) +} diff --git a/gemini/src/http/handler.rs b/gemini/src/http/handler.rs index adb346203..be9694ea9 100644 --- a/gemini/src/http/handler.rs +++ b/gemini/src/http/handler.rs @@ -2,7 +2,7 @@ use common::model::CommonResult; use jupiter::context::Context; use crate::{ - util::repo_alias_to_identifier, + util::{get_git_model_by_path, repo_alias_to_identifier}, ztm::{ agent::share_repo, get_or_create_remote_mega_tunnel, send_get_request_to_peer_by_tunnel, }, @@ -14,24 +14,15 @@ pub async fn repo_provide( context: Context, path: String, alias: String, + origin: String, ) -> Result { let url = format!("{bootstrap_node}/api/v1/repo_provide"); - let git_model = context - .services - .git_db_storage - .find_git_repo_exact_match(path.as_str()) - .await; - let git_model = match git_model { - Ok(r) => { - if let Some(m) = r { - m - } else { - return Err(String::from("Repo not found")); - } - } - Err(_) => return Err(String::from("Repo not found")), + let git_model = match get_git_model_by_path(context.clone(), path).await { + Some(r) => r, + None => return Err(String::from("Repo not found")), }; + let git_ref = context .services .git_db_storage @@ -41,13 +32,12 @@ pub async fn repo_provide( .unwrap(); let name = git_model.repo_name; - let (peer_id, _) = vault::init(); let identifier = repo_alias_to_identifier(alias); let update_time = git_model.created_at.and_utc().timestamp(); let repo_info = RepoInfo { name, identifier, - origin: peer_id, + origin, update_time, commit: git_ref.ref_git_id, peer_online: true, diff --git a/gemini/src/http/mod.rs b/gemini/src/http/mod.rs index 062ae9d9b..691230598 100644 --- a/gemini/src/http/mod.rs +++ b/gemini/src/http/mod.rs @@ -1 +1,2 @@ +pub mod cache_repo; pub mod handler; diff --git a/gemini/src/lib.rs b/gemini/src/lib.rs index a60af9cf4..38e9c585a 100644 --- a/gemini/src/lib.rs +++ b/gemini/src/lib.rs @@ -10,8 +10,6 @@ pub mod nostr; pub mod util; pub mod ztm; - - #[derive(Deserialize, Debug)] pub struct RelayGetParams { pub peer_id: Option, diff --git a/gemini/src/util.rs b/gemini/src/util.rs index 5af2cb575..1474b6d09 100644 --- a/gemini/src/util.rs +++ b/gemini/src/util.rs @@ -1,3 +1,5 @@ +use callisto::git_repo; +use jupiter::context::Context; use std::{ net::TcpListener, time::{SystemTime, UNIX_EPOCH}, @@ -60,3 +62,16 @@ pub fn get_ztm_app_tunnel_bound_name(remote_peer_id: String) -> String { get_short_peer_id(remote_peer_id) ) } + +pub async fn get_git_model_by_path(context: Context, path: String) -> Option { + let git_model = context + .services + .git_db_storage + .find_git_repo_exact_match(path.as_str()) + .await; + + match git_model { + Ok(r) => r, + Err(_) => None, + } +} From 580b3fd96368ed76081525ab7238a2b450ac6e69 Mon Sep 17 00:00:00 2001 From: wujian0327 <353981613@qq.com> Date: Wed, 9 Oct 2024 10:53:05 +0800 Subject: [PATCH 2/2] fix clippy --- gemini/src/util.rs | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/gemini/src/util.rs b/gemini/src/util.rs index 1474b6d09..8f8196d5c 100644 --- a/gemini/src/util.rs +++ b/gemini/src/util.rs @@ -70,8 +70,5 @@ pub async fn get_git_model_by_path(context: Context, path: String) -> Option r, - Err(_) => None, - } + git_model.unwrap_or_default() }