liminal/channel/
schema.rs

1use std::collections::BTreeMap;
2use std::sync::Arc;
3
4use jsonschema::Validator;
5use serde_json::{Map, Value};
6use uuid::Uuid;
7
8/// The schema keyword that makes a channel carry opaque bytes.
9///
10/// A document with `"x-liminal-opaque": true` admits and delivers every payload
11/// unchanged, JSON or not. JSON Schema ignores unknown keywords, so the document
12/// stays a valid schema; without the keyword a channel admits JSON only.
13pub const OPAQUE_KEYWORD: &str = "x-liminal-opaque";
14
15/// Unique identifier for a schema version that validated a message.
16#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
17pub struct SchemaId(Uuid);
18
19impl SchemaId {
20    /// Generates a new schema identifier.
21    #[must_use]
22    pub fn new() -> Self {
23        Self(Uuid::new_v4())
24    }
25
26    /// Wraps an existing UUID as a schema identifier.
27    #[must_use]
28    pub const fn from_uuid(uuid: Uuid) -> Self {
29        Self(uuid)
30    }
31
32    /// Returns the underlying UUID.
33    #[must_use]
34    pub const fn as_uuid(self) -> Uuid {
35        self.0
36    }
37}
38
39impl Default for SchemaId {
40    fn default() -> Self {
41        Self::new()
42    }
43}
44
45/// JSON Schema-backed message contract for a channel.
46#[derive(Clone, Debug)]
47pub struct Schema {
48    id: SchemaId,
49    definition: Arc<Value>,
50    validator: Arc<Validator>,
51    defaults: Arc<BTreeMap<String, Value>>,
52}
53
54impl Schema {
55    /// Builds a schema from a JSON Schema definition.
56    ///
57    /// # Errors
58    ///
59    /// Returns [`SchemaValidationError::InvalidSchema`] when the definition cannot be compiled.
60    pub fn new(definition: Value) -> Result<Self, SchemaValidationError> {
61        Self::with_id(SchemaId::new(), definition)
62    }
63
64    /// Builds a schema from a JSON Schema definition with an explicit identifier.
65    ///
66    /// # Errors
67    ///
68    /// Returns [`SchemaValidationError::InvalidSchema`] when the definition cannot be compiled.
69    pub fn with_id(id: SchemaId, definition: Value) -> Result<Self, SchemaValidationError> {
70        let defaults = collect_object_defaults(&definition)?;
71        Self::from_parts(id, definition, defaults)
72    }
73
74    /// Returns the schema version identifier.
75    #[must_use]
76    pub const fn id(&self) -> SchemaId {
77        self.id
78    }
79
80    /// Returns the wrapped JSON Schema definition.
81    #[must_use]
82    pub fn definition(&self) -> &Value {
83        &self.definition
84    }
85
86    /// Validates JSON payload bytes after applying known schema defaults.
87    ///
88    /// # Errors
89    ///
90    /// Returns [`SchemaValidationError`] when the payload is not JSON, cannot receive required
91    /// defaults, or does not match this schema.
92    pub fn validate<Payload>(&self, payload: Payload) -> Result<(), SchemaValidationError>
93    where
94        Payload: AsRef<[u8]>,
95    {
96        if self.opaque() {
97            return Ok(());
98        }
99        let value = self.normalized_value(payload)?;
100        self.validate_value(&value)
101    }
102
103    /// Validates payload bytes and returns the normalized JSON bytes to deliver.
104    ///
105    /// # Errors
106    ///
107    /// Returns [`SchemaValidationError`] when the payload cannot be parsed, normalized, validated,
108    /// or serialized for delivery.
109    pub fn validate_and_apply_defaults<Payload>(
110        &self,
111        payload: Payload,
112    ) -> Result<Vec<u8>, SchemaValidationError>
113    where
114        Payload: AsRef<[u8]>,
115    {
116        let bytes = payload.as_ref();
117        if self.opaque() {
118            return Ok(bytes.to_vec());
119        }
120        let value = self.normalized_value(bytes)?;
121        self.validate_value(&value)?;
122        if self.defaults.is_empty() {
123            return Ok(bytes.to_vec());
124        }
125        super::canonical::to_vec(value)
126            .map_err(|source| SchemaValidationError::Serialize { source })
127    }
128
129    /// Evolves an object schema by adding a required field with a default value.
130    ///
131    /// # Errors
132    ///
133    /// Returns [`SchemaValidationError`] when this is not an object schema, the field schema is
134    /// invalid, the default does not match the field schema, or the evolved schema cannot compile.
135    pub fn evolve_add_field(
136        &self,
137        name: impl Into<String>,
138        field_schema: Value,
139        default: Value,
140    ) -> Result<Self, SchemaValidationError> {
141        let name = name.into();
142        if name.is_empty() {
143            return Err(SchemaValidationError::EmptyFieldName);
144        }
145        validate_default(&field_schema, &default)?;
146
147        let mut definition = (*self.definition).clone();
148        if !is_object_schema(&definition) {
149            return Err(SchemaValidationError::NotObjectSchema);
150        }
151
152        let document = definition
153            .as_object_mut()
154            .ok_or(SchemaValidationError::NotObjectSchema)?;
155        insert_property_schema(document, &name, field_schema, &default)?;
156        insert_required_field(document, &name)?;
157
158        let mut defaults = (*self.defaults).clone();
159        defaults.insert(name, default);
160        Self::from_parts(SchemaId::new(), definition, defaults)
161    }
162
163    fn from_parts(
164        id: SchemaId,
165        definition: Value,
166        defaults: BTreeMap<String, Value>,
167    ) -> Result<Self, SchemaValidationError> {
168        let validator = jsonschema::validator_for(&definition).map_err(|error| {
169            SchemaValidationError::InvalidSchema {
170                message: error.to_string(),
171            }
172        })?;
173
174        Ok(Self {
175            id,
176            definition: Arc::new(definition),
177            validator: Arc::new(validator),
178            defaults: Arc::new(defaults),
179        })
180    }
181
182    fn normalized_value<Payload>(&self, payload: Payload) -> Result<Value, SchemaValidationError>
183    where
184        Payload: AsRef<[u8]>,
185    {
186        let mut value = serde_json::from_slice(payload.as_ref())
187            .map_err(|source| SchemaValidationError::InvalidJson { source })?;
188        self.apply_defaults(&mut value)?;
189        Ok(value)
190    }
191
192    /// A channel carries opaque bytes only when its schema document says so
193    /// with [`OPAQUE_KEYWORD`]; `{}` and `true` still mean any JSON value.
194    fn opaque(&self) -> bool {
195        self.definition.get(OPAQUE_KEYWORD) == Some(&Value::Bool(true))
196    }
197
198    fn apply_defaults(&self, value: &mut Value) -> Result<(), SchemaValidationError> {
199        if self.defaults.is_empty() {
200            return Ok(());
201        }
202
203        let object = value
204            .as_object_mut()
205            .ok_or(SchemaValidationError::PayloadNotObject)?;
206        for (field, default) in self.defaults.iter() {
207            object
208                .entry(field.clone())
209                .or_insert_with(|| default.clone());
210        }
211        Ok(())
212    }
213
214    fn validate_value(&self, value: &Value) -> Result<(), SchemaValidationError> {
215        self.validator
216            .validate(value)
217            .map_err(|error| SchemaValidationError::Mismatch {
218                message: error.to_string(),
219            })
220    }
221}
222
223/// Errors returned while compiling schemas, validating payloads, or evolving schemas.
224#[derive(Debug, thiserror::Error)]
225pub enum SchemaValidationError {
226    /// The payload was not syntactically valid JSON.
227    #[error("invalid JSON payload: {source}")]
228    InvalidJson { source: serde_json::Error },
229    /// The JSON Schema definition could not be compiled.
230    #[error("invalid JSON Schema: {message}")]
231    InvalidSchema { message: String },
232    /// The payload did not satisfy the JSON Schema definition.
233    #[error("payload does not match schema: {message}")]
234    Mismatch { message: String },
235    /// Schema evolution can only add fields to object schemas.
236    #[error("schema evolution only supports object schemas")]
237    NotObjectSchema,
238    /// Schema evolution field names must be non-empty.
239    #[error("schema evolution field name must not be empty")]
240    EmptyFieldName,
241    /// The schema's properties member was not an object.
242    #[error("object schema properties must be an object")]
243    InvalidProperties,
244    /// The schema's required member was not an array of strings.
245    #[error("object schema required must be an array of strings")]
246    InvalidRequired,
247    /// The payload cannot receive object-field defaults because it is not an object.
248    #[error("payload must be a JSON object to apply schema defaults")]
249    PayloadNotObject,
250    /// The normalized JSON payload could not be serialized for delivery.
251    #[error("failed to serialize normalized payload: {source}")]
252    Serialize { source: serde_json::Error },
253}
254
255fn collect_object_defaults(
256    definition: &Value,
257) -> Result<BTreeMap<String, Value>, SchemaValidationError> {
258    let Some(document) = definition.as_object() else {
259        return Ok(BTreeMap::new());
260    };
261    let Some(properties) = document.get("properties") else {
262        return Ok(BTreeMap::new());
263    };
264    let properties = properties
265        .as_object()
266        .ok_or(SchemaValidationError::InvalidProperties)?;
267
268    let defaults = properties
269        .iter()
270        .filter_map(|(field, schema)| {
271            schema
272                .as_object()
273                .and_then(|field_schema| field_schema.get("default"))
274                .map(|default| (field.clone(), default.clone()))
275        })
276        .collect();
277    Ok(defaults)
278}
279
280fn validate_default(field_schema: &Value, default: &Value) -> Result<(), SchemaValidationError> {
281    let validator = jsonschema::validator_for(field_schema).map_err(|error| {
282        SchemaValidationError::InvalidSchema {
283            message: error.to_string(),
284        }
285    })?;
286    validator
287        .validate(default)
288        .map_err(|error| SchemaValidationError::Mismatch {
289            message: format!("default value does not match field schema: {error}"),
290        })
291}
292
293fn is_object_schema(definition: &Value) -> bool {
294    let Some(document) = definition.as_object() else {
295        return false;
296    };
297
298    match document.get("type") {
299        Some(Value::String(schema_type)) => schema_type == "object",
300        Some(Value::Array(schema_types)) => schema_types
301            .iter()
302            .any(|schema_type| schema_type.as_str() == Some("object")),
303        Some(_) => false,
304        None => document.contains_key("properties"),
305    }
306}
307
308fn insert_property_schema(
309    document: &mut Map<String, Value>,
310    name: &str,
311    mut field_schema: Value,
312    default: &Value,
313) -> Result<(), SchemaValidationError> {
314    let field_document =
315        field_schema
316            .as_object_mut()
317            .ok_or_else(|| SchemaValidationError::InvalidSchema {
318                message: "field schema must be a JSON Schema object".to_owned(),
319            })?;
320    field_document.insert("default".to_owned(), default.clone());
321
322    let properties = document
323        .entry("properties".to_owned())
324        .or_insert_with(|| Value::Object(Map::new()));
325    let properties = properties
326        .as_object_mut()
327        .ok_or(SchemaValidationError::InvalidProperties)?;
328    properties.insert(name.to_owned(), field_schema);
329    Ok(())
330}
331
332fn insert_required_field(
333    document: &mut Map<String, Value>,
334    name: &str,
335) -> Result<(), SchemaValidationError> {
336    let required = document
337        .entry("required".to_owned())
338        .or_insert_with(|| Value::Array(Vec::new()));
339    let required = required
340        .as_array_mut()
341        .ok_or(SchemaValidationError::InvalidRequired)?;
342
343    if required.iter().any(|item| item.as_str().is_none()) {
344        return Err(SchemaValidationError::InvalidRequired);
345    }
346    if !required.iter().any(|item| item.as_str() == Some(name)) {
347        required.push(Value::String(name.to_owned()));
348    }
349    Ok(())
350}
351
352#[cfg(test)]
353mod tests {
354    use super::{Schema, SchemaId, SchemaValidationError};
355    use serde_json::{Value, json};
356
357    #[test]
358    fn opaque_channels_preserve_every_byte() -> Result<(), SchemaValidationError> {
359        let schema = Schema::new(json!({super::OPAQUE_KEYWORD: true}))?;
360        for bytes in [
361            b"plain prose\r\nexecute nothing".as_slice(),
362            &[0, 255, 10],
363            b"{ \"z\": 1, \"a\":2 }".as_slice(),
364        ] {
365            assert_eq!(schema.validate_and_apply_defaults(bytes)?, bytes);
366            schema.validate(bytes)?;
367        }
368        Ok(())
369    }
370
371    #[test]
372    fn an_empty_schema_still_admits_json_only() -> Result<(), SchemaValidationError> {
373        let schema = Schema::new(json!({}))?;
374        assert!(schema.validate(b"plain prose").is_err());
375        assert!(schema.validate_and_apply_defaults(b"plain prose").is_err());
376        schema.validate(b"\"json\"")?;
377        Ok(())
378    }
379
380    #[test]
381    fn contracts_without_defaults_preserve_json_spelling() -> Result<(), SchemaValidationError> {
382        let schema = Schema::new(json!({"type":"object"}))?;
383        let bytes = b"{ \"z\": 1, \"a\":2 }";
384        assert_eq!(schema.validate_and_apply_defaults(bytes)?, bytes);
385        Ok(())
386    }
387
388    #[test]
389    fn schema_is_clone_send_sync() {
390        fn assert_bounds<T: Clone + Send + Sync + std::fmt::Debug>() {}
391
392        assert_bounds::<Schema>();
393    }
394
395    #[test]
396    fn validates_payload_against_json_schema() -> Result<(), SchemaValidationError> {
397        let schema = order_schema()?;
398
399        schema.validate(br#"{"order_id":"A1","quantity":3}"#)?;
400        let result = schema.validate(br#"{"order_id":"A1","quantity":0}"#);
401
402        assert!(matches!(
403            result,
404            Err(SchemaValidationError::Mismatch { .. })
405        ));
406        Ok(())
407    }
408
409    #[test]
410    fn evolution_adds_defaulted_field_and_changes_schema_id() -> Result<(), SchemaValidationError> {
411        let schema = order_schema()?;
412        let old_id = schema.id();
413        let evolved =
414            schema.evolve_add_field("priority", json!({"type":"string"}), json!("normal"))?;
415        let normalized =
416            evolved.validate_and_apply_defaults(br#"{"order_id":"A1","quantity":3}"#)?;
417        let payload: Value = serde_json::from_slice(&normalized)
418            .map_err(|source| SchemaValidationError::InvalidJson { source })?;
419
420        assert_ne!(evolved.id(), old_id);
421        assert_eq!(payload.get("priority"), Some(&json!("normal")));
422        Ok(())
423    }
424
425    #[test]
426    fn evolution_rejects_non_object_schema() -> Result<(), SchemaValidationError> {
427        let schema = Schema::new(json!({"type":"array"}))?;
428        let result = schema.evolve_add_field("priority", json!({"type":"string"}), json!("normal"));
429
430        assert!(matches!(
431            result,
432            Err(SchemaValidationError::NotObjectSchema)
433        ));
434        Ok(())
435    }
436
437    #[test]
438    fn explicit_schema_id_is_preserved() -> Result<(), SchemaValidationError> {
439        let id = SchemaId::new();
440        let schema = Schema::with_id(id, json!({"type":"object"}))?;
441
442        assert_eq!(schema.id(), id);
443        Ok(())
444    }
445
446    fn order_schema() -> Result<Schema, SchemaValidationError> {
447        Schema::new(json!({
448            "type": "object",
449            "properties": {
450                "order_id": {"type": "string"},
451                "quantity": {"type": "integer", "minimum": 1}
452            },
453            "required": ["order_id", "quantity"],
454            "additionalProperties": false
455        }))
456    }
457}
liminal/channel/schema.rs

liminal/channel/
schema.rs
