web: admin storage updates, admin stats updates

Cargo.toml

@@ -51,10 +51,10 @@ land-sdk = { path = "crates/sdk", version = "0.1.0-b17" }
 land-core = { path = "crates/core", version = "0.1.0-b17" }
 land-dao = { path = "crates/dao", version = "0.1.0-b17" }
 land-storage = { path = "crates/storage", version = "0.1.0-b17" }
-anyhow = "1.0.72"
+anyhow = "1.0.73"
 axum = { version = "0.6.20", features = ["headers", "ws"] }
 clap = { version = "4.3.21", features = ["derive", "env"] }
-tokio = { version = "1.30.0", features = ["full"] }
+tokio = { version = "1.31.0", features = ["full"] }
 tracing = "0.1.37"
 lazy_static = "1.4.0"
 time = "0.3.25"

binary/center/src/restapi/admin.rs

@@ -65,3 +65,44 @@ pub async fn update_settings_domain(
     );
     Ok(StatusCode::OK)
 }
+
+#[tracing::instrument(name = "[list_settings_storage]", skip_all)]
+pub async fn list_settings_storage(
+    Extension(current_user): Extension<CurrentUser>,
+) -> Result<(StatusCode, Json<params::SettingsStorageResponse>), AppError> {
+    is_admin(&current_user)?;
+    let (key, local_config, s3_config) = crate::settings::load_storage_settings().await?;
+    let response = params::SettingsStorageResponse {
+        storage_type: key.clone(),
+        local: local_config,
+        s3: s3_config,
+    };
+    info!("success, key:{}", key);
+    Ok((StatusCode::OK, Json(response)))
+}
+
+#[tracing::instrument(name = "[update_settings_storage]", skip_all)]
+pub async fn update_settings_storage(
+    Extension(current_user): Extension<CurrentUser>,
+    body: axum::body::Bytes,
+) -> Result<StatusCode, AppError> {
+    is_admin(&current_user)?;
+    let config = serde_json::from_slice::<land_storage::s3::Config>(&body)?;
+    crate::settings::reload_s3(&config).await?;
+    info!("success, config:{:?}", config);
+    Ok(StatusCode::OK)
+}
+
+#[tracing::instrument(name = "[stats_handler]", skip_all)]
+pub async fn stats_handler(
+    Extension(current_user): Extension<CurrentUser>,
+) -> Result<(StatusCode, Json<params::StatsResponse>), AppError> {
+    is_admin(&current_user)?;
+    let response = params::StatsResponse {
+        deployments: land_dao::deployment::get_stats().await?,
+        projects: land_dao::project::get_stats().await?,
+        users: land_dao::user::get_stats().await?,
+        regions: land_dao::region::get_stats().await?,
+    };
+    Ok((StatusCode::OK, Json(response)))
+}

binary/center/src/restapi/deployment.rs

@@ -4,6 +4,20 @@ use axum::{extract::Path, http::StatusCode, Extension, Json};
 use tracing::{debug_span, info, warn, Instrument};
 use validator::Validate;
 
