diff --git a/aries/src/service/relay_server.rs b/aries/src/service/relay_server.rs index ce5f220ff..7a0f1921f 100644 --- a/aries/src/service/relay_server.rs +++ b/aries/src/service/relay_server.rs @@ -8,12 +8,12 @@ use axum::http::StatusCode; use axum::response::IntoResponse; use axum::routing::{get, post}; use axum::{Json, Router}; -use callisto::{ztm_node, ztm_repo_info}; +use callisto::{ztm_lfs_info, ztm_node, ztm_repo_info}; use clap::Parser; use common::config::Config; use gemini::ztm::hub::{LocalHub, ZTMUserPermit, ZTMCA}; use gemini::ztm::send_get_request_to_peer_by_tunnel; -use gemini::{Node, RelayGetParams, RelayResultRes, RepoInfo}; +use gemini::{LFSInfo, LFSInfoPostBody, Node, RelayGetParams, RelayResultRes, RepoInfo}; use jupiter::context::Context; use tower::ServiceBuilder; use tower_http::cors::{Any, CorsLayer}; @@ -46,7 +46,7 @@ pub struct RelayOptions { pub only_agent: bool, #[arg(long, short)] - pub config: Option + pub config: Option, } #[derive(Clone)] @@ -90,7 +90,9 @@ pub fn routers() -> Router { .route("/node_list", get(node_list)) .route("/repo_provide", post(repo_provide)) .route("/repo_list", get(repo_list)) - .route("/test/send", get(send_message)); + .route("/test/send", get(send_message)) + .route("/lfs_share", post(lfs_share)) + .route("/lfs_list", get(lfs_list)); Router::new() .merge(router) @@ -209,6 +211,52 @@ pub async fn repo_list( Ok(Json(repo_info_list_result)) } +pub async fn lfs_share( + state: State, + Json(lfs_info): Json, +) -> Result, (StatusCode, String)> { + let ztm_lfs_model: ztm_lfs_info::Model = lfs_info.into(); + let storage = state.context.services.ztm_storage.clone(); + match storage.insert_lfs_info(ztm_lfs_model).await { + Ok(_) => Ok(Json(RelayResultRes { success: true })), + Err(_) => Err(( + StatusCode::INTERNAL_SERVER_ERROR, + "invalid paras".to_string(), + )), + } +} + +pub async fn lfs_list( + Query(_query): Query, + state: State, +) -> Result>, (StatusCode, String)> { + let storage = state.context.services.ztm_storage.clone(); + let lfs_info_list: Vec = storage + .get_all_lfs_info() + .await + .unwrap() + .into_iter() + .map(|x| x.into()) + .collect(); + let nodelist: Vec = storage + .get_all_node() + .await + .unwrap() + .into_iter() + .map(|x| x.into()) + .collect(); + let mut lfs_info_list_result = vec![]; + for mut lfs in lfs_info_list { + for node in &nodelist { + if lfs.peer_id == node.peer_id { + lfs.peer_online = node.online; + } + } + lfs_info_list_result.push(lfs.clone()); + } + Ok(Json(lfs_info_list_result)) +} + async fn send_message( Query(query): Query>, state: State, diff --git a/common/src/model.rs b/common/src/model.rs index aa7a56c69..b4107082b 100644 --- a/common/src/model.rs +++ b/common/src/model.rs @@ -16,7 +16,7 @@ pub struct ZtmOptions { pub bootstrap_node: Option, #[arg(long, default_value_t = false)] - pub cache_repo: bool, + pub cache: bool, } #[derive(Deserialize, Debug)] @@ -75,4 +75,4 @@ pub struct RequestParams { pub struct CommonPage { pub total: u64, pub items: Vec, -} \ No newline at end of file +} diff --git a/gateway/src/https_server.rs b/gateway/src/https_server.rs index 54addf2e1..6ebb7faa0 100644 --- a/gateway/src/https_server.rs +++ b/gateway/src/https_server.rs @@ -8,7 +8,7 @@ use axum::{http, Router}; use axum_server::tls_rustls::RustlsConfig; use clap::Args; -use gemini::http::cache_repo::cache_public_repository; +use gemini::cache::cache_public_repo_and_lfs; use tower::ServiceBuilder; use tower_http::cors::{Any, CorsLayer}; use tower_http::decompression::RequestDecompressionLayer; @@ -208,10 +208,10 @@ pub fn check_run_with_ztm(context: Context, ztm: ZtmOptions, http_port: u16) { .await }); - if ztm.cache_repo { + if ztm.cache { thread::sleep(time::Duration::from_secs(3)); tokio::spawn(async move { - cache_public_repository(bootstrap_node, context, ztm_agent).await + cache_public_repo_and_lfs(bootstrap_node, context, ztm_agent, http_port).await }); } } diff --git a/gemini/src/cache/mod.rs b/gemini/src/cache/mod.rs new file mode 100755 index 000000000..2e8c0c936 --- /dev/null +++ b/gemini/src/cache/mod.rs @@ -0,0 +1,449 @@ +use std::{ + fs::{self, create_dir_all, File}, + io::{Read, Write}, + path::PathBuf, + time::Duration, +}; + +use crate::{ + http::handler::{repo_folk_alias, repo_provide}, + lfs::share_lfs, + util::{get_git_model_by_path, handle_response}, + ztm::{agent::LocalZTMAgent, get_or_create_remote_mega_tunnel}, + LFSInfo, RepoInfo, +}; +use callisto::ztm_path_mapping; +use common::utils::generate_id; +use jupiter::context::Context; +use reqwest::{get, Client}; +use tokio::process::Command; +use vault::get_peerid; + +pub async fn cache_public_repo_and_lfs( + bootstrap_node: String, + context: Context, + agent: LocalZTMAgent, + http_port: u16, +) { + 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(), + http_port, + ) + .await; + cache_public_lfs_handler( + bootstrap_node.clone(), + context.clone(), + agent.clone(), + http_port, + ) + .await; + } +} + +async fn cache_public_repository_handler( + bootstrap_node: String, + context: Context, + agent: LocalZTMAgent, + http_port: u16, +) { + 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 move { + clone_and_share_repo(bootstrap_node, context, agent, repo, http_port).await; + }); + handle.await.unwrap(); + } +} + +async fn clone_and_share_repo( + bootstrap_node: String, + context: Context, + agent: LocalZTMAgent, + repo: RepoInfo, + http_port: u16, +) { + 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(), http_port).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, http_port: u16) -> 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, http_port).await?; + push_to_new_remote(target_dir).await?; + Ok(()) +} + +async fn change_remote_url(repo_path: PathBuf, name: String, http_port: u16) -> 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:{}/third-part/{}", http_port, 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(()) +} + +async fn cache_public_lfs_handler( + bootstrap_node: String, + context: Context, + agent: LocalZTMAgent, + http_port: u16, +) { + tracing::info!("Start caching public lfs"); + // get public lfs by bootstrap_node + let url = format!("{bootstrap_node}/api/v1/lfs_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 lfs_list: Vec = match serde_json::from_slice(response_text.as_bytes()) { + Ok(p) => p, + Err(e) => { + tracing::error!("{}", e); + return; + } + }; + for lfs in lfs_list { + if !lfs.peer_online || lfs.peer_id == get_peerid() { + continue; + } + let lfs_object = context + .services + .lfs_db_storage + .get_lfs_object(lfs.file_hash.clone()) + .await + .unwrap(); + if lfs_object.is_some() { + tracing::info!("lfs {} exists, skip", lfs.file_hash); + continue; + } + let bootstrap_node = bootstrap_node.clone(); + let agent = agent.clone(); + let handle = tokio::spawn(async move { + download_and_upload_lfs(bootstrap_node, agent, lfs, http_port).await; + }); + handle.await.unwrap(); + } +} + +async fn download_and_upload_lfs( + bootstrap_node: String, + agent: LocalZTMAgent, + lfs: LFSInfo, + http_port: u16, +) { + let file_hash = lfs.file_hash.clone(); + let local_port = + match get_or_create_remote_mega_tunnel(agent.agent_port, lfs.peer_id.clone()).await { + Ok(local_port) => local_port, + Err(e) => { + tracing::error!("Open tunnel to {} failed,{}", lfs.peer_id, e); + return; + } + }; + let url = format!("http://localhost:{}/objects/{}", local_port, lfs.file_hash); + let response = match get(url.clone()).await { + Ok(response) => response, + Err(_) => { + tracing::error!("Download lfs failed {}", url); + return; + } + }; + if !response.status().is_success() { + tracing::error!("Download lfs failed {}", url); + return; + } + + // create temp file + let base_dir = + PathBuf::from(std::env::var("MEGA_BASE_DIR").unwrap_or_else(|_| "/tmp/.mega".to_string())); + let base_dir = base_dir.join("tmp"); + if !base_dir.exists() { + create_dir_all(base_dir.clone()).unwrap(); + } + let target_path = base_dir.join(file_hash.clone()); + + if target_path.exists() { + fs::remove_file(&target_path).unwrap(); + } + + let mut file = File::create(target_path.clone()).unwrap(); + let bytes = response.bytes().await.unwrap(); + file.write_all(&bytes).unwrap(); + + //upload to local mega lfs + let client = Client::new(); + let json_str = format!( + r#" + {{ + "operation": "upload", + "transfers": [], + "hash_algo": "", + "objects": [ + {{ + "oid":"{}", + "size":{} + }} + ] + }}"#, + file_hash, lfs.file_size + ); + + let url = format!("http://localhost:{}/objects/batch", http_port); + let response = client + .post(url) + .header("content-type", "application/json") + .body(json_str) + .send() + .await + .unwrap(); + + if !response.status().is_success() { + tracing::error!("Failed to send POST request: HTTP {}", response.status()); + return; + } + + let mut file = File::open(target_path.clone()).unwrap(); + let mut file_content = Vec::new(); + file.read_to_end(&mut file_content).unwrap(); + + let url = format!( + "http://localhost:{}/objects/{}", + http_port, + file_hash.clone() + ); + let response = client.put(url).body(file_content).send().await.unwrap(); + + if response.status().is_success() { + tracing::info!("LFS {} uploaded successfully!", file_hash); + } else { + tracing::error!("Failed to upload file: HTTP {}", response.status()); + } + + fs::remove_file(target_path).unwrap(); + + share_lfs( + bootstrap_node, + lfs.file_hash, + lfs.hash_type, + lfs.file_size, + lfs.origin, + ) + .await; +} + +#[cfg(test)] +mod tests { + // use std::{fs::File, io::Read}; + + // use reqwest::Client; + + // #[tokio::test] + // async fn lfs_upload() { + // let client = Client::new(); + // let json_str = r#" + // { + // "operation": "upload", + // "transfers": [], + // "hash_algo": "", + // "objects": [ + // { + // "oid":"52c90a86cb034b7a1c4beb79304fa76bd0a6cbb7b168c3a935076c714bd1c6b6", + // "size":199246498 + // } + // ] + // }"#; + + // let response = client + // .post("http://222.20.126.106:8000/objects/batch") + // .header("content-type", "application/json") + // .body(json_str) + // .send() + // .await + // .unwrap(); + + // if response.status().is_success() { + // println!("POST request successful!"); + // } else { + // println!("Failed to send POST request: HTTP {}", response.status()); + // return; + // } + + // let url = "http://222.20.126.106:8000/objects/52c90a86cb034b7a1c4beb79304fa76bd0a6cbb7b168c3a935076c714bd1c6b6"; + // let client = Client::new(); + + // let mut file = File::open("/home/wujian/pr/mega/bbb.mp4").unwrap(); + // let mut file_content = Vec::new(); + // file.read_to_end(&mut file_content).unwrap(); + + // let response = client.put(url).body(file_content).send().await.unwrap(); + + // if response.status().is_success() { + // println!("File uploaded successfully!"); + // } else { + // println!("Failed to upload file: HTTP {}", response.status()); + // } + // } +} diff --git a/gemini/src/http/cache_repo.rs b/gemini/src/http/cache_repo.rs deleted file mode 100644 index 7ba6da33c..000000000 --- a/gemini/src/http/cache_repo.rs +++ /dev/null @@ -1,222 +0,0 @@ -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/mod.rs b/gemini/src/http/mod.rs index 691230598..062ae9d9b 100644 --- a/gemini/src/http/mod.rs +++ b/gemini/src/http/mod.rs @@ -1,2 +1 @@ -pub mod cache_repo; pub mod handler; diff --git a/gemini/src/lfs/mod.rs b/gemini/src/lfs/mod.rs new file mode 100644 index 000000000..056dafc7b --- /dev/null +++ b/gemini/src/lfs/mod.rs @@ -0,0 +1,114 @@ +use std::collections::HashSet; + +use reqwest::Client; + +use crate::{ + util::handle_response, ztm::get_or_create_remote_mega_tunnel, LFSInfo, LFSInfoPostBody, +}; + +pub async fn share_lfs( + bootstrap_node: String, + file_hash: String, + hash_type: String, + file_size: i64, + origin: String, +) { + let lfs = LFSInfoPostBody { + file_hash, + hash_type, + file_size, + peer_id: vault::get_peerid(), + origin, + }; + tracing::info!("Share lfs {:?}", lfs); + let json = serde_json::to_string(&lfs).unwrap(); + + let client = Client::new(); + let url = format!("{}/api/v1/lfs_share", bootstrap_node); + let response = client + .post(url) + .header("content-type", "application/json") + .body(json) + .send() + .await + .unwrap(); + + if response.status().is_success() { + tracing::info!("Share lfs {} successfully!", lfs.file_hash); + } else { + let context = response.text().await.unwrap(); + tracing::error!("Share lfs {} failed,{}", lfs.file_hash, context); + } +} + +pub async fn create_lfs_download_tunnel( + bootstrap_node: String, + ztm_agent_port: u16, + file_uri: String, +) -> Result, String> { + let file_hash = match get_file_hash_from_origin(file_uri) { + Ok(file_hash) => file_hash, + Err(_) => { + return Err("invalid file_uri".to_string()); + } + }; + // get public lfs by bootstrap_node + let url = format!("{bootstrap_node}/api/v1/lfs_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 Err(s); + } + }; + let lfs_list: Vec = match serde_json::from_slice(response_text.as_bytes()) { + Ok(p) => p, + Err(e) => { + tracing::error!("{}", e); + return Err(e.to_string()); + } + }; + let peer_list: HashSet = lfs_list + .iter() + .filter(|x| x.file_hash == file_hash && x.peer_online && x.peer_id != vault::get_peerid()) + .map(|x| x.peer_id.clone()) + .collect(); + tracing::info!("Search lfs[{}] download peer:{:?}", file_hash, peer_list); + + let mut tunnel_list: Vec = vec![]; + for peer_id in peer_list { + match get_or_create_remote_mega_tunnel(ztm_agent_port, peer_id).await { + Ok(port) => { + tunnel_list.push(format!("http://localhost:{}", port)); + } + Err(s) => { + tracing::error!("{}", s); + } + } + } + Ok(tunnel_list) +} + +pub fn get_file_hash_from_origin(origin: String) -> Result { + // p2p://t14id7uQxwneJ2PnPtaA3GSUwxTx6HTaq1UkayQVWSPT/sha256/52c90a86cb034b7a1c4beb79304fa76bd0a6cbb7b168c3a935076c714bd1c6b6 + let words: Vec<&str> = origin.split('/').collect(); + if words.len() <= 4 { + return Err("invalid origin".to_string()); + } + return Ok(words.get(4).unwrap().to_string()); +} + +#[cfg(test)] +mod tests { + // use crate::lfs::create_lfs_download_tunnel; + + // #[tokio::test] + // async fn create_lfs_download_tunnel_test() { + // let result = create_lfs_download_tunnel("http://222.20.126.106:8001".to_string() + // , 7777, + // "p2p://t14id7uQxwneJ2PnPtaA3GSUwxTx6HTaq1UkayQVWSPT/sha256/52c90a86cb034b7a1c4beb79304fa76bd0a6cbb7b168c3a935076c714bd1c6b6".to_string() + // ).await.unwrap(); + // println!("{:?}", result); + // } +} diff --git a/gemini/src/lib.rs b/gemini/src/lib.rs index 38e9c585a..914730827 100644 --- a/gemini/src/lib.rs +++ b/gemini/src/lib.rs @@ -1,11 +1,15 @@ use std::fmt; -use callisto::{ztm_node, ztm_repo_info}; +use callisto::{ztm_lfs_info, ztm_node, ztm_repo_info}; use chrono::Utc; +use common::utils::generate_id; use serde::{Deserialize, Serialize}; +use util::get_utc_timestamp; pub mod ca; +pub mod cache; pub mod http; +pub mod lfs; pub mod nostr; pub mod util; pub mod ztm; @@ -175,3 +179,91 @@ impl From for RepoInfo { } } } + +#[derive(Serialize, Deserialize, Debug, Clone, Default)] +pub struct LFSInfo { + pub file_hash: String, + pub hash_type: String, + pub file_size: i64, + pub creation_time: i64, + pub peer_id: String, + pub origin: String, + pub peer_online: bool, +} + +impl From for ztm_lfs_info::Model { + fn from(r: LFSInfo) -> Self { + ztm_lfs_info::Model { + id: generate_id(), + file_hash: r.file_hash, + hash_type: r.hash_type, + file_size: r.file_size, + creation_time: r.creation_time, + peer_id: r.peer_id, + origin: r.origin, + } + } +} + +impl From for LFSInfo { + fn from(r: ztm_lfs_info::Model) -> Self { + LFSInfo { + file_hash: r.file_hash, + hash_type: r.hash_type, + file_size: r.file_size, + creation_time: r.creation_time, + peer_id: r.peer_id, + origin: r.origin, + peer_online: false, + } + } +} + +impl From for LFSInfoPostBody { + fn from(r: LFSInfo) -> Self { + LFSInfoPostBody { + file_hash: r.file_hash, + hash_type: r.hash_type, + file_size: r.file_size, + peer_id: r.peer_id, + origin: r.origin, + } + } +} + +#[derive(Serialize, Deserialize, Debug, Clone, Default)] +pub struct LFSInfoPostBody { + pub file_hash: String, + pub hash_type: String, + pub file_size: i64, + pub peer_id: String, + pub origin: String, +} + +impl From for ztm_lfs_info::Model { + fn from(r: LFSInfoPostBody) -> Self { + ztm_lfs_info::Model { + id: generate_id(), + file_hash: r.file_hash, + hash_type: r.hash_type, + file_size: r.file_size, + creation_time: get_utc_timestamp(), + peer_id: r.peer_id, + origin: r.origin, + } + } +} + +impl From for LFSInfo { + fn from(r: LFSInfoPostBody) -> Self { + LFSInfo { + file_hash: r.file_hash, + hash_type: r.hash_type, + file_size: r.file_size, + creation_time: get_utc_timestamp(), + peer_id: r.peer_id, + origin: r.origin, + peer_online: false, + } + } +} diff --git a/jupiter/callisto/src/lib.rs b/jupiter/callisto/src/lib.rs index 462b8a349..1154dd0ca 100644 --- a/jupiter/callisto/src/lib.rs +++ b/jupiter/callisto/src/lib.rs @@ -27,6 +27,7 @@ pub mod mq_storage; pub mod raw_blob; pub mod ssh_keys; pub mod user; +pub mod ztm_lfs_info; pub mod ztm_node; pub mod ztm_nostr_event; pub mod ztm_nostr_req; diff --git a/jupiter/callisto/src/prelude.rs b/jupiter/callisto/src/prelude.rs index c937787d8..575975f4c 100644 --- a/jupiter/callisto/src/prelude.rs +++ b/jupiter/callisto/src/prelude.rs @@ -23,6 +23,7 @@ pub use crate::mega_tree::Entity as MegaTree; pub use crate::raw_blob::Entity as RawBlob; pub use crate::ssh_keys::Entity as SshKeys; pub use crate::user::Entity as User; +pub use crate::ztm_lfs_info::Entity as ZtmLFSInfo; pub use crate::ztm_node::Entity as ZtmNode; pub use crate::ztm_path_mapping::Entity as ZtmPathMapping; pub use crate::ztm_repo_info::Entity as ZtmRepoInfo; diff --git a/jupiter/callisto/src/ztm_lfs_info.rs b/jupiter/callisto/src/ztm_lfs_info.rs new file mode 100644 index 000000000..a4b77ea01 --- /dev/null +++ b/jupiter/callisto/src/ztm_lfs_info.rs @@ -0,0 +1,21 @@ +//! `SeaORM` Entity. Generated by sea-orm-codegen 0.11.3 + +use sea_orm::entity::prelude::*; + +#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq)] +#[sea_orm(table_name = "ztm_lfs_info")] +pub struct Model { + #[sea_orm(primary_key, auto_increment = true)] + pub id: i64, + pub file_hash: String, + pub hash_type: String, + pub file_size: i64, + pub creation_time: i64, + pub peer_id: String, + pub origin: String, +} + +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation {} + +impl ActiveModelBehavior for ActiveModel {} diff --git a/jupiter/src/storage/ztm_storage.rs b/jupiter/src/storage/ztm_storage.rs index a1abe7f7e..264933608 100644 --- a/jupiter/src/storage/ztm_storage.rs +++ b/jupiter/src/storage/ztm_storage.rs @@ -1,6 +1,8 @@ use std::sync::Arc; -use callisto::{ztm_node, ztm_nostr_event, ztm_nostr_req, ztm_path_mapping, ztm_repo_info}; +use callisto::{ + ztm_lfs_info, ztm_node, ztm_nostr_event, ztm_nostr_req, ztm_path_mapping, ztm_repo_info, +}; use common::errors::MegaError; use sea_orm::InsertResult; use sea_orm::{ColumnTrait, DatabaseConnection, EntityTrait, IntoActiveModel, QueryFilter, Set}; @@ -171,6 +173,43 @@ impl ZTMStorage { } } + pub async fn insert_lfs_info( + &self, + lfs_info: ztm_lfs_info::Model, + ) -> Result { + let list = self + .get_lfs_info_by_origin_and_peerid(&lfs_info.origin, &lfs_info.peer_id) + .await + .unwrap(); + if list.is_empty() { + ztm_lfs_info::Entity::insert(lfs_info.clone().into_active_model()) + .exec(self.get_connection()) + .await + .unwrap(); + } + Ok(lfs_info) + } + + pub async fn get_lfs_info_by_origin_and_peerid( + &self, + origin: &str, + peer_id: &str, + ) -> Result, MegaError> { + let model = ztm_lfs_info::Entity::find() + .filter(ztm_lfs_info::Column::Origin.eq(origin)) + .filter(ztm_lfs_info::Column::PeerId.eq(peer_id)) + .all(self.get_connection()) + .await; + Ok(model?) + } + + pub async fn get_all_lfs_info(&self) -> Result, MegaError> { + Ok(ztm_lfs_info::Entity::find() + .all(self.get_connection()) + .await + .unwrap()) + } + pub async fn insert_nostr_event( &self, nostr_event: ztm_nostr_event::Model, diff --git a/sql/postgres/pg_20241023__init.sql b/sql/postgres/pg_20241023__init.sql index 30ecf28eb..e510587b3 100644 --- a/sql/postgres/pg_20241023__init.sql +++ b/sql/postgres/pg_20241023__init.sql @@ -239,6 +239,16 @@ CREATE TABLE IF NOT EXISTS "ztm_repo_info" ( "commit" VARCHAR(64) ); +CREATE TABLE IF NOT EXISTS "ztm_lfs_info" ( + "id" BIGINT PRIMARY KEY, + "file_hash" VARCHAR(256), + "hash_type" VARCHAR(64), + "file_size" BIGINT NOT NULL, + "creation_time" BIGINT NOT NULL, + "peer_id" VARCHAR(64), + "origin" VARCHAR(256) +); + CREATE TABLE IF NOT EXISTS "ztm_nostr_event" ( "id" VARCHAR(128) PRIMARY KEY, "pubkey" VARCHAR(128), diff --git a/sql/sqlite/sqlite_20241023_init.sql b/sql/sqlite/sqlite_20241023_init.sql index 0f02858fd..e54a13416 100644 --- a/sql/sqlite/sqlite_20241023_init.sql +++ b/sql/sqlite/sqlite_20241023_init.sql @@ -238,6 +238,16 @@ CREATE TABLE IF NOT EXISTS "ztm_repo_info" ( "commit" TEXT ); +CREATE TABLE IF NOT EXISTS "ztm_lfs_info" ( + "id" BIGINT PRIMARY KEY, + "file_hash" VARCHAR(256), + "hash_type" VARCHAR(64), + "file_size" BIGINT NOT NULL, + "creation_time" BIGINT NOT NULL, + "peer_id" VARCHAR(64), + "origin" VARCHAR(256) +); + CREATE TABLE IF NOT EXISTS "ztm_nostr_event" ( "id" VARCHAR(128) PRIMARY KEY, "pubkey" VARCHAR(128),