wip: custom kube config path

.gitignore

@@ -15,4 +15,5 @@ target
 /logs
 /dist
 /queue
-/cache
+/cache
+/data

Cargo.toml

@@ -110,7 +110,8 @@ sysinfo = { version = "0.33.1" }
 tempfile = { version = "3.15.0" }
 image = { version = "0.25.5" }
 webp = { version = "0.3.0", features = ["image"] }
-hex = "0.4.3"
+hex = { version = "0.4.3" }
+lettre = { version = "0.11" }
 
 [profile.release]
 opt-level = 3

crates/cluster/Cargo.toml

@@ -17,4 +17,5 @@ once_cell = { workspace = true }
 tokio-util = { workspace = true }
 tracing = { workspace = true }
 thiserror = { workspace = true }
-wsrx = { workspace = true }
+wsrx = { workspace = true }
+anyhow = { workspace = true }

crates/cluster/src/lib.rs

@@ -1,7 +1,8 @@
 pub mod traits;
 
 use std::{collections::BTreeMap, process};
-
+use std::path::Path;
+use anyhow::anyhow;
 use axum::extract::ws::WebSocket;
 use k8s_openapi::{
     api::core::v1::{
@@ -15,6 +16,7 @@ use kube::{
     api::{Api, DeleteParams, ListParams, PostParams},
     runtime::wait::conditions,
 };
+use kube::config::Kubeconfig;
 use once_cell::sync::OnceCell;
 use tokio_util::codec::Framed;
 use tracing::{error, info};
@@ -27,17 +29,19 @@ pub fn get_k8s_client() -> K8sClient {
     K8S_CLIENT.get().unwrap().clone()
 }
 
-pub async fn init() {
-    let result = Config::from_kubeconfig(&Default::default()).await;
+pub async fn init() -> Result<(), ClusterError> {
+    let result = Config::from_custom_kubeconfig(Kubeconfig::read_from(
+        Path::new(cds_env::get_env().cluster.kube_config_path.as_str())
+    )?, &Default::default()).await;
     if let Err(e) = result {
         error!(
             "Failed to create Kubernetes client from custom config: {:?}",
             e
         );
         process::exit(1);
     }
-    let config = result.unwrap();
-    let client = K8sClient::try_from(config).unwrap();
+    let config = result?;
+    let client = K8sClient::try_from(config)?;
     if let Err(_) = client.apiserver_version().await {
         error!("Failed to connect to Kubernetes API server.");
         process::exit(1);
@@ -46,7 +50,7 @@ pub async fn init() {
     info!("Kubernetes client initialized successfully.");
 
     let namespace_api: Api<Namespace> = Api::all(get_k8s_client().clone());
-    let namespaces = namespace_api.list(&ListParams::default()).await.unwrap();
+    let namespaces = namespace_api.list(&ListParams::default()).await?;
     if !namespaces.items.iter().any(|namespace| {
         namespace.metadata.name == Some(cds_env::get_env().clone().cluster.namespace)
     }) {
@@ -60,8 +64,10 @@ pub async fn init() {
         let _ = namespace_api
             .create(&PostParams::default(), &namespace)
             .await;
-        info!("Namespace is created successfully.")
+        info!("Namespace is created successfully.");
     }
+
+    Ok(())
 }
 
 pub async fn create(
@@ -151,13 +157,13 @@ pub async fn create(
 
     let mut nats: Vec<cds_db::entity::pod::Nat> = Vec::new();
 
-    match cds_env::get_env().cluster.proxy.enabled {
+    match cds_env::get_env().cluster.proxy.is_enabled {
         true => {
             for port in env.ports {
                 nats.push(cds_db::entity::pod::Nat {
                     src: format!("{}", port),
                     dst: None,
-                    proxy: cds_env::get_env().cluster.proxy.enabled,
+                    proxy: cds_env::get_env().cluster.proxy.is_enabled,
                     entry: None,
                 });
             }
@@ -202,7 +208,7 @@ pub async fn create(
                             nats.push(cds_db::entity::pod::Nat {
                                 src: format!("{}", port.port),
                                 dst: Some(format!("{}", node_port)),
-                                proxy: cds_env::get_env().cluster.proxy.enabled,
+                                proxy: cds_env::get_env().cluster.proxy.is_enabled,
                                 entry: Some(format!(
                                     "{}:{}",
                                     cds_config::get_config().await.cluster.entry,

crates/cluster/src/traits.rs

@@ -12,4 +12,6 @@ pub enum ClusterError {
     KubeRuntimeWaitError(#[from] kube::runtime::wait::Error),
     #[error("proxy error: {0}")]
     ProxyError(#[from] wsrx::Error),
+    #[error(transparent)]
+    OtherError(#[from] anyhow::Error),
 }

crates/config/src/cluster/mod.rs

@@ -1,4 +1,4 @@
-pub mod proxy;
+
 pub mod strategy;
 
 use sea_orm::FromJsonQueryResult;

crates/env/src/cluster/mod.rs

@@ -5,5 +5,6 @@ use serde::{Deserialize, Serialize};
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct Env {
     pub namespace: String,
+    pub kube_config_path: String,
     pub proxy: proxy::Env,
 }

crates/env/src/cluster/proxy/mod.rs

@@ -2,6 +2,6 @@ use serde::{Deserialize, Serialize};
 
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct Env {
-    pub enabled: bool,
+    pub is_enabled: bool,
     pub traffic_capture: bool,
 }

crates/env/src/lib.rs

@@ -1,10 +1,11 @@
-pub mod axum;
+pub mod server;
 pub mod cache;
 pub mod cluster;
-pub mod consts;
 pub mod db;
 pub mod metric;
 pub mod queue;
+pub mod media;
+pub mod auth;
 
 use std::{path::Path, process};
 
@@ -17,12 +18,14 @@ static APP_ENV: OnceCell<Env> = OnceCell::new();
 
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct Env {
-    pub axum: axum::Env,
+    pub server: server::Env,
+    pub auth: auth::Env,
     pub db: db::Env,
     pub queue: queue::Env,
     pub cache: cache::Env,
     pub metric: metric::Env,
     pub cluster: cluster::Env,
+    pub media: media::Env,
 }
 
 pub async fn init() {
@@ -36,6 +39,6 @@ pub async fn init() {
     }
 }
 
-pub fn get_env() -> &'static Env {
-    APP_ENV.get().unwrap()
+pub fn get_env() -> Env {
+    APP_ENV.get().unwrap().clone()
 }

crates/env/src/media/mod.rs

@@ -0,0 +1,6 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct Env {
+    pub path: String,
+}

crates/env/src/metric/mod.rs

@@ -2,6 +2,6 @@ use serde::{Deserialize, Serialize};
 
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct Env {
-    pub enabled: bool,
+    pub is_enabled: bool,
     pub namespace: String,
 }

crates/env/src/queue/mod.rs

@@ -4,7 +4,7 @@ use serde::{Deserialize, Serialize};
 pub struct Env {
     pub host: String,
     pub port: u16,
-    pub user: String,
+    pub username: String,
     pub password: String,
     pub token: String,
     pub tls: bool,

crates/env/src/axum/mod.rs → crates/env/src/server/mod.rs

@@ -1,10 +1,7 @@
-pub mod jwt;
-
 use serde::{Deserialize, Serialize};
 
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct Env {
     pub host: String,
     pub port: u16,
-    pub jwt: jwt::Env,
 }

crates/media/src/lib.rs

@@ -12,7 +12,7 @@ use crate::traits::MediaError;
 
 pub async fn get(path: String, filename: String) -> Result<Vec<u8>, MediaError> {
     let filepath =
-        PathBuf::from(cds_env::consts::path::MEDIA).join(format!("{}/{}", path, filename));
+        PathBuf::from(cds_env::get_env().media.path).join(format!("{}/{}", path, filename));
 
     match File::open(&filepath).await {
         Ok(mut file) => {
@@ -27,7 +27,7 @@ pub async fn get(path: String, filename: String) -> Result<Vec<u8>, MediaError>
 }
 
 pub async fn scan_dir(path: String) -> Result<Vec<(String, u64)>, MediaError> {
-    let filepath = PathBuf::from(cds_env::consts::path::MEDIA).join(path);
+    let filepath = PathBuf::from(cds_env::get_env().media.path).join(path);
     let mut files = Vec::new();
 
     if metadata(&filepath).await.is_err() {
@@ -50,7 +50,7 @@ pub async fn scan_dir(path: String) -> Result<Vec<(String, u64)>, MediaError> {
 
 pub async fn save(path: String, filename: String, data: Vec<u8>) -> Result<(), MediaError> {
     let filepath =
-        PathBuf::from(cds_env::consts::path::MEDIA).join(format!("{}/{}", path, filename));
+        PathBuf::from(cds_env::get_env().media.path).join(format!("{}/{}", path, filename));
     if let Some(parent) = filepath.parent() {
         if metadata(parent).await.is_err() {
             create_dir_all(parent).await?;
@@ -63,15 +63,15 @@ pub async fn save(path: String, filename: String, data: Vec<u8>) -> Result<(), M
 
 pub async fn delete(path: String, filename: String) -> Result<(), MediaError> {
     let filepath =
-        PathBuf::from(cds_env::consts::path::MEDIA).join(format!("{}/{}", path, filename));
+        PathBuf::from(cds_env::get_env().media.path).join(format!("{}/{}", path, filename));
     if metadata(&filepath).await.is_ok() {
         remove_file(&filepath).await?;
     }
     Ok(())
 }
 
 pub async fn delete_dir(path: String) -> Result<(), MediaError> {
-    let filepath = PathBuf::from(cds_env::consts::path::MEDIA).join(path);
+    let filepath = PathBuf::from(cds_env::get_env().media.path).join(path);
     if metadata(&filepath).await.is_ok() {
         remove_dir_all(&filepath).await?;
     }

crates/queue/src/lib.rs

@@ -59,7 +59,7 @@ pub async fn init() {
     let client = async_nats::ConnectOptions::new()
         .require_tls(cds_env::get_env().queue.tls)
         .user_and_password(
-            cds_env::get_env().queue.user.clone(),
+            cds_env::get_env().queue.username.clone(),
             cds_env::get_env().queue.password.clone(),
         )
         .token(cds_env::get_env().queue.token.clone())

crates/server/src/main.rs

@@ -34,13 +34,13 @@ async fn bootstrap() {
     migrator::run().await;
 
     cds_config::init().await;
-    cds_cluster::init().await;
+    let _ = cds_cluster::init().await;
     cds_web::init().await;
 
     let addr = format!(
         "{}:{}",
-        cds_env::get_env().axum.host,
-        cds_env::get_env().axum.port
+        cds_env::get_env().server.host,
+        cds_env::get_env().server.port
     );
     let listener = tokio::net::TcpListener::bind(&addr).await;
 
@@ -55,7 +55,7 @@ async fn bootstrap() {
     )
     .with_graceful_shutdown(shutdown_signal())
     .await
-    .expect("Failed to start axum server");
+    .expect("Failed to start server server");
 }
 
 async fn shutdown_signal() {

crates/web/src/util/jwt.rs

@@ -9,15 +9,15 @@ pub struct Claims {
     pub exp: usize,
 }
 
-pub async fn get_jwt_config() -> cds_env::axum::jwt::Env {
-    if let Some(jwt) = cds_cache::get::<cds_env::axum::jwt::Env>("jwt")
+pub async fn get_jwt_config() -> cds_env::auth::Env {
+    if let Some(jwt) = cds_cache::get::<cds_env::auth::Env>("jwt")
         .await
         .unwrap()
     {
         return jwt;
     }
 
-    let mut jwt = cds_env::get_env().axum.jwt.clone();
+    let mut jwt = cds_env::get_env().auth.clone();
     let re = Regex::new(r"\[([Uu][Uu][Ii][Dd])]").unwrap();
     jwt.secret = re
         .replace_all(&jwt.secret, Uuid::new_v4().simple().to_string())