feat(etl-api, etl-validation): connection validation endpoints

Cargo.toml

@@ -10,6 +10,7 @@ members = [
     "etl-postgres",
     "etl-replicator",
     "etl-telemetry",
+    "etl-validation",
 ]
 
 [workspace.package]
@@ -27,6 +28,7 @@ etl-destinations = { path = "etl-destinations", default-features = false }
 etl-postgres = { path = "etl-postgres", default-features = false }
 etl-replicator = { path = "etl-replicator", default-features = false }
 etl-telemetry = { path = "etl-telemetry", default-features = false }
+etl-validation = { path = "etl-validation", default-features = false }
 
 actix-web = { version = "4.11.0", default-features = false }
 actix-web-httpauth = { version = "0.8.2", default-features = false }

etl-api/Cargo.toml

@@ -18,6 +18,7 @@ name = "etl-api"
 etl-config = { workspace = true, features = ["utoipa"] }
 etl-postgres = { workspace = true, features = ["replication"] }
 etl-telemetry = { workspace = true }
+etl-validation = { workspace = true }
 
 actix-web = { workspace = true, features = ["macros", "http2"] }
 actix-web-httpauth = { workspace = true }

etl-api/src/configs/destination.rs

@@ -171,6 +171,13 @@ impl StoredDestinationConfig {
     }
 }
 
