use std::borrow::Cow;
use std::collections::BTreeMap;
use std::fmt;
use std::num::NonZeroU32;
use std::str::FromStr;
use schemars::JsonSchema;
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
use serde_json::Value;
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct SchemaId {
namespace: Cow<'static, str>,
name: Cow<'static, str>,
version: u32,
}
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
#[error("{text:?} is not a schema id: {why}; a schema id is <namespace>.<name>@<version>, e.g. billing.invoice@1")]
pub struct SchemaIdError {
pub text: String,
pub why: String,
}
impl SchemaId {
#[must_use]
pub const fn literal(namespace: &'static str, name: &'static str, version: u32) -> Self {
assert!(
is_part(namespace),
"SchemaId::literal: the namespace is not a non-empty run of ASCII letters, digits, `-` and `_`"
);
assert!(
is_part(name),
"SchemaId::literal: the name is not a non-empty run of ASCII letters, digits, `-` and `_`"
);
assert!(
version > 0,
"SchemaId::literal: the version is not a positive integer"
);
Self {
namespace: Cow::Borrowed(namespace),
name: Cow::Borrowed(name),
version,
}
}
pub fn new(namespace: &str, name: &str, version: u32) -> Result<Self, SchemaIdError> {
format!("{namespace}.{name}@{version}").parse()
}
#[must_use]
pub fn namespace(&self) -> &str {
&self.namespace
}
#[must_use]
pub fn name(&self) -> &str {
&self.name
}
#[must_use]
pub fn version(&self) -> u32 {
self.version
}
#[must_use]
pub fn family(&self) -> String {
format!("{}.{}", self.namespace, self.name)
}
#[must_use]
pub fn at(&self, version: NonZeroU32) -> Self {
Self {
namespace: self.namespace.clone(),
name: self.name.clone(),
version: version.get(),
}
}
}
const fn is_part(text: &str) -> bool {
let bytes = text.as_bytes();
if bytes.is_empty() {
return false;
}
let mut at = 0;
while at < bytes.len() {
let byte = bytes[at];
if !(byte.is_ascii_alphanumeric() || byte == b'-' || byte == b'_') {
return false;
}
at += 1;
}
true
}
impl FromStr for SchemaId {
type Err = SchemaIdError;
fn from_str(text: &str) -> Result<Self, Self::Err> {
let refuse = |why: &str| SchemaIdError {
text: text.to_owned(),
why: why.to_owned(),
};
let (family, version) = text
.rsplit_once('@')
.ok_or_else(|| refuse("it names no version"))?;
let (namespace, name) = family
.split_once('.')
.ok_or_else(|| refuse("it names no namespace"))?;
if !is_part(namespace) {
return Err(refuse(if namespace.is_empty() {
"its namespace is empty"
} else {
"its namespace is not letters, digits, `-` and `_`"
}));
}
if !is_part(name) {
return Err(refuse(if name.is_empty() {
"its name is empty"
} else {
"its name is not letters, digits, `-` and `_`"
}));
}
let version: u32 = version
.parse()
.ok()
.filter(|version| *version > 0)
.ok_or_else(|| refuse("its version is not a positive integer"))?;
Ok(Self {
namespace: Cow::Owned(namespace.to_owned()),
name: Cow::Owned(name.to_owned()),
version,
})
}
}
impl fmt::Display for SchemaId {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "{}.{}@{}", self.namespace, self.name, self.version)
}
}
impl Serialize for SchemaId {
fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
serializer.collect_str(self)
}
}
impl<'de> Deserialize<'de> for SchemaId {
fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
let text = String::deserialize(deserializer)?;
text.parse().map_err(serde::de::Error::custom)
}
}
impl JsonSchema for SchemaId {
fn schema_name() -> Cow<'static, str> {
Cow::Borrowed("SchemaId")
}
fn json_schema(_: &mut schemars::SchemaGenerator) -> schemars::Schema {
schemars::json_schema!({
"type": "string",
"description": "<namespace>.<name>@<version>: which schema a message is.",
"pattern": "^[A-Za-z0-9_-]+\\.[A-Za-z0-9_-]+@[1-9][0-9]*$"
})
}
}
pub trait Message: Serialize + DeserializeOwned + JsonSchema {
const SCHEMA: SchemaId;
}
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum RegistryError {
#[error(
"{id} is already registered with a different document; register a new version instead"
)]
Conflict {
id: SchemaId,
},
#[error("{id}: the document is not a usable JSON Schema: {why}")]
NotASchema {
id: SchemaId,
why: String,
},
#[error("{id} is not a registered schema; registered: {known}")]
Unknown {
id: SchemaId,
known: String,
},
}
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
#[error("{id}: at {pointer}: {message}")]
pub struct SchemaViolation {
pub id: SchemaId,
pub pointer: String,
pub message: String,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Read {
At(u32),
Unknown(UnknownVersion),
}
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
#[error(
"{family}@{declared} is not a version this build reads; it reads {}",
read_set_text(read_set)
)]
pub struct UnknownVersion {
pub family: String,
pub declared: u32,
pub read_set: Vec<u32>,
}
fn read_set_text(read_set: &[u32]) -> String {
if read_set.is_empty() {
return "no version of it".to_owned();
}
let versions: Vec<String> = read_set.iter().map(u32::to_string).collect();
format!("[{}]", versions.join(", "))
}
struct Registered {
document: Value,
validator: jsonschema::Validator,
}
#[derive(Default)]
pub struct Registry {
schemas: BTreeMap<SchemaId, Registered>,
}
impl fmt::Debug for Registry {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Registry")
.field("ids", &self.ids())
.finish()
}
}
impl Registry {
#[must_use]
pub fn new() -> Self {
Self::default()
}
pub fn register<M: Message>(&mut self) -> Result<(), RegistryError> {
let schema = schemars::schema_for!(M);
self.register_schema(M::SCHEMA, schema.to_value())
}
pub fn register_schema(&mut self, id: SchemaId, schema: Value) -> Result<(), RegistryError> {
if let Some(held) = self.schemas.get(&id) {
if held.document == schema {
return Ok(());
}
return Err(RegistryError::Conflict { id });
}
let validator =
jsonschema::validator_for(&schema).map_err(|failure| RegistryError::NotASchema {
id: id.clone(),
why: failure.to_string(),
})?;
self.schemas.insert(
id,
Registered {
document: schema,
validator,
},
);
Ok(())
}
#[must_use]
pub fn ids(&self) -> Vec<SchemaId> {
self.schemas.keys().cloned().collect()
}
#[must_use]
pub fn schema(&self, id: &SchemaId) -> Option<&Value> {
self.schemas.get(id).map(|held| &held.document)
}
pub fn check(&self, id: &SchemaId, payload: &Value) -> Result<(), CheckError> {
let held = self.schemas.get(id).ok_or_else(|| RegistryError::Unknown {
id: id.clone(),
known: self.known(),
})?;
match held.validator.iter_errors(payload).next() {
None => Ok(()),
Some(violation) => Err(CheckError::Violation(SchemaViolation {
id: id.clone(),
pointer: violation.instance_path().to_string(),
message: violation.to_string(),
})),
}
}
#[must_use]
pub fn read_set(&self, family: &str) -> Vec<u32> {
let mut versions: Vec<u32> = self
.schemas
.keys()
.filter(|id| id.family() == family)
.map(SchemaId::version)
.collect();
versions.sort_unstable_by(|a, b| b.cmp(a));
versions
}
#[must_use]
pub fn writes(&self, family: &str) -> Option<u32> {
self.read_set(family).first().copied()
}
#[must_use]
pub fn read_at(&self, family: &str, declared: u32) -> Read {
let read_set = self.read_set(family);
match (read_set.contains(&declared), read_set.first()) {
(true, Some(writes)) => Read::At(*writes),
_ => Read::Unknown(UnknownVersion {
family: family.to_owned(),
declared,
read_set,
}),
}
}
fn known(&self) -> String {
let ids: Vec<String> = self.schemas.keys().map(ToString::to_string).collect();
if ids.is_empty() {
"nothing".to_owned()
} else {
ids.join(", ")
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum CheckError {
#[error(transparent)]
Registry(#[from] RegistryError),
#[error(transparent)]
Violation(#[from] SchemaViolation),
}