Returning mapped connection model schemas only (#32)

* Returning mapped connection model schemas only

* Fixing tests

api/src/endpoints/connection_model_schema.rs

@@ -1,23 +1,32 @@
-use super::{create, delete, read, update, ApiResult, HookExt, PublicExt, RequestExt};
+use super::{
+    create, delete, read, update, ApiError, ApiResult, HookExt, PublicExt, ReadResponse, RequestExt,
+};
 use crate::{
+    api_payloads::ErrorResponse,
     internal_server_error,
     server::{AppState, AppStores},
+    util::shape_mongo_filter,
 };
 use axum::{
-    extract::{Path, State},
+    extract::{Path, Query, State},
     routing::{patch, post},
-    Json, Router,
+    Extension, Json, Router,
 };
+use futures::try_join;
+use http::StatusCode;
 use integrationos_domain::{
     algebra::MongoStore,
-    connection_model_schema::{ConnectionModelSchema, Mappings, SchemaPaths},
+    connection_model_schema::{
+        ConnectionModelSchema, Mappings, PublicConnectionModelSchema, SchemaPaths,
+    },
+    event_access::EventAccess,
     id::{prefix::IdPrefix, Id},
     json_schema::JsonSchema,
 };
 use mongodb::bson::doc;
-use serde::{Deserialize, Serialize};
+use serde::{de::DeserializeOwned, Deserialize, Serialize};
 use serde_json::Value;
-use std::sync::Arc;
+use std::{collections::BTreeMap, sync::Arc};
 use tracing::error;
 
 pub fn get_router() -> Router<Arc<AppState>> {
@@ -34,6 +43,79 @@ pub fn get_router() -> Router<Arc<AppState>> {
         )
 }
 
+#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
+#[cfg_attr(feature = "dummy", derive(fake::Dummy))]
+#[serde(rename_all = "camelCase")]
+pub struct PublicGetConnectionModelSchema;
+
+pub async fn public_get_connection_model_schema<T, U>(
+    event_access: Option<Extension<Arc<EventAccess>>>,
+    query: Option<Query<BTreeMap<String, String>>>,
+    State(state): State<Arc<AppState>>,
+) -> Result<Json<ReadResponse<U>>, ApiError>
+where
+    T: RequestExt<Output = U>,
+    U: Serialize + DeserializeOwned + Unpin + Sync + Send + 'static,
+{
+    match query.as_ref().and_then(|q| q.get("connectionDefinitionId")) {
+        Some(id) => id.to_string(),
+        None => {
+            return Err((
+                StatusCode::BAD_REQUEST,
+                Json(ErrorResponse {
+                    error: "connectionDefinitionId is required".to_string(),
+                }),
+            ))
+        }
+    };
+
+    let mut query = shape_mongo_filter(
+        query,
+        event_access.map(|e| {
+            let Extension(e) = e;
+            e
+        }),
+        None,
+    );
+
+    query.filter.remove("ownership.buildableId");
+    query.filter.remove("environment");
+    query.filter.insert("mapping", doc! { "$ne": null });
+
+    let store = T::get_store(state.app_stores.clone());
+    let count = store.count(query.filter.clone(), None);
+    let find = store.get_many(
+        Some(query.filter),
+        None,
+        None,
+        Some(query.limit),
+        Some(query.skip),
+    );
+
+    let res = match try_join!(count, find) {
+        Ok((total, rows)) => ReadResponse {
+            rows,
+            skip: query.skip,
+            limit: query.limit,
+            total,
+        },
+        Err(e) => {
+            error!("Error reading from store: {e}");
+            return Err(internal_server_error!());
+        }
+    };
+
+    Ok(Json(res))
+}
+
+impl RequestExt for PublicGetConnectionModelSchema {
+    type Output = PublicConnectionModelSchema;
+
+    fn get_store(stores: AppStores) -> MongoStore<Self::Output> {
+        stores.public_model_schema.clone()
+    }
+}
+
 pub async fn get_platform_models(
     Path(platform_name): Path<String>,
     State(state): State<Arc<AppState>>,

api/src/routes/protected.rs

@@ -1,7 +1,11 @@
 use crate::{
     endpoints::{
-        connection, connection_model_definition::test_connection_model_definition, event_access,
-        events, metrics, oauth, passthrough, pipeline, transactions, unified,
+        connection,
+        connection_model_definition::test_connection_model_definition,
+        connection_model_schema::{
+            public_get_connection_model_schema, PublicGetConnectionModelSchema,
+        },
+        event_access, events, metrics, oauth, passthrough, pipeline, transactions, unified,
     },
     middleware::{
         auth,
@@ -11,9 +15,13 @@ use crate::{
     server::AppState,
 };
 use axum::{
-    error_handling::HandleErrorLayer, middleware::from_fn_with_state, routing::post, Router,
+    error_handling::HandleErrorLayer,
+    middleware::from_fn_with_state,
+    routing::{get, post},
+    Router,
 };
 use http::HeaderName;
+use integrationos_domain::connection_model_schema::PublicConnectionModelSchema;
 use std::{iter::once, sync::Arc};
 use tower::{filter::FilterLayer, ServiceBuilder};
 use tower_governor::{governor::GovernorConfigBuilder, GovernorLayer};
@@ -31,6 +39,13 @@ pub async fn get_router(state: &Arc<AppState>) -> Router<Arc<AppState>> {
             "/connection-model-definitions/test/:id",
             post(test_connection_model_definition),
         )
+        .route(
+            "/connection-model-schema",
+            get(public_get_connection_model_schema::<
+                PublicGetConnectionModelSchema,
+                PublicConnectionModelSchema,
+            >),
+        )
         .nest("/oauth", oauth::get_router())
         .nest("/unified", unified::get_router())
         .layer(TraceLayer::new_for_http())