validate incoming JSON (#1699)

Author: phiSgr

apps/framework-cli/src/cli/local_webserver.rs

@@ -37,10 +37,12 @@ use rdkafka::message::OwnedMessage;
 use rdkafka::producer::future_producer::OwnedDeliveryResult;
 use rdkafka::producer::{DeliveryFuture, FutureRecord};
 use rdkafka::util::Timeout;
-use serde::Deserialize;
 use serde::Serialize;
-use serde_json::Value;
+use serde::{Deserialize, Deserializer};
+use serde_json::Deserializer as JsonDeserializer;
 
+use crate::framework::data_model::model::DataModel;
+use crate::utilities::validate_passthrough::{DataModelArrayVisitor, DataModelVisitor};
 use lazy_static::lazy_static;
 use log::Level::{Debug, Trace};
 use std::collections::{HashMap, HashSet};
@@ -412,12 +414,10 @@ fn route_not_found_response() -> hyper::http::Result<Response<Full<Bytes>>> {
 async fn send_payload_to_topic(
     configured_producer: &ConfiguredProducer,
     topic_name: &str,
-    payload: Value,
+    payload: Vec<u8>,
     metrics: Arc<Metrics>,
     route: PathBuf,
 ) -> Result<(i32, i64), (KafkaError, OwnedMessage)> {
-    let payload = serde_json::to_vec(&payload).unwrap();
-
     debug!("Sending payload {:?} to topic: {}", payload, topic_name);
 
     metrics
@@ -447,6 +447,7 @@ async fn to_reader(req: Request<Incoming>) -> bytes::buf::Reader<impl Buf + Size
 async fn handle_json_req(
     configured_producer: &ConfiguredProducer,
     topic_name: &str,
+    data_model: &DataModel,
     req: Request<Incoming>,
     metrics: Arc<Metrics>,
     route: PathBuf,
@@ -456,7 +457,9 @@ async fn handle_json_req(
     let url = req.uri().to_string();
     let number_of_bytes = req.body().size_hint().exact().unwrap();
     let body = to_reader(req).await;
-    let parsed: Result<Value, serde_json::Error> = serde_json::from_reader(body);
+
+    let parsed = JsonDeserializer::from_reader(body)
+        .deserialize_any(&mut DataModelVisitor::new(&data_model.columns));
 
     metrics
         .send_metric(MetricsMessage::PutIngestedBytesCount {
@@ -508,6 +511,7 @@ async fn wait_for_batch_complete(
 async fn handle_json_array_body(
     configured_producer: &ConfiguredProducer,
     topic_name: &str,
+    data_model: &DataModel,
     req: Request<Incoming>,
     metrics: Arc<Metrics>,
     route: PathBuf,
@@ -522,7 +526,9 @@ async fn handle_json_array_body(
         "starting to parse json array with length {} for {}",
         number_of_bytes, topic_name
     );
-    let parsed: Result<Vec<Value>, serde_json::Error> = serde_json::from_reader(body);
+    let parsed = JsonDeserializer::from_reader(body).deserialize_seq(&mut DataModelArrayVisitor {
+        inner: DataModelVisitor::new(&data_model.columns),
+    });
 
     debug!("parsed json array for {}", topic_name);
     metrics
@@ -541,8 +547,6 @@ async fn handle_json_array_body(
     let mut temp_res: Vec<Result<DeliveryFuture, KafkaError>> = Vec::new();
 
     for (count, payload) in parsed.ok().unwrap().into_iter().enumerate() {
-        let payload = serde_json::to_vec(&payload).unwrap();
-
         debug!("Sending payload {:?} to topic: {}", payload, topic_name);
         let record = FutureRecord::to(topic_name)
             .key(topic_name) // This should probably be generated by the client that pushes data to the API
@@ -644,6 +648,7 @@ async fn ingest_route(
             EndpointIngestionFormat::Json => Ok(handle_json_req(
                 &configured_producer,
                 &route_meta.topic_name,
+                &route_meta.data_model,
                 req,
                 metrics,
                 route,
@@ -652,6 +657,7 @@ async fn ingest_route(
             EndpointIngestionFormat::JsonArray => Ok(handle_json_array_body(
                 &configured_producer,
                 &route_meta.topic_name,
+                &route_meta.data_model,
                 req,
                 metrics,
                 route,
@@ -859,11 +865,15 @@ impl Webserver {
                     ApiChange::ApiEndpoint(Change::Added(api_endpoint)) => {
                         log::info!("Adding route: {:?}", api_endpoint.path);
                         match api_endpoint.api_type {
-                            APIType::INGRESS { target_topic } => {
+                            APIType::INGRESS {
+                                target_topic,
+                                data_model,
+                            } => {
                                 route_table.insert(
                                     api_endpoint.path.clone(),
                                     RouteMeta {
                                         format: api_endpoint.format.clone(),
+                                        data_model: data_model.unwrap(),
                                         topic_name: target_topic,
                                     },
                                 );
@@ -879,14 +889,18 @@ impl Webserver {
                     }
                     ApiChange::ApiEndpoint(Change::Updated { before, after }) => {
                         match &after.api_type {
-                            APIType::INGRESS { target_topic } => {
+                            APIType::INGRESS {
+                                target_topic,
+                                data_model,
+                            } => {
                                 log::info!("Replacing route: {:?} with {:?}", before, after);
 
                                 route_table.remove(&before.path);
                                 route_table.insert(
                                     after.path.clone(),
                                     RouteMeta {
                                         format: after.format.clone(),
+                                        data_model: data_model.as_ref().unwrap().clone(),
                                         topic_name: target_topic.clone(),
                                     },
                                 );

apps/framework-cli/src/cli/watcher.rs

@@ -214,6 +214,7 @@ async fn process_data_models_changes(
             ingest_route,
             RouteMeta {
                 topic_name: fo.topic.clone(),
+                data_model: fo.data_model.clone(),
                 format: fo.data_model.config.ingestion.format.clone(),
             },
         );

apps/framework-cli/src/framework/controller.rs

@@ -2,6 +2,7 @@ use super::core::code_loader::FrameworkObject;
 use super::data_model::config::EndpointIngestionFormat;
 use crate::framework::core::infrastructure::table::Table;
 use crate::framework::core::plan::PlanningError;
+use crate::framework::data_model::model::DataModel;
 use crate::infrastructure::migration::InitialDataLoadError;
 use crate::infrastructure::olap;
 use crate::infrastructure::olap::clickhouse::config::ClickHouseConfig;
@@ -31,6 +32,7 @@ use std::sync::Arc;
 #[derive(Debug, Clone)]
 pub struct RouteMeta {
     pub topic_name: String,
+    pub data_model: DataModel,
     pub format: EndpointIngestionFormat,
 }
 
@@ -481,6 +483,7 @@ pub async fn set_up_topic_and_tables_and_route(
         ingest_route.clone(),
         RouteMeta {
             topic_name,
+            data_model: fo.data_model.clone(),
             format: fo.data_model.config.ingestion.format.clone(),
         },
     );

apps/framework-cli/src/framework/core/infrastructure/api_endpoint.rs

@@ -11,7 +11,13 @@ use super::{topic::Topic, DataLineage, InfrastructureSignature};
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub enum APIType {
-    INGRESS { target_topic: String },
+    INGRESS {
+        target_topic: String,
+        // in previous versions this is not stored,
+        // so deserialization may fail if the value is not optional
+        // our code does not depend on the stored field
+        data_model: Option<DataModel>,
+    },
     EGRESS,
 }
 
@@ -41,9 +47,10 @@ impl ApiEndpoint {
             name: data_model.name.clone(),
             api_type: APIType::INGRESS {
                 target_topic: topic.id(),
+                data_model: Some(data_model.clone()),
             },
-            // This implementation is actually removing the functionaliy of nestedness of paths in
-            // data model to change the ingest path. Howevever, we are changin how this works with an
+            // This implementation is actually removing the functionality of nestedness of paths in
+            // data model to change the ingest path. However, we are changing how this works with an
             // explicit in ingest path and we have not seen people use that functionality yet.
             path: PathBuf::from("ingest")
                 .join(data_model.name.clone())
@@ -95,7 +102,7 @@ impl DataLineage for ApiEndpoint {
 
     fn pushes_data_to(&self) -> Vec<InfrastructureSignature> {
         match &self.api_type {
-            APIType::INGRESS { target_topic } => {
+            APIType::INGRESS { target_topic, .. } => {
                 vec![InfrastructureSignature::Topic {
                     id: target_topic.clone(),
                 }]

apps/framework-cli/src/infrastructure/olap/clickhouse/queries.rs

@@ -272,6 +272,7 @@ fn basic_field_type_to_string(
                 .iter()
                 .map(|enum_member| match &enum_member.value {
                     EnumValue::Int(int) => format!("'{}' = {}", enum_member.name, int),
+                    // "Numbers are assigned starting from 1 by default."
                     EnumValue::String(string) => format!("'{}'", string),
                 })
                 .collect::<Vec<String>>()

apps/framework-cli/src/utilities.rs

@@ -7,6 +7,7 @@ pub mod git;
 pub mod package_managers;
 pub mod retry;
 pub mod system;
+pub mod validate_passthrough;
 
 pub trait PathExt {
     fn ext_is_supported_lang(&self) -> bool;