diff --git a/api/src/main.rs b/api/src/main.rs index e9f0da1..1510af6 100644 --- a/api/src/main.rs +++ b/api/src/main.rs @@ -4,7 +4,7 @@ use async_nats::{ jetstream::{self}, }; use axum::{ - routing::{get, post}, + routing::{get, post, delete}, Router, }; use db::seed::seed_database; @@ -169,6 +169,10 @@ async fn main() -> io::Result<()> { "/pipeline-nodes/:id", post(routes::api::v0::pipeline_nodes::update), ) + .route( + "/pipeline-nodes/:id", + delete(routes::api::v0::pipeline_nodes::delete) + ) .route( "/pipeline-node-connections", post(routes::api::v0::pipeline_node_connections::create), diff --git a/api/src/routes/api/v0/pipeline_nodes.rs b/api/src/routes/api/v0/pipeline_nodes.rs index bbcce35..600e2a0 100644 --- a/api/src/routes/api/v0/pipeline_nodes.rs +++ b/api/src/routes/api/v0/pipeline_nodes.rs @@ -1,7 +1,10 @@ use axum::{extract::Path, Json}; use hyper::StatusCode; use serde::{Deserialize, Serialize}; +use serde_json::json; +use tracing::info; use uuid::Uuid; +use sqlx::Acquire; use crate::{ app_state::{Coords, DatabaseConnection}, @@ -49,6 +52,60 @@ pub struct PipelineNodeCreationParams { coords: Coords, } +#[derive(Debug, Serialize)] +struct ApiError { + error: String, +} + +impl From for ApiError { + fn from(err: sqlx::Error) -> Self { + ApiError { + error: err.to_string(), + } + } +} + +pub async fn delete( + DatabaseConnection(mut conn): DatabaseConnection, + Path(id): Path, +) -> Result)> { + info!("Attempting to delete pipeline node: {id}"); + + let mut tx = conn + .begin() + .await + .map_err(internal_error)?; + + // Check if node exists + let node_exists = sqlx::query!("SELECT id FROM pipeline_nodes WHERE id = $1", id) + .fetch_optional(&mut *tx) + .await + .map_err(internal_error)? + .is_some(); + + if !node_exists { + return Err(( + StatusCode::NOT_FOUND, + Json(ApiError { + error: "Pipeline node not found".to_string(), + }), + )); + } + + // Delete the node (dependencies handled by ON DELETE CASCADE) + sqlx::query!("DELETE FROM pipeline_nodes WHERE id = $1", id) + .execute(&mut *tx) + .await + .map_err(internal_error)?; + + tx.commit() + .await + .map_err(internal_error)?; + + info!("Successfully deleted pipeline node: {id}"); + Ok(StatusCode::NO_CONTENT) +} + pub async fn create( DatabaseConnection(mut conn): DatabaseConnection, Json(payload): Json, @@ -163,3 +220,59 @@ pub async fn update( outputs: vec![], })) } + +#[cfg(test)] +mod tests { + use super::*; + use sqlx::PgPool; + + #[tokio::test] + async fn test_delete_pipeline_node_success() { + // Replace with your test database URL + let pool = PgPool::connect("postgres://user:pass@localhost/test_db") + .await + .unwrap(); + let mut conn = DatabaseConnection(pool.acquire().await.unwrap()); + + // Insert test data + let id = Uuid::new_v4(); + sqlx::query!( + "INSERT INTO pipeline_nodes (id, pipeline_id, node_id, node_version, coords) VALUES ($1, $2, $3, $4, $5)", + id, + Uuid::new_v4, + Uuid::new_v4, + "1.0", + serde_json::json!({"x": 0, "y": 0}) + ) + .execute(&mut *conn.0) + .await + .unwrap(); + + // Delete the node + let result = delete(conn, Path(id)).await; + assert!(matches!(result, Ok(StatusCode::NO_CONTENT))); + + // Verify deletion + let exists = sqlx::query!("SELECT id FROM pipeline_nodes WHERE id = $1", id) + .fetch_optional(&mut *pool.acquire().await.unwrap()) + .await + .unwrap() + .is_none(); + assert!(exists); + } + + #[tokio::test] + async fn test_delete_pipeline_node_not_found() { + let pool = PgPool::connect("postgres://user:pass@localhost/test_db") + .await + .unwrap(); + let conn = DatabaseConnection(pool.acquire().await.unwrap()); + + let id = Uuid::new_v4(); + let result = delete(conn, Path(id)).await; + assert!(matches!( + result, + Err((StatusCode::NOT_FOUND, _)) + )); + } +}