+/// upload_chunks uploads deploy chunks to storage
+async fn upload_chunks(id: i32, storage_path: &str, deploy_chunk: Vec<u8>) -> anyhow::Result<()> {
+    let upload_res = land_storage::write_global(storage_path, deploy_chunk).await;
+    if upload_res.is_err() {
+        land_dao::deployment::set_deploy_status(id, land_dao::deployment::DeployStatus::Failed)
+            .await?;
+        return Err(upload_res.err().unwrap());
+    }
+    let _ = land_dao::deployment::set_storage_success(id, storage_path.to_string()).await?;
+    // trigger build conf
+    conf::trigger().await;
+    Ok(())
+}
+
 #[tracing::instrument(name = "[create_dp]", skip_all)]
 pub async fn create_handler(
     Extension(current_user): Extension<CurrentUser>,
@@ -43,36 +57,10 @@ pub async fn create_handler(
     let deployment_id = deployment.id;
     tokio::task::spawn(
         async move {
-            match land_storage::write(&storage_path, payload.deploy_chunk).await {
-                Ok(_) => {
-                    info!(
-                        "success to upload deploy wasm to storage, storage_path:{}",
-                        storage_path,
-                    );
-                }
-                Err(err) => {
-                    info!(
-                        "failed to upload deploy wasm to storage, storage_path:{}, err:{}",
-                        storage_path, err
-                    );
-                }
+            match upload_chunks(deployment_id, &storage_path, payload.deploy_chunk).await{
+                Ok(_) => info!("success to upload deploy chunk to storage, deployment_id:{}, storage_path:{}", deployment_id, storage_path),
+                Err(e) => warn!("failed to upload deploy chunk to storage, deployment_id:{}, storage_path:{}, err:{}", deployment_id, storage_path, e),
             }
-
-            // then update storage_path and deploy status
-            match land_dao::deployment::set_storage_success(deployment_id, storage_path.clone())
-                .await
-            {
-                Ok(_) => {}
-                Err(err) => {
-                    warn!(
-                        "failed to update deployment storage_path, id:{}, storage_path:{}, err:{}",
-                        deployment_id, storage_path, err
-                    );
-                }
-            }
-
-            // trigger build conf
-            conf::trigger().await;
         }
         .instrument(debug_span!("[upload_deploy_chunk]")),
     );

binary/center/src/restapi/mod.rs

@@ -51,6 +51,9 @@ fn api_router() -> Router {
         .route("/v1/regions", get(admin::list_regions))
         .route("/v1/settings/domains", get(admin::list_settings_domains))
         .route("/v1/settings/domains", post(admin::update_settings_domain))
+        .route("/v1/settings/storage", get(admin::list_settings_storage))
+        .route("/v1/settings/storage", post(admin::update_settings_storage))
+        .route("/v1/settings/stats", get(admin::stats_handler))
         .route_layer(middleware::from_fn(auth::middleware))
 }
 

binary/center/src/restapi/params.rs

@@ -129,8 +129,23 @@ pub struct RegionResponse {
     pub status: String,
 }
 
-#[derive(Serialize, Deserialize,Validate, Debug)]
+#[derive(Serialize, Deserialize, Validate, Debug)]
 pub struct SettingsDomainRequest {
     pub domain: String,
     pub protocol: String,
 }
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct SettingsStorageResponse {
+    pub storage_type: String,
+    pub local: land_storage::local::Config,
+    pub s3: land_storage::s3::Config,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct StatsResponse {
+    pub deployments: i32,
+    pub projects: i32,
+    pub users: i32,
+    pub regions: i32,
+}

binary/center/src/settings/mod.rs

@@ -9,6 +9,8 @@ use tracing::{debug, info};
 
 mod storage;
 pub use storage::init as init_storage;
+pub use storage::load_settings as load_storage_settings;
+pub use storage::reload_s3;
 
 /// DOMAIN is the domain to access the function
 pub static DOMAIN: Lazy<Mutex<DomainSetting>> = Lazy::new(|| {

binary/center/src/settings/storage.rs

@@ -5,11 +5,8 @@ use tracing::debug;
 
 /// first_init_s3 init s3 storage settings to db
 pub async fn first_init_s3() -> Result<()> {
-    let s3_key = settings::Key::S3Storage.to_string();
-    let content = serde_json::to_string(&land_storage::s3::Config::default())?;
-    let values: HashMap<String, String> = vec![(s3_key.clone(), content)].into_iter().collect();
-    settings::update_maps(values).await?;
-    Ok(())
+    let cfg = land_storage::s3::Config::default();
+    update_s3(&cfg).await
 }
 
 /// first_init_local init local storage settings to db
@@ -21,10 +18,15 @@ pub async fn first_init_local() -> Result<()> {
     Ok(())
 }
 
-async fn load_settings() -> Result<(land_storage::local::Config, land_storage::s3::Config)> {
+pub async fn load_settings() -> Result<(
+    String,
+    land_storage::local::Config,
+    land_storage::s3::Config,
+)> {
+    let type_key = settings::Key::StorageType.to_string();
     let s3_key = settings::Key::S3Storage.to_string();
     let local_storage_key = settings::Key::LocalStorage.to_string();
-    let keys = vec![s3_key.clone(), local_storage_key.clone()];
+    let keys = vec![type_key.clone(), s3_key.clone(), local_storage_key.clone()];
     let settings_map = settings::list_maps(keys).await?;
     let s3_config = if let Some(s3_content) = settings_map.get(&s3_key) {
         serde_json::from_str::<land_storage::s3::Config>(s3_content)?
@@ -36,7 +38,9 @@ async fn load_settings() -> Result<(land_storage::local::Config, land_storage::s
     } else {
         land_storage::local::Config::default()
     };
-    Ok((local_config, s3_config))
+    let default_value = "local".to_string();
+    let type_key_value = settings_map.get(&type_key).unwrap_or(&default_value);
+    Ok((type_key_value.to_string(), local_config, s3_config))
 }
 
 /// update_storage_type update storage type
@@ -50,8 +54,8 @@ async fn update_storage_type(stype: String) -> Result<()> {
 /// init storage
 #[tracing::instrument(name = "[STORAGE]")]
 pub async fn init() -> Result<()> {
-    let (local_config, s3_config) = load_settings().await?;
-    let type_name = std::env::var("STORAGE_TYPE").unwrap_or_else(|_| "local".to_string());
+    let (current_type, local_config, s3_config) = load_settings().await?;
+    let type_name = std::env::var("STORAGE_TYPE").unwrap_or_else(|_| current_type.clone());
     match type_name.as_str() {
         "local" => {
             debug!("Init, STORAGE_TYPE:{}", "local");
@@ -68,3 +72,18 @@ pub async fn init() -> Result<()> {
     update_storage_type(type_name).await?;
     Ok(())
 }
+
+async fn update_s3(cfg: &land_storage::s3::Config) -> Result<()> {
+    let s3_key = settings::Key::S3Storage.to_string();
+    let content = serde_json::to_string(cfg)?;
+    let values: HashMap<String, String> = vec![(s3_key.clone(), content)].into_iter().collect();
+    settings::update_maps(values).await?;
+    Ok(())
+}
+
+/// reload_s3 reload s3 storage settings to db
+pub async fn reload_s3(cfg: &land_storage::s3::Config) -> Result<()> {
+    update_s3(cfg).await?;
+    land_storage::s3::reload_global(cfg).await?;
+    Ok(())
+}

crates/dao/src/deployment.rs

@@ -81,6 +81,21 @@ pub async fn set_storage_success(id: i32, storage_path: String) -> Result<deploy
     Ok(deployment)
 }
 
+/// set_deploy_status sets deploy status
+pub async fn set_deploy_status(id: i32, status: DeployStatus) -> Result<()> {
+    let db = DB.get().unwrap();
+    let deployment = deployment::Entity::find_by_id(id)
+        .one(db)
+        .await?
+        .ok_or(anyhow::anyhow!("deployment not found"))?;
+
+    let mut active_model: deployment::ActiveModel = deployment.into();
+    active_model.deploy_status = Set(status.to_string());
+    active_model.updated_at = Set(chrono::Utc::now());
+    active_model.update(db).await?;
+    Ok(())
+}
+
 /// find_by_uuid finds a deployment by uuid
 pub async fn publish(owner_id: i32, uuid: String) -> Result<deployment::Model> {
     let db = DB.get().unwrap();
@@ -228,3 +243,17 @@ pub async fn update_prod_domain(id: i32, domain: String) -> Result<()> {
         .await?;
     Ok(())
 }
+
+/// get_stats gets the stats of deployments
+pub async fn get_stats() -> Result<i32> {
+    let db = DB.get().unwrap();
+    let values: Vec<JsonValue> = JsonValue::find_by_statement(Statement::from_sql_and_values(
+        DbBackend::MySql,
+        r#"select count(id) as counter from deployment where status != 'deleted'"#,
+        [],
+    ))
+    .all(db)
+    .await?;
+    let counter = values[0]["counter"].as_i64().unwrap() as i32;
+    Ok(counter)
+}

crates/dao/src/project.rs

@@ -1,7 +1,10 @@
 use crate::{model::project, DB};
 use anyhow::Result;
 use rand::{thread_rng, Rng};
-use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QueryOrder, Set};
+use sea_orm::{
+    ActiveModelTrait, ColumnTrait, DbBackend, EntityTrait, FromQueryResult, JsonValue, QueryFilter,
+    QueryOrder, Set, Statement,
+};
 
 #[derive(strum::Display)]
 #[strum(serialize_all = "lowercase")]
@@ -136,3 +139,17 @@ pub async fn rename(owner_id: i32, old_name: String, new_name: String) -> Result
     let project = active_model.update(db).await?;
     Ok(project)
 }
+
+/// get_stats gets the stats of deployments
+pub async fn get_stats() -> Result<i32> {
+    let db = DB.get().unwrap();
+    let values: Vec<JsonValue> = JsonValue::find_by_statement(Statement::from_sql_and_values(
+        DbBackend::MySql,
+        r#"select count(id) as counter from project where status != 'deleted'"#,
+        [],
+    ))
+    .all(db)
+    .await?;
+    let counter = values[0]["counter"].as_i64().unwrap() as i32;
+    Ok(counter)
+}

crates/dao/src/region.rs

@@ -1,6 +1,9 @@
 use crate::{model::region, DB};
 use anyhow::Result;
-use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QueryOrder, Set};
+use sea_orm::{
+    ActiveModelTrait, ColumnTrait, DbBackend, EntityTrait, FromQueryResult, JsonValue, QueryFilter,
+    QueryOrder, Set, Statement,
+};
 use std::collections::HashMap;
 
 #[derive(strum::Display)]
@@ -82,3 +85,17 @@ pub async fn set_inactive(key: String) -> Result<()> {
     active_model.update(db).await?;
     Ok(())
 }
+
+/// get_stats gets the stats of deployments
+pub async fn get_stats() -> Result<i32> {
+    let db = DB.get().unwrap();
+    let values: Vec<JsonValue> = JsonValue::find_by_statement(Statement::from_sql_and_values(
+        DbBackend::MySql,
+        r#"select count(id) as counter from region where status != 'deleted'"#,
+        [],
+    ))
+    .all(db)
+    .await?;
+    let counter = values[0]["counter"].as_i64().unwrap() as i32;
+    Ok(counter)
+}

crates/dao/src/user.rs

@@ -8,7 +8,10 @@ use crate::{
 use anyhow::Result;
 use rand::distributions::Alphanumeric;
 use rand::{thread_rng, Rng};
-use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter};
+use sea_orm::{
+    ActiveModelTrait, ColumnTrait, DbBackend, EntityTrait, FromQueryResult, JsonValue, QueryFilter,
+    Statement,
+};
 
 #[derive(strum::Display)]
 #[strum(serialize_all = "lowercase")]
@@ -132,3 +135,17 @@ pub async fn signup_by_oauth(
     .await?;
     Ok((user_model, token))
 }
+
+/// get_stats gets the stats of deployments
+pub async fn get_stats() -> Result<i32> {
+    let db = DB.get().unwrap();
+    let values: Vec<JsonValue> = JsonValue::find_by_statement(Statement::from_sql_and_values(
+        DbBackend::MySql,
+        r#"select count(id) as counter from user_info where status != 'deleted'"#,
+        [],
+    ))
+    .all(db)
+    .await?;
+    let counter = values[0]["counter"].as_i64().unwrap() as i32;
+    Ok(counter)
+}