+impl From<FullApiDestinationConfig> for DestinationConfig {
+    fn from(value: FullApiDestinationConfig) -> Self {
+        let stored_config: StoredDestinationConfig = value.into();
+        stored_config.into_etl_config()
+    }
+}
+
 impl From<FullApiDestinationConfig> for StoredDestinationConfig {
     fn from(value: FullApiDestinationConfig) -> Self {
         match value {

etl-api/src/routes/destinations.rs

@@ -1,20 +1,21 @@
+use crate::configs::destination::FullApiDestinationConfig;
+use crate::configs::encryption::EncryptionKey;
+use crate::db;
+use crate::db::destinations::DestinationsDbError;
+use crate::routes::{ErrorMessage, TenantIdError, TestConnectionResponse, extract_tenant_id};
 use actix_web::{
     HttpRequest, HttpResponse, Responder, ResponseError, delete, get,
     http::{StatusCode, header::ContentType},
     post,
     web::{Data, Json, Path},
 };
+use etl_config::shared::DestinationConfig;
+use etl_validation::DestinationValidator;
 use serde::{Deserialize, Serialize};
 use sqlx::PgPool;
 use thiserror::Error;
 use utoipa::ToSchema;
 
-use crate::configs::destination::FullApiDestinationConfig;
-use crate::configs::encryption::EncryptionKey;
-use crate::db;
-use crate::db::destinations::DestinationsDbError;
-use crate::routes::{ErrorMessage, TenantIdError, extract_tenant_id};
-
 #[derive(Debug, Error)]
 pub enum DestinationError {
     #[error("The destination with id {0} was not found")]
@@ -283,3 +284,30 @@ pub async fn read_all_destinations(
 
     Ok(Json(response))
 }
+
+#[utoipa::path(
+    summary = "Test a destination connection (pre-creation)",
+    description = "Tests a destination connection without creating a resource. Validates connectivity, permissions, and configuration for BigQuery or Iceberg destinations. Useful for validating credentials before creating a destination.",
+    request_body = FullApiDestinationConfig,
+    responses(
+        (status = 200, description = "Connection test successful", body = TestConnectionResponse),
+        (status = 400, description = "Bad request or connection test failed", body = ErrorMessage),
+        (status = 500, description = "Internal server error", body = ErrorMessage),
+    ),
+    tag = "Destinations"
+)]
+#[post("/destinations/test-connection")]
+pub async fn test_destination_connection(
+    config: Json<FullApiDestinationConfig>,
+) -> Result<impl Responder, DestinationError> {
+    let destination_config: DestinationConfig = config.into_inner().into();
+
+    destination_config.validate().await.map_err(|e| {
+        DestinationsDbError::Database(sqlx::Error::Configuration(Box::new(std::io::Error::new(
+            std::io::ErrorKind::Other,
+            e,
+        ))))
+    })?;
+
+    Ok(Json(TestConnectionResponse { valid: true }))
+}

etl-api/src/routes/mod.rs

@@ -21,6 +21,12 @@ const MIN_POOL_CONNECTIONS: u32 = 1;
 /// Maximum number of connections for the source Postgres connection pool.
 const MAX_POOL_CONNECTIONS: u32 = 1;
 
+#[derive(Debug, Serialize, Deserialize, ToSchema)]
+pub struct TestConnectionResponse {
+    #[schema(example = true)]
+    pub valid: bool,
+}
+
 #[derive(Debug, Serialize, Deserialize, ToSchema)]
 pub struct ErrorMessage {
     #[schema(example = "an error occurred in the api")]

etl-api/src/routes/sources.rs

@@ -1,14 +1,15 @@
 use crate::configs::encryption::EncryptionKey;
-use crate::configs::source::{FullApiSourceConfig, StrippedApiSourceConfig};
+use crate::configs::source::{FullApiSourceConfig, StoredSourceConfig, StrippedApiSourceConfig};
 use crate::db;
 use crate::db::sources::SourcesDbError;
-use crate::routes::{ErrorMessage, TenantIdError, extract_tenant_id};
+use crate::routes::{ErrorMessage, TenantIdError, TestConnectionResponse, extract_tenant_id};
 use actix_web::{
     HttpRequest, HttpResponse, Responder, ResponseError, delete, get,
     http::{StatusCode, header::ContentType},
     post,
     web::{Data, Json, Path},
 };
+use etl_validation::SourceValidator;
 use serde::{Deserialize, Serialize};
 use sqlx::PgPool;
 use thiserror::Error;
@@ -282,3 +283,30 @@ pub async fn read_all_sources(
 
     Ok(Json(response))
 }
+
+#[utoipa::path(
+    summary = "Test a source connection (pre-creation)",
+    description = "Tests a PostgreSQL source connection without creating a resource. Validates connectivity, version, WAL level, permissions, and replication slot availability. Useful for validating credentials before creating a source.",
+    request_body = FullApiSourceConfig,
+    responses(
+        (status = 200, description = "Connection test successful", body = TestConnectionResponse),
+        (status = 400, description = "Bad request or connection test failed", body = ErrorMessage),
+        (status = 500, description = "Internal server error", body = ErrorMessage),
+    ),
+    tag = "Sources"
+)]
+#[post("/sources/test-connection")]
+pub async fn test_source_connection(
+    config: Json<FullApiSourceConfig>,
+) -> Result<impl Responder, SourceError> {
+    let config = config.into_inner();
+    let stored_config: StoredSourceConfig = config.into();
+    let pg_config = stored_config.into_connection_config();
+
+    pg_config
+        .validate()
+        .await
+        .map_err(|e| SourcesDbError::Database(sqlx::Error::Configuration(Box::new(e))))?;
+
+    Ok(Json(TestConnectionResponse { valid: true }))
+}

etl-api/src/startup.rs

@@ -24,10 +24,12 @@ use crate::{
     configs::encryption,
     db::publications::Publication,
     routes::{
+        TestConnectionResponse,
         destinations::{
             CreateDestinationRequest, CreateDestinationResponse, ReadDestinationResponse,
             ReadDestinationsResponse, UpdateDestinationRequest, create_destination,
-            delete_destination, read_all_destinations, read_destination, update_destination,
+            delete_destination, read_all_destinations, read_destination,
+            test_destination_connection, update_destination,
         },
         destinations_pipelines::{
             CreateDestinationPipelineRequest, CreateDestinationPipelineResponse,
@@ -60,7 +62,7 @@ use crate::{
             },
             read_all_sources, read_source,
             tables::read_table_names,
-            update_source,
+            test_source_connection, update_source,
         },
         tenants::{
             CreateOrUpdateTenantRequest, CreateOrUpdateTenantResponse, CreateTenantRequest,
@@ -242,6 +244,7 @@ pub async fn run(
             ReadTenantsResponse,
             CreateSourceRequest,
             CreateSourceResponse,
+            TestConnectionResponse,
             UpdateSourceRequest,
             ReadSourceResponse,
             ReadSourcesResponse,
@@ -293,6 +296,7 @@ pub async fn run(
         crate::routes::sources::update_source,
         crate::routes::sources::delete_source,
         crate::routes::sources::read_all_sources,
+        crate::routes::sources::test_source_connection,
         crate::routes::sources::publications::create_publication,
         crate::routes::sources::publications::read_publication,
         crate::routes::sources::publications::update_publication,
@@ -304,6 +308,7 @@ pub async fn run(
         crate::routes::destinations::update_destination,
         crate::routes::destinations::delete_destination,
         crate::routes::destinations::read_all_destinations,
+        crate::routes::destinations::test_destination_connection,
         crate::routes::tenants_sources::create_tenant_and_source,
         crate::routes::destinations_pipelines::create_destination_and_pipeline,
         crate::routes::destinations_pipelines::update_destination_and_pipeline,
@@ -343,16 +348,18 @@ pub async fn run(
                     .service(read_all_tenants)
                     //sources
                     .service(create_source)
+                    .service(read_all_sources)
+                    .service(test_source_connection)
                     .service(read_source)
                     .service(update_source)
                     .service(delete_source)
-                    .service(read_all_sources)
                     //destinations
                     .service(create_destination)
+                    .service(read_all_destinations)
+                    .service(test_destination_connection)
                     .service(read_destination)
                     .service(update_destination)
                     .service(delete_destination)
-                    .service(read_all_destinations)
                     //pipelines
                     .service(create_pipeline)
                     .service(read_pipeline)

etl-api/tests/support/test_app.rs

@@ -248,6 +248,28 @@ impl TestApp {
             .expect("failed to execute request")
     }
 
+    pub async fn test_source_connection(
+        &self,
+        config: &etl_api::configs::source::FullApiSourceConfig,
+    ) -> reqwest::Response {
+        self.post_authenticated(format!("{}/v1/sources/test-connection", &self.address))
+            .json(config)
+            .send()
+            .await
+            .expect("Failed to execute request.")
+    }
+
+    pub async fn test_destination_connection(
+        &self,
+        config: &etl_api::configs::destination::FullApiDestinationConfig,
+    ) -> reqwest::Response {
+        self.post_authenticated(format!("{}/v1/destinations/test-connection", &self.address))
+            .json(config)
+            .send()
+            .await
+            .expect("Failed to execute request.")
+    }
+
     pub async fn create_pipeline(
         &self,
         tenant_id: &str,
@@ -501,12 +523,21 @@ impl Drop for TestApp {
         // First, abort the server task to ensure it's terminated.
         self.server_handle.abort();
 
+        // Clone the config to avoid move issues
+        let db_config = self.config.database.clone();
+
         // To use `block_in_place,` we need a multithreaded runtime since when a blocking
         // task is issued, the runtime will offload existing tasks to another worker.
-        tokio::task::block_in_place(move || {
-            Handle::current()
-                .block_on(async move { drop_pg_database(&self.config.database).await });
-        });
+        // Wrap in catch_unwind to ensure panics during cleanup don't propagate
+        let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+            tokio::task::block_in_place(|| {
+                Handle::current().block_on(async {
+                    // Give server time to shut down gracefully
+                    tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
+                    drop_pg_database(&db_config).await;
+                });
+            });
+        }));
     }
 }