From bc76a01cccaf18290b63caf417d27131b683445c Mon Sep 17 00:00:00 2001 From: "benjamin.747" Date: Mon, 9 Sep 2024 16:17:24 +0800 Subject: [PATCH] [mono] added file api for fuse, remove unused config `raw_obj_storage_type`, `big_obj_threshold` --- aries/src/service/api/nostr_router.rs | 7 +- ceres/src/api_service/import_api_service.rs | 81 +++-------------- ceres/src/api_service/mod.rs | 22 ++++- ceres/src/api_service/mono_api_service.rs | 11 +-- ceres/src/lfs/handler.rs | 87 ++++++++---------- ceres/src/pack/import_repo.rs | 9 +- ceres/src/pack/monorepo.rs | 2 +- common/src/config.rs | 6 -- docker/config.toml | 8 -- docs/development.md | 16 ++-- jupiter/src/context.rs | 53 +++++------ .../local_storage.rs | 11 +-- .../src/{raw_storage => lfs_storage}/mod.rs | 10 +- jupiter/src/lib.rs | 2 +- jupiter/src/storage/git_db_storage.rs | 71 ++++----------- jupiter/src/storage/git_fs_storage.rs | 91 ------------------- .../{lfs_storage.rs => lfs_db_storage.rs} | 8 +- jupiter/src/storage/mod.rs | 29 +----- jupiter/src/storage/mono_storage.rs | 37 +------- jupiter/src/storage/raw_db_storage.rs | 61 +++++++++++++ mega/config.toml | 8 -- mono/config.toml | 8 -- mono/src/api/api_router.rs | 89 ++++++++++++++---- mono/src/api/lfs/lfs_router.rs | 8 +- mono/src/api/mr_router.rs | 12 +-- mono/src/server/https_server.rs | 3 +- 26 files changed, 298 insertions(+), 452 deletions(-) rename jupiter/src/{raw_storage => lfs_storage}/local_storage.rs (95%) rename jupiter/src/{raw_storage => lfs_storage}/mod.rs (94%) delete mode 100644 jupiter/src/storage/git_fs_storage.rs rename jupiter/src/storage/{lfs_storage.rs => lfs_db_storage.rs} (97%) create mode 100644 jupiter/src/storage/raw_db_storage.rs diff --git a/aries/src/service/api/nostr_router.rs b/aries/src/service/api/nostr_router.rs index 4ead97572..763131812 100644 --- a/aries/src/service/api/nostr_router.rs +++ b/aries/src/service/api/nostr_router.rs @@ -1,7 +1,4 @@ -use std::{ - collections::{HashMap, HashSet}, - sync::Arc, -}; +use std::collections::{HashMap, HashSet}; use axum::{ extract::{Query, State}, @@ -118,7 +115,7 @@ async fn recieve( } async fn transfer_event_to_subscribed_nodes( - storage: Arc, + storage: ZTMStorage, nostr_event: NostrEvent, ztm_agent_port: u16, ) { diff --git a/ceres/src/api_service/import_api_service.rs b/ceres/src/api_service/import_api_service.rs index 33a52b5b9..9f9b132dd 100644 --- a/ceres/src/api_service/import_api_service.rs +++ b/ceres/src/api_service/import_api_service.rs @@ -6,8 +6,6 @@ use std::path::PathBuf; use axum::async_trait; -use callisto::raw_blob; -use common::errors::MegaError; use jupiter::context::Context; use mercury::errors::GitError; use mercury::internal::object::commit::Commit; @@ -26,19 +24,16 @@ pub struct ImportApiService { #[async_trait] impl ApiHandler for ImportApiService { + fn get_context(&self) -> Context { + self.context.clone() + } + async fn create_monorepo_file(&self, _: CreateFileInfo) -> Result<(), GitError> { return Err(GitError::CustomError( "import dir does not support create file".to_string(), )); } - async fn get_raw_blob_by_hash(&self, hash: &str) -> Result, MegaError> { - self.context - .services - .mono_storage - .get_raw_blob_by_hash(hash) - .await - } fn strip_relative(&self, path: &Path) -> Result { if let Ok(relative_path) = path.strip_prefix(self.repo.repo_path.clone()) { @@ -49,9 +44,14 @@ impl ApiHandler for ImportApiService { )) } } + async fn get_root_commit(&self) -> Commit { let storage = self.context.services.git_db_storage.clone(); - let refs = storage.get_default_ref(self.repo.repo_id).await.unwrap().unwrap(); + let refs = storage + .get_default_ref(self.repo.repo_id) + .await + .unwrap() + .unwrap(); storage .get_commit_by_hash(self.repo.repo_id, &refs.ref_git_id) .await @@ -62,7 +62,11 @@ impl ApiHandler for ImportApiService { async fn get_root_tree(&self) -> Tree { let storage = self.context.services.git_db_storage.clone(); - let refs = storage.get_default_ref(self.repo.repo_id).await.unwrap().unwrap(); + let refs = storage + .get_default_ref(self.repo.repo_id) + .await + .unwrap() + .unwrap(); let root_commit = storage .get_commit_by_hash(self.repo.repo_id, &refs.ref_git_id) @@ -179,58 +183,3 @@ impl ApiHandler for ImportApiService { target_commit } } - -impl ImportApiService { - // pub async fn get_objects_data( - // &self, - // object_id: &str, - // repo_path: &str, - // ) -> Result { - // let node = match self.storage.get_node_by_hash(object_id, repo_path).await { - // Ok(Some(node)) => node, - // _ => return Err((StatusCode::NOT_FOUND, "Blob not found".to_string())), - // }; - // let raw_data = match self.storage.get_obj_data_by_id(object_id).await { - // Ok(Some(model)) => model, - // _ => return Err((StatusCode::NOT_FOUND, "Blob not found".to_string())), - // }; - // let file_name = format!("inline; filename=\"{}\"", node.name.unwrap()); - // let res = Response::builder() - // .header("Content-Type", "application/octet-stream") - // .header("Content-Disposition", file_name) - // .body(raw_data.data.into()) - // .unwrap(); - // Ok(res) - // } - - // pub async fn count_object_num( - // &self, - // repo_path: &str, - // ) -> Result, (StatusCode, String)> { - // let query_res = self.storage.count_obj_from_node(repo_path).await.unwrap(); - // let tree = query_res - // .iter() - // .find(|x| x.node_type == "tree") - // .map(|x| x.count) - // .unwrap_or_default() - // .try_into() - // .unwrap(); - // let blob = query_res - // .iter() - // .find(|x| x.node_type == "blob") - // .map(|x| x.count) - // .unwrap_or_default() - // .try_into() - // .unwrap(); - // let commit = self.storage.count_obj_from_commit(repo_path).await.unwrap().try_into().unwrap(); - // let counter = GitTypeCounter { - // commit, - // tree, - // blob, - // tag: 0, - // ofs_delta: 0, - // ref_delta: 0, - // }; - // Ok(Json(counter)) - // } -} diff --git a/ceres/src/api_service/mod.rs b/ceres/src/api_service/mod.rs index c03112d3f..4f8e454eb 100644 --- a/ceres/src/api_service/mod.rs +++ b/ceres/src/api_service/mod.rs @@ -7,12 +7,13 @@ use axum::async_trait; use callisto::raw_blob; use common::errors::MegaError; -use jupiter::utils::converter::generate_git_keep_with_timestamp; +use jupiter::{context::Context, utils::converter::generate_git_keep_with_timestamp}; use mercury::{ errors::GitError, internal::object::{ commit::Commit, tree::{Tree, TreeItem, TreeItemMode}, + ObjectTrait, }, }; @@ -26,9 +27,18 @@ pub mod mono_api_service; #[async_trait] pub trait ApiHandler: Send + Sync { + fn get_context(&self) -> Context; + async fn create_monorepo_file(&self, file_info: CreateFileInfo) -> Result<(), GitError>; - async fn get_raw_blob_by_hash(&self, hash: &str) -> Result, MegaError>; + async fn get_raw_blob_by_hash(&self, hash: &str) -> Result, MegaError> { + let context = self.get_context(); + context + .services + .raw_db_storage + .get_raw_blob_by_hash(hash) + .await + } fn strip_relative(&self, path: &Path) -> Result; @@ -36,6 +46,14 @@ pub trait ApiHandler: Send + Sync { async fn get_root_tree(&self) -> Tree; + async fn get_tree_as_data(&self, path: &Path) -> Result, GitError> { + let res = self.search_tree_by_path(path).await.unwrap(); + if let Some(tree) = res { + return tree.to_data(); + } + Ok(vec![]) + } + async fn get_tree_by_hash(&self, hash: &str) -> Tree; async fn get_tree_relate_commit(&self, t_hash: &str) -> Commit; diff --git a/ceres/src/api_service/mono_api_service.rs b/ceres/src/api_service/mono_api_service.rs index be661b77c..cda4d35c3 100644 --- a/ceres/src/api_service/mono_api_service.rs +++ b/ceres/src/api_service/mono_api_service.rs @@ -28,6 +28,10 @@ pub struct MonoApiService { #[async_trait] impl ApiHandler for MonoApiService { + fn get_context(&self) -> Context { + self.context.clone() + } + /// Creates a new file or directory in the monorepo based on the provided file information. /// /// # Arguments @@ -113,13 +117,6 @@ impl ApiHandler for MonoApiService { Ok(()) } - async fn get_raw_blob_by_hash(&self, hash: &str) -> Result, MegaError> { - self.context - .services - .mono_storage - .get_raw_blob_by_hash(hash) - .await - } fn strip_relative(&self, path: &Path) -> Result { Ok(path.to_path_buf()) diff --git a/ceres/src/lfs/handler.rs b/ceres/src/lfs/handler.rs index 636367fe5..12b3e6a48 100644 --- a/ceres/src/lfs/handler.rs +++ b/ceres/src/lfs/handler.rs @@ -1,6 +1,5 @@ use std::cmp::min; use std::collections::HashMap; -use std::sync::Arc; use anyhow::Result; use bytes::Bytes; @@ -10,18 +9,17 @@ use rand::prelude::*; use callisto::{lfs_locks, lfs_objects, lfs_split_relations}; use common::errors::{GitLFSError, MegaError}; use jupiter::context::Context; -use jupiter::storage::lfs_storage::LfsStorage; +use jupiter::storage::lfs_db_storage::LfsDbStorage; +use crate::lfs::lfs_structs::ChunkRepresentation; use crate::lfs::lfs_structs::{ BatchRequest, LockList, LockRequest, ObjectError, UnlockRequest, VerifiableLockList, VerifiableLockRequest, }; - -use crate::lfs::lfs_structs::ChunkRepresentation; use crate::lfs::lfs_structs::{Link, Lock, LockListQuery, MetaObject, Representation, RequestVars}; pub async fn lfs_retrieve_lock( - storage: Arc, + storage: LfsDbStorage, query: LockListQuery, ) -> Result { let mut lock_list = LockList { @@ -49,7 +47,7 @@ pub async fn lfs_retrieve_lock( } pub async fn lfs_verify_lock( - storage: Arc, + storage: LfsDbStorage, req: VerifiableLockRequest, ) -> Result { let mut limit = req.limit.unwrap_or(0); @@ -88,7 +86,7 @@ pub async fn lfs_verify_lock( } pub async fn lfs_create_lock( - storage: Arc, + storage: LfsDbStorage, req: LockRequest, ) -> Result { let res = lfs_get_filtered_locks( @@ -139,7 +137,7 @@ pub async fn lfs_create_lock( } pub async fn lfs_delete_lock( - storage: Arc, + storage: LfsDbStorage, id: &str, unlock_request: UnlockRequest, ) -> Result { @@ -187,7 +185,7 @@ pub async fn lfs_process_batch( request.authorization = "".to_string(); } let mut response_objects = Vec::::new(); - let storage = context.services.lfs_storage.clone(); + let storage = context.services.lfs_db_storage.clone(); let config = context.config.lfs.clone(); let server_url = context.config.lfs.url.clone(); @@ -247,7 +245,7 @@ pub async fn lfs_fetch_chunk_ids( "Server didn't run in `split` mode, didn't support chunk ids".to_string(), )); } - let storage = context.services.lfs_storage.clone(); + let storage = context.services.lfs_db_storage.clone(); let meta = lfs_get_meta(storage.clone(), &fetch_vars.oid) .await @@ -298,10 +296,10 @@ pub async fn lfs_upload_object( body_bytes: &[u8], ) -> Result<(), GitLFSError> { let config = context.config.lfs.clone(); + let storage = context.services.lfs_db_storage.clone(); let lfs_storage = context.services.lfs_storage.clone(); - let raw_storage = context.services.raw_storage.clone(); - let meta = lfs_get_meta(lfs_storage.clone(), &request_vars.oid) + let meta = lfs_get_meta(storage.clone(), &request_vars.oid) .await .unwrap(); if config.enable_split && meta.splited { @@ -312,9 +310,9 @@ pub async fn lfs_upload_object( for chunk in body_bytes.chunks(config.split_size) { // sha256 let sub_id = sha256::digest(chunk); - let res = raw_storage.put_object(&sub_id, chunk).await; + let res = lfs_storage.put_object(&sub_id, chunk).await; if res.is_err() { - lfs_delete_meta(lfs_storage.clone(), request_vars) + lfs_delete_meta(storage.clone(), request_vars) .await .unwrap(); // TODO: whether/how to delete the uploaded blocks. @@ -327,18 +325,17 @@ pub async fn lfs_upload_object( // save the relationship to database let mut offset = 0; for sub_id in sub_ids { - // let db = config.context.services.lfs_storage.clone(); let size = min(config.split_size as i64, body_bytes.len() as i64 - offset); - lfs_put_relation(lfs_storage.clone(), &meta.oid, &sub_id, offset, size) + lfs_put_relation(storage.clone(), &meta.oid, &sub_id, offset, size) .await .unwrap(); offset += size; } } else { // normal mode - let res = raw_storage.put_object(&meta.oid, body_bytes).await; + let res = lfs_storage.put_object(&meta.oid, body_bytes).await; if res.is_err() { - lfs_delete_meta(lfs_storage.clone(), request_vars) + lfs_delete_meta(storage.clone(), request_vars) .await .unwrap(); return Err(GitLFSError::GeneralError(String::from( @@ -351,13 +348,10 @@ pub async fn lfs_upload_object( /// Download object from storage. /// when server enable split, if OID is a complete object, then splice the object and return it. -pub async fn lfs_download_object( - context: Context, - oid: &String, -) -> Result { +pub async fn lfs_download_object(context: Context, oid: &String) -> Result { let config = context.config.lfs; - let stg = context.services.lfs_storage.clone(); - let raw_storage = context.services.raw_storage.clone(); + let stg = context.services.lfs_db_storage.clone(); + let lfs_storage = context.services.lfs_storage.clone(); if config.enable_split { let meta = lfs_get_meta(stg.clone(), oid).await; // let relation_db = context.services.lfs_storage.clone(); @@ -373,7 +367,7 @@ pub async fn lfs_download_object( } let mut bytes = vec![0u8; meta.size as usize]; for relation in relations { - let sub_bytes = raw_storage.get_object(&relation.sub_oid).await.unwrap(); + let sub_bytes = lfs_storage.get_object(&relation.sub_oid).await.unwrap(); let offset = relation.offset as usize; let size = relation.size as usize; bytes[offset..offset + size].copy_from_slice(&sub_bytes); @@ -388,13 +382,13 @@ pub async fn lfs_download_object( )); } - let bytes = raw_storage.get_object(oid).await.unwrap(); + let bytes = lfs_storage.get_object(oid).await.unwrap(); Ok(bytes) } } } else { let meta = lfs_get_meta(stg, oid).await.unwrap(); - let bytes = raw_storage.get_object(&meta.oid).await.unwrap(); + let bytes = lfs_storage.get_object(&meta.oid).await.unwrap(); Ok(bytes) } } @@ -467,26 +461,23 @@ fn create_link(href: &str, header: &HashMap) -> Link { /// check if meta file exist in storage. async fn lfs_file_exist(context: &Context, meta: &MetaObject) -> bool { let config = context.config.lfs.clone(); - let storage = context.services.lfs_storage.clone(); - let raw_storage = context.services.raw_storage.clone(); + let storage = context.services.lfs_db_storage.clone(); + let lfs_storage = context.services.lfs_storage.clone(); if meta.splited && config.enable_split { - let relations = storage - .get_lfs_relations(meta.oid.clone()) - .await - .unwrap(); + let relations = storage.get_lfs_relations(meta.oid.clone()).await.unwrap(); if relations.is_empty() { return false; } relations .iter() - .all(|relation| raw_storage.exist_object(&relation.sub_oid)) + .all(|relation| lfs_storage.exist_object(&relation.sub_oid)) } else { - raw_storage.exist_object(&meta.oid) + lfs_storage.exist_object(&meta.oid) } } async fn lfs_get_filtered_locks( - storage: Arc, + storage: LfsDbStorage, refspec: &str, path: &str, cursor: &str, @@ -542,7 +533,10 @@ async fn lfs_get_filtered_locks( Ok((locks, next)) } -async fn lfs_get_locks(storage: Arc, refspec: &str) -> Result, GitLFSError> { +async fn lfs_get_locks( + storage: LfsDbStorage, + refspec: &str, +) -> Result, GitLFSError> { let result = storage.get_lock_by_id(refspec).await.unwrap(); match result { Some(val) => { @@ -555,7 +549,7 @@ async fn lfs_get_locks(storage: Arc, refspec: &str) -> Result, + storage: LfsDbStorage, repo: &str, locks: Vec, ) -> Result<(), GitLFSError> { @@ -611,10 +605,7 @@ async fn lfs_add_lock( } } -async fn lfs_get_meta( - storage: Arc, - oid: &str, -) -> Result { +async fn lfs_get_meta(storage: LfsDbStorage, oid: &str) -> Result { let result = storage.get_lfs_object(oid.to_owned()).await.unwrap(); match result { @@ -629,7 +620,7 @@ async fn lfs_get_meta( } async fn lfs_put_meta( - storage: Arc, + storage: LfsDbStorage, v: &RequestVars, splited: bool, ) -> Result { @@ -666,7 +657,7 @@ async fn lfs_put_meta( } } -async fn lfs_delete_meta(storage: Arc, v: &RequestVars) -> Result<(), GitLFSError> { +async fn lfs_delete_meta(storage: LfsDbStorage, v: &RequestVars) -> Result<(), GitLFSError> { let res = storage.delete_lfs_object(v.oid.to_owned()).await; lfs_delete_all_relations(storage.clone(), &v.oid) .await @@ -678,7 +669,7 @@ async fn lfs_delete_meta(storage: Arc, v: &RequestVars) -> Result<() } async fn delete_lock( - storage: Arc, + storage: LfsDbStorage, repo: &str, _user: Option, id: &str, @@ -751,7 +742,7 @@ async fn delete_lock( /// put relation, ignore if already exist. async fn lfs_put_relation( - storage: Arc, + storage: LfsDbStorage, ori_oid: &String, sub_oid: &String, offset: i64, @@ -778,7 +769,7 @@ async fn lfs_put_relation( /// delete all relations of an object if it exists. do nothing if not. async fn lfs_delete_all_relations( - storage: Arc, + storage: LfsDbStorage, ori_oid: &String, ) -> Result<(), GitLFSError> { let relations = storage.get_lfs_relations(ori_oid.to_owned()).await.unwrap(); @@ -789,7 +780,7 @@ async fn lfs_delete_all_relations( } async fn lfs_check_sub_oid_exist( - storage: Arc, + storage: LfsDbStorage, sub_oid: &String, ) -> Result { let result = storage.get_lfs_relations_ori_oid(sub_oid).await.unwrap(); diff --git a/ceres/src/pack/import_repo.rs b/ceres/src/pack/import_repo.rs index 656710da7..3b3e67074 100644 --- a/ceres/src/pack/import_repo.rs +++ b/ceres/src/pack/import_repo.rs @@ -17,7 +17,7 @@ use callisto::{mega_tree, raw_blob}; use common::errors::MegaError; use jupiter::{ context::Context, - storage::{batch_save_model, GitStorageProvider}, + storage::{batch_save_model}, }; use mercury::{ errors::GitError, @@ -81,6 +81,7 @@ impl PackHandler for ImportRepo { let (stream_tx, stream_rx) = mpsc::channel(pack_config.channel_message_size); let storage = self.context.services.git_db_storage.clone(); + let raw_storage = self.context.services.raw_db_storage.clone(); let total = storage.get_obj_count_by_repo_id(self.repo.repo_id).await; let encoder = PackEncoder::new(total, 0, stream_tx); encoder.encode_async(entry_rx).await.unwrap(); @@ -125,11 +126,11 @@ impl PackHandler for ImportRepo { let mut blob_handler = vec![]; for chunk in bids.chunks(10000) { - let stg_clone = storage.clone(); + let raw_storage = raw_storage.clone(); let sender_clone = entry_tx.clone(); let chunk_clone = chunk.to_vec(); let handler = tokio::spawn(async move { - let mut blob_stream = stg_clone.get_raw_blobs(chunk_clone).await.unwrap(); + let mut blob_stream = raw_storage.get_raw_blobs_stream(chunk_clone).await.unwrap(); while let Some(model) = blob_stream.next().await { match model { Ok(m) => { @@ -279,7 +280,7 @@ impl PackHandler for ImportRepo { ) -> Result, MegaError> { self.context .services - .mono_storage + .raw_db_storage .get_raw_blobs_by_hashes(hashes) .await } diff --git a/ceres/src/pack/monorepo.rs b/ceres/src/pack/monorepo.rs index 3cd31e766..026483748 100644 --- a/ceres/src/pack/monorepo.rs +++ b/ceres/src/pack/monorepo.rs @@ -319,7 +319,7 @@ impl PackHandler for MonoRepo { ) -> Result, MegaError> { self.context .services - .mono_storage + .raw_db_storage .get_raw_blobs_by_hashes(hashes) .await } diff --git a/common/src/config.rs b/common/src/config.rs index 0a6164290..0a54ff697 100644 --- a/common/src/config.rs +++ b/common/src/config.rs @@ -203,9 +203,6 @@ impl Default for SshConfig { #[derive(Serialize, Deserialize, Debug, Clone)] pub struct StorageConfig { - pub raw_obj_storage_type: String, - pub big_obj_threshold: usize, - pub raw_obj_local_path: PathBuf, pub obs_access_key: String, pub obs_secret_key: String, pub obs_region: String, @@ -215,9 +212,6 @@ pub struct StorageConfig { impl Default for StorageConfig { fn default() -> Self { Self { - raw_obj_storage_type: String::from("LOCAL"), - big_obj_threshold: 1024, - raw_obj_local_path: PathBuf::from("/tmp/.mega/objects"), obs_access_key: String::new(), obs_secret_key: String::new(), obs_region: String::from("cn-east-3"), diff --git a/docker/config.toml b/docker/config.toml index 5556eb647..4c4023cb7 100644 --- a/docker/config.toml +++ b/docker/config.toml @@ -40,14 +40,6 @@ sqlx_logging = false ssh_key_path = "${base_dir}/etc/ssh" [storage] -# raw object stroage type, can be `local` or `remote` -raw_obj_storage_type = "LOCAL" - -## If the object file size exceeds the threshold value, it will be handled by file storage instead of the database, Unit is KB -big_obj_threshold = 1024 - -# set the local path of the project storage -raw_obj_local_path = "${base_dir}/objects" obs_access_key = "" obs_secret_key = "" diff --git a/docs/development.md b/docs/development.md index 9f0518635..06a2c6d83 100644 --- a/docs/development.md +++ b/docs/development.md @@ -277,16 +277,6 @@ sqlx_logging = false ssh_key_path = "${base_dir}/ssh" [storage] -# raw object stroage type, can be `local` or `remote` -raw_obj_storage_type = "LOCAL" - -## If the object file size exceeds the threshold value, it will be handled by file storage instead of the database, Unit is KB -big_obj_threshold = 1024 - -# set the local path of the project storage -raw_obj_local_path = "${base_dir}/objects" - -lfs_obj_local_path = "${base_dir}/lfs" obs_access_key = "" obs_secret_key = "" @@ -323,6 +313,12 @@ hub = "http://127.0.0.1:8888" agent = "http://127.0.0.1:7777" [lfs] +# LFS Server url +url = "https://git.gitmono.com" + +# set the local path of the lfs storage +lfs_obj_local_path = "${base_dir}/lfs" + ## IMPORTANT: The 'enable_split' feature can only be enabled for new databases. Existing databases do not support this feature. # Enable or disable splitting large files into smaller chunks enable_split = false # Default is disabled. Set to true to enable file splitting. diff --git a/jupiter/src/context.rs b/jupiter/src/context.rs index 1c06f7922..7d68e3432 100644 --- a/jupiter/src/context.rs +++ b/jupiter/src/context.rs @@ -3,11 +3,11 @@ use std::{env, path::PathBuf, sync::Arc}; use common::config::Config; use crate::{ - raw_storage::{local_storage::LocalStorage, RawStorage}, + lfs_storage::{local_storage::LocalStorage, LfsStorage}, storage::{ - git_db_storage::GitDbStorage, init::database_connection, lfs_storage::LfsStorage, - mono_storage::MonoStorage, mq_storage::MQStorage, user_storage::UserStorage, - ztm_storage::ZTMStorage, + git_db_storage::GitDbStorage, init::database_connection, lfs_db_storage::LfsDbStorage, + mono_storage::MonoStorage, mq_storage::MQStorage, raw_db_storage::RawDbStorage, + user_storage::UserStorage, ztm_storage::ZTMStorage, }, }; @@ -34,32 +34,28 @@ impl Context { #[derive(Clone)] pub struct Service { - pub mono_storage: Arc, - pub git_db_storage: Arc, - pub lfs_storage: Arc, - pub ztm_storage: Arc, - pub mq_storage: Arc, + pub mono_storage: MonoStorage, + pub git_db_storage: GitDbStorage, + pub raw_db_storage: RawDbStorage, + pub lfs_db_storage: LfsDbStorage, + pub ztm_storage: ZTMStorage, + pub mq_storage: MQStorage, pub user_storage: UserStorage, - pub raw_storage: Arc, + pub lfs_storage: Arc, } impl Service { async fn new(config: &Config) -> Service { let connection = Arc::new(database_connection(&config.database).await); Service { - mono_storage: Arc::new( - MonoStorage::new(connection.clone(), config.storage.clone()).await, - ), - git_db_storage: Arc::new( - GitDbStorage::new(connection.clone(), config.storage.clone()).await, - ), - lfs_storage: Arc::new(LfsStorage::new(connection.clone()).await), - ztm_storage: Arc::new(ZTMStorage::new(connection.clone()).await), - mq_storage: Arc::new(MQStorage::new(connection.clone()).await), + mono_storage: MonoStorage::new(connection.clone()).await, + git_db_storage:GitDbStorage::new(connection.clone()).await, + raw_db_storage: RawDbStorage::new(connection.clone()).await, + lfs_db_storage: LfsDbStorage::new(connection.clone()).await, + ztm_storage: ZTMStorage::new(connection.clone()).await, + mq_storage: MQStorage::new(connection.clone()).await, user_storage: UserStorage::new(connection.clone()).await, - raw_storage: Arc::new(LocalStorage::init( - config.lfs.lfs_obj_local_path.clone(), - )), + lfs_storage: Arc::new(LocalStorage::init(config.lfs.lfs_obj_local_path.clone())), } } @@ -69,13 +65,14 @@ impl Service { fn mock() -> Arc { Arc::new(Self { - mono_storage: Arc::new(MonoStorage::mock()), - git_db_storage: Arc::new(GitDbStorage::mock()), - lfs_storage: Arc::new(LfsStorage::mock()), - ztm_storage: Arc::new(ZTMStorage::mock()), - mq_storage: Arc::new(MQStorage::mock()), + mono_storage: MonoStorage::mock(), + git_db_storage: GitDbStorage::mock(), + raw_db_storage: RawDbStorage::mock(), + lfs_db_storage: LfsDbStorage::mock(), + ztm_storage: ZTMStorage::mock(), + mq_storage: MQStorage::mock(), user_storage: UserStorage::mock(), - raw_storage: Arc::new(LocalStorage::init( + lfs_storage: Arc::new(LocalStorage::init( PathBuf::from(env::current_dir().unwrap().parent().unwrap()).join("tests"), )), }) diff --git a/jupiter/src/raw_storage/local_storage.rs b/jupiter/src/lfs_storage/local_storage.rs similarity index 95% rename from jupiter/src/raw_storage/local_storage.rs rename to jupiter/src/lfs_storage/local_storage.rs index 7f0161970..3919b0173 100644 --- a/jupiter/src/raw_storage/local_storage.rs +++ b/jupiter/src/lfs_storage/local_storage.rs @@ -5,10 +5,9 @@ use std::path::{Path, PathBuf}; use async_trait::async_trait; use bytes::Bytes; -use callisto::db_enums::StorageType; use common::errors::MegaError; -use crate::raw_storage::RawStorage; +use crate::lfs_storage::LfsStorage; #[derive(Default)] pub struct LocalStorage { @@ -23,10 +22,7 @@ impl LocalStorage { } #[async_trait] -impl RawStorage for LocalStorage { - fn get_storage_type(&self) -> StorageType { - StorageType::LocalFs - } +impl LfsStorage for LocalStorage { async fn get_ref(&self, repo_id: i64, ref_name: &str) -> Result { let path = Path::new(&self.base_path).join(repo_id.to_string()).join(ref_name); @@ -108,9 +104,8 @@ mod tests { use std::path::Path; use std::{env, path::PathBuf}; - use crate::raw_storage::{local_storage::LocalStorage, RawStorage}; + use crate::lfs_storage::{local_storage::LocalStorage, LfsStorage}; - // #[test] #[tokio::test] async fn test_content_store() { let oid = "6ae8a75555209fd6c44157c0aed8016e763ff435a19cf186f76863140143ff72".to_owned(); diff --git a/jupiter/src/raw_storage/mod.rs b/jupiter/src/lfs_storage/mod.rs similarity index 94% rename from jupiter/src/raw_storage/mod.rs rename to jupiter/src/lfs_storage/mod.rs index 6de5ba020..a6e3bba70 100644 --- a/jupiter/src/raw_storage/mod.rs +++ b/jupiter/src/lfs_storage/mod.rs @@ -6,10 +6,9 @@ use std::{ use async_trait::async_trait; use bytes::Bytes; -use callisto::db_enums::StorageType; use common::errors::MegaError; -use crate::raw_storage::local_storage::LocalStorage; +use crate::lfs_storage::local_storage::LocalStorage; pub mod local_storage; @@ -22,8 +21,7 @@ pub struct BlobLink { } #[async_trait] -pub trait RawStorage: Sync + Send { - fn get_storage_type(&self) -> StorageType; +pub trait LfsStorage: Sync + Send { async fn get_ref(&self, repo_id: i64, ref_name: &str) -> Result; @@ -113,7 +111,7 @@ pub trait RawStorage: Sync + Send { } } -pub async fn init(storage_type: String, base_path: PathBuf) -> Arc { +pub async fn init(storage_type: String, base_path: PathBuf) -> Arc { match storage_type.as_str() { "LOCAL" => { Arc::new(LocalStorage::init(base_path)) @@ -125,6 +123,6 @@ pub async fn init(storage_type: String, base_path: PathBuf) -> Arc Arc { +pub fn mock() -> Arc { Arc::new(LocalStorage::init(PathBuf::from("/"))) } diff --git a/jupiter/src/lib.rs b/jupiter/src/lib.rs index 2198f32b3..982170302 100644 --- a/jupiter/src/lib.rs +++ b/jupiter/src/lib.rs @@ -1,4 +1,4 @@ pub mod context; -pub mod raw_storage; +pub mod lfs_storage; pub mod storage; pub mod utils; diff --git a/jupiter/src/storage/git_db_storage.rs b/jupiter/src/storage/git_db_storage.rs index f341382a8..089ddca89 100644 --- a/jupiter/src/storage/git_db_storage.rs +++ b/jupiter/src/storage/git_db_storage.rs @@ -1,6 +1,5 @@ use std::sync::{Arc, Mutex}; -use async_trait::async_trait; use futures::{stream, Stream, StreamExt}; use sea_orm::sea_query::Expr; use sea_orm::{ @@ -10,23 +9,15 @@ use sea_orm::{ use sea_orm::{PaginatorTrait, QueryOrder}; use callisto::{git_blob, git_commit, git_repo, git_tag, git_tree, import_refs, raw_blob}; -use common::config::StorageConfig; use common::errors::MegaError; use mercury::internal::object::GitObjectModel; use mercury::internal::pack::entry::Entry; -use crate::{ - raw_storage::{self, RawStorage}, - storage::GitStorageProvider, -}; - use crate::storage::batch_save_model; #[derive(Clone)] pub struct GitDbStorage { - pub raw_storage: Arc, pub connection: Arc, - pub raw_obj_threshold: usize, } #[derive(Debug)] @@ -38,10 +29,22 @@ struct GitObjects { tags: Vec, } -#[async_trait] -impl GitStorageProvider for GitDbStorage { - async fn save_ref(&self, repo_id: i64, mut refs: import_refs::Model) -> Result<(), MegaError> { - // let mut model: import_refs::Model = refs.clone().into(); +impl GitDbStorage { + pub fn get_connection(&self) -> &DatabaseConnection { + &self.connection + } + + pub async fn new(connection: Arc) -> Self { + GitDbStorage { connection } + } + + pub fn mock() -> Self { + GitDbStorage { + connection: Arc::new(DatabaseConnection::default()), + } + } + + pub async fn save_ref(&self, repo_id: i64, mut refs: import_refs::Model) -> Result<(), MegaError> { refs.repo_id = repo_id; let a_model = refs.into_active_model(); import_refs::Entity::insert(a_model) @@ -51,7 +54,7 @@ impl GitStorageProvider for GitDbStorage { Ok(()) } - async fn remove_ref(&self, repo_id: i64, ref_name: &str) -> Result<(), MegaError> { + pub async fn remove_ref(&self, repo_id: i64, ref_name: &str) -> Result<(), MegaError> { import_refs::Entity::delete_many() .filter(import_refs::Column::RepoId.eq(repo_id)) .filter(import_refs::Column::RefName.eq(ref_name)) @@ -60,7 +63,7 @@ impl GitStorageProvider for GitDbStorage { Ok(()) } - async fn get_ref(&self, repo_id: i64) -> Result, MegaError> { + pub async fn get_ref(&self, repo_id: i64) -> Result, MegaError> { let result = import_refs::Entity::find() .filter(import_refs::Column::RepoId.eq(repo_id)) .order_by_asc(import_refs::Column::RefName) @@ -69,7 +72,7 @@ impl GitStorageProvider for GitDbStorage { Ok(result) } - async fn update_ref( + pub async fn update_ref( &self, repo_id: i64, ref_name: &str, @@ -88,29 +91,6 @@ impl GitStorageProvider for GitDbStorage { ref_data.update(self.get_connection()).await.unwrap(); Ok(()) } -} - -impl GitDbStorage { - pub fn get_connection(&self) -> &DatabaseConnection { - &self.connection - } - - pub async fn new(connection: Arc, config: StorageConfig) -> Self { - GitDbStorage { - connection, - raw_storage: raw_storage::init(config.raw_obj_storage_type, config.raw_obj_local_path) - .await, - raw_obj_threshold: config.big_obj_threshold, - } - } - - pub fn mock() -> Self { - GitDbStorage { - connection: Arc::new(DatabaseConnection::default()), - raw_storage: raw_storage::mock(), - raw_obj_threshold: 1024, - } - } pub async fn get_default_ref( &self, @@ -324,7 +304,7 @@ impl GitDbStorage { } pub async fn get_blobs_by_repo_id( - & self, + &self, repo_id: i64, ) -> Result> + '_ + Send, MegaError> { Ok(git_blob::Entity::find() @@ -347,17 +327,6 @@ impl GitDbStorage { .unwrap()) } - pub async fn get_raw_blobs( - &self, - hashes: Vec, - ) -> Result> + '_ + Send, MegaError> { - Ok(raw_blob::Entity::find() - .filter(raw_blob::Column::Sha1.is_in(hashes)) - .stream(self.get_connection()) - .await - .unwrap()) - } - pub async fn get_tags_by_repo_id( &self, repo_id: i64, diff --git a/jupiter/src/storage/git_fs_storage.rs b/jupiter/src/storage/git_fs_storage.rs deleted file mode 100644 index 4b6a3dab3..000000000 --- a/jupiter/src/storage/git_fs_storage.rs +++ /dev/null @@ -1,91 +0,0 @@ -use std::{io::Cursor, sync::Arc}; - -use async_trait::async_trait; - -use callisto::import_refs; -use common::{config::StorageConfig, errors::MegaError}; -use mercury::{ - hash::SHA1, - internal::{ - object::{types::ObjectType, utils}, - pack::entry::Entry, - }, -}; - -use crate::{ - raw_storage::{self, RawStorage}, - storage::GitStorageProvider, -}; - -pub struct GitFsStorage { - pub raw_storage: Arc, -} - -#[async_trait] -impl GitStorageProvider for GitFsStorage { - async fn save_ref(&self, repo_id: i64, refs: import_refs::Model) -> Result<(), MegaError> { - self.raw_storage - .put_ref(repo_id, &refs.ref_name, &refs.ref_git_id) - .await - } - - async fn remove_ref(&self, repo_id: i64, ref_name: &str) -> Result<(), MegaError> { - self.raw_storage.delete_ref(repo_id, ref_name).await - } - - async fn get_ref(&self, _repo_id: i64) -> Result, MegaError> { - // let ref_hash = self.raw_storage.get_ref(&repo.repo_name, ref_name).await; - // if let Some() - todo!() - } - - async fn update_ref( - &self, - repo_id: i64, - ref_name: &str, - new_id: &str, - ) -> Result<(), MegaError> { - self.raw_storage.update_ref(repo_id, ref_name, new_id).await - } -} - -impl GitFsStorage { - pub async fn new(config: StorageConfig) -> Self { - GitFsStorage { - raw_storage: raw_storage::init(config.raw_obj_storage_type, config.raw_obj_local_path) - .await, - } - } - - pub fn mock() -> Self { - Self { - raw_storage: raw_storage::mock(), - } - } - - pub async fn save_entry(&self, entry_list: Vec) -> Result<(), MegaError> { - for entry in entry_list { - self.raw_storage - .put_object(&entry.hash.to_plain_str(), &entry.data) - .await - .unwrap(); - } - Ok(()) - } - - pub async fn get_entry_by_sha1(&self, sha1_vec: Vec<&str>) -> Result, MegaError> { - let mut res: Vec = Vec::new(); - for sha1 in sha1_vec { - let data = self.raw_storage.get_object(sha1).await.unwrap(); - let (type_num, _) = utils::read_type_and_size(&mut Cursor::new(&data)).unwrap(); - let obj_type = ObjectType::from_u8(type_num).unwrap(); - let hash = SHA1::new(&data.to_vec()); - res.push(Entry { - obj_type, - data: data.to_vec(), - hash, - }) - } - Ok(res) - } -} diff --git a/jupiter/src/storage/lfs_storage.rs b/jupiter/src/storage/lfs_db_storage.rs similarity index 97% rename from jupiter/src/storage/lfs_storage.rs rename to jupiter/src/storage/lfs_db_storage.rs index 4b5ff4b15..ae858eb0a 100644 --- a/jupiter/src/storage/lfs_storage.rs +++ b/jupiter/src/storage/lfs_db_storage.rs @@ -8,21 +8,21 @@ use callisto::{lfs_locks, lfs_objects, lfs_split_relations}; use common::errors::MegaError; #[derive(Clone)] -pub struct LfsStorage { +pub struct LfsDbStorage { pub connection: Arc, } -impl LfsStorage { +impl LfsDbStorage { pub fn get_connection(&self) -> &DatabaseConnection { &self.connection } pub async fn new(connection: Arc) -> Self { - LfsStorage { connection } + LfsDbStorage { connection } } pub fn mock() -> Self { - LfsStorage { + LfsDbStorage { connection: Arc::new(DatabaseConnection::default()), } } diff --git a/jupiter/src/storage/mod.rs b/jupiter/src/storage/mod.rs index 098f9ccad..377ce5195 100644 --- a/jupiter/src/storage/mod.rs +++ b/jupiter/src/storage/mod.rs @@ -1,40 +1,15 @@ pub mod git_db_storage; -pub mod git_fs_storage; pub mod init; -pub mod lfs_storage; +pub mod lfs_db_storage; pub mod mono_storage; pub mod mq_storage; +pub mod raw_db_storage; pub mod user_storage; pub mod ztm_storage; -use async_trait::async_trait; use sea_orm::{sea_query::OnConflict, ActiveModelTrait, ConnectionTrait, EntityTrait}; -use callisto::import_refs; use common::errors::MegaError; -/// -/// This interface is designed to handle the commonalities between the database storage and -/// file system storage. -/// -#[async_trait] -pub trait GitStorageProvider: Send + Sync { - async fn save_ref(&self, repo_id: i64, refs: import_refs::Model) -> Result<(), MegaError>; - - async fn remove_ref(&self, repo_id: i64, ref_name: &str) -> Result<(), MegaError>; - - async fn get_ref(&self, repo_id: i64) -> Result, MegaError>; - - async fn update_ref(&self, repo_id: i64, ref_name: &str, new_id: &str) - -> Result<(), MegaError>; - - // async fn save_entry(&self, repo: &Repo, entry_list: Vec) -> Result<(), MegaError>; - - // async fn get_entry_by_sha1( - // &self, - // repo: Repo, - // sha1_vec: Vec<&str>, - // ) -> Result, MegaError>; -} /// Performs batch saving of models in the database. /// diff --git a/jupiter/src/storage/mono_storage.rs b/jupiter/src/storage/mono_storage.rs index b513eb552..a6fe77bca 100644 --- a/jupiter/src/storage/mono_storage.rs +++ b/jupiter/src/storage/mono_storage.rs @@ -12,21 +12,17 @@ use callisto::{ mega_blob, mega_commit, mega_mr, mega_mr_comment, mega_mr_conv, mega_refs, mega_tag, mega_tree, raw_blob, }; -use common::config::StorageConfig; use common::errors::MegaError; use common::utils::generate_id; use mercury::internal::object::MegaObjectModel; use mercury::internal::{object::commit::Commit, pack::entry::Entry}; -use crate::raw_storage::{self, RawStorage}; use crate::storage::batch_save_model; use crate::utils::converter::MegaModelConverter; #[derive(Clone)] pub struct MonoStorage { - pub raw_storage: Arc, pub connection: Arc, - pub raw_obj_threshold: usize, } #[derive(Debug)] @@ -43,20 +39,13 @@ impl MonoStorage { &self.connection } - pub async fn new(connection: Arc, config: StorageConfig) -> Self { - MonoStorage { - connection, - raw_storage: raw_storage::init(config.raw_obj_storage_type, config.raw_obj_local_path) - .await, - raw_obj_threshold: config.big_obj_threshold, - } + pub async fn new(connection: Arc) -> Self { + MonoStorage { connection } } pub fn mock() -> Self { MonoStorage { connection: Arc::new(DatabaseConnection::default()), - raw_storage: raw_storage::mock(), - raw_obj_threshold: 1024, } } @@ -387,28 +376,6 @@ impl MonoStorage { .await .unwrap()) } - - pub async fn get_raw_blobs_by_hashes( - &self, - hashes: Vec, - ) -> Result, MegaError> { - Ok(raw_blob::Entity::find() - .filter(raw_blob::Column::Sha1.is_in(hashes)) - .all(self.get_connection()) - .await - .unwrap()) - } - - pub async fn get_raw_blob_by_hash( - &self, - hash: &str, - ) -> Result, MegaError> { - Ok(raw_blob::Entity::find() - .filter(raw_blob::Column::Sha1.eq(hash)) - .one(self.get_connection()) - .await - .unwrap()) - } } #[cfg(test)] diff --git a/jupiter/src/storage/raw_db_storage.rs b/jupiter/src/storage/raw_db_storage.rs new file mode 100644 index 000000000..809c4f58b --- /dev/null +++ b/jupiter/src/storage/raw_db_storage.rs @@ -0,0 +1,61 @@ +use std::sync::Arc; + +use futures::Stream; +use sea_orm::{ColumnTrait, DatabaseConnection, DbErr, EntityTrait, QueryFilter}; + +use callisto::raw_blob; +use common::errors::MegaError; + +#[derive(Clone)] +pub struct RawDbStorage { + pub connection: Arc, +} + +impl RawDbStorage { + pub fn get_connection(&self) -> &DatabaseConnection { + &self.connection + } + + pub async fn new(connection: Arc) -> Self { + RawDbStorage { connection } + } + + pub fn mock() -> Self { + RawDbStorage { + connection: Arc::new(DatabaseConnection::default()), + } + } + + pub async fn get_raw_blobs_by_hashes( + &self, + hashes: Vec, + ) -> Result, MegaError> { + Ok(raw_blob::Entity::find() + .filter(raw_blob::Column::Sha1.is_in(hashes)) + .all(self.get_connection()) + .await + .unwrap()) + } + + pub async fn get_raw_blob_by_hash( + &self, + hash: &str, + ) -> Result, MegaError> { + Ok(raw_blob::Entity::find() + .filter(raw_blob::Column::Sha1.eq(hash)) + .one(self.get_connection()) + .await + .unwrap()) + } + + pub async fn get_raw_blobs_stream( + &self, + hashes: Vec, + ) -> Result> + '_ + Send, MegaError> { + Ok(raw_blob::Entity::find() + .filter(raw_blob::Column::Sha1.is_in(hashes)) + .stream(self.get_connection()) + .await + .unwrap()) + } +} diff --git a/mega/config.toml b/mega/config.toml index f75f2ca81..f939074f3 100644 --- a/mega/config.toml +++ b/mega/config.toml @@ -40,14 +40,6 @@ sqlx_logging = false ssh_key_path = "${base_dir}/ssh" [storage] -# raw object stroage type, can be `local` or `remote` -raw_obj_storage_type = "LOCAL" - -## If the object file size exceeds the threshold value, it will be handled by file storage instead of the database, Unit is KB -big_obj_threshold = 1024 - -# set the local path of the project storage -raw_obj_local_path = "${base_dir}/objects" obs_access_key = "" obs_secret_key = "" diff --git a/mono/config.toml b/mono/config.toml index 2cdb5d7c3..ac0dffa15 100644 --- a/mono/config.toml +++ b/mono/config.toml @@ -40,14 +40,6 @@ sqlx_logging = false ssh_key_path = "${base_dir}/ssh" [storage] -# raw object stroage type, can be `local` or `remote` -raw_obj_storage_type = "LOCAL" - -## If the object file size exceeds the threshold value, it will be handled by file storage instead of the database, Unit is KB -big_obj_threshold = 1024 - -# set the local path of the project storage -raw_obj_local_path = "${base_dir}/objects" obs_access_key = "" obs_secret_key = "" diff --git a/mono/src/api/api_router.rs b/mono/src/api/api_router.rs index 890aa9ef7..8fde2a8f4 100644 --- a/mono/src/api/api_router.rs +++ b/mono/src/api/api_router.rs @@ -1,19 +1,24 @@ use axum::{ - extract::{Query, State}, - http::StatusCode, - response::IntoResponse, + body::Body, + extract::{Path, Query, State}, + response::{IntoResponse, Response}, routing::{get, post}, Json, Router, }; -use ceres::model::{ - create_file::CreateFileInfo, - query::{BlobContentQuery, CodePreviewQuery}, - tree::{LatestCommitInfo, TreeBriefItem, TreeCommitItem}, +use ceres::{ + api_service::ApiHandler, + model::{ + create_file::CreateFileInfo, + query::{BlobContentQuery, CodePreviewQuery}, + tree::{LatestCommitInfo, TreeBriefItem, TreeCommitItem}, + }, }; use common::model::CommonResult; +use http::StatusCode; use taurus::event::api_request::{ApiRequestEvent, ApiType}; +use crate::api::error::ApiError; use crate::api::mr_router; use crate::api::user::user_router; use crate::api::MonoApiServiceState; @@ -25,7 +30,9 @@ pub fn routers() -> Router { .route("/latest-commit", get(get_latest_commit)) .route("/tree/commit-info", get(get_tree_commit_info)) .route("/tree", get(get_tree_info)) - .route("/blob", get(get_blob_object)); + .route("/blob", get(get_blob_string)) + .route("/file/blob/:object_id", get(get_blob_file)) + .route("/file/tree", get(get_tree_file)); Router::new() .merge(router) @@ -33,10 +40,10 @@ pub fn routers() -> Router { .merge(user_router::routers()) } -async fn get_blob_object( +async fn get_blob_string( Query(query): Query, state: State, -) -> Result>, (StatusCode, String)> { +) -> Result>, ApiError> { ApiRequestEvent::notify(ApiType::Blob, &state.0.context.config); let res = state .api_handler(query.path.clone().into()) @@ -51,14 +58,14 @@ async fn get_blob_object( Ok(Json(res)) } -async fn life_cycle_check() -> Result { +async fn life_cycle_check() -> Result { Ok(Json("http ready")) } async fn create_file( state: State, Json(json): Json, -) -> Result>, (StatusCode, String)> { +) -> Result>, ApiError> { ApiRequestEvent::notify(ApiType::CreateFile, &state.0.context.config); let res = state .api_handler(json.path.clone().into()) @@ -75,21 +82,20 @@ async fn create_file( async fn get_latest_commit( Query(query): Query, state: State, -) -> Result, (StatusCode, String)> { +) -> Result, ApiError> { ApiRequestEvent::notify(ApiType::LastestCommit, &state.0.context.config); let res = state .api_handler(query.path.clone().into()) .await .get_latest_commit(query.path.into()) - .await - .unwrap(); + .await?; Ok(Json(res)) } async fn get_tree_info( Query(query): Query, state: State, -) -> Result>>, (StatusCode, String)> { +) -> Result>>, ApiError> { ApiRequestEvent::notify(ApiType::TreeInfo, &state.0.context.config); let res = state .api_handler(query.path.clone().into()) @@ -106,7 +112,7 @@ async fn get_tree_info( async fn get_tree_commit_info( Query(query): Query, state: State, -) -> Result>>, (StatusCode, String)> { +) -> Result>>, ApiError> { ApiRequestEvent::notify(ApiType::CommitInfo, &state.0.context.config); let res = state .api_handler(query.path.clone().into()) @@ -119,3 +125,52 @@ async fn get_tree_commit_info( }; Ok(Json(res)) } + +pub async fn get_blob_file( + state: State, + Path(oid): Path, +) -> Result { + let api_handler = state.monorepo(); + + let result = api_handler.get_raw_blob_by_hash(&oid).await.unwrap(); + let file_name = format!("inline; filename=\"{}\"", oid); + match result { + Some(model) => Ok(Response::builder() + .header("Content-Type", "application/octet-stream") + .header("Content-Disposition", file_name) + .body(Body::from(model.data.unwrap())) + .unwrap()), + None => Ok({ + Response::builder() + .status(StatusCode::NOT_FOUND) + .body(Body::empty()) + .unwrap() + }), + } +} + +pub async fn get_tree_file( + state: State, + Query(query): Query, +) -> Result { + let res = state + .api_handler(query.path.clone().into()) + .await + .get_tree_as_data(std::path::Path::new(&query.path)) + .await; + + let file_name = format!("inline; filename=\"{}\"", ""); + match res { + Ok(data) => Ok(Response::builder() + .header("Content-Type", "application/octet-stream") + .header("Content-Disposition", file_name) + .body(Body::from(data)) + .unwrap()), + Err(_) => Ok({ + Response::builder() + .status(StatusCode::NOT_FOUND) + .body(Body::empty()) + .unwrap() + }), + } +} diff --git a/mono/src/api/lfs/lfs_router.rs b/mono/src/api/lfs/lfs_router.rs index 5f89725fc..7f4bef491 100644 --- a/mono/src/api/lfs/lfs_router.rs +++ b/mono/src/api/lfs/lfs_router.rs @@ -83,7 +83,7 @@ pub async fn list_locks( Query(query): Query, ) -> Result, (StatusCode, String)> { let result: Result = - handler::lfs_retrieve_lock(state.context.services.lfs_storage.clone(), query).await; + handler::lfs_retrieve_lock(state.context.services.lfs_db_storage.clone(), query).await; match result { Ok(lock_list) => { let body = serde_json::to_string(&lock_list).unwrap_or_default(); @@ -100,7 +100,7 @@ pub async fn list_locks_for_verification( state: State, Json(json): Json, ) -> Result, (StatusCode, String)> { - let result = handler::lfs_verify_lock(state.context.services.lfs_storage.clone(), json).await; + let result = handler::lfs_verify_lock(state.context.services.lfs_db_storage.clone(), json).await; match result { Ok(lock_list) => { let body = serde_json::to_string(&lock_list).unwrap_or_default(); @@ -122,7 +122,7 @@ pub async fn create_lock( state: State, Json(json): Json, ) -> Result, (StatusCode, String)> { - let result = handler::lfs_create_lock(state.context.services.lfs_storage.clone(), json).await; + let result = handler::lfs_create_lock(state.context.services.lfs_db_storage.clone(), json).await; match result { Ok(lock) => { let lock_response = LockResponse { @@ -151,7 +151,7 @@ pub async fn delete_lock( Json(json): Json, ) -> Result { let result = - handler::lfs_delete_lock(state.context.services.lfs_storage.clone(), &id, json).await; + handler::lfs_delete_lock(state.context.services.lfs_db_storage.clone(), &id, json).await; match result { Ok(lock) => { diff --git a/mono/src/api/mr_router.rs b/mono/src/api/mr_router.rs index 71cde07c6..9da10f019 100644 --- a/mono/src/api/mr_router.rs +++ b/mono/src/api/mr_router.rs @@ -2,15 +2,15 @@ use std::{collections::HashMap, path::PathBuf}; use axum::{ extract::{Path, Query, State}, - http::StatusCode, routing::{get, post}, Json, Router, }; use ceres::model::mr::{MRDetail, MrInfoItem}; use common::model::CommonResult; - use taurus::event::api_request::{ApiRequestEvent, ApiType}; + +use crate::api::error::ApiError; use crate::api::MonoApiServiceState; pub fn routers() -> Router { @@ -24,7 +24,7 @@ pub fn routers() -> Router { async fn merge( Path(mr_id): Path, state: State, -) -> Result>, (StatusCode, String)> { +) -> Result>, ApiError> { ApiRequestEvent::notify(ApiType::MergeRequest, &state.0.context.config); let res = state.monorepo().merge_mr(mr_id).await; @@ -39,7 +39,7 @@ async fn merge( async fn get_mr_list( Query(query): Query>, state: State, -) -> Result>>, (StatusCode, String)> { +) -> Result>>, ApiError> { ApiRequestEvent::notify(ApiType::MergeList, &state.0.context.config); let status = query.get("status").unwrap(); let res = state.monorepo().mr_list(status).await; @@ -53,7 +53,7 @@ async fn get_mr_list( async fn mr_detail( Path(mr_id): Path, state: State, -) -> Result>>, (StatusCode, String)> { +) -> Result>>, ApiError> { ApiRequestEvent::notify(ApiType::MergeDetail, &state.0.context.config); let res = state.monorepo().mr_detail(mr_id).await; let res = match res { @@ -66,7 +66,7 @@ async fn mr_detail( async fn get_mr_files( Path(mr_id): Path, state: State, -) -> Result>>, (StatusCode, String)> { +) -> Result>>, ApiError> { ApiRequestEvent::notify(ApiType::MergeFiles, &state.0.context.config); let res = state.monorepo().mr_tree_files(mr_id).await; let res = match res { diff --git a/mono/src/server/https_server.rs b/mono/src/server/https_server.rs index 2691f3704..e8dc9ba00 100644 --- a/mono/src/server/https_server.rs +++ b/mono/src/server/https_server.rs @@ -124,7 +124,8 @@ pub async fn app(config: Config, host: String, port: u16, common: CommonOptions) // add TraceLayer for log record // add CorsLayer to add cors header Router::new() - .nest("/", lfs_router::routers().with_state(api_state.clone())) + .nest("/", lfs_router::routers()) + .with_state(api_state.clone()) .nest( "/api/v1", api_router::routers().with_state(api_state.clone()),