Merge pull request #3 from Embucket/CP-5

Implement warehouses and profiles

Author: osipovartem

crates/nexus/src/http/ui/handlers/databases.rs

@@ -96,7 +96,7 @@ pub async fn get_database(
             created_at: Default::default(),
             updated_at: Default::default(),
             storage_profile: storage_profile::StorageProfile {
-                r#type: aws::CloudProvider::S3,
+                r#type: aws::CloudProvider::AWS,
                 region: "".to_string(),
                 bucket: "".to_string(),
                 credentials: Default::default(),

crates/nexus/src/http/ui/handlers/profiles.rs

@@ -2,6 +2,7 @@ use crate::error::AppError;
 use crate::http::ui::models::{aws, storage_profile};
 use crate::state::AppState;
 use axum::{extract::Path, extract::State, Json};
+use control_plane::models::{StorageProfile, StorageProfileCreateRequest};
 use utoipa::OpenApi;
 use uuid::Uuid;
 
@@ -44,17 +45,13 @@ pub async fn create_storage_profile(
     State(state): State<AppState>,
     Json(payload): Json<storage_profile::CreateStorageProfilePayload>,
 ) -> Result<Json<storage_profile::StorageProfile>, AppError> {
-    Ok(Json(storage_profile::StorageProfile {
-        r#type: aws::CloudProvider::S3,
-        region: "2".to_string(),
-        bucket: "".to_string(),
-        credentials: Default::default(),
-        sts_role_arn: None,
-        endpoint: None,
-        id: Default::default(),
-        created_at: Default::default(),
-        updated_at: Default::default(),
-    }))
+    let request: StorageProfileCreateRequest = payload.into();
+    let profile: StorageProfile = state
+        .control_svc
+        .create_profile(&request)
+        .await
+        .map_err(|e| AppError::from(e))?;
+    Ok(Json(profile.into()))
 }
 
 #[utoipa::path(
@@ -71,19 +68,14 @@ pub async fn create_storage_profile(
 )]
 pub async fn get_storage_profile(
     State(state): State<AppState>,
-    Path(id): Path<Uuid>,
+    Path(storage_profile_id): Path<Uuid>,
 ) -> Result<Json<storage_profile::StorageProfile>, AppError> {
-    Ok(Json(storage_profile::StorageProfile {
-        r#type: aws::CloudProvider::S3,
-        region: "1".to_string(),
-        bucket: "".to_string(),
-        credentials: Default::default(),
-        sts_role_arn: None,
-        endpoint: None,
-        id: Default::default(),
-        created_at: Default::default(),
-        updated_at: Default::default(),
-    }))
+    let profile: StorageProfile = state
+        .control_svc
+        .get_profile(storage_profile_id)
+        .await
+        .map_err(|e| AppError::from(e))?;
+    Ok(Json(profile.into()))
 }
 
 #[utoipa::path(
@@ -100,8 +92,13 @@ pub async fn get_storage_profile(
 )]
 pub async fn delete_storage_profile(
     State(state): State<AppState>,
-    Path(id): Path<Uuid>,
+    Path(storage_profile_id): Path<Uuid>,
 ) -> Result<Json<()>, AppError> {
+    state
+        .control_svc
+        .delete_profile(storage_profile_id)
+        .await
+        .map_err(|e| AppError::from(e))?;
     Ok(Json(()))
 }
 
