ref: major refactoring

Author: raphael-goetz

Cargo.lock

@@ -15,6 +15,8 @@ base = { workspace = true }
 anyhow = { workspace = true }
 base64 = "0.22.1"
 ring = "0.17.14"
+form_urlencoded = "1.2.1"
+percent-encoding = "2.3.1"
 hyper-util = "0.1.19"
 hyper = "1.8.1"
 http-body-util = "0.1.3"

adapter/rest/Cargo.toml

@@ -1,32 +1,20 @@
 use base::{
     runner::{ServerContext, ServerRunner},
-    store::{FlowExecutionResult, FlowIdentifyResult},
     traits::Server as ServerTrait,
 };
 use code0_flow::flow_service::ModuleDefinitionAppendix;
-use http_body_util::{BodyExt, Full};
 use hyper::server::conn::http1;
-use hyper::{Request, Response};
-use hyper::{
-    StatusCode,
-    body::{Bytes, Incoming},
-};
 use hyper_util::rt::TokioIo;
-use std::convert::Infallible;
 use std::net::SocketAddr;
 use std::sync::Arc;
 use tokio::net::TcpListener;
 use tonic::async_trait;
-use tucana::shared::{
-    Endpoint, ModuleDefinition, Struct, ValidationFlow, Value, helper::value::ToValue, value::Kind,
-};
-
-use crate::auth::{authenticate_header_name, validate_flow_auth};
-use crate::response::{error_to_http_response, value_to_http_response};
+use tucana::shared::{Endpoint, ModuleDefinition};
 
 mod auth;
 mod config;
 mod content_type;
+mod request;
 mod response;
 mod route;
 
@@ -69,129 +57,6 @@ struct HttpServer {
     addr: Option<SocketAddr>,
 }
 
