Add autocomplete when creating connections and Kafka topic autocomplete

arroyo-api/Cargo.toml

@@ -42,8 +42,8 @@ tower-http = {version = "0.4", features = ["trace", "fs", "cors", "validate-requ
 axum = {version = "0.6.12", features = ["headers", "tokio", "macros"]}
 axum-extra = "0.7.4"
 thiserror = "1.0.40"
-utoipa = "3"
-utoipa-swagger-ui = { version = "3", features = ["axum"] }
+utoipa = "4"
+utoipa-swagger-ui = { version = "4", features = ["axum"] }
 
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"

arroyo-api/src/connection_profiles.rs

@@ -1,10 +1,11 @@
 use axum::extract::{Path, State};
 use axum::Json;
 use axum_extra::extract::WithRejection;
+use std::collections::BTreeMap;
 
 use arroyo_connectors::connector_for_type;
 use arroyo_rpc::api_types::connections::{
-    ConnectionProfile, ConnectionProfilePost, TestSourceMessage,
+    ConnectionAutocompleteResp, ConnectionProfile, ConnectionProfilePost, TestSourceMessage,
 };
 use arroyo_rpc::api_types::ConnectionProfileCollection;
 use cornucopia_async::GenericClient;
@@ -181,6 +182,47 @@ pub(crate) async fn delete_connection_profile(
     Ok(())
 }
 
+/// Get autocomplete suggestions for a connection profile
+#[utoipa::path(
+    get,
+    path = "/v1/connection_profiles/{id}/autocomplete",
+    tag = "connection_profiles",
+    params(
+       ("id" = String, Path, description = "Connection Profile id")
+    ),
+    responses(
+       (status = 200, description = "Autocomplete suggestions for connection profile"),
+    ),
+)]
+pub(crate) async fn get_connection_profile_autocomplete(
+    State(state): State<AppState>,
+    bearer_auth: BearerAuth,
+    Path(pub_id): Path<String>,
+) -> Result<Json<ConnectionAutocompleteResp>, ErrorResp> {
+    let client = client(&state.pool).await?;
+    let auth_data = authenticate(&state.pool, bearer_auth).await?;
+
+    let connection_profile = api_queries::get_connection_profile_by_pub_id()
+        .bind(&client, &auth_data.organization_id, &pub_id)
+        .opt()
+        .await
+        .map_err(log_and_map)?
+        .ok_or_else(|| not_found("Connection profile"))?;
+
+    let connector = connector_for_type(&connection_profile.r#type).unwrap();
+
+    let result = connector
+        .get_autocomplete(&connection_profile.config)
+        .unwrap()
+        .await
+        .map_err(log_and_map)?
+        .map_err(|e| bad_request(format!("Failed to get autocomplete suggestions: {}", e)))?;
+
+    Ok(Json(ConnectionAutocompleteResp {
+        values: BTreeMap::from_iter(result.into_iter()),
+    }))
+}
+
 pub(crate) async fn get_all_connection_profiles<C: GenericClient>(
     auth: &AuthData,
     client: &C,

arroyo-api/src/lib.rs

@@ -6,7 +6,8 @@ use utoipa::OpenApi;
 
 use crate::connection_profiles::{
     __path_create_connection_profile, __path_delete_connection_profile,
-    __path_get_connection_profiles, __path_test_connection_profile,
+    __path_get_connection_profile_autocomplete, __path_get_connection_profiles,
+    __path_test_connection_profile,
 };
 use crate::connection_tables::{
     __path_create_connection_table, __path_delete_connection_table, __path_get_connection_tables,
@@ -148,6 +149,7 @@ pub(crate) fn to_micros(dt: OffsetDateTime) -> u64 {
         get_connection_profiles,
         test_connection_profile,
         delete_connection_profile,
+        get_connection_profile_autocomplete,
         get_connection_tables,
         create_connection_table,
         create_connection_profile,
@@ -186,6 +188,7 @@ pub(crate) fn to_micros(dt: OffsetDateTime) -> u64 {
         Connector,
         ConnectionProfile,
         ConnectionProfilePost,
+        ConnectionAutocompleteResp,
         ConnectionProfileCollection,
         ConnectionTable,
         ConnectionTablePost,

arroyo-api/src/rest.rs

@@ -18,8 +18,8 @@ use utoipa::OpenApi;
 use utoipa_swagger_ui::SwaggerUi;
 
 use crate::connection_profiles::{
-    create_connection_profile, delete_connection_profile, get_connection_profiles,
-    test_connection_profile,
+    create_connection_profile, delete_connection_profile, get_connection_profile_autocomplete,
+    get_connection_profiles, test_connection_profile,
 };
 use crate::connection_tables::{
     create_connection_table, delete_connection_table, get_connection_tables, test_connection_table,
@@ -117,6 +117,10 @@ pub fn create_rest_app(pool: Pool, controller_addr: &str) -> Router {
             "/connection_profiles/:id",
             delete(delete_connection_profile),
         )
+        .route(
+            "/connection_profiles/:id/autocomplete",
+            get(get_connection_profile_autocomplete),
+        )
         .route("/connection_tables", get(get_connection_tables))
         .route("/connection_tables", post(create_connection_table))
         .route("/connection_tables/test", post(test_connection_table))

arroyo-connectors/src/confluent.rs

@@ -123,6 +123,14 @@ impl Connector for ConfluentConnector {
         }
     }
 
+    fn get_autocomplete(
+        &self,
+        profile: Self::ProfileT,
+    ) -> Receiver<anyhow::Result<HashMap<String, Vec<String>>>> {
+        let profile = profile.into();
+        KafkaConnector {}.get_autocomplete(profile)
+    }
+
     fn test(
         &self,
         _: &str,