feat(api): Add feature flag to control maximum pipelines (#486)

Author: iambriccardo

etl-api/src/feature_flags.rs

@@ -1,3 +1,4 @@
+use actix_web::web::Data;
 use tracing::info;
 
 /// Initializes the ConfigCat client for feature flag evaluation in the API.
@@ -19,3 +20,23 @@ pub fn init_feature_flags(
         }
     }
 }
+
+/// Returns the maximum number of pipelines allowed per tenant.
+///
+/// Checks the `maximumPipelinesPerTenant` feature flag and falls back to
+/// the default if the flag is not set or the client is unavailable.
+pub async fn get_max_pipelines_per_tenant(
+    client: Option<&Data<configcat::Client>>,
+    tenant_id: &str,
+    default_value: i64,
+) -> i64 {
+    match client {
+        Some(client) => {
+            let user = configcat::User::new(tenant_id);
+            client
+                .get_value("maximumPipelinesPerTenant", default_value, Some(user))
+                .await
+        }
+        None => default_value,
+    }
+}

etl-api/src/routes/destinations_pipelines.rs

@@ -22,6 +22,7 @@ use crate::db::pipelines::{
     MAX_PIPELINES_PER_TENANT, PipelinesDbError, count_pipelines_for_tenant, read_pipeline,
 };
 use crate::db::sources::{SourcesDbError, source_exists};
+use crate::feature_flags::get_max_pipelines_per_tenant;
 
 #[derive(Debug, Error)]
 enum DestinationPipelineError {
@@ -192,6 +193,7 @@ pub async fn create_destination_and_pipeline(
     pool: Data<PgPool>,
     destination_and_pipeline: Json<CreateDestinationPipelineRequest>,
     encryption_key: Data<EncryptionKey>,
+    feature_flags_client: Option<Data<configcat::Client>>,
 ) -> Result<impl Responder, DestinationPipelineError> {
     let tenant_id = extract_tenant_id(&req)?;
     let destination_and_pipeline = destination_and_pipeline.into_inner();
@@ -210,10 +212,16 @@ pub async fn create_destination_and_pipeline(
         ));
     }
 
+    let max_pipelines = get_max_pipelines_per_tenant(
+        feature_flags_client.as_ref(),
+        tenant_id,
+        MAX_PIPELINES_PER_TENANT,
+    )
+    .await;
     let pipeline_count = count_pipelines_for_tenant(txn.deref_mut(), tenant_id).await?;
-    if pipeline_count >= MAX_PIPELINES_PER_TENANT {
+    if pipeline_count >= max_pipelines {
         return Err(DestinationPipelineError::PipelineLimitReached {
-            limit: MAX_PIPELINES_PER_TENANT,
+            limit: max_pipelines,
         });
     }
 

etl-api/src/routes/pipelines.rs

@@ -22,6 +22,7 @@ use crate::db::images::ImagesDbError;
 use crate::db::pipelines::{MAX_PIPELINES_PER_TENANT, PipelinesDbError, read_pipeline_components};
 use crate::db::replicators::ReplicatorsDbError;
 use crate::db::sources::{SourcesDbError, source_exists};
+use crate::feature_flags::get_max_pipelines_per_tenant;
 use crate::k8s::core::{
     create_k8s_object_prefix, create_or_update_pipeline_resources_in_k8s,
     delete_pipeline_resources_in_k8s,
@@ -463,6 +464,7 @@ pub async fn create_pipeline(
     req: HttpRequest,
     pool: Data<PgPool>,
     pipeline: Json<CreatePipelineRequest>,
+    feature_flags_client: Option<Data<configcat::Client>>,
 ) -> Result<impl Responder, PipelineError> {
     let tenant_id = extract_tenant_id(&req)?;
     let pipeline = pipeline.into_inner();
@@ -476,11 +478,17 @@ pub async fn create_pipeline(
         return Err(PipelineError::DestinationNotFound(pipeline.destination_id));
     }
 
+    let max_pipelines = get_max_pipelines_per_tenant(
+        feature_flags_client.as_ref(),
+        tenant_id,
+        MAX_PIPELINES_PER_TENANT,
+    )
+    .await;
     let pipeline_count =
         db::pipelines::count_pipelines_for_tenant(txn.deref_mut(), tenant_id).await?;
-    if pipeline_count >= MAX_PIPELINES_PER_TENANT {
+    if pipeline_count >= max_pipelines {
         return Err(PipelineError::PipelineLimitReached {
-            limit: MAX_PIPELINES_PER_TENANT,
+            limit: max_pipelines,
         });
     }