Merge remote-tracking branch 'origin/main'

Author: nicosammito

Cargo.lock

@@ -7,8 +7,8 @@ version = "0.0.0"
 edition = "2024"
 
 [workspace.dependencies]
-code0-flow = { version = "0.0.25" }
-tucana = { version = "0.0.52", features = ["aquila"] }
+code0-flow = { version = "0.0.27" }
+tucana = { version = "0.0.56", features = ["aquila"] }
 serde_json = { version = "1.0.138" }
 log = "0.4.27"
 env_logger = "0.11.8"

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "rest"
-version = "0.1.0"
+version.workspace = true
 edition.workspace = true
 
 [dependencies]

adapter/rest/Cargo.toml

@@ -14,6 +14,7 @@ use hyper::{
     server::conn::http1,
 };
 use hyper_util::rt::TokioIo;
+use serde_json::{Map, Number, Value as J};
 use std::collections::HashMap;
 use std::convert::Infallible;
 use std::net::SocketAddr;
@@ -58,6 +59,45 @@ fn json_error(status: StatusCode, msg: &str) -> Response<Full<Bytes>> {
         .unwrap()
 }
 
+fn pb_value_to_json(v: &Value) -> serde_json::Value {
+    match v.kind.as_ref() {
+        None => J::Null,
+
+        Some(Kind::NullValue(_)) => J::Null,
+
+        Some(Kind::BoolValue(b)) => J::Bool(*b),
+
+        Some(Kind::StringValue(s)) => J::String(s.clone()),
+
+        Some(Kind::NumberValue(n)) => {
+            if let Some(num) = Number::from_f64(*n) {
+                J::Number(num)
+            } else if n.is_nan() {
+                J::String("NaN".to_string())
+            } else if n.is_infinite() {
+                if n.is_sign_positive() {
+                    J::String("Infinity".to_string())
+                } else {
+                    J::String("-Infinity".to_string())
+                }
+            } else {
+                // Fallback for unexpected cases where from_f64 returns None.
+                J::Null
+            }
+        }
+
+        Some(Kind::ListValue(lv)) => J::Array(lv.values.iter().map(pb_value_to_json).collect()),
+
+        Some(Kind::StructValue(st)) => {
+            let mut obj = Map::new();
+            for (k, v) in &st.fields {
+                obj.insert(k.clone(), pb_value_to_json(v));
+            }
+            J::Object(obj)
+        }
+    }
+}
+
 fn build_response(
     status: StatusCode,
     headers: HashMap<String, String>,
@@ -210,8 +250,13 @@ async fn execute_flow_to_hyper_response(
             };
 
             // payload -> json bytes
-            let json = serde_json::to_vec_pretty(payload_val).unwrap_or_else(|err| {
-                format!(r#"{{"error":"Serialization failed: {}"}}"#, err).into_bytes()
+            let json_val = pb_value_to_json(payload_val);
+            let json = serde_json::to_vec_pretty(&json_val).unwrap_or_else(|err| {
+                let fallback = serde_json::json!({
+                    "error": format!("Serialization failed: {}", err),
+                });
+                serde_json::to_vec(&fallback)
+                    .unwrap_or_else(|_| br#"{"error":"Serialization failed"}"#.to_vec())
             });
 
             let status =

adapter/rest/src/main.rs

@@ -1,6 +1,5 @@
 use crate::traits::IdentifiableFlow;
 use async_nats::jetstream::kv::Config;
-use code0_flow::flow_validator::verify_flow;
 use futures_lite::StreamExt;
 use prost::Message;
 use tucana::shared::{ExecutionFlow, ValidationFlow, Value};
@@ -119,17 +118,7 @@ impl AdapterStore {
         flow: ValidationFlow,
         input_value: Option<Value>,
     ) -> Option<Value> {
-        if let Some(body) = input_value.clone() {
-            let verify_result = verify_flow(flow.clone(), body);
-
-            match verify_result {
-                Ok(()) => {}
-                Err(_err) => {
-                    return None;
-                }
-            };
-        }
-
+        // TODO: Replace body vaidation with triangulus when its ready
         let uuid = uuid::Uuid::new_v4().to_string();
         let flow_id = flow.flow_id;
         let execution_flow: ExecutionFlow = Self::convert_validation_flow(flow, input_value);