feat(rust): Add validate_json() method in Rust

Cargo.toml

@@ -31,6 +31,8 @@ serde = { version = "1.0", features = ["derive"]}
 serde_json = "1.0"
 serde_yaml = "0.9"
 regress = { version = "0.5.0", optional = true }  # required by generated code
+jsonschema = "0.17.1"
+thiserror = "1.0.49"
 
 # redeclare all build-dependencies here so that the build.rs can be used
 # standalone to view types
@@ -45,3 +47,5 @@ schemars = { version = "0.8.0", optional = true }
 serde_json = "1.0"
 syn = { version = "2.0.11", optional = true }
 typify = { version = "0.0.11", optional = true }
+jsonschema = "0.17.1"
+thiserror = "1.0.49"

rust/lib.rs

@@ -1,8 +1,8 @@
+use jsonschema::JSONSchema;
 use serde::{Deserialize, Serialize};
-use std::error;
-use std::fmt;
 use std::fs::{read_to_string, File};
 use std::path::{Path, PathBuf};
+use thiserror::Error;
 
 // This file is supposed to be auto-generated via rust/build.rs
 pub mod schema_types {
@@ -25,30 +25,23 @@ pub enum CompatibilityMode {
     Backward,
 }
 
-#[derive(Debug, PartialEq, Eq)]
+#[derive(Debug, Error)]
 #[non_exhaustive]
 pub enum SchemaError {
+    #[error("Topic not found")]
     TopicNotFound,
+    #[error("Invalid version")]
     InvalidVersion,
+    #[error("Invalid type")]
     InvalidType,
+    #[error("Invalid schema")]
     InvalidSchema,
+    #[error("Invalid message")]
+    InvalidMessage,
 }
 
-impl fmt::Display for SchemaError {
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        match self {
-            SchemaError::TopicNotFound => write!(f, "Topic not found"),
-            SchemaError::InvalidVersion => write!(f, "Invalid version"),
-            SchemaError::InvalidType => write!(f, "Invalid type"),
-            SchemaError::InvalidSchema => write!(f, "Invalid schema"),
-        }
-    }
-}
-
-impl error::Error for SchemaError {}
-
 #[derive(Debug, PartialEq, Serialize, Deserialize)]
-struct TopicSchema {
+pub struct TopicSchema {
     version: u16,
     #[serde(rename = "type")]
     schema_type: SchemaType,
@@ -70,25 +63,42 @@ impl TopicData {
     }
 }
 
-#[derive(Debug, PartialEq, Serialize, Deserialize)]
+#[derive(Debug, Serialize, Deserialize)]
 pub struct Schema {
     pub version: u16,
     pub schema_type: SchemaType,
     pub compatibility_mode: CompatibilityMode,
-    // TODO: Actually parse the json schema
     pub schema: String,
     pub schema_filepath: PathBuf,
+    #[serde(skip)]
+    compiled_json_schema: Option<JSONSchema>,
 }
 
-/// Returns the schema for a topic. If version is passed, return the schema for
-/// the specified version, otherwise the latest version is returned.
-///
-/// Only JSON schemas are currently supported.
-///
-/// # Errors
-///
-/// Will return `Err` if `topic` or `version` is not found or if schema data is invalid.
-pub fn get_schema(topic: &str, version: Option<u16>) -> Result<Schema, SchemaError> {
+impl PartialEq for Schema {
+    fn eq(&self, other: &Self) -> bool {
+        self.version == other.version
+            && self.schema_type == other.schema_type
+            && self.compatibility_mode == other.compatibility_mode
+            && self.schema == other.schema
+            && self.schema_filepath == other.schema_filepath
+    }
+}
+
+impl Schema {
+    pub fn validate_json(&self, input: &[u8]) -> Result<(), SchemaError> {
+        let message = serde_json::from_slice(input).map_err(|_| SchemaError::InvalidMessage)?;
+
+        if let Some(compiled) = &self.compiled_json_schema {
+            compiled
+                .validate(&message)
+                .map_err(|_| SchemaError::InvalidMessage)
+        } else {
+            Err(SchemaError::InvalidSchema)
+        }
+    }
+}
+
+pub fn get_topic_schema(topic: &str, version: Option<u16>) -> Result<TopicSchema, SchemaError> {
     let topic_path = Path::new(env!("CARGO_MANIFEST_DIR")).join(format!("topics/{topic}.yaml"));
     let mut topic_data = TopicData::load(&topic_path)?;
     topic_data.schemas.sort_by_key(|x| x.version);
@@ -109,16 +119,36 @@ pub fn get_schema(topic: &str, version: Option<u16>) -> Result<Schema, SchemaErr
         return Err(SchemaError::InvalidType);
     }
 
+    Ok(schema_metadata)
+}
+
+/// Returns the schema for a topic. If version is passed, return the schema for
+/// the specified version, otherwise the latest version is returned.
+///
+/// Only JSON schemas are currently supported.
+///
+/// # Errors
+///
+/// Will return `Err` if `topic` or `version` is not found or if schema data is invalid.
+pub fn get_schema(topic: &str, version: Option<u16>) -> Result<Schema, SchemaError> {
+    let schema_metadata = get_topic_schema(topic, version)?;
     let json_schema_path =
         Path::new(env!("CARGO_MANIFEST_DIR")).join(format!("schemas/{}", schema_metadata.resource));
 
+    let schema = read_to_string(&json_schema_path).map_err(|_| SchemaError::InvalidSchema)?;
+
+    let s = serde_json::from_str(&schema).map_err(|_| SchemaError::InvalidSchema)?;
+    let compiled_json_schema =
+        Some(JSONSchema::compile(&s).map_err(|_| SchemaError::InvalidSchema)?);
+
     Ok({
         Schema {
             version: schema_metadata.version,
             schema_type: schema_metadata.schema_type,
             compatibility_mode: schema_metadata.compatibility_mode,
-            schema: read_to_string(&json_schema_path).map_err(|_| SchemaError::InvalidSchema)?,
+            schema,
             schema_filepath: json_schema_path.to_owned(),
+            compiled_json_schema,
         }
     })
 }
@@ -129,9 +159,33 @@ mod tests {
 
     #[test]
     fn test_get_schema() {
-        assert_eq!(get_schema("asdf", None), Err(SchemaError::TopicNotFound));
+        assert!(matches!(
+            get_schema("asdf", None),
+            Err(SchemaError::TopicNotFound)
+        ));
         let schema = get_schema("snuba-queries", None).unwrap();
         assert_eq!(schema.version, 1);
         assert_eq!(schema.schema_type, SchemaType::Json);
+
+        // Did not error
+        get_schema("snuba-queries", Some(1)).unwrap();
+    }
+
+    #[test]
+    fn test_validate() {
+        let topic = "snuba-queries";
+        let topic_schema = get_topic_schema(topic, None).unwrap();
+        let schema = get_schema(topic, None).unwrap();
+        let example_dir = topic_schema.examples.first().unwrap();
+        let example_path = Path::new(env!("CARGO_MANIFEST_DIR"))
+            .join(format!("examples/{}{}", example_dir, "snuba-queries1.json"));
+        let example = read_to_string(&example_path).unwrap();
+        schema.validate_json(example.as_bytes()).unwrap();
+
+        let invalid = "{}";
+        assert!(matches!(
+            schema.validate_json(invalid.as_bytes()),
+            Err(SchemaError::InvalidMessage)
+        ));
     }
 }