-async fn execute_flow_to_hyper_response(
-    flow: ValidationFlow,
-    body: Value,
-    store: Arc<base::store::AdapterStore>,
-) -> Response<Full<Bytes>> {
-    match store.execute_flow_with_emitter(flow, Some(body)).await {
-        FlowExecutionResult::Ongoing(result) => {
-            log::debug!("Received first ongoing response from emitter");
-            value_to_http_response(result)
-        }
-        FlowExecutionResult::Failed => {
-            log::error!("Flow execution failed event received from emitter");
-            error_to_http_response(StatusCode::INTERNAL_SERVER_ERROR, "Internal server error")
-        }
-        FlowExecutionResult::FinishedWithoutOngoing => Response::builder()
-            .status(StatusCode::NO_CONTENT)
-            .body(Full::new(Bytes::new()))
-            .unwrap(),
-        FlowExecutionResult::TransportError => {
-            log::error!("Flow execution transport error");
-            error_to_http_response(StatusCode::INTERNAL_SERVER_ERROR, "Internal server error")
-        }
-    }
-}
-
-pub async fn handle_request(
-    req: Request<Incoming>,
-    store: Arc<base::store::AdapterStore>,
-) -> Result<Response<Full<Bytes>>, Infallible> {
-    let method = req.method().clone();
-    let path = req.uri().path().to_string();
-    let headers = req.headers().clone();
-
-    // Read full body
-    let body_bytes = match BodyExt::collect(req.into_body()).await {
-        Ok(collected) => collected.to_bytes().to_vec(),
-        Err(err) => {
-            log::error!("Failed to read request body: {}", err);
-            return Ok(error_to_http_response(
-                StatusCode::BAD_REQUEST,
-                "Failed to read request body",
-            ));
-        }
-    };
-
-    // slug matching
-    let Some(slug) = route::extract_slug_from_path(&path) else {
-        return Ok(error_to_http_response(
-            StatusCode::BAD_REQUEST,
-            "Missing slug in path",
-        ));
-    };
-
-    let pattern = format!("REST.{}.*", slug);
-    let route = route::RequestRoute {
-        url: path.clone(),
-        method,
-    };
-
-    let resp = match store.get_possible_flow_match(pattern, route).await {
-        FlowIdentifyResult::Single(flow) => {
-            if let Err(err) = validate_flow_auth(&flow, &headers) {
-                let mut response = error_to_http_response(err.status_code(), err.message());
-                response
-                    .headers_mut()
-                    .insert(authenticate_header_name(), err.challenge());
-                return Ok(response);
-            }
-
-            let request_body_value =
-                match content_type::parse_body_from_headers(&headers, &body_bytes) {
-                    Ok(value) => value,
-                    Err(err) => {
-                        log::warn!("Failed to parse request body: {}", err);
-                        let status_code = match err {
-                            content_type::BodyParseError::UnsupportedContentType { .. } => {
-                                StatusCode::UNSUPPORTED_MEDIA_TYPE
-                            }
-                            _ => StatusCode::BAD_REQUEST,
-                        };
-
-                        return Ok(error_to_http_response(status_code, &err.to_string()));
-                    }
-                };
-
-            let mut header_fields = std::collections::HashMap::new();
-            let mut fields = std::collections::HashMap::new();
-
-            for (name, value) in headers.iter() {
-                let key = name.as_str().to_owned();
-                let value_str = value
-                    .to_str()
-                    .map(str::to_owned)
-                    .unwrap_or_else(|_| String::from_utf8_lossy(value.as_bytes()).into_owned());
-
-                header_fields.insert(key, value_str.to_value());
-            }
-
-            if let Some(v) = request_body_value {
-                fields.insert(String::from("payload"), v);
-            };
-
-            fields.insert(
-                String::from("headers"),
-                Value {
-                    kind: Some(Kind::StructValue(Struct {
-                        fields: header_fields,
-                    })),
-                },
-            );
-
-            let input = Value {
-                kind: Some(Kind::StructValue(Struct { fields })),
-            };
-
-            execute_flow_to_hyper_response(flow, input, store).await
-        }
-        _ => error_to_http_response(StatusCode::NOT_FOUND, "No flow found for path"),
-    };
-
-    Ok(resp)
-}
-
 #[async_trait]
 impl ServerTrait<config::HttpServerConfig> for HttpServer {
     async fn init(&mut self, ctx: &ServerContext<config::HttpServerConfig>) -> anyhow::Result<()> {
@@ -247,7 +112,7 @@ impl ServerTrait<config::HttpServerConfig> for HttpServer {
             tokio::spawn(async move {
                 let svc = hyper::service::service_fn(move |req| {
                     let store = Arc::clone(&store);
-                    async move { handle_request(req, store).await }
+                    async move { request::handle(req, store).await }
                 });
 
                 let conn = http1::Builder::new().serve_connection(io, svc);

adapter/rest/src/main.rs

@@ -0,0 +1,167 @@
+use hyper::{HeaderMap, header::HeaderValue};
+use std::collections::HashMap;
+use tucana::shared::{Struct, ValidationFlow, Value, helper::value::ToValue, value::Kind};
+
+use crate::route;
+
+pub(super) fn build_flow_input(
+    flow: &ValidationFlow,
+    path: &str,
+    query: Option<&str>,
+    headers: &HeaderMap<HeaderValue>,
+    payload: Option<Value>,
+) -> Value {
+    let mut fields = HashMap::new();
+
+    if let Some(payload) = payload {
+        fields.insert(String::from("payload"), payload);
+    }
+
+    fields.insert(
+        String::from("headers"),
+        string_map_to_value(header_map(headers)),
+    );
+    fields.insert(
+        String::from("query_params"),
+        string_map_to_value(query_params(query)),
+    );
+    fields.insert(
+        String::from("path_params"),
+        string_map_to_value(route::extract_path_params(flow, path)),
+    );
+
+    Value {
+        kind: Some(Kind::StructValue(Struct { fields })),
+    }
+}
+
+fn header_map(headers: &HeaderMap<HeaderValue>) -> HashMap<String, String> {
+    headers
+        .iter()
+        .map(|(name, value)| {
+            let value = value
+                .to_str()
+                .map(str::to_owned)
+                .unwrap_or_else(|_| String::from_utf8_lossy(value.as_bytes()).into_owned());
+
+            (name.as_str().to_owned(), value)
+        })
+        .collect()
+}
+
+fn query_params(query: Option<&str>) -> HashMap<String, String> {
+    let Some(query) = query else {
+        return HashMap::new();
+    };
+
+    // Repeated query keys currently use last-write-wins because Taurus receives
+    // a simple object here, not a multi-map.
+    form_urlencoded::parse(query.as_bytes())
+        .map(|(key, value)| (key.into_owned(), value.into_owned()))
+        .collect()
+}
+
+fn string_map_to_value(map: HashMap<String, String>) -> Value {
+    Value {
+        kind: Some(Kind::StructValue(Struct {
+            fields: map
+                .into_iter()
+                .map(|(key, value)| (key, value.to_value()))
+                .collect(),
+        })),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{build_flow_input, query_params, string_map_to_value};
+    use hyper::HeaderMap;
+    use tucana::shared::{FlowSetting, Struct, ValidationFlow, Value, value::Kind};
+
+    #[test]
+    fn query_params_are_percent_decoded() {
+        let params = query_params(Some("search=hello+world&tag=a%2Fb&empty="));
+
+        assert_eq!(
+            params.get("search").map(String::as_str),
+            Some("hello world")
+        );
+        assert_eq!(params.get("tag").map(String::as_str), Some("a/b"));
+        assert_eq!(params.get("empty").map(String::as_str), Some(""));
+    }
+
+    #[test]
+    fn string_map_is_converted_to_struct_value() {
+        let value =
+            string_map_to_value([("id".to_string(), "42".to_string())].into_iter().collect());
+
+        let Value {
+            kind: Some(Kind::StructValue(Struct { fields })),
+        } = value
+        else {
+            panic!("expected struct value");
+        };
+
+        assert_eq!(
+            fields.get("id").and_then(|value| value.kind.as_ref()),
+            Some(&Kind::StringValue("42".to_string()))
+        );
+    }
+
+    #[test]
+    fn flow_input_contains_query_and_path_params() {
+        let flow = ValidationFlow {
+            flow_id: 1,
+            project_slug: "project".to_string(),
+            settings: vec![FlowSetting {
+                database_id: None,
+                flow_setting_id: "httpURL".to_string(),
+                value: Some(Value {
+                    kind: Some(Kind::StringValue("/users/:user_id".to_string())),
+                }),
+                cast: None,
+            }],
+            ..ValidationFlow::default()
+        };
+
+        let input = build_flow_input(
+            &flow,
+            "/project/users/42",
+            Some("search=hello+world"),
+            &HeaderMap::new(),
+            None,
+        );
+
+        assert_eq!(
+            nested_string_field(&input, "query_params", "search"),
+            Some("hello world")
+        );
+        assert_eq!(
+            nested_string_field(&input, "path_params", "user_id"),
+            Some("42")
+        );
+    }
+
+    fn nested_string_field<'a>(value: &'a Value, field: &str, nested: &str) -> Option<&'a str> {
+        let Some(Kind::StructValue(Struct { fields })) = value.kind.as_ref() else {
+            return None;
+        };
+        let Some(Value {
+            kind:
+                Some(Kind::StructValue(Struct {
+                    fields: nested_fields,
+                })),
+        }) = fields.get(field)
+        else {
+            return None;
+        };
+        let Some(Value {
+            kind: Some(Kind::StringValue(value)),
+        }) = nested_fields.get(nested)
+        else {
+            return None;
+        };
+
+        Some(value.as_str())
+    }
+}