ref(limits): Limit pipelines to 1 (#477)

Author: iambriccardo

etl-api/src/db/pipelines.rs

@@ -18,7 +18,10 @@ use std::ops::DerefMut;
 use thiserror::Error;
 
 /// Maximum number of pipelines allowed per tenant.
-pub const MAX_PIPELINES_PER_TENANT: i64 = 3;
+///
+/// For now, we keep the maximum to 1, this way, we give us a simpler surface area for breaking changes
+/// to the `etl` schema in the source database since only one pipeline will use it.
+pub const MAX_PIPELINES_PER_TENANT: i64 = 1;
 
 pub struct Pipeline {
     pub id: i64,

etl-api/tests/destinations_pipelines.rs

@@ -3,7 +3,7 @@ use etl_api::routes::destinations_pipelines::{
     CreateDestinationPipelineRequest, CreateDestinationPipelineResponse,
     UpdateDestinationPipelineRequest,
 };
-use etl_api::routes::pipelines::{CreatePipelineRequest, ReadPipelineResponse};
+use etl_api::routes::pipelines::ReadPipelineResponse;
 use etl_telemetry::tracing::init_test_tracing;
 use reqwest::StatusCode;
 
@@ -134,15 +134,18 @@ async fn iceberg_supabase_destination_and_pipeline_can_be_created() {
 }
 
 #[tokio::test(flavor = "multi_thread")]
-async fn tenant_cannot_create_more_than_three_destinations_pipelines() {
+async fn tenant_cannot_create_more_than_max_destinations_pipelines() {
+    use etl_api::db::pipelines::MAX_PIPELINES_PER_TENANT;
+
     init_test_tracing();
     // Arrange
     let app = spawn_test_app().await;
     let tenant_id = &create_tenant(&app).await;
     let source_id = create_source(&app, tenant_id).await;
     create_default_image(&app).await;
 
-    for idx in 0..3 {
+    // Create the maximum allowed pipelines
+    for idx in 0..MAX_PIPELINES_PER_TENANT {
         let destination_pipeline = CreateDestinationPipelineRequest {
             destination_name: format!("BigQuery Destination {idx}"),
             destination_config: new_bigquery_destination_config(),
@@ -155,8 +158,9 @@ async fn tenant_cannot_create_more_than_three_destinations_pipelines() {
         assert!(response.status().is_success());
     }
 
+    // Attempt to create one more pipeline should fail
     let destination_pipeline = CreateDestinationPipelineRequest {
-        destination_name: "BigQuery Destination 3".to_string(),
+        destination_name: format!("BigQuery Destination {MAX_PIPELINES_PER_TENANT}"),
         destination_config: new_bigquery_destination_config(),
         source_id,
         pipeline_config: new_pipeline_config(),
@@ -531,43 +535,44 @@ async fn destination_and_pipeline_with_another_tenants_pipeline_cannot_be_update
     assert_eq!(response.status(), StatusCode::BAD_REQUEST);
 }
 
-#[tokio::test(flavor = "multi_thread")]
-async fn duplicate_destination_pipeline_with_same_source_cannot_be_created() {
-    init_test_tracing();
-    // Arrange
-    let app = spawn_test_app().await;
-    create_default_image(&app).await;
-    let tenant_id = &create_tenant(&app).await;
-    let source_id = create_source(&app, tenant_id).await;
-
-    // Create first destination and pipeline
-    let destination_pipeline = CreateDestinationPipelineRequest {
-        destination_name: new_name(),
-        destination_config: new_bigquery_destination_config(),
-        source_id,
-        pipeline_config: new_pipeline_config(),
-    };
-    let response = app
-        .create_destination_pipeline(tenant_id, &destination_pipeline)
-        .await;
-    assert!(response.status().is_success());
-    let response: CreateDestinationPipelineResponse = response
-        .json()
-        .await
-        .expect("failed to deserialize response");
-    let first_destination_id = response.destination_id;
-
-    // Act - Try to create another pipeline with same source and the first destination
-    let pipeline_request = CreatePipelineRequest {
-        source_id,
-        destination_id: first_destination_id,
-        config: updated_pipeline_config(),
-    };
-    let response = app.create_pipeline(tenant_id, &pipeline_request).await;
-
-    // Assert
-    assert_eq!(response.status(), StatusCode::CONFLICT);
-}
+// TODO: Re-enable these tests once MAX_PIPELINES_PER_TENANT is lifted from 1.
+// #[tokio::test(flavor = "multi_thread")]
+// async fn duplicate_destination_pipeline_with_same_source_cannot_be_created() {
+//     init_test_tracing();
+//     // Arrange
+//     let app = spawn_test_app().await;
+//     create_default_image(&app).await;
+//     let tenant_id = &create_tenant(&app).await;
+//     let source_id = create_source(&app, tenant_id).await;
+//
+//     // Create first destination and pipeline
+//     let destination_pipeline = CreateDestinationPipelineRequest {
+//         destination_name: new_name(),
+//         destination_config: new_bigquery_destination_config(),
+//         source_id,
+//         pipeline_config: new_pipeline_config(),
+//     };
+//     let response = app
+//         .create_destination_pipeline(tenant_id, &destination_pipeline)
+//         .await;
+//     assert!(response.status().is_success());
+//     let response: CreateDestinationPipelineResponse = response
+//         .json()
+//         .await
+//         .expect("failed to deserialize response");
+//     let first_destination_id = response.destination_id;
+//
+//     // Act - Try to create another pipeline with same source and the first destination
+//     let pipeline_request = CreatePipelineRequest {
+//         source_id,
+//         destination_id: first_destination_id,
+//         config: updated_pipeline_config(),
+//     };
+//     let response = app.create_pipeline(tenant_id, &pipeline_request).await;
+//
+//     // Assert
+//     assert_eq!(response.status(), StatusCode::CONFLICT);
+// }
 
 #[tokio::test(flavor = "multi_thread")]
 async fn destination_and_pipeline_can_be_deleted() {

etl-api/tests/pipelines.rs

@@ -223,14 +223,17 @@ async fn pipeline_can_be_created() {
 }
 
 #[tokio::test(flavor = "multi_thread")]
-async fn tenant_cannot_create_more_than_three_pipelines() {
+async fn tenant_cannot_create_more_than_max_pipelines() {
+    use etl_api::db::pipelines::MAX_PIPELINES_PER_TENANT;
+
     init_test_tracing();
     // Arrange
     let app = spawn_test_app().await;
     create_default_image(&app).await;
     let tenant_id = &create_tenant(&app).await;
 
-    for _ in 0..3 {
+    // Create the maximum allowed pipelines
+    for _ in 0..MAX_PIPELINES_PER_TENANT {
         let source_id = create_source(&app, tenant_id).await;
         let destination_id = create_destination(&app, tenant_id).await;
         let pipeline = CreatePipelineRequest {
@@ -242,6 +245,7 @@ async fn tenant_cannot_create_more_than_three_pipelines() {
         assert!(response.status().is_success());
     }
 
+    // Attempt to create one more pipeline should fail
     let source_id = create_source(&app, tenant_id).await;
     let destination_id = create_destination(&app, tenant_id).await;
     let pipeline = CreatePipelineRequest {
@@ -575,27 +579,17 @@ async fn all_pipelines_can_be_read() {
     let app = spawn_test_app().await;
     create_default_image(&app).await;
     let tenant_id = &create_tenant(&app).await;
-    let source1_id = create_source(&app, tenant_id).await;
-    let source2_id = create_source(&app, tenant_id).await;
-    let destination1_id = create_destination(&app, tenant_id).await;
-    let destination2_id = create_destination(&app, tenant_id).await;
+    let source_id = create_source(&app, tenant_id).await;
+    let destination_id = create_destination(&app, tenant_id).await;
 
-    let pipeline1_id = create_pipeline_with_config(
+    let pipeline_id = create_pipeline_with_config(
         &app,
         tenant_id,
-        source1_id,
-        destination1_id,
+        source_id,
+        destination_id,
         new_pipeline_config(),
     )
     .await;
-    let pipeline2_id = create_pipeline_with_config(
-        &app,
-        tenant_id,
-        source2_id,
-        destination2_id,
-        updated_pipeline_config(),
-    )
-    .await;
 
     // Act
     let response = app.read_all_pipelines(tenant_id).await;
@@ -606,98 +600,95 @@ async fn all_pipelines_can_be_read() {
         .json()
         .await
         .expect("failed to deserialize response");
-    for pipeline in response.pipelines {
-        if pipeline.id == pipeline1_id {
-            assert_eq!(&pipeline.tenant_id, tenant_id);
-            assert_eq!(pipeline.source_id, source1_id);
-            assert_eq!(pipeline.destination_id, destination1_id);
-            insta::assert_debug_snapshot!(pipeline.config);
-        } else if pipeline.id == pipeline2_id {
-            assert_eq!(&pipeline.tenant_id, tenant_id);
-            assert_eq!(pipeline.source_id, source2_id);
-            assert_eq!(pipeline.destination_id, destination2_id);
-            insta::assert_debug_snapshot!(pipeline.config);
-        }
-    }
+    assert_eq!(response.pipelines.len(), 1);
+    let pipeline = &response.pipelines[0];
+    assert_eq!(pipeline.id, pipeline_id);
+    assert_eq!(&pipeline.tenant_id, tenant_id);
+    assert_eq!(pipeline.source_id, source_id);
+    assert_eq!(pipeline.destination_id, destination_id);
+    insta::assert_debug_snapshot!(pipeline.config);
 }
 
-#[tokio::test(flavor = "multi_thread")]
-async fn duplicate_pipeline_with_same_source_and_destination_cannot_be_created() {
-    init_test_tracing();
-    // Arrange
-    let app = spawn_test_app().await;
-    create_default_image(&app).await;
-    let tenant_id = &create_tenant(&app).await;
-    let source_id = create_source(&app, tenant_id).await;
-    let destination_id = create_destination(&app, tenant_id).await;
-
-    // Create first pipeline
-    let pipeline = CreatePipelineRequest {
-        source_id,
-        destination_id,
-        config: new_pipeline_config(),
-    };
-    let response = app.create_pipeline(tenant_id, &pipeline).await;
-    assert!(response.status().is_success());
-
-    // Act - Try to create duplicate pipeline with same source and destination
-    let duplicate_pipeline = CreatePipelineRequest {
-        source_id,
-        destination_id,
-        config: updated_pipeline_config(),
-    };
-    let response = app.create_pipeline(tenant_id, &duplicate_pipeline).await;
-
-    // Assert
-    assert_eq!(response.status(), StatusCode::CONFLICT);
-}
-
-#[tokio::test(flavor = "multi_thread")]
-async fn updating_pipeline_to_duplicate_source_destination_combination_fails() {
-    init_test_tracing();
-    // Arrange
-    let app = spawn_test_app().await;
-    create_default_image(&app).await;
-    let tenant_id = &create_tenant(&app).await;
-    let source1_id = create_source(&app, tenant_id).await;
-    let source2_id = create_source(&app, tenant_id).await;
-    let destination_id = create_destination(&app, tenant_id).await;
-
-    // Create first pipeline
-    let pipeline1 = CreatePipelineRequest {
-        source_id: source1_id,
-        destination_id,
-        config: new_pipeline_config(),
-    };
-    let response = app.create_pipeline(tenant_id, &pipeline1).await;
-    assert!(response.status().is_success());
-
-    // Create second pipeline with different source
-    let pipeline2 = CreatePipelineRequest {
-        source_id: source2_id,
-        destination_id,
-        config: new_pipeline_config(),
-    };
-    let response = app.create_pipeline(tenant_id, &pipeline2).await;
-    let response: CreatePipelineResponse = response
-        .json()
-        .await
-        .expect("failed to deserialize response");
-    let pipeline2_id = response.id;
-
-    // Act - Try to update second pipeline to have same source as first
-    let updated_config = UpdatePipelineRequest {
-        source_id: source1_id, // This would create a duplicate
-        destination_id,
-        config: updated_pipeline_config(),
-    };
-    let response = app
-        .update_pipeline(tenant_id, pipeline2_id, &updated_config)
-        .await;
-
-    // Assert
-    assert_eq!(response.status(), StatusCode::CONFLICT);
-}
+// TODO: Re-enable these tests once MAX_PIPELINES_PER_TENANT is lifted from 1.
+// These tests require multiple pipelines per tenant to function correctly.
+//
+// #[tokio::test(flavor = "multi_thread")]
+// async fn duplicate_pipeline_with_same_source_and_destination_cannot_be_created() {
+//     init_test_tracing();
+//     // Arrange
+//     let app = spawn_test_app().await;
+//     create_default_image(&app).await;
+//     let tenant_id = &create_tenant(&app).await;
+//     let source_id = create_source(&app, tenant_id).await;
+//     let destination_id = create_destination(&app, tenant_id).await;
+//
+//     // Create first pipeline
+//     let pipeline = CreatePipelineRequest {
+//         source_id,
+//         destination_id,
+//         config: new_pipeline_config(),
+//     };
+//     let response = app.create_pipeline(tenant_id, &pipeline).await;
+//     assert!(response.status().is_success());
+//
+//     // Act - Try to create duplicate pipeline with same source and destination
+//     let duplicate_pipeline = CreatePipelineRequest {
+//         source_id,
+//         destination_id,
+//         config: updated_pipeline_config(),
+//     };
+//     let response = app.create_pipeline(tenant_id, &duplicate_pipeline).await;
+//
+//     // Assert
+//     assert_eq!(response.status(), StatusCode::CONFLICT);
+// }
+//
+// #[tokio::test(flavor = "multi_thread")]
+// async fn updating_pipeline_to_duplicate_source_destination_combination_fails() {
+//     init_test_tracing();
+//     // Arrange
+//     let app = spawn_test_app().await;
+//     create_default_image(&app).await;
+//     let tenant_id = &create_tenant(&app).await;
+//     let source1_id = create_source(&app, tenant_id).await;
+//     let source2_id = create_source(&app, tenant_id).await;
+//     let destination_id = create_destination(&app, tenant_id).await;
+//
+//     // Create first pipeline
+//     let pipeline1 = CreatePipelineRequest {
+//         source_id: source1_id,
+//         destination_id,
+//         config: new_pipeline_config(),
+//     };
+//     let response = app.create_pipeline(tenant_id, &pipeline1).await;
+//     assert!(response.status().is_success());
+//
+//     // Create second pipeline with different source
+//     let pipeline2 = CreatePipelineRequest {
+//         source_id: source2_id,
+//         destination_id,
+//         config: new_pipeline_config(),
+//     };
+//     let response = app.create_pipeline(tenant_id, &pipeline2).await;
+//     let response: CreatePipelineResponse = response
+//         .json()
+//         .await
+//         .expect("failed to deserialize response");
+//     let pipeline2_id = response.id;
+//
+//     // Act - Try to update second pipeline to have same source as first
+//     let updated_config = UpdatePipelineRequest {
+//         source_id: source1_id, // This would create a duplicate
+//         destination_id,
+//         config: updated_pipeline_config(),
+//     };
+//     let response = app
+//         .update_pipeline(tenant_id, pipeline2_id, &updated_config)
+//         .await;
+//
+//     // Assert
+//     assert_eq!(response.status(), StatusCode::CONFLICT);
+// }
 
 #[tokio::test(flavor = "multi_thread")]
 async fn pipeline_version_can_be_updated() {