use std::collections::BTreeMap;
use std::path::Path;
use serde::{Deserialize, Serialize};
pub fn lock_version() -> u32 {
2
}
const LEGACY_EVENTS_ONLY_VERSION: u32 = 1;
#[derive(Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
pub struct SchemaRegistry {
version: u32,
#[serde(default)]
events: BTreeMap<String, EventEntry>,
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
agents: BTreeMap<String, AgentEntry>,
}
impl SchemaRegistry {
pub fn new() -> Self {
SchemaRegistry {
version: lock_version(),
events: BTreeMap::new(),
agents: BTreeMap::new(),
}
}
pub fn get(&self, key: &str) -> Option<&EventEntry> {
self.events.get(key)
}
pub fn insert(&mut self, key: String, entry: EventEntry) {
self.events.insert(key, entry);
}
pub fn get_agent(&self, key: &str) -> Option<&AgentEntry> {
self.agents.get(key)
}
pub fn insert_agent(&mut self, key: String, entry: AgentEntry) {
self.agents.insert(key, entry);
}
pub fn agents(&self) -> impl Iterator<Item = (&String, &AgentEntry)> {
self.agents.iter()
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct AgentEntry {
pub version: i64,
pub floor: i64,
pub shapes: Vec<AgentShape>,
}
impl AgentEntry {
fn validate(&self) -> Result<(), String> {
if self.floor < 1 || self.floor > self.version {
return Err(format!(
"has floor {} outside 1..={}",
self.floor, self.version
));
}
let versions: Vec<i64> = self.shapes.iter().map(|s| s.version).collect();
let expected: Vec<i64> = (self.floor..=self.version).collect();
if versions != expected {
return Err(format!(
"records shapes for versions {versions:?}, not one per version {}..={}",
self.floor, self.version
));
}
Ok(())
}
pub fn oldest_readable(&self) -> i64 {
let mut m = self.version;
while m > self.floor && self.shapes.iter().any(|s| s.version == m && s.additive) {
m -= 1;
}
m
}
pub fn current(&self) -> Option<&AgentShape> {
self.shapes.iter().find(|s| s.version == self.version)
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct AgentShape {
pub version: i64,
pub additive: bool,
pub fields: Vec<StoreShape>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct StoreShape {
pub name: String,
pub kind: String,
#[serde(rename = "type")]
pub ty: String,
pub refinement: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct EventEntry {
pub schema: i64,
pub fields: Vec<FieldShape>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct FieldShape {
pub name: String,
#[serde(rename = "type")]
pub ty: String,
pub default: bool,
}
pub fn parse(existing: Option<&str>, project_root: &Path) -> Result<SchemaRegistry, String> {
let Some(text) = existing else {
return Ok(SchemaRegistry::new());
};
let path = project_root.join("bynk.schema.lock");
if text.trim().is_empty() {
return Err(format!(
"schema registry `{}` is empty or truncated (corrupt); refusing \
to treat it as a fresh project — restore it from version control",
path.display()
));
}
let reg: SchemaRegistry = toml::from_str(text).map_err(|e| {
format!(
"schema registry `{}` is corrupt ({e}) — restore it from version control",
path.display()
)
})?;
let mut reg = reg;
if reg.version == LEGACY_EVENTS_ONLY_VERSION && reg.agents.is_empty() {
reg.version = lock_version();
}
if reg.version != lock_version() {
return Err(format!(
"unsupported schema registry version {} (`{}`)",
reg.version,
path.display()
));
}
for (key, entry) in ®.agents {
if let Err(why) = entry.validate() {
return Err(format!(
"schema registry `{}` is corrupt: agent `{key}` {why} — restore it from \
version control",
path.display()
));
}
}
Ok(reg)
}
pub fn serialize(reg: &SchemaRegistry) -> String {
toml::to_string_pretty(reg).expect("SchemaRegistry always serializes")
}
#[cfg(test)]
mod tests {
use super::*;
fn shape(name: &str, ty: &str, default: bool) -> FieldShape {
FieldShape {
name: name.to_string(),
ty: ty.to_string(),
default,
}
}
fn test_root() -> &'static Path {
Path::new("/project")
}
#[test]
fn new_registry_is_empty_at_the_current_lock_version() {
let reg = SchemaRegistry::new();
assert_eq!(reg.version, lock_version());
assert!(reg.events.is_empty());
}
#[test]
fn insert_then_get_round_trips_an_entry() {
let mut reg = SchemaRegistry::new();
let entry = EventEntry {
schema: 2,
fields: vec![shape("orderId", "String", false)],
};
reg.insert("commerce.order.PaymentConfirmed".to_string(), entry.clone());
assert_eq!(reg.get("commerce.order.PaymentConfirmed"), Some(&entry));
assert_eq!(reg.get("no.such.key"), None);
}
#[test]
fn parse_of_absent_content_is_an_empty_registry() {
let reg = parse(None, test_root()).unwrap();
assert!(reg.events.is_empty());
}
#[test]
fn parse_round_trips_serialized_content() {
let mut reg = SchemaRegistry::new();
reg.insert(
"commerce.order.PaymentConfirmed".to_string(),
EventEntry {
schema: 2,
fields: vec![shape("orderId", "String", false)],
},
);
let body = serialize(®);
assert_eq!(parse(Some(&body), test_root()).unwrap(), reg);
}
#[test]
fn a_truncated_file_is_corruption_not_a_fresh_project() {
let err = parse(Some(" \n"), test_root()).unwrap_err();
assert!(err.contains("restore it from version control"));
assert!(err.contains("bynk.schema.lock"));
}
#[test]
fn an_unparseable_file_is_corruption() {
let err = parse(Some("not valid toml {{{"), test_root()).unwrap_err();
assert!(err.contains("corrupt"));
assert!(err.contains("bynk.schema.lock"));
}
#[test]
fn a_v1_lock_reads_as_the_current_version_with_no_agents() {
let v1 = "version = 1\n\n[events]\n";
let reg = parse(Some(v1), test_root()).unwrap();
assert_eq!(reg.version, lock_version());
assert!(reg.agents.is_empty());
assert!(serialize(®).starts_with("version = 2"));
}
#[test]
fn a_v1_lock_that_somehow_carries_agents_is_refused() {
let text = "version = 1\n\n[agents.\"u.A\"]\nversion = 1\nfloor = 1\nshapes = []\n";
let err = parse(Some(text), test_root()).unwrap_err();
assert!(err.contains("unsupported schema registry version 1"));
}
#[test]
fn agent_entries_round_trip() {
let mut reg = SchemaRegistry::new();
reg.insert_agent(
"shop.Cart".to_string(),
AgentEntry {
version: 2,
floor: 1,
shapes: vec![
AgentShape {
version: 1,
additive: false,
fields: vec![StoreShape {
name: "owner".to_string(),
kind: "Cell".to_string(),
ty: "String".to_string(),
refinement: "String".to_string(),
}],
},
AgentShape {
version: 2,
additive: true,
fields: vec![],
},
],
},
);
let body = serialize(®);
assert_eq!(parse(Some(&body), test_root()).unwrap(), reg);
assert_eq!(
reg.get_agent("shop.Cart")
.unwrap()
.current()
.unwrap()
.version,
2
);
}
#[test]
fn an_agent_entry_with_no_snapshot_for_its_version_is_corruption() {
let text = "version = 2\n\n[agents.\"u.A\"]\nversion = 2\nfloor = 1\n\n\
[[agents.\"u.A\".shapes]]\nversion = 1\nadditive = false\nfields = []\n";
let err = parse(Some(text), test_root()).unwrap_err();
assert!(err.contains("corrupt"), "{err}");
assert!(err.contains("agent `u.A`"), "{err}");
}
#[test]
fn an_agent_entry_with_its_floor_above_its_version_is_corruption() {
let text = "version = 2\n\n[agents.\"u.A\"]\nversion = 1\nfloor = 2\nshapes = []\n";
let err = parse(Some(text), test_root()).unwrap_err();
assert!(err.contains("floor 2"), "{err}");
}
#[test]
fn oldest_readable_walks_back_over_additive_snapshots_only() {
let shape = |version, additive| AgentShape {
version,
additive,
fields: vec![],
};
let entry = |version, floor, shapes| AgentEntry {
version,
floor,
shapes,
};
assert_eq!(entry(1, 1, vec![shape(1, false)]).oldest_readable(), 1);
let all = entry(3, 1, vec![shape(1, false), shape(2, true), shape(3, true)]);
assert_eq!(all.oldest_readable(), 1);
let broken = entry(3, 1, vec![shape(1, false), shape(2, true), shape(3, false)]);
assert_eq!(broken.oldest_readable(), 3);
let mid = entry(3, 1, vec![shape(1, false), shape(2, false), shape(3, true)]);
assert_eq!(mid.oldest_readable(), 2);
let floored = entry(3, 2, vec![shape(2, true), shape(3, true)]);
assert_eq!(floored.oldest_readable(), 2);
}
#[test]
fn a_registry_with_no_agents_writes_no_agents_table() {
assert!(!serialize(&SchemaRegistry::new()).contains("agents"));
}
#[test]
fn an_unsupported_lock_version_is_corruption() {
let err = parse(Some("version = 99\n"), test_root()).unwrap_err();
assert!(err.contains("unsupported schema registry version 99"));
}
}