@@ -117,5 +114,6 @@ pub async fn delete_storage_profile(
 pub async fn list_storage_profiles(
     State(state): State<AppState>,
 ) -> Result<Json<Vec<storage_profile::StorageProfile>>, AppError> {
-    Ok(Json(vec![]))
+    let profiles = state.control_svc.list_profiles().await?;
+    Ok(Json(profiles.into_iter().map(|p| p.into()).collect()))
 }

crates/nexus/src/http/ui/handlers/tables.rs

@@ -90,7 +90,7 @@ pub async fn get_table(
                 statistics: None,
                 compaction_summary: None,
                 storage_profile: storage_profile::StorageProfile {
-                    r#type: aws::CloudProvider::S3,
+                    r#type: aws::CloudProvider::AWS,
                     region: "22".to_string(),
                     bucket: "2".to_string(),
                     credentials: Default::default(),
@@ -155,7 +155,7 @@ pub async fn create_table(
                 statistics: None,
                 compaction_summary: None,
                 storage_profile: storage_profile::StorageProfile {
-                    r#type: aws::CloudProvider::S3,
+                    r#type: aws::CloudProvider::AWS,
                     region: "2".to_string(),
                     bucket: "2".to_string(),
                     credentials: Default::default(),

crates/nexus/src/http/ui/handlers/warehouses.rs

@@ -2,6 +2,7 @@ use crate::error::AppError;
 use crate::http::ui::models::warehouse;
 use crate::state::AppState;
 use axum::{extract::Path, extract::State, Json};
+use control_plane::models::{Warehouse as WarehouseModel, WarehouseCreateRequest};
 use utoipa::OpenApi;
 use uuid::Uuid;
 
@@ -31,14 +32,33 @@ pub struct ApiDoc;
     path = "/ui/warehouses",
     operation_id = "webWarehousesDashboard",
     responses(
-        (status = 200, description = "List all warehouses", body = Vec<warehouse::WarehousesDashboard>),
+        (status = 200, description = "List all warehouses", body = warehouse::WarehousesDashboard),
         (status = 500, description = "Internal server error")
     )
 )]
 pub async fn list_warehouses(
     State(state): State<AppState>,
-) -> Result<Json<Vec<warehouse::WarehousesDashboard>>, AppError> {
-    Ok(Json(vec![]))
+) -> Result<Json<warehouse::WarehousesDashboard>, AppError> {
+    let warehouses = state.control_svc.list_warehouses().await?;
+    let storage_profiles = state.control_svc.list_profiles().await?;
+    let mut extended_warehouses = Vec::new();
+
+    for warehouse in warehouses {
+        let mut extended_warehouse =
+            warehouse::WarehouseExtended::new(warehouse.clone().into(), Default::default());
+        if let Some(profile) = storage_profiles
+            .iter()
+            .find(|p| p.id == extended_warehouse.storage_profile_id)
+        {
+            extended_warehouse.storage_profile = profile.clone().into();
+            extended_warehouses.push(extended_warehouse)
+        }
+    }
+    Ok(Json(warehouse::WarehousesDashboard {
+        warehouses: extended_warehouses,
+        statistics: Default::default(),
+        compaction_summary: None,
+    }))
 }
 
 #[utoipa::path(
@@ -49,25 +69,28 @@ pub async fn list_warehouses(
         ("warehouseId" = Uuid, Path, description = "Warehouse ID")
     ),
     responses(
-        (status = 200, description = "Warehouse found", body = warehouse::Warehouse),
+        (status = 200, description = "Warehouse found", body = warehouse::WarehouseExtended),
         (status = 404, description = "Warehouse not found")
     )
 )]
 pub async fn get_warehouse(
     State(state): State<AppState>,
     Path(warehouse_id): Path<Uuid>,
-) -> Result<Json<warehouse::Warehouse>, AppError> {
-    // let warehouse = state.warehouse_service.get_warehouse(warehouse_id).await?;
-    Ok(Json(warehouse::Warehouse {
-        name: "".to_string(),
-        storage_profile_id: Default::default(),
-        key_prefix: "key".to_string(),
-        id: Default::default(),
-        external_id: Default::default(),
-        location: "".to_string(),
-        created_at: Default::default(),
-        updated_at: Default::default(),
-    }))
+) -> Result<Json<warehouse::WarehouseExtended>, AppError> {
+    let warehouse = state.control_svc.get_warehouse(warehouse_id).await?;
+
+    let mut extended_warehouse =
+        warehouse::WarehouseExtended::new(warehouse.into(), Default::default());
+
+    if let Ok(profile) = state
+        .control_svc
+        .get_profile(extended_warehouse.storage_profile_id)
+        .await
+    {
+        extended_warehouse.storage_profile = profile.into();
+    }
+
+    Ok(Json(extended_warehouse))
 }
 
 #[utoipa::path(
@@ -85,22 +108,25 @@ pub async fn create_warehouse(
     State(state): State<AppState>,
     Json(payload): Json<warehouse::CreateWarehousePayload>,
 ) -> Result<Json<warehouse::Warehouse>, AppError> {
-    Ok(Json(warehouse::Warehouse {
-        name: "".to_string(),
-        storage_profile_id: Default::default(),
-        key_prefix: "".to_string(),
-        id: Default::default(),
-        external_id: Default::default(),
-        location: "".to_string(),
-        created_at: Default::default(),
-        updated_at: Default::default(),
-    }))
+    let request: WarehouseCreateRequest = payload.into();
+
+    state
+        .control_svc
+        .get_profile(request.storage_profile_id)
+        .await
+        .map_err(|e| AppError::from(e))?;
+    let warehouse: WarehouseModel = state
+        .control_svc
+        .create_warehouse(&request)
+        .await
+        .map_err(|e| AppError::from(e))?;
+    Ok(Json(warehouse.into()))
 }
 
 #[utoipa::path(
     delete,
     path = "/ui/warehouses/{warehouseId}",
-    operation_id = "webCeleteWarehouse",
+    operation_id = "webCreteWarehouse",
     params(
         ("warehouseId" = Uuid, Path, description = "Warehouse ID")
     ),
@@ -112,6 +138,11 @@ pub async fn create_warehouse(
 pub async fn delete_warehouse(
     State(state): State<AppState>,
     Path(warehouse_id): Path<Uuid>,
-) -> Result<(), AppError> {
-    Ok(())
+) -> Result<Json<()>, AppError> {
+    state
+        .control_svc
+        .delete_warehouse(warehouse_id)
+        .await
+        .map_err(|e| AppError::from(e))?;
+    Ok(Json(()))
 }

crates/nexus/src/http/ui/mod.rs

@@ -1,3 +1,3 @@
-pub mod router;
-pub mod models;
 pub mod handlers;
+pub mod models;
+pub mod router;

crates/nexus/src/http/ui/models/aws.rs

@@ -1,3 +1,4 @@
+use control_plane::models;
 use serde::{Deserialize, Serialize};
 use utoipa::ToSchema;
 use validator::Validate;
@@ -20,6 +21,23 @@ impl AwsAccessKeyCredential {
     }
 }
 
+impl From<AwsAccessKeyCredential> for models::AwsAccessKeyCredential {
+    fn from(credential: AwsAccessKeyCredential) -> Self {
+        models::AwsAccessKeyCredential {
+            aws_access_key_id: credential.aws_access_key_id,
+            aws_secret_access_key: credential.aws_secret_access_key,
+        }
+    }
+}
+impl From<models::AwsAccessKeyCredential> for AwsAccessKeyCredential {
+    fn from(credential: models::AwsAccessKeyCredential) -> Self {
+        AwsAccessKeyCredential {
+            aws_access_key_id: credential.aws_access_key_id,
+            aws_secret_access_key: credential.aws_secret_access_key,
+        }
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Validate, ToSchema)]
 pub struct AwsRoleCredential {
     #[validate(length(min = 1))]
@@ -38,26 +56,108 @@ impl AwsRoleCredential {
     }
 }
 
+impl From<AwsRoleCredential> for models::AwsRoleCredential {
+    fn from(credential: AwsRoleCredential) -> Self {
+        models::AwsRoleCredential {
+            role_arn: credential.role_arn,
+            external_id: credential.external_id,
+        }
+    }
+}
+impl From<models::AwsRoleCredential> for AwsRoleCredential {
+    fn from(credential: models::AwsRoleCredential) -> Self {
+        AwsRoleCredential {
+            role_arn: credential.role_arn,
+            external_id: credential.external_id,
+        }
+    }
+}
+
 #[derive(
-    Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, Hash, ToSchema,
+    Debug,
+    Clone,
+    Copy,
+    PartialEq,
+    Eq,
+    PartialOrd,
+    Ord,
+    Serialize,
+    Deserialize,
+    Hash,
+    Default,
+    ToSchema,
 )]
 pub enum CloudProvider {
-    #[serde(rename = "s3")]
-    S3,
+    #[serde(rename = "aws")]
+    #[default]
+    AWS,
     #[serde(rename = "gcs")]
-    Gcs,
+    GCS,
     #[serde(rename = "azure")]
-    Azure,
+    AZURE,
+}
+
+impl From<models::CloudProvider> for CloudProvider {
+    fn from(provider: models::CloudProvider) -> Self {
+        match provider {
+            models::CloudProvider::AWS => CloudProvider::AWS,
+            models::CloudProvider::GCS => CloudProvider::GCS,
+            models::CloudProvider::AZURE => CloudProvider::AZURE,
+        }
+    }
+}
+
+impl From<CloudProvider> for models::CloudProvider {
+    fn from(provider: CloudProvider) -> Self {
+        match provider {
+            CloudProvider::AWS => models::CloudProvider::AWS,
+            CloudProvider::GCS => models::CloudProvider::GCS,
+            CloudProvider::AZURE => models::CloudProvider::AZURE,
+        }
+    }
 }
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, ToSchema)]
 pub enum Credentials {
-    AwsAccessKeyCredential(AwsAccessKeyCredential),
-    AwsRoleCredential(AwsRoleCredential),
+    #[serde(rename = "access_key")]
+    AccessKey(AwsAccessKeyCredential),
+    #[serde(rename = "role")]
+    Role(AwsRoleCredential),
 }
 
 impl Default for Credentials {
     fn default() -> Self {
-        Credentials::AwsAccessKeyCredential(AwsAccessKeyCredential::default())
+        Credentials::AccessKey(AwsAccessKeyCredential::default())
+    }
+}
+impl From<models::Credentials> for Credentials {
+    fn from(credentials: models::Credentials) -> Self {
+        match credentials {
+            models::Credentials::AccessKey(creds) => Credentials::AccessKey(
+                AwsAccessKeyCredential::new(creds.aws_access_key_id, creds.aws_secret_access_key),
+            ),
+            models::Credentials::Role(creds) => {
+                Credentials::Role(AwsRoleCredential::new(creds.role_arn, creds.external_id))
+            }
+        }
+    }
+}
+
+impl From<Credentials> for models::Credentials {
+    fn from(credentials: Credentials) -> Self {
+        match credentials {
+            Credentials::AccessKey(creds) => models::Credentials::AccessKey(
+                models::AwsAccessKeyCredential {
+                    aws_access_key_id: creds.aws_access_key_id,
+                    aws_secret_access_key: creds.aws_secret_access_key,
+                },
+            ),
+            Credentials::Role(creds) => {
+                models::Credentials::Role(models::AwsRoleCredential {
+                    role_arn: creds.role_arn,
+                    external_id: creds.external_id,
+                })
+            }
+        }
     }
 }